# HTML page scraping for feeds without RSS/Atom/JSON.
#
# Used when a feed row carries a `scrape_config` JSON blob. The scrape mode
# turns a plain server-rendered news list page into entries, so sources that
# publish no feed at all (vakbonden, gemeenten, OV-bedrijven) can still be
# followed with the exact same pipeline (OV-filter, title dedup, WhatsApp).
#
# Selector syntax is deliberately small: one element per selector, no
# combinators. Supported: "tag" ".class" "tag.class" "#id" "*"
# Config keys:
# item (required) element selector for each news card/item
# title (required) element selector for the headline inside the item
# link (required) element selector for the inside the item
# date (optional) element selector for the date text; the element's
# datetime/content attributes win over its text
# date_format (optional) strptime format for the date text. Dutch month
# names are translated to numbers first, so "%d %B %Y" works
# for "11 september 2026" regardless of the process locale.
# summary (optional) element selector for a short text inside the item
#
# Items without a resolvable link or a non-empty title are skipped; the caller
# counts skips for sanity checking.
from __future__ import annotations
import re
import urllib.parse
from datetime import datetime, timezone
from html.parser import HTMLParser
_NL_MONTHS = {
"januari": 1, "februari": 2, "maart": 3, "april": 4, "mei": 5, "juni": 6,
"juli": 7, "augustus": 8, "september": 9, "oktober": 10, "november": 11,
"december": 12,
}
_WS = re.compile(r"\s+")
class _Node:
__slots__ = ("tag", "attrs", "children", "text")
def __init__(self, tag: str, attrs: dict[str, str]) -> None:
self.tag = tag
self.attrs = attrs
self.children: list[_Node] = []
self.text = ""
class _Builder(HTMLParser):
"""Turns an HTML document into a small DOM tree (stdlib only)."""
def __init__(self) -> None:
super().__init__(convert_charrefs=True)
self.root = _Node("#root", {})
self._stack = [self.root]
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
node = _Node(tag, {k: (v or "") for k, v in attrs})
self._stack[-1].children.append(node)
self._stack.append(node)
def handle_startendtag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
node = _Node(tag, {k: (v or "") for k, v in attrs})
self._stack[-1].children.append(node)
def handle_endtag(self, tag: str) -> None:
if len(self._stack) > 1:
self._stack.pop()
def handle_data(self, data: str) -> None:
self._stack[-1].text += data
def _match(node: _Node, sel: str) -> bool:
if sel == "*":
return True
cls = None
if sel.startswith("."):
tag, cls = None, sel[1:]
elif sel.startswith("#"):
return node.attrs.get("id", "") == sel[1:]
elif "." in sel:
tag, cls = sel.split(".", 1)
else:
tag = sel
if tag and node.tag != tag:
return False
if cls:
classes = node.attrs.get("class", "").split()
if cls not in classes:
return False
return True
def _iter_tree(node: _Node):
for child in node.children:
yield child
yield from _iter_tree(child)
def _node_text(node: _Node) -> str:
parts = [node.text]
def walk(n: _Node) -> None:
for c in n.children:
parts.append(c.text)
walk(c)
walk(node)
return _WS.sub(" ", "".join(parts)).strip()
def _parse_date(text: str, fmt: str | None, attrs: dict[str, str]) -> datetime | None:
raw = attrs.get("datetime") or attrs.get("content") or ""
for candidate in (raw.strip(), text):
if not candidate:
continue
try:
return datetime.fromisoformat(candidate).astimezone(timezone.utc)
except ValueError:
pass
if fmt:
# Dutch month names -> numbers so %B works under any locale.
# The format's %B is translated to %m alongside (the text now
# carries the number, not the name).
translated = candidate
for name, num in _NL_MONTHS.items():
translated = re.sub(
rf"\b{name}\b", str(num), translated, flags=re.IGNORECASE
)
translated_fmt = fmt.replace("%B", "%m")
try:
return datetime.strptime(translated, translated_fmt).replace(
tzinfo=timezone.utc
)
except ValueError:
pass
return None
def _first(node: _Node, sel: str) -> _Node | None:
for n in _iter_tree(node):
if _match(n, sel):
return n
return None
def _href(node: _Node, sel: str, base: str) -> str:
# The item node itself may be the (whole card is a link): match it first.
targets = [node]
targets.extend(_iter_tree(node))
for n in targets:
if _match(n, sel):
href = (n.attrs.get("href") or "").strip()
if href:
return urllib.parse.urljoin(base, href)
return ""
def page_title(html_text: str) -> str:
"""First text, if any."""
m = re.search(r"<title[^>]*>(.*?)", html_text, re.IGNORECASE | re.DOTALL)
return _WS.sub(" ", m.group(1)).strip() if m else ""
def extract_entries(
html_text: str, base_url: str, cfg: dict[str, object]
) -> tuple[list[dict[str, object]], int]:
"""Returns (entries, skipped). Each entry dict:
{id, date, title, summary, link, paywalled} with date as tz-aware datetime."""
item_sel = str(cfg.get("item") or "").strip()
title_sel = str(cfg.get("title") or "").strip()
link_sel = str(cfg.get("link") or "").strip()
if not (item_sel and title_sel and link_sel):
raise ValueError("scrape_config requires item, title and link selectors")
date_sel = str(cfg.get("date") or "").strip()
date_format = str(cfg.get("date_format") or "").strip() or None
summary_sel = str(cfg.get("summary") or "").strip()
builder = _Builder()
try:
builder.feed(html_text)
builder.close()
except Exception as e: # malformed HTML must not kill the whole poll
raise ValueError(f"HTML parse failed: {e}") from e
entries: list[dict[str, object]] = []
skipped = 0
for node in _iter_tree(builder.root):
if not _match(node, item_sel):
continue
link = _href(node, link_sel, base_url)
title_node = _first(node, title_sel)
title = _node_text(title_node) if title_node else ""
if not link or not title:
skipped += 1
continue
date = None
if date_sel:
date_node = _first(node, date_sel)
if date_node:
date = _parse_date(_node_text(date_node), date_format, date_node.attrs)
entries.append(
{
"id": link,
"date": date or datetime.now(timezone.utc),
"title": title,
"summary": _node_text(_first(node, summary_sel)) if summary_sel else "",
"link": link,
"paywalled": False,
}
)
return entries, skipped