# rss - A maubot plugin to subscribe to RSS/Atom feeds.
# Copyright (C) 2022 Tulir Asokan
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
from __future__ import annotations

from typing import Any
from datetime import datetime, timedelta, timezone
from string import Template
from time import mktime, monotonic, time
import asyncio
import hashlib
import html
import io
import logging
import json
import re

import aiohttp
import attr
import feedparser

from maubot import MessageEvent, Plugin, __version__ as maubot_version
from maubot.handlers import command, event
from mautrix.types import (
    EventID,
    EventType,
    MessageType,
    PowerLevelStateEventContent,
    RoomID,
    StateEvent,
)
from mautrix.util.async_db import UpgradeTable
from mautrix.util.config import BaseProxyConfig, ConfigUpdateHelper

from .db import DBManager, Entry, Feed, Subscription, normalize_title
from .html_scrape import extract_entries as _scrape_entries
from .html_scrape import page_title as _scrape_page_title
from .migrations import upgrade_table

rss_change_level = EventType.find("xyz.maubot.rss", t_class=EventType.Class.STATE)


# Gewogen OV-trefwoord-families voor het match-filter. Elke tuple = (regex, gewicht).
# Per familie wordt 1× geteld per ARTIKEL (titel én samenvatting samen; herhaling
# of het woord in beide velden = één signaal — 'busstation' 60× weegt 1×, besluit
# 2026-09-11). Eén term die meerdere families matcht telt óók maar 1× (hoogste
# gewicht wint): 'ov-chipkaart' bevat 'chipkaart', 'openbaar vervoer' bevat
# 'vervoer', 'treinstoring' matcht de zwakke \w*trein\w*-familie mee — geen
# dubbeltelling (besluit 2026-09-13; 'busbaan' stond in twee families en is
# uit de halte-familie verwijderd, refactor 2026-09-13). Drempel (3): doorlaten
# als de som van de unieke familie-gewichten >= MATCH_FILTER_THRESHOLD
# (titel weegt niet extra).
# MAX-GEWICHT 2 — besluit 2026-09-12: een enkele FAMILIE-hit kan nooit solo
# doorlaten (2 < 3); écht OV-nieuws combineert in de praktijk altijd met een
# tweede signaal (gevalideerd op 205 productie-artikelen; zie beleidsblok).
# Gewicht 2 (onmiskenbaar OV: merken, sectorjargon, samenstellingen als bushalte/
# spoorlijn/veerboot — de vroegere 'sterk(3)' én 'medium(2)' zijn hierin
# opgegaan, want beide moeten nu combineren).
# Gewicht 1 (zwak/dubbelzinnig): moet met een andere OV-familie combineren
# (een tweede, andere familie — dezelfde zwakke familie in titel én samenvatting
# telt niet als combinatie) EN minstens één zwak woord moet in de TITEL staan:
# een combinatie van alleen zwakke woorden in de samenvatting is geen bewijs
# (het golfbericht met "station" + "pendelbus" + "bus" in de samenvatting is
# geen OV-nieuws). Weert zo "VW-bus"/"schoolbus", "weer op de rails", het
# werkwoord "brengen", het idioom "goed met elkaar overweg kunnen", het werkwoord
# "overwegen", het werkwoord "opwaarderen" en brandweerjargon ("sein brandmeester"),
# zonder eenduidige bushalte/overweg/NS/trein-berichten te raken.
# Inflectie: families matchen met \w* aan het eind, zodat meervoud en vervoeging
#   meegenomen worden ("spoorlijnen", "seinstoringen", "ontspoorde", "chipkaarten").
#   Een koppelteken vóór een match is géén OV-signaal: "VW-bus" en "radio-station"
#   vallen af (lookbehind bij 'bus' resp. 'station').
# MATCH_FILTER_BLACKLIST: gematchte woorden die tóch geen OV-waarde hebben (o.a.
#   straatnamen rond stations — "de Stationsstraat afsluiten voor een festival"
#   is geen OV-nieuws). Zwartgelijste woorden leveren geen score op, maar
#   blokkeren de familie niet: een écht "station"-woord ernaast scoort gewoon.
# Bewust geaccepteerde botsingen: krant "De Metro"/café "De Tram" (gewicht 2,
#   moet combineren). 'spoorbrug' weegt 2: een brug is vaak louter locatie
#   ('bij de spoorbrug tussen Liempde en Boxtel', Stuifmail 2026-09-12) — écht
#   spoorbrug-nieuws combineert met trein/spoor/verkeer (besluit 2026-09-12).
# Naam-botsingen die bewust zijn afgezwakt (2026-09-12): 'hermes' (vervoerder
#   Hermes vs. achternaam Ton Hermes — Margraten-false positive), 'breng'
#   (vervoerder vs. werkwoord 'brengen'), 'spoor' (figuurlijk 'op het spoor van').
# Losse algemene woorden zijn zwak (moeten combineren): "trein", "sprinter"
#   (atleet), "vervoerder" (logistiek), "reiziger"/"passagier" (vliegen),
#   "inchecken"/"uitchecken" (Schiphol/hotel), "station(s)" (omroepen), "bus",
#   "sein" (brandweer), "overweg", "breng", "kaartje", "opwaarderen", "rails",
#   "staking" (alle sectoren), "vervoer" (vracht/auto), "abonnement", "ritten".
# Multi-woord-frasen en merknamen (ice-berlijn, deutsche bahn, european sleeper,
# ns international, teso, italo, nightjet, maaslijn) zijn onmiskenbaar OV en
# wegen 2; een los 'ice' is NIET toevoegbaar (case-insensitive botsing met 'ijs',
# besluit 2026-09-12).
# =====================================================================
# FILTERCLASSIFICATIE-BELEID (bij elke aanpassing aan het filter volgen)
# =====================================================================
# Doel: precisie boven recall. Liever een OV-bericht te veel gefilterd dan
# een niet-OV-bericht te veel doorgelaten (gebruikersbesluit, 2026-09-10).
#
# Beslissing, in drie stappen (refactor 2026-09-13; logica equivalent met de
# voorheen gelaagde takken — MAX-2-stelsel, besluit 2026-09-12):
#   1. STERK SIGNAAL (whitelist-term óf dienststoring-zin in de TITEL) -> DOORLATEN.
#      Whitelist = termen die in élke context OV betekenen (merken:
#      NS/ProRail/Arriva; onmiskenbaar jargon: spoorstoring, perron, bushalte,
#      conducteur, ovpay; multi-woord-frasen: nederlandse spoorwegen, openbaar
#      vervoer, ice-berlijn ...). Een whitelist-hit in de titel telt +2,
#      bovenop het familie-gewicht van dezelfde term (elke whitelist-term zit
#      óók in een familie) — samen altijd >= drempel, dus eenduidige OV-titels
#      ('Reactie Nederlandse Spoorwegen', 'Werkzaamheden aan perron') passeren
#      alleen. REGEL: twijfel je of de term ook buiten het OV kan voorkomen ->
#      NIET hier, maar zwak (eigen familie). Een whitelist-term ALLEEN in de
#      SAMENVATTING blijft inert: een bijzin als 'de zakelijke bezoeker is ook
#      nodig voor het openbaar vervoer' in een economiebericht is geen
#      OV-nieuws (besluit 2026-09-12); zo'n hit scoort alleen via de familie.
#   2. CONTEXT: som van unieke OV-signalen over titel + samenvatting
#      >= drempel (3). Per familie 1× per artikel (herhaling of beide velden
#      tellen niet dubbel, besluit 2026-09-11); overlappende familie-treffers
#      van dezelfde term wegen 1× met het hoogste gewicht (span-merge in
#      _fam_hits, besluit 2026-09-13 — 'ov-chipkaart' bevat 'chipkaart' maar
#      weegt als één term 2, geen 3).
#      - sterk (2):    onmiskenbaar OV-jargon/merken/samenvoegingen (spoorlijn,
#                      bushalte, busverkeer, intercity, veerboot, ertms,
#                      rangeer..., ice-frasen, ...) én het dienststoring-
#                      contextpatroon 'vervoerswoord + rijdt/rijden (niet) +
#                      niet' ('sprinters rijden niet', 'metro rijdt niet').
#                      Eén hit (titel óf samenvatting) is onvoldoende: moet
#                      combineren (2 + 1 = 3). De vroegere 'één sterk woord
#                      volstaat'-regel is vervallen — 'hermes' als achternaam
#                      passeerde zo op één vervoerder-hit (Margraten,
#                      2026-09-12).
#      - zwak (1):     DUBBELZINNIGHEID. Woorden die (ook) buiten het OV
#                      voorkomen. REGEL: elk dubbelzinnig woord krijgt een EIGEN
#                      familie (niet samengevoegd), zodat twee verschillende
#                      zwakke woorden wél als combinatie tellen, maar één enkel
#                      woord nooit de doorslag kan geven. Bekende dubbelzinnige:
#                      trein, bus, station(s), sprinter, reiziger/passagier,
#                      vervoerder, inchecken/uitchecken, sein, overweg, breng,
#                      kaartje, opwaarderen, rails, dienstregeling, chipkaart,
#                      reisproduct, vervoer(s)bewijs, ontsporen,
#                      pendelbus/shuttlebus (evenement-/hotel-/parkeerpendel;
#                      synoniemen -> bewust één familie), busongeluk/busvervoer
#                      (touringcar/schoolbus), staking, vervoer, vervoersbedrijf,
#                      vervoerregio, ritten, bovenleiding (industrie),
#                      abonnement, hermes, spoor.
#   3. TITEL-ANKER: staat er geen enkel OV-signaal in de titel (geen
#      familie-hit, geen whitelist-term), dan is de score uitsluitend
#      samenvatting-bewijs en mag die geen zwak woord bevatten. Twee/meer
#      zwakke woorden in de samenvatting ('station' + 'pendelbus' + 'bus' in
#      een golfverslag) en sterk+zwak uitsluitend in de samenvatting
#      ('openbaar vervoer' + 'zakelijke reizigers' in een economiebericht,
#      beide besluiten 2026-09-12) zijn geen OV-bewijs. Alleen twee sterke
#      termen in de samenvatting blijven bewust door (delftopzondag
#      'tram 1, halte Buitenhofdreef', besluit 2026-09-13).
#   BLACKLIST       -> gematchte woorden zonder OV-waarde, restant van
#      straatnamen rond spoorlijnen die de zwakke \b\w*trein\w*\b-familie zou
#      matchen ('treinstraat'/'treinweg'); blokkeren de familie niet. De
#      oudere stations*-varianten zijn sinds de woordgrens-familie
#      \bstations?\b inert en verwijderd (2026-09-13) — gedrag vastgelegd in
#      validate_filter.py.
#
# BEOORDELEN VAN EEN NIEUW WOORD (incl. COMPLETEHEIDS-AUDIT, 2026-09-12):
#   a) Kan het ook buiten het OV voorkomen?  ja -> zwak (1), eigen familie.
#   b) Is het een samenvoeging/merk die alleen OV kan betekenen? -> sterk (2)
#      of whitelist (frasen die in élke context OV betekenen).
#   c) Bij twijfel: zwak (1). Conservatief = precision-first.
#   d) Ter controle: leg nieuwe woordkandidaten voor aan minimaal één extra
#      taalmodel (de audit op 2026-09-10 gebruikte 3 modellen; enige stem
#      'TWIJFEL'/'NIET-OV' => behandelen als dubbelzinnig).
#   e) COMPLETEHEID (besluit 2026-09-12): check bij elke filterwijziging óók of
#      de families compleet zijn — draai de OV-gap-scan over de afgewezen
#      artikelen van OV-specialist-feeds (SpoorPro/Treinreiziger/Mobiliteit),
#      classificeer nieuwe kandidaten via stap a-d en vul aan. De audit van
#      2026-09-12 leverde o.a. spoorsector, spoorbeheerder, spoornet,
#      spoorvervoerder, treinvertraging/uitval, ertms, teso, rangeer, ice-frasen,
#      maaslijn, ns international, deutsche bahn, european sleeper, italo,
#      nightjet (sterk) en 11 zwakke woorden op.
# Dienststoring-contextpatroon: vervoerswoord + 'rijdt/rijden (niet)' + 'niet'
# ('sprinters rijden niet', 'metro rijdt niet tussen Rotterdam en Schiedam').
# Gedeeld door de familie én het titel-anker (_disruption_hit): in de TITEL is de
# storingszin een sterk OV-signaal (net als een whitelist-term), in de samenvatting
# weegt hij als familie (2). 'rijden' an sich is géén OV-signaal. 'schoolbus'
# matcht niet ('bus' niet aan het woordbegin), refactor 2026-09-13.
_DISRUPTION_RE_SRC = (
    r"\b(?:sprinters?|stoptrein(?:en)?|intercity'?s?|trein(?:en)?|trams?|bussen|bus|"
    r"metro'?s?|veerdienst(?:en)?)\s+(?:rijdt|rijden|reed|reden)\s+niet\b"
)

