buzz-rss/validate_filter.py

#!/usr/bin/env python3
"""Blijvende filter-invarianten + regressie voor het OV-matchfilter.

Draaien (van de repo-root): python3 validate_filter.py   [alleen stdlib]

Waarom dit bestaat (zie AGENTS.md):
- het classificatiebeleid (precision-first) staat als commentaar in buzzrss/bot.py;
- DIT script maakt dat beleid afdwingbaar: een latere filterwijziging die een
  dubbelzinnig woord sterk/whitelist maakt, of een onmiskenbare OV-term verwakt,
  faalt hierop met exit 1 — vóór commit/deploy.
"""
import ast
import re
import sys
import textwrap
from pathlib import Path

ROOT = Path(__file__).resolve().parent
BOT = ROOT / "buzzrss" / "bot.py"

# --- extractie zonder maubot-imports (zelfde techniek als de audits) ---
src = BOT.read_text()
tree = ast.parse(src)
WANT_ASSIGN = {
    "MATCH_FILTER_FAMILIES", "MATCH_FILTER_THRESHOLD", "MATCH_FILTER_BLACKLIST",
    "MATCH_FILTER_WHITELIST", "_compiled_match_families",
    "_WHITELIST_RE", "_WHITELIST_PHRASES",
}
WANT_METHODS = {"_fam_hits", "_filter_passes", "_filter_text", "_whitelist_hit"}

nodes = []
for n in tree.body:
    if isinstance(n, ast.AnnAssign) and getattr(n.target, "id", "") in WANT_ASSIGN:
        nodes.append(ast.unparse(n))
    elif isinstance(n, ast.Assign) and any(getattr(t, "id", "") in WANT_ASSIGN for t in n.targets):
        nodes.append(ast.unparse(n))
    if isinstance(n, ast.ClassDef) and n.name == "RSSBot":
        methods = [
            ast.unparse(m)
            for m in n.body
            if isinstance(m, (ast.FunctionDef, ast.AsyncFunctionDef)) and m.name in WANT_METHODS
        ]
        nodes.append("class RSSBot:\n" + textwrap.indent("\n".join(methods), "    "))

ns = {}
exec(compile("from __future__ import annotations\nimport re, html\n" + "\n".join(nodes) + "\n", "<bot>", "exec"), ns)
R = ns["RSSBot"]
FAMILIES = ns["MATCH_FILTER_FAMILIES"]           # [(regex, gewicht)]
WHITELIST = ns["MATCH_FILTER_WHITELIST"]         # frozenset
COMPILED = [(re.compile(p, re.I), w) for p, w in FAMILIES]


def weights_for(word: str) -> set[int]:
    """gewichten van families die `word` matchen (case-insensitive)."""
    return {w for pat, w in COMPILED if pat.search(word)}


def whitelisted(word: str) -> bool:
    return ns["_WHITELIST_RE"].search(word) is not None or any(
        ph in word.lower() for ph in ns["_WHITELIST_PHRASES"]
    )


failures = []


def check(cond: bool, msg: str):
    print(("  OK   " if cond else "  FOUT ") + msg)
    if not cond:
        failures.append(msg)


# --- Invariant 1: dubbelzinnige woorden mogen nooit sterk/medium of whitelist zijn
AMBIGUOUS = [
    "trein", "bus", "station", "stations", "sprinter", "sprinters",
    "reiziger", "reizigers", "vervoerder", "vervoerders",
    "inchecken", "uitchecken", "sein", "overweg", "overwegen", "breng",
    "kaartje", "opwaarderen", "rails", "dienstregeling", "chipkaart",
    "reisproduct", "vervoersbewijs", "vervoerbewijs",
    "pendelbus", "shuttlebus",
    "ontspor", "ontspoor", "ontspoord", "ontspoorde", "ontsporing",
]
print(f"== invariant 1: {len(AMBIGUOUS)} dubbelzinnige woorden -> alleen zwak (1), nooit whitelist ==")
for w in AMBIGUOUS:
    gew = weights_for(w)
    check(1 in gew, f"{w!r} heeft een zwakke familie")
    check(not (gew - {1}), f"{w!r} zit NIET in een sterke/medium familie (gewichten {sorted(gew)})")
    check(not whitelisted(w), f"{w!r} staat NIET op de whitelist")

# --- Invariant 2: onmiskenbare OV-termen mogen niet uitsluitend zwak zijn
UNAMBIGUOUS = [
    "spoorstoring", "spoorlijn", "perron", "bushalte", "bushokje",
    "metrostation", "busstation", "conducteur", "ovpay", "ns", "prorail",
    "treinstoring", "halte", "openbaar vervoer", "intercity", "buurtbus",
    "veerdienst", "lijn 12", "ov-chipkaart",
]
print(f"== invariant 2: {len(UNAMBIGUOUS)} onmiskenbare OV-termen -> sterk/medium of whitelist ==")
for w in UNAMBIGUOUS:
    gew = weights_for(w)
    check(whitelisted(w) or any(g >= 2 for g in gew),
          f"{w!r} heeft whitelist of gewicht >= 2 (gewichten {sorted(gew)})")

