from __future__ import annotations
from collections.abc import Iterator
from pathlib import Path
from typing import ClassVar
from selectolax.parser import HTMLParser
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 HTMLExtractor(BaseExtractor):
"""
Ultra-fast HTML/XHTML extractor using selectolax.
Handles:
- Headings (h1 - h6) preserving document tree hierarchy
- Text paragraphs (
,
)
- HTML tags -> TableData + Markdown tables
- Strips script, style, and navigation junk tags
"""
supported_types: ClassVar[list[FileType]] = [FileType.HTML]
def stream(self, path: str | Path) -> Iterator[Element]:
path_obj = Path(path)
try:
with open(path_obj, encoding="utf-8", errors="replace") as f:
html_content = f.read()
if not html_content.strip():
return
parser = HTMLParser(html_content)
# Strip noise (script, style, svg)
for tag in parser.css("script, style, noscript, svg"):
tag.decompose()
body = parser.body or parser.root
if body is None:
return
for node in body.iter():
tag_name = node.tag.lower() if node.tag else ""
# Headings
if tag_name in ("h1", "h2", "h3", "h4", "h5", "h6"):
level = int(tag_name[1])
text = node.text(strip=True)
if text:
yield Element(
type="heading",
level=level,
text=text,
markdown_repr=f"{'#' * level} {text}",
confidence=1.0,
)
# Paragraphs and Lists
elif tag_name in ("p", "li"):
text = node.text(strip=True)
if text:
prefix = "- " if tag_name == "li" else ""
elem_type = "list_item" if tag_name == "li" else "paragraph"
yield Element(
type=elem_type,
text=text,
markdown_repr=f"{prefix}{text}",
confidence=1.0,
)
# Tables
elif tag_name == "table":
headers = []
rows = []
for th in node.css("th"):
h_text = th.text(strip=True)
if h_text:
headers.append(h_text)
for tr in node.css("tr"):
tds = [td.text(strip=True) for td in tr.css("td")]
if any(tds):
rows.append(tds)
if not headers and rows:
headers = [f"Column_{i + 1}" for i in range(len(rows[0]))]
if headers or rows:
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_obj.name,
data=TableData(headers=headers, rows=rows),
markdown_repr=markdown_repr,
confidence=1.0,
)
except Exception: # noqa: BLE001
return