# =====================================================================
MATCH_FILTER_FAMILIES: list[tuple[str, int]] = [
    # sterk (2) — onmiskenbaar OV, maar één hit is onvoldoende (max-2-stelsel,
    # besluit 2026-09-12): elke echte OV-bericht combineert met een tweede signaal.
    (r"\b(?:ns|arriva|qbuzz|connexxion|cxx|transdev|keolis|ret|gvb|ebs|htm|syntus|u-ov|allgo|flixbus|blauwnet|r-net|valleilijn|waterbus|eurostar|thalys|randstadrail|gvu|prorail|rrreis|qliner|ov|teso|deutsche bahn|european sleeper|ns international|italo|nightjet)\b", 2),  # vervoerders; 'hermes' is zwak (achternaam-botsing, Margraten 2026-09-12), 'Breng' ook (werkwoord "brengen"); internationale vervoerders: teso (veer), deutsche bahn/european sleeper/italo/nightjet/ns international (audit 2026-09-12)
    (r"\b(?:spoorweg|spoorba(?:an|nen)|spoorlijn|spoorverkeer|spoorwegovergang|spoorstoring|perron|seinstoring|wisselstoring|spoorstaking|spoorsector|spoorbeheerder|spoornet|spoorvervoerder|ertms)\w*\b", 2),  # spoorinfra; let op de klinkerwisseling: 'ontsporing' (open lettergreep, één o) vs 'ontspoord'/'ontspoorde' (dicht, dubbel o) — vandaar beide stammen. Géén 'spoorbrug': die is 2 maar vaak louter locatie (besluit 2026-09-12). Uitbreiding 2026-09-12 (completheids-audit): spoorsector, spoorbeheerder, spoornet, spoorvervoerder, ertms
    (r"\b(?:treinstation|treinstations|metrostation|metrostations|busstation|busstations)\b", 2),  # station-samenstellingen: onmiskenbaar OV, maar één station-woord náást een ander onderwerp ('pand naast busstation', besluit 2026-09-11) is geen OV-bericht; in de samenvatting moet een tweede OV-woord bijkomen
    (r"(?<![\w-])\bstations?\b", 1),  # los 'station(s)' is ZWAK: ook tv-/radio-/regionale omroepstations ('de lokale stations'), moet combineren met een andere OV-familie
    (r"\b(?:bus|tram|metro)-?(?:halte|haltes|lijn|lijnen|verkeer|net|staking|ba(?:an|nen)|hok(?:je|jes)|chauffeur)\w*\b", 2),  # voertuig-samenstelling (bushalte/buslijn/bushokje/busbanen/buschauffeur/…); géén 'station': *station-samenstellingen zitten in de station-familie, anders telde 'busstation' dubbel mee — fix 2026-09-11. Géén 'ongeluk'/'vervoer': die zijn bewust zwak (busongeluk = ook touringcar, busvervoer = ook bedrijfsvervoer, audit 2026-09-12). 'busbaan'/'bushokje' stonden óók in de halte-familie — verwijderd bij de refactor 2026-09-13 (geen dubbelwoorden in twee families)
    (r"\b(?:treinstoring|treinverkeer|treinstaking|treinongeval|treinbotsing|treinongeluk|treinvertraging|treinhinder|treinuitval|treinverstoring)\w*\b", 2),  # trein-jargon-samenstellingen: onmiskenbaar OV; uitgebreid met vertraging/hinder/uitval/verstoring (audit 2026-09-12)
    (r"\b\w*trein\w*\b", 1),  # losse 'trein' en algemene samenstellingen (stoptrein, goederentrein, treinverkeer al in bovenstaande familie) — ZWAK: 'de trein naar mijn nieuwe huis' is geen OV-jargon, moet combineren met een andere familie
    (r"\b(?:intercity|streekvervoer|stadsvervoer|veerdienst|veerboot|veerboten|veerpont|voetveer|veerverbinding|buurtbus|belbus|maaslijn|nedersaksenlijn|kamperlijn)\w*\b", 2),  # treintypes, vervoerssectoren, veren en lijnnamen ('sprinter' staat in de zwakke groep: atleet-botsing); maaslijn/nedersaksenlijn/kamperlijn via audit 2026-09-12
    (r"\blijn\s?\d{1,3}\b", 2),  # 'lijn 12' — vervoerslijn met nummer
    (r"(?:\brangeer\w*|\brangeren\w*)\b", 2),  # rangeren/rangerend/rangeerterrein/rangeerbedrijf: onmiskenbaar spoorjargon (audit 2026-09-12; stam 'rangeren' = ranger+en, 'rangeerterrein' = rangeer+...)
    (r"\bice[\s\-–—]*(?:berlijn|hamburg|brussel|bremen|parijs|frankfurt|keulen|amsterdam|bazel|milaan|kopenhagen|zurich|zürich|duitsland|international)\b|\bice-trein\w*\b", 2),  # ICE-frasen: onmiskenbaar trein; los 'ice' is NIET toevoegbaar (case-insensitive botsing met 'ijs', besluit 2026-09-12)
    # onmiskenbaar OV maar iets algemener (2, moet ook combineren)
    (r"\btrams?\b", 2),
    (r"\bmetro(?:'?s)?\b", 2),
    (r"\b(?:openbaar vervoer|lightrail|halte)\w*\b", 2),  # frasen en halte-hits; 'bushokje'/'busbaan' matchen al de bus-samenstelling-familie — bewust niet dubbel in twee families (refactor 2026-09-13)
    (r"\b(?:ov-chipkaart|ovpay|ov-pay|studentenreisproduct)\w*\b", 2),
    (r"\b(?:ov-staking|conducteursstaking)\w*\b", 2),
    (_DISRUPTION_RE_SRC, 2),  # dienststoring-context ('sprinters rijden niet', 'metro rijdt niet...'); zie _DISRUPTION_RE_SRC
    (r"\bconduct(?:eur|eurs)\b", 2),  # trein-/buspersoneel: onmiskenbaar OV-jargon
    (r"\b(?:pont|ponten)\b", 2),  # waterbus/veerdienst
    (r"\bspoorbrug(?:gen)?\b", 2),  # spoorinfra, maar vaak locatie-referentie ('bij de spoorbrug', Stuifmail 2026-09-12); moet combineren — zelfde regel als station-samenstellingen
    # zwak (1) — dubbelzinnig, moet combineren
    (r"(?<!-)\bbus(?:sen)?\b", 1),  # ná een koppelteken geen OV: 'VW-bus', 'school-bus' 
    (r"\b(?:pendelbus|shuttlebus)\w*\b", 1),  # evenement-/hotel-/parkeerpendel ('pendelbus naar de golfbaan') is geen OV; synoniemen bewust in één familie
    (r"\b(?:inchecken|uitchecken)\w*\b", 1),  # ook op luchthavens/hotels — geen OV-jargon, moet combineren
    (r"\bsprinters?\b", 1),  # ook atleet ('sprinter verpulvert record') — aparte familie, moet combineren
    (r"\bvervoerders?\b", 1),  # ook logistiek ('vrachtvervoerder') — aparte familie, moet combineren
    (r"\b(?:reizigers?|passagiers?)\b", 1),  # ook 'zakenreizigers vliegen' / vliegtuigpassagiers — synoniemen bewust één familie (audit 2026-09-12)
    (r"\bdienstregeling\w*\b", 1),  # dubbelzinnig (modelaudit) — meestal OV, maar moet combineren
    (r"\bchipkaart(?:en)?\b", 1),  # dubbelzinnig (bankpas/algemeen) — aparte familie
    (r"\breisproduct(?:en)?\b", 1),  # dubbelzinnig — aparte familie
    (r"\bvervoer(?:s)?bewijs\w*\b", 1),  # dubbelzinnig — aparte familie
    (r"\b(?:ontspor|ontspoor)\w*\b", 1),  # werkwoord 'ontsporen' is dubbelzinnig ('het plan ontspoorde'); echte trein-ontsporing combineert met trein/spoor 
    (r"\bbreng\b", 1),  # vervoerder Breng vs werkwoord "brengen" ("hoe breng je ... in beeld")
    (r"\boverweg\b", 1),  # spoorwegovergang vs idioom "goed met elkaar overweg kunnen"
    (r"\boverwegen\b", 1),  # meervoud van overweg vs werkwoord ("een gratieverlening overwegen")
    (r"\bsein\b", 1),  # spoorsein vs brandweerjargon ("sein brandmeester") en figuurlijk gebruik
    (r"\bkaart(?:je|jes)\b", 1),
    (r"\bopwaarderen\b", 1),  # ov-chipkaart opladen vs werkwoord "een wijk/park opwaarderen"
    (r"\brails\b", 1),
    (r"\bbusongeluk(?:ken)?\b", 1),  # audit 2026-09-12: ook touringcar/schoolbus; combineert met buschauffeur/bus → door (busongeluk Zwitserland)
    (r"\bbusvervoer\b", 1),  # audit 2026-09-12: ook touringcar-/bedrijfsvervoer
    (r"\bspoor\b", 1),  # figuurlijk 'op het spoor van de daders', 'het spoor bijster'; écht spoornieuws combineert met spoorlijn/trein/perron
    (r"\bstaking(?:en)?\b", 1),  # arbeidsstaking in alle sectoren; 'spoorstaking'/'busstaking'/'ov-staking' zitten in de sterke families (besluit 2026-09-12)
    (r"\bvervoer\b", 1),  # ook vracht-/autovervoer; 'openbaar vervoer' zit in de sterke familie; samenstellingen (streekvervoer e.d.) matchen hier niet (geen \b)
    (r"\bvervoersbedrijf(?:ven)?\b", 1),  # ook goederen-/taxibedrijf (audit 2026-09-12)
    (r"\bvervoerregio'?s?\b", 1),  # OV-autoriteit ('Vervoerregio Amsterdam'), maar audit: ook algemener vervoersgebied
    (r"\britten\b", 1),  # ook fiets-/taxiritten (audit 2026-09-12)
    (r"\bbovenleiding(?:en)?\b", 1),  # audit 2026-09-12: ook water/gas/elektra-bovenleiding; spoor-combinatie (spoorlijn/ertms) haalt de drempel
    (r"\babonnement(?:en)?\b", 1),  # gym/museum/telefoon (audit 2026-09-12)
    (r"\bhermes\b", 1),  # vervoerder Hermes vs. achternaam (Ton Hermes, stichting Adoptie Graven Margraten, 2026-09-12)
]

