|
1 | 1 | from __future__ import annotations |
2 | 2 |
|
| 3 | +import json |
3 | 4 | from abc import ABC, abstractmethod |
| 5 | +from dataclasses import dataclass, field |
| 6 | +from html.parser import HTMLParser as _StdHTMLParser |
| 7 | +from typing import Any |
| 8 | + |
| 9 | + |
| 10 | +@dataclass(frozen=True, slots=True) |
| 11 | +class Section: |
| 12 | + """A heading boundary discovered by a structure-aware parser. |
| 13 | +
|
| 14 | + ``start_offset`` is the character index into ``ParsedDocument.text`` where the |
| 15 | + section body begins, so heading-aware chunking can split on real boundaries. |
| 16 | + """ |
| 17 | + |
| 18 | + heading: str |
| 19 | + level: int |
| 20 | + start_offset: int |
| 21 | + |
| 22 | + |
| 23 | +@dataclass(frozen=True, slots=True) |
| 24 | +class ParsedDocument: |
| 25 | + """Structured result of parsing a raw payload. |
| 26 | +
|
| 27 | + ``text`` is the extracted plain text. ``metadata`` carries parser-derived, |
| 28 | + string-valued attributes (page counts, document title, heading path). |
| 29 | + ``sections`` is the heading structure used by heading-aware chunking; it is |
| 30 | + empty for formats without headings. |
| 31 | + """ |
| 32 | + |
| 33 | + text: str |
| 34 | + metadata: dict[str, str] = field(default_factory=dict) |
| 35 | + sections: list[Section] = field(default_factory=list) |
4 | 36 |
|
5 | 37 |
|
6 | 38 | class DocumentParser(ABC): |
7 | 39 | """Base parser contract for source document extraction.""" |
8 | 40 |
|
9 | 41 | @abstractmethod |
10 | | - def parse(self, raw_payload: bytes) -> str: |
11 | | - """Return extracted textual content from raw payload.""" |
| 42 | + def parse(self, raw_payload: bytes) -> ParsedDocument: |
| 43 | + """Return structured text + metadata extracted from a raw payload.""" |
12 | 44 |
|
13 | 45 |
|
14 | | -class PdfParser(DocumentParser): |
15 | | - def parse(self, raw_payload: bytes) -> str: |
16 | | - raise NotImplementedError("PDF parser not implemented yet") |
| 46 | +class MissingParserDependencyError(RuntimeError): |
| 47 | + """Raised when a parser's optional dependency is not installed.""" |
| 48 | + |
| 49 | + |
| 50 | +class TextParser(DocumentParser): |
| 51 | + def parse(self, raw_payload: bytes) -> ParsedDocument: |
| 52 | + return ParsedDocument(text=raw_payload.decode("utf-8", errors="replace")) |
| 53 | + |
| 54 | + |
| 55 | +def _strip_frontmatter(text: str) -> tuple[str, dict[str, str]]: |
| 56 | + """Strip a leading YAML (``---``) or TOML (``+++``) frontmatter block. |
| 57 | +
|
| 58 | + Frontmatter keys are surfaced as ``frontmatter.<key>`` metadata using a |
| 59 | + deliberately small, dependency-free ``key: value`` scan (no nested YAML). |
| 60 | + """ |
| 61 | + fences = {"---": "---", "+++": "+++"} |
| 62 | + lines = text.splitlines(keepends=True) |
| 63 | + if not lines: |
| 64 | + return text, {} |
| 65 | + opener = lines[0].strip() |
| 66 | + if opener not in fences: |
| 67 | + return text, {} |
| 68 | + |
| 69 | + closer = fences[opener] |
| 70 | + meta: dict[str, str] = {} |
| 71 | + for index in range(1, len(lines)): |
| 72 | + if lines[index].strip() == closer: |
| 73 | + for raw_line in lines[1:index]: |
| 74 | + if ":" in raw_line: |
| 75 | + key, _, value = raw_line.partition(":") |
| 76 | + key = key.strip() |
| 77 | + value = value.strip().strip("'\"") |
| 78 | + if key and value: |
| 79 | + meta[f"frontmatter.{key}"] = value |
| 80 | + remainder = "".join(lines[index + 1 :]) |
| 81 | + return remainder.lstrip("\n"), meta |
| 82 | + # No closing fence: treat the whole thing as body. |
| 83 | + return text, {} |
17 | 84 |
|
18 | 85 |
|
19 | 86 | class MarkdownParser(DocumentParser): |
20 | | - def parse(self, raw_payload: bytes) -> str: |
21 | | - raise NotImplementedError("Markdown parser not implemented yet") |
| 87 | + def parse(self, raw_payload: bytes) -> ParsedDocument: |
| 88 | + body, metadata = _strip_frontmatter(raw_payload.decode("utf-8", errors="replace")) |
| 89 | + sections: list[Section] = [] |
| 90 | + offset = 0 |
| 91 | + for line in body.splitlines(keepends=True): |
| 92 | + stripped = line.lstrip() |
| 93 | + if stripped.startswith("#"): |
| 94 | + level = len(stripped) - len(stripped.lstrip("#")) |
| 95 | + heading = stripped[level:].strip() |
| 96 | + if heading and 1 <= level <= 6: |
| 97 | + sections.append(Section(heading=heading, level=level, start_offset=offset)) |
| 98 | + offset += len(line) |
| 99 | + if sections and "frontmatter.title" not in metadata: |
| 100 | + metadata["title"] = sections[0].heading |
| 101 | + return ParsedDocument(text=body, metadata=metadata, sections=sections) |
| 102 | + |
| 103 | + |
| 104 | +class _HtmlTextExtractor(_StdHTMLParser): |
| 105 | + _SKIP = {"script", "style"} |
| 106 | + |
| 107 | + def __init__(self) -> None: |
| 108 | + super().__init__(convert_charrefs=True) |
| 109 | + self.parts: list[str] = [] |
| 110 | + self.title_parts: list[str] = [] |
| 111 | + self.headings: list[str] = [] |
| 112 | + self._skip_depth = 0 |
| 113 | + self._in_title = False |
| 114 | + self._heading_buffer: list[str] | None = None |
| 115 | + |
| 116 | + def handle_starttag(self, tag: str, attrs: Any) -> None: |
| 117 | + if tag in self._SKIP: |
| 118 | + self._skip_depth += 1 |
| 119 | + if tag == "title": |
| 120 | + self._in_title = True |
| 121 | + if tag in {"h1", "h2", "h3", "h4", "h5", "h6"}: |
| 122 | + self._heading_buffer = [] |
| 123 | + |
| 124 | + def handle_endtag(self, tag: str) -> None: |
| 125 | + if tag in self._SKIP and self._skip_depth: |
| 126 | + self._skip_depth -= 1 |
| 127 | + if tag == "title": |
| 128 | + self._in_title = False |
| 129 | + if tag in {"h1", "h2", "h3", "h4", "h5", "h6"} and self._heading_buffer is not None: |
| 130 | + heading = "".join(self._heading_buffer).strip() |
| 131 | + if heading: |
| 132 | + self.headings.append(heading) |
| 133 | + self._heading_buffer = None |
| 134 | + |
| 135 | + def handle_data(self, data: str) -> None: |
| 136 | + if self._skip_depth: |
| 137 | + return |
| 138 | + if self._in_title: |
| 139 | + self.title_parts.append(data) |
| 140 | + if self._heading_buffer is not None: |
| 141 | + self._heading_buffer.append(data) |
| 142 | + if data.strip(): |
| 143 | + self.parts.append(data) |
22 | 144 |
|
23 | 145 |
|
24 | 146 | class HtmlParser(DocumentParser): |
25 | | - def parse(self, raw_payload: bytes) -> str: |
26 | | - raise NotImplementedError("HTML parser not implemented yet") |
| 147 | + """Dependency-free HTML text extraction via the stdlib HTML parser.""" |
| 148 | + |
| 149 | + def parse(self, raw_payload: bytes) -> ParsedDocument: |
| 150 | + extractor = _HtmlTextExtractor() |
| 151 | + extractor.feed(raw_payload.decode("utf-8", errors="replace")) |
| 152 | + extractor.close() |
| 153 | + text = " ".join(part.strip() for part in extractor.parts if part.strip()) |
| 154 | + metadata: dict[str, str] = {} |
| 155 | + title = "".join(extractor.title_parts).strip() |
| 156 | + if title: |
| 157 | + metadata["title"] = title |
| 158 | + if extractor.headings: |
| 159 | + metadata["headings"] = " > ".join(extractor.headings) |
| 160 | + sections = [ |
| 161 | + Section(heading=heading, level=1, start_offset=0) for heading in extractor.headings |
| 162 | + ] |
| 163 | + return ParsedDocument(text=text, metadata=metadata, sections=sections) |
| 164 | + |
| 165 | + |
| 166 | +class PdfParser(DocumentParser): |
| 167 | + """PDF text extraction behind the optional ``ingestion-pdf`` extra (pypdf).""" |
| 168 | + |
| 169 | + def parse(self, raw_payload: bytes) -> ParsedDocument: |
| 170 | + try: |
| 171 | + from pypdf import PdfReader |
| 172 | + except ImportError as exc: # pragma: no cover - exercised only without the extra |
| 173 | + raise MissingParserDependencyError( |
| 174 | + "PDF parsing requires the optional dependency 'pypdf'. " |
| 175 | + 'Install it with: pip install "llm-observability-analytics[ingestion-pdf]"' |
| 176 | + ) from exc |
| 177 | + |
| 178 | + import io |
| 179 | + |
| 180 | + reader = PdfReader(io.BytesIO(raw_payload)) |
| 181 | + pages = [(page.extract_text() or "").strip() for page in reader.pages] |
| 182 | + text = "\n\n".join(page for page in pages if page) |
| 183 | + metadata = {"page_count": str(len(reader.pages))} |
| 184 | + return ParsedDocument(text=text, metadata=metadata) |
27 | 185 |
|
28 | 186 |
|
29 | 187 | class JsonParser(DocumentParser): |
30 | | - def parse(self, raw_payload: bytes) -> str: |
31 | | - raise NotImplementedError("JSON parser not implemented yet") |
| 188 | + """Structured JSON extraction. |
32 | 189 |
|
| 190 | + With ``text_fields`` set, concatenates those top-level string fields (useful |
| 191 | + for record-shaped documents). Otherwise re-serializes deterministically so |
| 192 | + content hashes stay stable across runs. |
| 193 | + """ |
33 | 194 |
|
34 | | -class TextParser(DocumentParser): |
35 | | - def parse(self, raw_payload: bytes) -> str: |
36 | | - return raw_payload.decode("utf-8", errors="replace") |
| 195 | + def __init__(self, text_fields: list[str] | None = None) -> None: |
| 196 | + self.text_fields = text_fields |
| 197 | + |
| 198 | + def parse(self, raw_payload: bytes) -> ParsedDocument: |
| 199 | + try: |
| 200 | + payload = json.loads(raw_payload.decode("utf-8", errors="replace")) |
| 201 | + except json.JSONDecodeError as exc: |
| 202 | + raise ValueError(f"Invalid JSON payload: {exc}") from exc |
| 203 | + |
| 204 | + metadata: dict[str, str] = {} |
| 205 | + if isinstance(payload, dict): |
| 206 | + for candidate in ("title", "name", "id"): |
| 207 | + if candidate in payload and str(payload[candidate]).strip(): |
| 208 | + metadata["title"] = str(payload[candidate]).strip() |
| 209 | + break |
| 210 | + |
| 211 | + if self.text_fields and isinstance(payload, dict): |
| 212 | + parts = [ |
| 213 | + str(payload[field_name]) |
| 214 | + for field_name in self.text_fields |
| 215 | + if field_name in payload and str(payload[field_name]).strip() |
| 216 | + ] |
| 217 | + text = "\n".join(parts) if parts else _json_to_content(payload) |
| 218 | + else: |
| 219 | + text = _json_to_content(payload) |
| 220 | + return ParsedDocument(text=text, metadata=metadata) |
| 221 | + |
| 222 | + |
| 223 | +def _json_to_content(payload: Any) -> str: |
| 224 | + """Render JSON deterministically so hashes remain stable across runs.""" |
| 225 | + return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) |
0 commit comments