Spaces:
Running on Zero
Running on Zero
File size: 3,835 Bytes
53d9817 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | from __future__ import annotations
import json
import xml.etree.ElementTree as ET
from collections.abc import Iterator
from pathlib import Path
from typing import ClassVar
from universal_parser.core.router import register
from universal_parser.core.schema import Element, TableData
from universal_parser.core.sniffer import FileType
from universal_parser.extractors.base import BaseExtractor
@register
class JSONXMLExtractor(BaseExtractor):
"""
Extractor for JSON (.json) and XML (.xml) files.
Handles:
- List of objects in JSON -> Table element (if tabular schema)
- Nested JSON objects -> formatted code block element
- XML -> element nodes parsed into headings, paragraphs, and code blocks
"""
supported_types: ClassVar[list[FileType]] = [FileType.JSON, FileType.XML]
def stream(self, path: str | Path) -> Iterator[Element]:
path_obj = Path(path)
ext = path_obj.suffix.lower()
if ext == ".json":
yield from self._stream_json(path_obj)
elif ext == ".xml":
yield from self._stream_xml(path_obj)
def _stream_json(self, path: Path) -> Iterator[Element]:
try:
with open(path, encoding="utf-8", errors="replace") as f:
data = json.load(f)
# Check if it's a list of uniform dictionaries (tabular JSON)
if isinstance(data, list) and data and all(isinstance(row, dict) for row in data):
headers = list(data[0].keys())
rows = [[str(row.get(h, "")).strip() for h in headers] for row in data]
md_header = "| " + " | ".join(headers) + " |"
md_separator = "| " + " | ".join(["---"] * len(headers)) + " |"
md_rows = ["| " + " | ".join(r) + " |" for r in rows]
markdown_repr = "\n".join([md_header, md_separator] + md_rows)
yield Element(
type="table",
text=path.name,
data=TableData(headers=headers, rows=rows),
markdown_repr=markdown_repr,
confidence=1.0,
)
else:
# General JSON: render formatted code block
formatted_json = json.dumps(data, indent=2)
yield Element(
type="code_block",
text=formatted_json,
markdown_repr=f"```json\n{formatted_json}\n```",
confidence=1.0,
)
except Exception: # noqa: BLE001
return
def _stream_xml(self, path: Path) -> Iterator[Element]:
try:
tree = ET.parse(path)
root = tree.getroot()
# Yield root tag as level 1 heading
yield Element(
type="heading",
level=1,
text=root.tag,
markdown_repr=f"# {root.tag}",
confidence=1.0,
)
for child in root:
text_val = (child.text or "").strip()
if text_val:
yield Element(
type="paragraph",
text=f"{child.tag}: {text_val}",
markdown_repr=f"**{child.tag}**: {text_val}",
confidence=1.0,
)
else:
child_str = ET.tostring(child, encoding="unicode").strip()
if child_str:
yield Element(
type="code_block",
text=child_str,
markdown_repr=f"```xml\n{child_str}\n```",
confidence=1.0,
)
except Exception: # noqa: BLE001
return
|