# Drempel voor doorlaten: som van unieke familie-gewichten >= 3. Max-2-stelsel
# (besluit 2026-09-12): geen enkele familie weegt meer dan 2, dus een enkele hit
# kan nooit solo doorlaten; een whitelist-term in de titel telt +2 (zie
# _filter_passes). De oude '2×titel + 1×samenvatting'-rekensom is vervallen
# (2026-09-11: de titel weegt niet extra).
MATCH_FILTER_THRESHOLD = 3

# Woorden die een familie wél matcht maar die geen OV-waarde hebben. Een hit dat
# hierin staat telt niet mee voor de score (en blokkeert de familie niet). Vorm:
# exact woord (lowercase), zoals door de familie-regex wordt gematcht.
# Refactor 2026-09-13: alleen woorden die een HUIDIGE familie nog matchen blijven
# staan ('treinstraat'/'treinweg' via de zwakke \b\w*trein\w*\b-familie). De
# stations*-varianten (stationsstraat, stationair, stationeren, stationcar,
# stationwagen, stationsplein, ...) waren inert sinds de woordgrens-familie
# \bstations?\b en zijn verwijderd — dat gedrag is in validate_filter.py
# vastgelegd.
MATCH_FILTER_BLACKLIST: frozenset[str] = frozenset({
    # Straatnamen rond spoorlijnen die de \b\w*trein\w*\b-familie zou matchen:
    # een straat die toevallig 'Trein' heet maakt een bericht geen OV-bericht.
    "treinstraat", "treinweg",
})

