"""HTML parsing with selectolax: title, meta, canonical, Open Graph, JSON-LD, embedded JSON (Next.js/Nuxt/data-props),
headings, tables, links, cleaned text."""
from __future__ import annotations
import html as htmlmod
import json
import re
from dataclasses import dataclass, field
from typing import Any
from urllib.parse import urljoin
from selectolax.parser import HTMLParser, Node
_WS = re.compile(r"[ \t\r\f\v]+")
_NL = re.compile(r"\n{3,}")
_SKIP_TAGS = {"script", "style", "noscript", "svg", "template", "iframe", "canvas", "nav", "footer", "form", "button"}
@dataclass
class HtmlDoc:
url: str
title: str | None = None
canonical: str | None = None
description: str | None = None
lang: str | None = None
meta: dict[str, str] = field(default_factory=dict)
og: dict[str, str] = field(default_factory=dict)
json_ld: list[Any] = field(default_factory=list)
embedded_json: dict[str, Any] = field(default_factory=dict) # id/key -> parsed JSON
headings: list[tuple[int, str]] = field(default_factory=list)
tables: list[dict[str, Any]] = field(default_factory=list) # {"headers": [...], "rows": [[...]], "caption": str}
links: list[tuple[str, str]] = field(default_factory=list) # (absolute href, anchor text)
text: str = ""
published_at: str | None = None
modified_at: str | None = None
tree: HTMLParser | None = field(default=None, repr=False)
def structured(self) -> dict[str, Any]:
"""JSON-serialisable summary stored in `snapshots.structured`."""
return {
"title": self.title, "canonical": self.canonical, "description": self.description, "lang": self.lang,
"meta": {k: v for k, v in self.meta.items() if len(v) < 500}, "og": self.og,
"json_ld": self.json_ld[:20], "embedded_json_keys": list(self.embedded_json)[:50],
"headings": self.headings[:200], "tables": self.tables[:40], "published_at": self.published_at, "modified_at": self.modified_at,
"link_count": len(self.links), "text_length": len(self.text),
}
def css(self, selector: str) -> list[Node]:
return self.tree.css(selector) if self.tree else []
def css_first(self, selector: str) -> Node | None:
return self.tree.css_first(selector) if self.tree else None
def links_matching(self, pattern: str | re.Pattern[str]) -> list[tuple[str, str]]:
rx = re.compile(pattern) if isinstance(pattern, str) else pattern
seen: set[str] = set()
out: list[tuple[str, str]] = []
for href, text in self.links:
if href not in seen and rx.search(href):
seen.add(href)
out.append((href, text))
return out
def clean_text(s: str) -> str:
s = htmlmod.unescape(s)
s = _WS.sub(" ", s)
s = "\n".join(line.strip() for line in s.split("\n"))
return _NL.sub("\n\n", s).strip()
def node_text(node: Node, *, separator: str = " ") -> str:
return clean_text(node.text(separator=separator, strip=True))
def parse_html(content: str | bytes, url: str = "", *, keep_tree: bool = True, max_links: int = 5000) -> HtmlDoc:
tree = HTMLParser(content)
doc = HtmlDoc(url=url, tree=tree if keep_tree else None)
if (html_node := tree.css_first("html")) is not None:
doc.lang = html_node.attributes.get("lang")
if (t := tree.css_first("title")) is not None:
doc.title = clean_text(t.text()) or None
for m in tree.css("meta"):
a = m.attributes
key = a.get("property") or a.get("name") or a.get("itemprop")
val = a.get("content")
if not key or val is None:
continue
key = key.strip().lower()
val = val.strip()
if key.startswith(("og:", "twitter:", "article:")):
doc.og[key] = val
else:
doc.meta.setdefault(key, val)
doc.description = doc.meta.get("description") or doc.og.get("og:description")
if not doc.title:
doc.title = doc.og.get("og:title")
doc.published_at = (doc.og.get("article:published_time") or doc.meta.get("date") or doc.meta.get("pubdate")
or doc.meta.get("publish_date") or doc.meta.get("datepublished") or doc.meta.get("dc.date.issued"))
doc.modified_at = doc.og.get("article:modified_time") or doc.meta.get("last-modified") or doc.meta.get("datemodified")
for link in tree.css("link[rel]"):
rel = (link.attributes.get("rel") or "").lower()
href = link.attributes.get("href")
if "canonical" in rel and href:
doc.canonical = urljoin(url, href)
for s in tree.css("script"):
stype = (s.attributes.get("type") or "").lower()
sid = s.attributes.get("id") or ""
raw = s.text() or ""
if not raw.strip():
continue
if "ld+json" in stype:
parsed = _loads_lenient(raw)
if parsed is not None:
if isinstance(parsed, list):
doc.json_ld.extend(parsed)
else:
doc.json_ld.append(parsed)
elif stype in ("application/json", "text/json") or sid in ("__NEXT_DATA__", "__NUXT_DATA__", "__remixContext"):
parsed = _loads_lenient(raw)
if parsed is not None:
doc.embedded_json[sid or f"json_{len(doc.embedded_json)}"] = parsed
# Hugging Face / Svelte style `data-props` attributes and generic data-* JSON blobs
for n in tree.css("[data-props]"):
raw = n.attributes.get("data-props") or ""
parsed = _loads_lenient(raw)
if parsed is not None:
key = f"data-props:{n.attributes.get('data-target') or len(doc.embedded_json)}"
doc.embedded_json.setdefault(key, parsed)
for level in range(1, 5):
for h in tree.css(f"h{level}"):
txt = node_text(h)
if txt:
doc.headings.append((level, txt[:300]))
for table in tree.css("table")[:60]:
parsed_table = _parse_table(table)
if parsed_table["rows"]:
doc.tables.append(parsed_table)
seen: set[str] = set()
for a in tree.css("a[href]"):
href = a.attributes.get("href") or ""
if not href or href.startswith(("#", "javascript:", "mailto:", "tel:")):
continue
absolute = urljoin(url, href.strip())
if absolute in seen:
continue
seen.add(absolute)
doc.links.append((absolute, node_text(a)[:200]))
if len(doc.links) >= max_links:
break
doc.text = extract_text(HTMLParser(content)) # fresh parse: extract_text prunes nav/footer/button/noscript, doc.tree must stay intact
return doc
def extract_text(tree: HTMLParser) -> str:
body = tree.body or tree.root
if body is None:
return ""
for tag in _SKIP_TAGS:
for n in body.css(tag):
n.decompose()
for n in body.css("[aria-hidden='true'], .sr-only, [hidden]"):
n.decompose()
parts: list[str] = []
block_tags = {"p", "div", "section", "article", "li", "h1", "h2", "h3", "h4", "h5", "h6", "tr", "br", "pre", "blockquote", "td", "th", "dd", "dt"}
for node in body.traverse(include_text=True):
if node.tag == "-text":
t = node.text(deep=False)
if t and t.strip():
parts.append(t)
elif node.tag in block_tags:
parts.append("\n")
return clean_text("".join(parts))
def _parse_table(table: Node) -> dict[str, Any]:
headers: list[str] = []
rows: list[list[str]] = []
caption_node = table.css_first("caption")
caption = node_text(caption_node) if caption_node else None
for tr in table.css("tr"):
cells = tr.css("th, td")
if not cells:
continue
values = [node_text(c)[:500] for c in cells]
if not headers and all(c.tag == "th" for c in cells):
headers = values
else:
rows.append(values)
if len(rows) > 500:
break
return {"caption": caption, "headers": headers, "rows": rows}
def _loads_lenient(raw: str) -> Any | None:
raw = raw.strip()
if not raw:
return None
try:
return json.loads(raw)
except json.JSONDecodeError:
pass
try:
return json.loads(htmlmod.unescape(raw))
except json.JSONDecodeError:
return None
def find_in_json(obj: Any, key: str, *, max_hits: int = 50) -> list[Any]:
"""Depth-first search of every value under `key` inside a nested JSON structure."""
hits: list[Any] = []
def walk(o: Any) -> None:
if len(hits) >= max_hits:
return
if isinstance(o, dict):
for k, v in o.items():
if k == key:
hits.append(v)
walk(v)
elif isinstance(o, list):
for v in o:
walk(v)
walk(obj)
return hits
__all__ = ["HtmlDoc", "clean_text", "extract_text", "find_in_json", "node_text", "parse_html"]