4adfff5b2a
Nieuwe module buzzrss/html_scrape.py (stdlib-only DOM + simpele selectors: tag/.class/tag.class/#id) die een server-rendered nieuwspagina omzet in entries. Per feed via feed.scrape_config (JSON; migratie v13): item/title/ link/date(+date_format, NL-maandnamen locale-onafhankelijk)/summary. Scrape-entries lopen door de bestaande pipeline (OV-filter, titel-dedup, WhatsApp-forward, run-log). Eerste scrape-feed: CNV Stads- en streekvervoer (feed 102).
214 lines
No EOL
7.2 KiB
Python
214 lines
No EOL
7.2 KiB
Python
# 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 <a href> 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 <a> (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 <title> text, if any."""
|
|
m = re.search(r"<title[^>]*>(.*?)</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 |