# MATCH_FILTER_WHITELIST: termen die een bericht per definitie OV maken. Sinds
# besluit 2026-09-12 is dit geen auto-pass meer: een whitelist-hit in de TITEL
# telt +2 (één onmiskenbare term 2, nog een signaal nodig voor de drempel 3), en
# in de SAMENVATTING helemaal niet (die scoort via de overeenkomstige familie —
# een whitelist-term zit altijd óók in een familie). Zie _filter_passes.
# "Onmiskenbaar" = in élke context OV: merken (NS/ProRail/Arriva), onmiskenbaar
# jargon (spoorstoring, perron, bushalte, conducteur, ovpay), multi-woord-frasen
# (nederlandse spoorwegen, openbaar vervoer, ice-berlijn ...).
# Regel: twijfel je of de term ook buiten het OV kan voorkomen -> NIET hier, maar
# zwak (eigen familie). Daarom 'hermes' hieruit (achternaam, besluit 2026-09-12).
# 'treinstel'/'veerdienst'/'buurtbus'/'lijn 12' staan hier alleen als TITEL-anker
# (= het oude effect van één onmiskenbare term in de titel, nu via +2) — in de
# samenvatting wegen ze als familie (2/1). Vergelijking is case-insensitive.
MATCH_FILTER_WHITELIST: frozenset[str] = frozenset({
    # Vervoerders en OV-merken (incl. multi-woord termen)
    "nederlandse spoorwegen", "ns", "prorail", "randstadrail", "arriva", "qbuzz",
    "connexxion", "transdev", "keolis", "ret", "gvb", "htm", "ebs", "syntus",
    "u-ov", "allgo", "blauwnet", "r-net", "valleilijn", "rrreis",
    "qliner", "flixbus", "eurostar", "thalys", "gvu", "ovpay",
    "deutsche bahn", "european sleeper", "ns international", "italo", "nightjet",
    "teso", "ice-berlijn",
    # Spoorinfra-jargon (samenvoegingen, onmiskenbaar OV)
    "spoorstoring", "spoorverkeer", "spoorwegovergang", "spoorlijn", "spoorbaan",
    "spoorstaking", "spoorsector", "spoorbeheerder", "spoornet", "spoorvervoerder",
    "ertms", "perron", "wisselstoring", "seinstoring", "treinstoring", "treinverkeer",
    "treinstaking", "treinongeval", "treinbotsing", "treinstel", "treinvertraging",
    "treinhinder", "treinuitval", "treinverstoring", "maaslijn", "nedersaksenlijn",
    "kamperlijn",
    # Bus/tram/metro-jargon
    "bushalte", "bushaltes", "busbaan", "buslijn", "busnet",
    "busstaking", "buurtbus", "veerdienst", "lijn 12",
    "tramhalte", "tramlijn", "tramnet", "tramverkeer",
    "metrolijn", "metroverkeer",
    # Tarief/vervoerder-jargon
    "ov-chipkaart", "ov-staking", "lightrail",
    "conducteur", "conducteurs",
    "openbaar vervoer",
})

_compiled_match_families: list[tuple[re.Pattern, int]] = [
    (re.compile(pat, re.IGNORECASE), w) for pat, w in MATCH_FILTER_FAMILIES
]

# Whitelist: enkel-woord termen als \\b-term\\b (geen substring-botsingen zoals
# 'ns' in 'kunstenaar'), multi-woord termen als exacte frase.
_WHITELIST_RE = re.compile(
    r"\b(?:" + "|".join(re.escape(w) for w in MATCH_FILTER_WHITELIST if " " not in w) + r")\b",
    re.IGNORECASE,
)
_WHITELIST_PHRASES: tuple[str, ...] = tuple(w for w in MATCH_FILTER_WHITELIST if " " in w)

# Dienststoring-context als sterk TITEL-signaal (zie _DISRUPTION_RE_SRC).
_DISRUPTION_RE = re.compile(_DISRUPTION_RE_SRC, re.IGNORECASE)


class Config(BaseProxyConfig):
    def do_update(self, helper: ConfigUpdateHelper) -> None:
        helper.copy("update_interval")
        helper.copy("max_backoff")
        helper.copy("poll_parallelism_limit")
        helper.copy("spam_sleep")
        helper.copy("command_prefix")
        helper.copy("notification_template")
        helper.copy("exclude_paywalled")
        helper.copy("title_dedup_days")
        helper.copy("run_log_file")
        helper.copy("whatsapp_api")
        helper.copy("whatsapp_session")
        helper.copy("whatsapp_chat_id")
        helper.copy("admins")


class BoolArgument(command.Argument):
    def __init__(self, name: str, label: str = None, *, required: bool = False) -> None:
        super().__init__(name, label, required=required, pass_raw=False)

    def match(self, val: str, **kwargs) -> tuple[str, Any]:
        part = val.split(" ")[0].lower()
        if part in ("f", "false", "n", "no", "0"):
            res = False
        elif part in ("t", "true", "y", "yes", "1"):
            res = True
        else:
            raise ValueError("invalid boolean")
        return val[len(part) :], res