# --- Invariant 3: patronen compileren en gewichten geldig
print("== invariant 3: alle 24+ families compileren, gewicht in {1,2,3} ==")
for p, w in FAMILIES:
    check(w in (1, 2, 3), f"familie {p[:40]!r} heeft geldig gewicht {w}")
check(len(COMPILED) == len(FAMILIES), "alle families compileren")

# --- Regressie: gedrag van _filter_passes op bekende cases (precision-first)
GOLDEN = [
    # (titel, samenvatting, verwacht_door)
    # geweerd — bekende ruis
    ("Na mooi seizoen in wereldstad Parijs wacht nu EK voor 'lange man' Tom Koops",
     "een weekje vakantie pak ik de trein naar mijn nieuwe huis. Berlijn", False),
    ("Amerikaanse talkshowhost Jimmy Kimmel schrapt gesprek 'na druk regering Trump'",
     "ABC, onze regionale omroepen en lokale stations.", False),
    ("Inchecken op Schiphol duurt langer", "Passagiers staan in de rij bij de balies.", False),
    ("Sprinter verpulvert record op de 100 meter", "De atleet liep een nationale toptijd.", False),
    ("Zakenreizigers vliegen weer massaal", "Het aantal vluchten is gestegen.", False),
    ("Vrachtvervoerder failliet", "De transporteur heeft uitstel van betaling aangevraagd.", False),
    ("Het plan voor de nieuwe wijk ontspoorde", "De gemeente moet de kosten herzien.", False),
    ("Sein brandmeester: brandweer rukt uit", "De brandweer kreeg een sein om uit te rukken.", False),
    ("VW-bus in brand op de A2", "De inzittenden kwamen er veilig uit.", False),
    ("Fietser gewond op de Stationsstraat", "Het slachtoffer is naar het ziekenhuis gebracht.", False),
    ("Kunstenaar maakt kaartje voor het museum", "", False),
    ("Benzineprijs stijgt opnieuw", "De pomp wordt duurder door de olieprijs.", False),
    # geweerd — zwakke woorden alléén in de samenvatting (geen titel-anker).
    # Echt geval: Omroep Brabant, Solheim Cup-golfverslag (11-09-2026) — 'pendelbus'
    # stond op de whitelist en 'station'+'pendelbus'+'bus' in de samenvatting was
    # samen score 3.
    ("Golf-fans uit de hele wereld naar Cromvoirt: 'Beter dan sterrenrestaurant'",
     "Op het station in Den Bosch stapte de oud-prof in de pendelbus naar Bernardus Golf. "
     "In de bus bleek ze zeker niet de enige Amerikaanse.", False),
    ("Toernooi trekt duizenden bezoekers", "Vanaf het station rijden pendelbussen en extra bussen.", False),
    # door — echte OV-berichten
    ("Trein vertraagd, reizigers pakken de bus", "De storing duurde een uur.", True),
    ("Station Amersfoort krijgt nieuwe perrons", "De NS laat de perrons vernieuwen.", True),
    ("Trein ontspoorde bij Weesp", "De spoorlijn was uren dicht.", True),
    ("Bushalte aan de Markt wordt verplaatst", "De halte gaat 200 meter opschuiven.", True),
    ("Conducteur bekogeld in stoptrein bij Den Haag", "De conducteur is nagekeken.", True),
    ("Nieuwe app voor ovpay", "De ovpay-app wordt dit najaar vernieuwd.", True),
    ("Metrostation Centraal tijdelijk gesloten", "De metro's rijden niet.", True),
    ("NS verlengt noodregeling op traject Zwolle", "Reizigers moeten rekening houden met minder treinen.", True),
    ("Spoorstoring bij Utrecht opgeheven", "ProRail heeft de wisselstoring verholpen.", True),
    ("Nieuwe intercity naar Berlijn", "De trein rijdt vanaf december vier keer per dag.", True),
    ("Streekvervoer kampt met personeelstekort", "Vervoerders schrappen ritten.", True),
    ("Buurtbus in de Achterhoek rijdt niet meer", "Er zijn te weinig vrijwillige chauffeurs.", True),
    ("Veerdienst Vlissingen-Breskens vaart uit de vaart", "De rederij zet een boot in.", True),
    ("Lijn 12 rijdt vanaf maandag een andere route", "De route gaat via het centrum.", True),
    ("Werkzaamheden aan perron", "Het perron wordt volgende maand opgeknapt.", True),
    ("Reactie Nederlandse Spoorwegen", "De Nederlandse Spoorwegen zeggen meer in te zetten op punctualiteit.", True),
]
print(f"== regressie: {len(GOLDEN)} cases (precision-first) ==")
for t, s, want in GOLDEN:
    got = R._filter_passes(t, s)
    check(got == want, f"'{t[:55]}' -> {'door' if got else 'geweerd'} (verwacht {'door' if want else 'geweerd'})")

print()
if failures:
    print(f"FILTER-INVALIDATIE: {len(failures)} afwijking(en) — corrigeer vóór commit/deploy.")
    sys.exit(1)
print("ALLES OK — filter volgt het precision-first-beleid.")