class RSSBot(Plugin):
    dbm: DBManager
    exclude_paywalled: bool = False
    poll_task: asyncio.Future
    _poll_started: bool = False
    poll_sema: asyncio.Semaphore | None
    _run_logger: logging.Logger | None = None
    _run_handler: logging.FileHandler | None = None
    http: aiohttp.ClientSession
    power_level_cache: dict[RoomID, tuple[int, PowerLevelStateEventContent, StateEvent]]

    @classmethod
    def get_config_class(cls) -> type[BaseProxyConfig]:
        return Config

    @classmethod
    def get_db_upgrade_table(cls) -> UpgradeTable:
        return upgrade_table

    async def start(self) -> None:
        # Idempotentie-guard: maubot kan internal_start() (her)aanroepen via
        # boot/client/upload/update-paden; Instance.start() zet zijn eigen
        # started-flag pas ná plugin.start(). Zonder guard ontstaat bij een
        # tweede (concurrente) aanroep een tweede poll-loop. Check+set is
        # synchronous (geen await ertussen) → atoom in de event-loop.
        if self._poll_started:
            self.log.warning("start() called twice, ignoring duplicate poll task")
            return
        self._poll_started = True
        try:
            await super().start()
            self.on_external_config_update()
            self.dbm = DBManager(self.database)
            self.http = self.client.api.session
            self.power_level_cache = {}
            self.poll_task = asyncio.create_task(self.poll_feeds())
        except Exception:
            # Niets geïnitialiseerd (geen dbm/http/poll_task): guard terugdraaien,
            # anders wordt een geldige herstart genegeerd en blijft de instance dood.
            self._poll_started = False
            raise

    def on_external_config_update(self) -> None:
        self.config.load_and_update()
        poll_parallelism_limit = self.config["poll_parallelism_limit"]
        self.poll_sema = (
            asyncio.Semaphore(poll_parallelism_limit) if poll_parallelism_limit > 0 else None
        )
        # single-instance fork: class-level since _parse_rss is a classmethod
        type(self).exclude_paywalled = bool(self.config["exclude_paywalled"])
        self._setup_run_log()

    def _setup_run_log(self) -> None:
        # De logger zelf is gedeeld over reloads (zelfde log-naam). Bij een
        # maubot-reload is self._run_logger op de nieuwe instance weer None,
        # terwijl de oude handler aan de geregistreerde logger blijft hangen.
        # Daarom ALTIJD op naam opruimen, niet op het instance-attribuut:
        # anders wordt elke poll-run eenmalig per reload gelogd (N× duplicaat).
        self._run_logger = logging.getLogger(f"{self.log.name}.runlog")
        for handler in list(self._run_logger.handlers):
            self._run_logger.removeHandler(handler)
            handler.close()
        self._run_handler = None
        run_log_file = self.config["run_log_file"] if "run_log_file" in self.config else ""
        if not run_log_file:
            return
        self._run_handler = logging.FileHandler(run_log_file, encoding="utf-8")
        self._run_handler.setFormatter(logging.Formatter("%(asctime)s %(message)s"))
        self._run_logger.setLevel(logging.INFO)
        self._run_logger.propagate = False
        self._run_logger.addHandler(self._run_handler)

    async def stop(self) -> None:
        await super().stop()
        self._poll_started = False
        self.poll_task.cancel()
        if self._run_handler:
            self._run_handler.close()

    async def poll_feeds(self) -> None:
        try:
            await self._poll_feeds()
        except asyncio.CancelledError:
            self.log.debug("Polling stopped")
        except Exception:
            self.log.exception("Fatal error while polling feeds")

    @staticmethod
    def _filter_text(raw: str) -> str:
        """Platte tekst voor het match-filter: eerst entities decoderen, dán tags
        strippen. Andersom (strip → unescape) laat dubbel-geëscapete feeds hun HTML
        als tekst houden — inclusief URL's en attributen, die dan vals scoren."""
        return re.sub(r"<[^>]+>", "", html.unescape(raw))

    @staticmethod
    def _whitelist_hit(text: str) -> bool:
        """True als een onmiskenbare OV-term uit MATCH_FILTER_WHITELIST in text zit.

        Enkele-woord termen matchen op woordgrens ('ns' telt NIET binnen 'kunstenaar'),
        multi-woord termen ('nederlandse spoorwegen') als exacte frase, beide
        case-insensitive."""
        if _WHITELIST_RE.search(text):
            return True
        low = text.lower()
        return any(ph in low for ph in _WHITELIST_PHRASES)

    @staticmethod
    def _disruption_hit(text: str) -> bool:
        """True als een dienststoring-zin ('sprinters rijden niet', 'metro rijdt
        niet tussen Rotterdam en Schiedam') in text staat. In de TITEL telt zo'n
        zin als sterk OV-signaal (+2, net als een whitelist-term); als familie
        (2) scoort hij overal — zie _DISRUPTION_RE_SRC."""
        return _DISRUPTION_RE.search(text) is not None

    @staticmethod
    def _fam_hits(text: str) -> list[tuple[int, int]]:
        """(familie-index, gewicht) per OV-signaal in text (1× per familie).

        Gematchte woorden uit MATCH_FILTER_BLACKLIST (o.a. 'stationsstraat')
        tellen niet mee en blokkeren de familie niet: een écht OV-woord in
        dezelfde tekst scoort gewoon verder.

        Eén term telt maximaal één signaal (besluit 2026-09-13): overlappende
        treffers van verschillende families op dezelfde plek worden samengevoegd
        met het hoogste gewicht. Voorheen telde één woord dubbel als het in
        meerdere families stond — 'ov-chipkaart' bevat 'chipkaart' (2+1=3),
        'openbaar vervoer' bevat 'vervoer' (2+1=3), 'treinstoring' matcht ook
        de zwakke \\w*trein\\w*-familie (2+1=3). Max-2-stelsel: één term kan
        nooit solo doorlaten."""
        spans: list[tuple[int, int, int, int]] = []  # (start, end, familie-idx, gewicht)
        for idx, (family, weight) in enumerate(_compiled_match_families):
            for m in family.finditer(text):
                if m.group(0).lower() in MATCH_FILTER_BLACKLIST:
                    continue
                spans.append((m.start(), m.end(), idx, weight))
                break
        spans.sort()
        merged: list[tuple[int, int, int, int]] = []
        for start, end, idx, weight in spans:
            if merged and start < merged[-1][1]:
                s0, e0, i0, w0 = merged[-1]
                if weight > w0:
                    merged[-1] = (s0, max(e0, end), idx, weight)
                else:
                    merged[-1] = (s0, max(e0, end), i0, w0)
            else:
                merged.append((start, end, idx, weight))
        return [(idx, w) for _, _, idx, w in merged]

    @staticmethod
    def _filter_passes(title: str, summary: str) -> bool:
        """Driestaps-beslissing, precision-first (refactor 2026-09-13; beslissing
        equivalent met v1.13.0, maar de drie historische uitzonderings-takken
        reduceren tot één regel — zie stap 3):

          1. STERK SIGNAAL: een whitelist-term óf een dienststoring-zin
             ('sprinters rijden niet') in de titel is voldoende. Elke
             whitelist-term staat óók in een familie, dus zo'n titel telt +2 én
             het familie-gewicht — samen altijd >= drempel ('NS meldt storing',
             'Werkzaamheden aan perron', 'Buurtbus rijdt niet meer'). In de
             samenvatting ALLEEN is zo'n term géén bewijs ('openbaar vervoer'
             als bijzin in een economiebericht, besluit 2026-09-12) en scoort
             die alleen via de overeenkomstige familie.

          2. CONTEXT: de som van unieke OV-signalen over titel + samenvatting
             >= drempel (3). Eén term telt één signaal per artikel ('busstation'
             60× of in beide velden = één keer); overlappende familie-treffers
             van dezelfde term wegen 1× met het hoogste gewicht (span-merge in
             _fam_hits, besluit 2026-09-13 — 'ov-chipkaart' bevat 'chipkaart'
             maar weegt als één term 2, geen 3). Geen
             familie weegt meer dan 2 (max-2-stelsel, besluit 2026-09-12): een
             enkel sterk woord kan nooit solo doorlaten ('hermes' als
             achternaam passeerde vroeger op één vervoerder-hit van 3 —
             Margraten; écht OV-nieuws combineert in de praktijk altijd met een
             tweede signaal, gevalideerd op 205 productie-artikelen).

          3. TITEL-ANKER: staat er geen enkel OV-signaal in de titel (geen
             familie-hit, geen whitelist-term), dan is de score uitsluitend
             samenvatting-bewijs en mag die geen zwak/ambigue woord bevatten —
             'station'+'pendelbus'+'bus' in een golfverslag, 'openbaar
             vervoer'+'zakelijke reizigers' in een economiebericht en 'sterk +
             zwak uitsluitend in de samenvatting' (beide besluiten 2026-09-12)
             vallen hieronder. Twee sterke termen in de samenvatting blijven
             bewust door (delftopzondag 'tram 1, halte Buitenhofdreef',
             besluit 2026-09-13).

        Weert zo 'VW-bus' (bus matcht niet ná een koppelteken), 'schoolbus',
        'bussen rijden om' in een woningbrand-lead, het werkwoord 'brengen',
        het idioom 'goed met elkaar overweg kunnen', 'een wijk opwaarderen',
        brandweerjargon 'sein brandmeester', figuratief 'weer op de rails',
        en een cultuurstukje over een 'stationspiano'. Bewust doorgelaten:
        krant 'De Metro'/café 'De Tram' (2, combineert) en atleet-'sprinter'
        in combinatie met ander OV-woord."""
        t_hits = RSSBot._fam_hits(title)
        s_hits = RSSBot._fam_hits(summary)
        # Unieke OV-signalen over beide velden: één term telt één signaal per
        # artikel ('busstation' in titel én samenvatting, of 60×, blijft één
        # treffer); _fam_hits levert per term al het hoogste gewicht bij
        # overlappende familie-treffers.
        combined = dict(t_hits + s_hits)
        score = sum(combined.values())
        strong_title = RSSBot._whitelist_hit(title) or RSSBot._disruption_hit(title)
        if strong_title:
            # Stap 1: sterk titel-signaal — +2 bovenop het familie-gewicht van
            # dezelfde term, waardoor een eenduidige OV-titel alleen voldoet.
            score += 2
        if score < MATCH_FILTER_THRESHOLD:
            return False
        if not t_hits and not strong_title and 1 in combined.values():
            # Stap 3: geen enkel titel-signaal én een zwak woord in de score —
            # samenvatting-bewijs zonder titel-anker is geen OV-bewijs.
            return False
        return True

    @staticmethod
    def _filter_summary() -> str:
        """Mensvriendelijke weergave van het vaste gewogen match-filter."""
        return (
            "**OV-matchfilter (vast in code; max-2-stelsel, precision-first):**\n"
            "- **Sterk signaal in de TITEL → door**: whitelist-term (NS, ProRail, Nederlandse Spoorwegen, spoorstoring, perron, "
            "bushalte, conducteur, ov-chipkaart, ovpay, openbaar vervoer, Deutsche Bahn, ICE Berlijn, ...) of een dienststoring-zin "
            "('sprinters rijden niet', 'metro rijdt niet') — die telt +2 bovenop het familie-gewicht. In de samenvatting alleen "
            "weegt zo'n term mee als gewone familie.\n"
            "- **Context (som ≥ 3)**: unieke OV-families over titel + samenvatting. Sterk (2): vervoerders (NS, Arriva, ProRail, Qbuzz, ...), "
            "spoorinfra (spoorlijn, spoorstoring, perron, ertms, spoorsector), bus-samenstellingen (bushalte, buslijn, buschauffeur, busbaan), "
            "treinstoring/treinverkeer/treinvertraging, intercity/streekvervoer/veerdienst/buurtbus/belbus, 'lijn 12', ICE-frasen, rangeer..., "
            "tram, metro, openbaar vervoer, lightrail, halte, pont, spoorbrug, ov-chipkaart/ovpay, ov-staking, conducteur, dienststoring-zin. "
            "Zwak (1): losse 'bus' (niet ná een koppelteken), 'pendelbus'/'shuttlebus', 'trein', 'station(s)', 'breng', 'overweg', "
            "'overwegen', 'sein', 'kaartje', 'opwaarderen', 'rails', 'inchecken'/'uitchecken', 'dienstregeling', 'chipkaart', "
            "'reisproduct', 'vervoersbewijs', 'ontsporen', 'reiziger'/'passagier', 'vervoerder', 'sprinter', 'spoor', 'staking', "
            "'vervoer', 'busongeluk', 'busvervoer', ... — elke dubbelzinnige term heeft een eigen familie en moet combineren.\n"
            "- **Eén term = één signaal**: herhaling, beide velden en overlappende familie-treffers wegen niet dubbel. "
            "Zonder enig titel-signaal (familie-hit, whitelist-term of storingszin in de titel) mag de score geen zwak woord "
            "bevatten — samenvatting-bewijs is geen OV-bewijs (twee sterke termen in de samenvatting blijven bewust door).\n"
            "- **Uitgesloten (blacklist)**: 'treinstraat'/'treinweg' (straatnamen rond spoorlijnen); historische stations*-gevallen "
            "('stationsstraat', 'stationair', 'stationcar', ...) worden al door de woordgrens-familie geweerd"
        )

    async def _send(
        self, feed: Feed, entry: Entry, sub: Subscription, *, check_filter: bool = True
    ) -> EventID | None:
        if check_filter:
            summary = self._filter_text(entry.summary)
            if not self._filter_passes(entry.title, summary):
                return None

        message = sub.notification_template.safe_substitute(
            {
                "feed_url": feed.url,
                "feed_title": feed.title,
                "feed_subtitle": feed.subtitle,
                "feed_link": feed.link,
                **attr.asdict(entry),
            }
        )
        msgtype = MessageType.NOTICE if sub.send_notice else MessageType.TEXT
        try:
            return await self.client.send_markdown(
                sub.room_id, message, msgtype=msgtype, allow_html=True
            )
        except Exception as e:
            self.log.warning(f"Failed to send {entry.id} of {feed.id} to {sub.room_id}: {e}")
            return None

    async def _send_whatsapp(self, feed: Feed, entry: Entry) -> bool:
        """Post entry.link naar de WhatsApp-groep via wwebjs-api (met dedup-check). True = nieuw geplaatst."""
        api = self.config.get("whatsapp_api", "")
        chat_id = self.config.get("whatsapp_chat_id", "")
        if not api or not chat_id:
            self.log.debug("WhatsApp forwarding disabled")
            return False
        link = entry.link
        if not link:
            self.log.warning(f"No link for {entry.id}, skipping WhatsApp")
            return False
        if await self.dbm.is_whatsapp_posted(link):
            self.log.info(f"{entry.id} already posted to WhatsApp before, skipping")
            return False
        session = self.config.get("whatsapp_session", "EdGpt")
        base = api.rstrip("/")
        # Dedup: sla over als de URL al in de doelgroep staat. Eerst de lokale
        # tabel (eigen posts; daarna de globale search (vreemde posts), gefilterd
        # op chat-id van het gevonden bericht.

        try:
            resp = await self.http.post(
                f"{base}/client/searchMessages/{session}",
                json={"query": link, "options": {"limit": 1}},
                timeout=aiohttp.ClientTimeout(total=20),
            )
            data = await resp.json()
            if not data.get("success"):
                # Checkfout (wwebjs antwoordt ook met HTTP 200 + success:false) =
                # niet posten: anders omzeilt de fout de dedup → dubbele link.
                self.log.warning(f"WhatsApp dedup check failed for {entry.id}: {data}")
                return False
            for msg in data.get("messages", []):
                msg_data = msg.get("_data", {})
                if (
                    msg_data.get("id", {}).get("remote") == chat_id
                    and link in msg_data.get("body", "")
                ):
                    self.log.info(f"{entry.id} already in WhatsApp group, skipping")
                    return False
        except Exception as e:
            self.log.warning(f"WhatsApp dedup check failed for {entry.id}: {e}")
            return False
        try:
            resp = await self.http.post(
                f"{base}/client/sendMessage/{session}",
                json={"chatId": chat_id, "content": link, "contentType": "string"},
                timeout=aiohttp.ClientTimeout(total=20),
            )
            data = await resp.json()
            ok = bool(data.get("success"))
            if ok:
                try:
                    await self.dbm.mark_whatsapp_posted(link)
                except Exception as e:
                    self.log.warning(f"Failed to record WhatsApp post {entry.id}: {e}")
            self.log.info(f"WhatsApp post {entry.id} to {chat_id}: {'ok' if ok else 'failed: ' + str(data)}")
            return ok
        except Exception as e:
            self.log.warning(f"Failed to send {entry.id} to WhatsApp {chat_id}: {e}")
            return False

    async def _broadcast(
        self, feed: Feed, entry: Entry, subscriptions: list[Subscription]
    ) -> tuple[int, int, int]:
        """Send entry to all subs. Returns (placed, filtered, failed).

        placed   = 1 when the entry was posted to at least one channel (Matrix room or newly posted WhatsApp link)
        filtered = 1 when the global match filter rejected the entry
        failed   = number of subscriptions where sending failed
        """
        self.log.debug(f"Broadcasting {entry.id} of {feed.id}")
        summary = self._filter_text(entry.summary)
        if not self._filter_passes(entry.title, summary):
            return 0, 1, 0
        spam_sleep = self.config["spam_sleep"]
        if spam_sleep >= 0:
            results: list[EventID | None] = []
            for sub in subscriptions:
                results.append(await self._send(feed, entry, sub, check_filter=False))
                await asyncio.sleep(spam_sleep)
        else:
            results = await asyncio.gather(
                *(self._send(feed, entry, sub, check_filter=False) for sub in subscriptions)
            )
        sent = sum(1 for result in results if result is not None)
        whatsapp_ok = await self._send_whatsapp(feed, entry)
        return (1 if (sent or whatsapp_ok) else 0, 0, len(results) - sent)

    async def _poll_once(self) -> None:
        subs = await self.dbm.get_feeds()
        if not subs:
            self._log_run_summary(0, 0, 0, 0, 0, 0, 0, 0.0)
            return
        now = int(time())
        start = monotonic()
        title_dedup_days = int(self.config.get("title_dedup_days", 30))
        posted_titles: set[str] = set()
        if title_dedup_days > 0:
            cutoff = datetime.now(timezone.utc) - timedelta(days=title_dedup_days)
            posted_titles = await self.dbm.get_recent_posted_titles(cutoff)
        tasks = [self.try_parse_feed(feed=feed) for feed in subs if feed.next_retry < now]
        feed: Feed
        entries: list[Entry] | None
        self.log.info(f"Polling {len(tasks)} feeds")
        evaluated_total = 0
        placed_total = 0
        filtered_total = 0
        deduped_total = 0
        failed_total = 0
        unchanged_total = 0
        for res in asyncio.as_completed(tasks):
            feed, entries = await res
            self.log.trace(
                f"Fetching {feed.id} (backoff: {feed.error_count} / {feed.next_retry}) "
                f"success: {entries is not None}"
            )
            if entries is None:
                # Feed returned 304: unchanged, no new entries, no error.
                unchanged_total += 1
                if feed.error_count > 0:
                    self.log.debug(f"Resetting backoff of {feed.id}")
                    await self.dbm.set_backoff(feed, error_count=0, next_retry=0)
                continue
            if not entries:
                error_count = feed.error_count + 1
                next_retry_delay = self.config["update_interval"] * 60 * error_count
                next_retry_delay = min(next_retry_delay, self.config["max_backoff"] * 60)
                next_retry = int(time() + next_retry_delay)
                self.log.debug(f"Setting backoff of {feed.id} to {error_count} / {next_retry}")
                await self.dbm.set_backoff(feed, error_count, next_retry)
                continue
            elif feed.error_count > 0:
                self.log.debug(f"Resetting backoff of {feed.id}")
                await self.dbm.set_backoff(feed, error_count=0, next_retry=0)
            try:
                new_entries = {entry.id: entry for entry in entries}
            except Exception:
                self.log.exception(f"Weird error in items of {feed.url}")
                continue
            await self.dbm.filter_entries(feed.id, new_entries)
            self.log.trace(
                f"Feed {feed.id} had {len(new_entries)} new entries (of {len(entries)} total)"
            )
            evaluated_total += len(entries)
            new_entry_list: list[Entry] = list(new_entries.values())
            new_entry_list.sort(key=lambda entry: (entry.date, entry.id))
            await self.dbm.add_entries(new_entry_list)
            for entry in new_entry_list:
                norm = normalize_title(entry.title)
                if title_dedup_days > 0 and norm and norm in posted_titles:
                    deduped_total += 1
                    self.log.info(
                        f"{entry.id} of feed {feed.id} title already posted ({norm[:60]}), skipping"
                    )
                    continue
                placed, filtered, failed = await self._broadcast(
                    feed, entry, feed.subscriptions
                )
                placed_total += placed
                filtered_total += filtered
                failed_total += failed
                if placed and norm:
                    # In-memory set: vangt ook duplicaten binnen dezelfde poll-run
                    # (twee sites van dezelfde uitgever in één run).
                    posted_titles.add(norm)
                    try:
                        await self.dbm.mark_entry_posted(feed.id, entry.id)
                    except Exception:
                        self.log.warning(
                            f"Failed to mark {entry.id} of feed {feed.id} as posted"
                        )
        duration = monotonic() - start
        self._log_run_summary(
            len(tasks),
            evaluated_total,
            placed_total,
            filtered_total,
            deduped_total,
            failed_total,
            unchanged_total,
            duration,
        )

    def _log_run_summary(
        self,
        feeds: int,
        evaluated: int,
        selected: int,
        filtered: int,
        deduped: int,
        failed: int,
        unchanged: int,
        duration: float,
    ) -> None:
        line = (
            f"run complete feeds={feeds} evaluated={evaluated} selected={selected} "
            f"filtered={filtered} deduped={deduped} failed={failed} unchanged={unchanged} "
            f"duration={duration:.2f}s"
        )
        self.log.info(f"Finished polling {feeds} feeds in {duration:.2f} seconds: {line}")
        if self._run_logger:
            self._run_logger.info(line)

    async def _poll_feeds(self) -> None:
        self.log.debug("Polling started")
        while True:
            try:
                await self._poll_once()
            except asyncio.CancelledError:
                self.log.debug("Polling stopped")
                raise
            except Exception:
                self.log.exception("Error while polling feeds")
            await asyncio.sleep(self.config["update_interval"] * 60)

    async def try_parse_feed(self, feed: Feed | None = None) -> tuple[Feed, list[Entry] | None]:
        if self.poll_sema is not None:
            async with self.poll_sema:
                return await self._try_parse_feed(feed)
        else:
            return await self._try_parse_feed(feed)

    async def _try_parse_feed(self, feed: Feed | None = None) -> tuple[Feed, list[Entry] | None]:
        try:
            self.log.trace(
                f"Trying to fetch {feed.id} / {feed.url} "
                f"(backoff: {feed.error_count} / {feed.next_retry})"
            )
            return await self.parse_feed(feed=feed)
        except Exception as e:
            self.log.warning(f"Failed to parse feed {feed.id} / {feed.url}: {e}")
            return feed, []

    @property
    def _feed_get_headers(self) -> dict[str, str]:
        return {
            "User-Agent": f"maubot/{maubot_version} +https://github.com/maubot/rss",
        }

    async def parse_feed(
        self, *, feed: Feed | None = None, url: str | None = None
    ) -> tuple[Feed, list[Entry] | None]:
        if feed is None:
            if url is None:
                raise ValueError("Either feed or url must be set")
            feed = Feed(id=-1, url=url, title="", subtitle="", link="")
        elif url is not None:
            raise ValueError("Only one of feed or url must be set")
        headers = self._feed_get_headers
        if feed.id > 0:
            if feed.etag:
                headers["If-None-Match"] = feed.etag
            if feed.last_modified:
                headers["If-Modified-Since"] = feed.last_modified
        resp = await self.http.get(
            feed.url,
            headers=headers,
            timeout=aiohttp.ClientTimeout(total=20),
        )
        if resp.status == 304:
            # Feed unchanged since last fetch: no new entries, no error.
            return feed, None
        etag = resp.headers.get("ETag", "")
        last_modified = resp.headers.get("Last-Modified", "")
        ct = resp.headers["Content-Type"].split(";")[0].strip()
        scrape_cfg = self._load_scrape_config(feed)
        if scrape_cfg:
            feed, entries = await self._parse_html(feed, resp, scrape_cfg)
        elif ct == "application/json" or ct == "application/feed+json":
            feed, entries = await self._parse_json(feed, resp)
        else:
            feed, entries = await self._parse_rss(feed, resp)
        # Validators pas bewaren als de body volledig gelezen én geparseerd is:
        # anders stuurt de volgende poll If-None-Match op een respons die nooit
        # verwerkt is (304 → 'unchanged', entries stil weg).
        if feed.id > 0 and (etag or last_modified):
            feed.etag, feed.last_modified = etag, last_modified
            try:
                await self.dbm.set_feed_headers(feed, etag, last_modified)
            except Exception:
                self.log.debug(f"Failed to store fetch headers of {feed.id}")
        return feed, entries

    @staticmethod
    def _load_scrape_config(feed: Feed) -> dict | None:
        """Parsed scrape_config JSON for HTML-page feeds; None for normal feeds.

        Invalid JSON silently falls back to normal feed parsing instead of
        taking the feed down; the validation happens when the config is set."""
        raw = (feed.scrape_config or "").strip()
        if not raw:
            return None
        try:
            cfg = json.loads(raw)
        except (ValueError, TypeError):
            return None
        return cfg if isinstance(cfg, dict) else None

    async def _parse_html(
        self, feed: Feed, resp: aiohttp.ClientResponse, cfg: dict
    ) -> tuple[Feed, list[Entry]]:
        content = await resp.read()
        html_text = content.decode("utf-8", errors="replace")
        raw_entries, skipped = _scrape_entries(html_text, feed.url, cfg)
        self.log.debug(
            f"Scraped {feed.url}: {len(raw_entries)} items, {skipped} skipped"
        )
        feed.title = _scrape_page_title(html_text) or feed.url
        feed.subtitle = feed.url
        feed.link = feed.url
        entries = [
            Entry(
                feed_id=feed.id,
                id=str(e["id"]),
                date=e["date"],
                title=e["title"],
                summary=e["summary"],
                link=e["link"],
                paywalled=e["paywalled"],
            )
            for e in raw_entries
        ]
        if self.exclude_paywalled:
            entries = [e for e in entries if not e.paywalled]
        return feed, entries

    @classmethod
    async def _parse_json(
        cls, feed: Feed, resp: aiohttp.ClientResponse
    ) -> tuple[Feed, list[Entry]]:
        content = await resp.json()
        if content["version"] not in (
            "https://jsonfeed.org/version/1",
            "https://jsonfeed.org/version/1.1",
        ):
            raise ValueError("Unsupported JSON feed version")
        if not isinstance(content["items"], list):
            raise ValueError("Feed is not a valid JSON feed (items is not a list)")
        feed.title = content["title"]
        feed.subtitle = content.get("subtitle", "")
        feed.link = content.get("home_page_url", "")
        entries = [cls._parse_json_entry(feed.id, entry) for entry in content["items"]]
        if cls.exclude_paywalled:
            entries = [e for e in entries if not e.paywalled]
        return feed, entries

    @classmethod
    def _parse_json_entry(cls, feed_id: int, entry: dict[str, Any]) -> Entry:
        try:
            date = datetime.fromisoformat(entry["date_published"])
        except (ValueError, KeyError):
            date = datetime.now(timezone.utc)
        title = entry.get("title", "")
        summary = (
            entry.get("summary") or entry.get("content_html") or entry.get("content_text") or ""
        ).strip()
        id = str(entry["id"])
        link = entry.get("url") or id
        return Entry(
            feed_id=feed_id, id=id, date=date, title=title, summary=summary, link=link,
            paywalled=cls._is_paywalled_link(link),
        )

    @classmethod
    async def _parse_rss(
        cls, feed: Feed, resp: aiohttp.ClientResponse
    ) -> tuple[Feed, list[Entry]]:
        content = await resp.read()
        headers = {"Content-Location": feed.url, **resp.headers, "Content-Encoding": "identity"}
        parsed_data = feedparser.parse(io.BytesIO(content), response_headers=headers)
        if parsed_data.bozo:
            if not isinstance(parsed_data.bozo_exception, feedparser.ThingsNobodyCaresAboutButMe):
                raise parsed_data.bozo_exception
        feed_data = parsed_data.get("feed", {})
        feed.title = feed_data.get("title", feed.url)
        feed.subtitle = feed_data.get("description", "")
        feed.link = feed_data.get("link", "")
        entries = [cls._parse_rss_entry(feed.id, entry) for entry in parsed_data.entries]
        if cls.exclude_paywalled:
            entries = [e for e in entries if not e.paywalled]
        return feed, entries

    @staticmethod
    def _is_paywalled_link(link: str) -> bool:
        # DPG Media premium articles live under /ps/ (Parool/VK/Trouw feeds
        # carry no paywall marker; the /ps/ path is the only reliable signal).
        return "/ps/" in (link or "")

    @classmethod
    def _parse_rss_entry(cls, feed_id: int, entry: Any) -> Entry:
        return Entry(
            feed_id=feed_id,
            id=(
                getattr(entry, "id", None)
                or getattr(entry, "link", None)
                or hashlib.sha1(
                    " ".join(
                        [
                            getattr(entry, "title", ""),
                            getattr(entry, "description", ""),
                        ]
                    ).encode("utf-8")
                ).hexdigest()
            ),
            date=cls._parse_rss_date(entry),
            title=getattr(entry, "title", ""),
            summary=getattr(entry, "description", "").strip(),
            link=getattr(entry, "link", ""),
            paywalled=(
                # Telegraaf/NHD convention: <premium>true</premium>
                str(getattr(entry, "premium", "")).lower() == "true"
                # DPG Media (AD/Volkskrant/Parool/Trouw): <dpp:paid>true</dpp:paid>
                or str(getattr(entry, "dpp_paid", "")).lower() == "true"
                # DPG Media /ps/ premium path (Parool/VK/Trouw carry no marker)
                or cls._is_paywalled_link(getattr(entry, "link", ""))
            ),
        )

    @staticmethod
    def _parse_rss_date(entry: Any) -> datetime:
        try:
            return datetime.fromtimestamp(mktime(entry["published_parsed"]))
        except (KeyError, TypeError, ValueError):
            pass
        try:
            return datetime.fromtimestamp(mktime(entry["date_parsed"]))
        except (KeyError, TypeError, ValueError):
            pass
        return datetime.now(timezone.utc)

    async def get_power_levels(
        self, room_id: RoomID
    ) -> tuple[PowerLevelStateEventContent, StateEvent]:
        try:
            expiry, pls, create = self.power_level_cache[room_id]
            if expiry < int(time()):
                return pls, create
        except KeyError:
            pass
        pls = await self.client.get_state_event(room_id, EventType.ROOM_POWER_LEVELS)
        create = await self.client.get_state_event(room_id, EventType.ROOM_CREATE, format="event")
        self.power_level_cache[room_id] = (int(time()) + 5 * 60, pls, create)
        return pls, create

    async def can_manage(self, evt: MessageEvent) -> bool:
        if evt.sender in self.config["admins"]:
            return True
        levels, create = await self.get_power_levels(evt.room_id)
        user_level = levels.get_user_level(evt.sender, create=create)
        state_level = levels.get_event_level(rss_change_level)
        if not isinstance(state_level, int):
            state_level = 50
        if user_level < state_level:
            await evt.reply(
                "You don't have the permission to manage the subscriptions of this room."
            )
            return False
        return True

    @command.new(
        name=lambda self: self.config["command_prefix"],
        help="Manage this RSS bot",
        require_subcommand=True,
    )
    async def rss(self) -> None:
        pass

    @rss.subcommand("subscribe", aliases=("s", "sub"), help="Subscribe this room to a feed.")
    @command.argument("url", "feed URL", pass_raw=True)
    async def subscribe(self, evt: MessageEvent, url: str) -> None:
        if not await self.can_manage(evt):
            return
        feed = await self.dbm.get_feed_by_url(url)
        if not feed:
            try:
                info, entries = await self.parse_feed(url=url)
            except Exception as e:
                await evt.reply(f"Failed to load feed: {e}")
                return
            feed = await self.dbm.create_feed(info)
            await self.dbm.add_entries(entries, override_feed_id=feed.id)
        elif feed.error_count > 0:
            await self.dbm.set_backoff(feed, error_count=feed.error_count, next_retry=0)
        feed_info = f"feed ID {feed.id}: [{feed.title}]({feed.url})"
        sub, _ = await self.dbm.get_subscription(feed.id, evt.room_id)
        if sub is not None:
            subscriber = (
                "You"
                if sub.user_id == evt.sender
                else f"[{sub.user_id}](https://matrix.to/#/{sub.user_id})"
            )
            await evt.reply(f"{subscriber} had already subscribed this room to {feed_info}")
        else:
            await self.dbm.subscribe(
                feed.id, evt.room_id, evt.sender, self.config["notification_template"]
            )
            await evt.reply(f"Subscribed to {feed_info}")

    @rss.subcommand(
        "set-filter", aliases=("f",), help="Show the fixed global OV match filter."
    )
    async def set_filter(self, evt: MessageEvent) -> None:
        await evt.reply(self._filter_summary())

    @rss.subcommand(
        "filter", aliases=("show-filter", "get-filter"), help="Show the current global match filter."
    )
    async def command_show_filter(self, evt: MessageEvent) -> None:
        await evt.reply(self._filter_summary())

    @rss.subcommand(
        "unsubscribe", aliases=("u", "unsub"), help="Unsubscribe this room from a feed."
    )
    @command.argument("feed_id", "feed ID", parser=int)
    async def unsubscribe(self, evt: MessageEvent, feed_id: int) -> None:
        if not await self.can_manage(evt):
            return
        sub, feed = await self.dbm.get_subscription(feed_id, evt.room_id)
        if not sub:
            await evt.reply("This room is not subscribed to that feed")
            return
        await self.dbm.unsubscribe(feed.id, evt.room_id)
        await evt.reply(f"Unsubscribed from feed ID {feed.id}: [{feed.title}]({feed.url})")

    @rss.subcommand(
        "template",
        aliases=("t", "tpl"),
        help="Change the notification template for a subscription in this room",
    )
    @command.argument("feed_id", "feed ID", parser=int)
    @command.argument("template", "new template", pass_raw=True, required=False)
    async def command_template(self, evt: MessageEvent, feed_id: int, template: str) -> None:
        if not await self.can_manage(evt):
            return
        sub, feed = await self.dbm.get_subscription(feed_id, evt.room_id)
        if not sub:
            await evt.reply("This room is not subscribed to that feed")
            return
        if not template:
            await evt.reply(
                '<p>Current template in this room:</p><pre><code language="markdown">'
                f"{html.escape(sub.notification_template.template)}"
                "</code></pre>",
                allow_html=True,
                markdown=False,
            )
            return
        await self.dbm.update_template(feed.id, evt.room_id, template)
        sub = Subscription(
            feed_id=feed.id,
            room_id=sub.room_id,
            user_id=sub.user_id,
            notification_template=Template(template),
            send_notice=sub.send_notice,
        )
        sample_entry = Entry(
            feed_id=feed.id,
            id="SAMPLE",
            date=datetime.now(timezone.utc),
            title="Sample entry",
            summary="This is a sample entry to demonstrate your new template",
            link="http://example.com",
        )
        await evt.reply(f"Template for feed ID {feed.id} updated. Sample notification:")
        await self._send(feed, sample_entry, sub)

    @rss.subcommand(
        "notice", aliases=("n",), help="Set whether or not the bot should send updates as m.notice"
    )
    @command.argument("feed_id", "feed ID", parser=int)
    @BoolArgument("setting", "true/false")
    async def command_notice(self, evt: MessageEvent, feed_id: int, setting: bool) -> None:
        if not await self.can_manage(evt):
            return
        sub, feed = await self.dbm.get_subscription(feed_id, evt.room_id)
        if not sub:
            await evt.reply("This room is not subscribed to that feed")
            return
        await self.dbm.set_send_notice(feed.id, evt.room_id, setting)
        send_type = "m.notice" if setting else "m.text"
        await evt.reply(f"Updates for feed ID {feed.id} will now be sent as `{send_type}`")

    @staticmethod
    def _format_subscription(feed: Feed, subscriber: str) -> str:
        msg = (
            f"* {feed.id} - [{feed.title}]({feed.url}) "
            f"(subscribed by [{subscriber}](https://matrix.to/#/{subscriber}))"
        )
        if feed.error_count > 1:
            msg += f"  \n  ⚠️ The last {feed.error_count} attempts to fetch the feed have failed!"
        return msg

    @rss.subcommand(
        "subscriptions",
        aliases=("ls", "list", "subs"),
        help="List the subscriptions in the current room.",
    )
    async def command_subscriptions(self, evt: MessageEvent) -> None:
        subscriptions = await self.dbm.get_feeds_by_room(evt.room_id)
        if len(subscriptions) == 0:
            await evt.reply("There are no RSS subscriptions in this room")
            return
        filter_line = self._filter_summary().replace("\n", "  \n")
        await evt.reply(
            "**Subscriptions in this room:**\n\n"
            + "\n".join(
                self._format_subscription(feed, subscriber)
                for feed, subscriber in subscriptions
            )
            + f"\n\n{filter_line}"
        )

    @event.on(EventType.ROOM_TOMBSTONE)
    async def tombstone(self, evt: StateEvent) -> None:
        if not evt.content.replacement_room:
            return
        await self.dbm.update_room_id(evt.room_id, evt.content.replacement_room)