politics-tracking/backend/politrack/sources/parlament/common.py

"""Shared spec and pure parsing helpers for the Austrian Parliament adapters (CC BY 4.0).

API (https://www.parlament.gv.at/recherchieren/open-data/):
  * list exports: POST /Filter/api/filter/data/<listeId>?js=eval&showAll=true  (JSON body = filter)
      409 persons since 1918 · 101 parliamentary items · 251 speeches of a person
      110 Parlamentskorrespondenz (press releases) · 600 calendar (Termine)
  * detail pages: <page URL>?json=TRUE
"""

from __future__ import annotations

import re
from datetime import date, datetime
from typing import Any

from politrack.core import Request, SourceSpec, register_spec
from politrack.text import html_to_text

BASE = "https://www.parlament.gv.at"
CHAMBER_SLUG = {"NR": "nationalrat", "BR": "bundesrat"}
CHAMBER_NAME_DE = {"NR": "Nationalrat", "BR": "Bundesrat"}
GOVERNMENT_RE = re.compile(r"Bundeskanzler|Vizekanzler|Bundesminister|Staatssekretär", re.IGNORECASE)

SPEC = register_spec(SourceSpec(
    slug="parlament",
    name="Parlament Österreich (Open Data)",
    kind="official",
    homepage="https://www.parlament.gv.at/recherchieren/open-data/",
    license="CC BY 4.0",
    access="api",
    trust=95,
    store_fulltext=True,
    min_interval=1.0,
    tos_notes="Official JSON API (?json=TRUE, /Filter/api/filter/data/<list>). Attribution required. The "
              "Parlamentsdirektion may prohibit uses incompatible with the dignity of parliament.",
))


def list_request(list_id: int, body: dict, kind: str, meta: dict | None = None) -> Request:
    return Request(url=f"{BASE}/Filter/api/filter/data/{list_id}?js=eval&showAll=true", method="POST", json=body,
                   kind=kind, meta=meta or {})


def detail_request(path: str, kind: str, meta: dict | None = None) -> Request:
    return Request(url=f"{BASE}{path}", params={"json": "TRUE"}, kind=kind, meta=meta or {})


def list_rows(data: dict) -> list[dict[str, Any]]:
    """Turn a filter-API export into dicts keyed by column label (duplicates get a _<index> suffix)."""
    labels: list[str] = []
    for i, header in enumerate(data.get("header", [])):
        label = header.get("label") or header.get("feld_name") or f"col{i}"
        labels.append(label if label not in labels else f"{label}_{i}")
    return [dict(zip(labels, row)) for row in data.get("rows", [])]


def parse_date(value: str | None) -> date | None:
    if not value:
        return None
    for fmt in ("%d.%m.%Y", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d"):
        try:
            return datetime.strptime(value[:19], fmt).date()
        except ValueError:
            continue
    return None


def walk_dicts(obj: Any, required_key: str):
    """Yield every nested dict that contains `required_key`."""
    if isinstance(obj, dict):
        if required_key in obj:
            yield obj
        for value in obj.values():
            yield from walk_dicts(value, required_key)
    elif isinstance(obj, list):
        for value in obj:
            yield from walk_dicts(value, required_key)


# -- faction votes ------------------------------------------------------------------------

PARTY_TOKEN = re.compile(r"^[A-ZÄÖÜ][A-Za-zÄÖÜäöüß.\- ]{0,30}$")
# Split votes ("teilweise", "bzw.", per-part results in parentheses) can't be attributed to factions.
SPLIT_VOTE = re.compile(r"\(|\bbzw\.|\bteilweise\b|\btlw\.|\bmehrstimmig\b|\bgetrennt", re.IGNORECASE)


def parse_party_list(value: str | None) -> list[str]:
    if not value:
        return []
    parts = [p.strip(" .;") for p in re.split(r",|\bund\b", value)]
    return [p for p in parts if PARTY_TOKEN.match(p)]


def parse_vote_text(text: str) -> dict[str, Any] | None:
    """'137. Sitzung …: Gesetzesvorschlag in dritter Lesung angenommen Dafür: ÖVP, SPÖ, dagegen: FPÖ'"""
    m = re.search(r"\b(angenommen|abgelehnt)\b", text, re.IGNORECASE)
    if not m:
        return None
    rest = text[m.end():]
    roll_call = "namentliche abstimmung" in text.lower()
    split = bool(SPLIT_VOTE.search(rest)) and not roll_call
    pro = re.search(r"Dafür:\s*(.*?)(?:,?\s*dagegen:|$)", rest, re.IGNORECASE)
    contra = re.search(r"dagegen:\s*(.*)$", rest, re.IGNORECASE)
    return {
        "stage": text[: m.start()].strip(" :,"),
        "result": "accepted" if m.group(1).lower() == "angenommen" else "rejected",
        "for": [] if split or roll_call else parse_party_list(pro.group(1) if pro else None),
        "against": [] if split or roll_call else parse_party_list(contra.group(1) if contra else None),
        "unanimous": "einstimmig" in text.lower(),
        "split": split,
        "roll_call": roll_call,
    }


# -- roll-call votes ----------------------------------------------------------------------

ROLL_CALL_HEADER = re.compile(r"Mit\s*[„\"“]?\s*(Ja|Nein)\s*[“\"”]?\s*stimmten[^:]*:", re.IGNORECASE)
NEW_VOTE = re.compile(r"\b(gelangen|kommen)\s+(?:\w+\s+){0,2}zur\s+Abstimmung\b", re.IGNORECASE)
PERSON_LINK = re.compile(r"<a[^>]*href=\"[^\"]*/person/(\d+)\"[^>]*>.*?</a>", re.IGNORECASE | re.DOTALL)


def parse_roll_call(html_text: str, anchor: str | None) -> dict[str, list[tuple[str, int | None]]]:
    """Extract the "Mit „Ja“ stimmten …" / "Mit „Nein“ stimmten …" lists that follow `anchor`.

    Returns {"for": [(label, pad|None)], "against": [...]}. Names are mostly surnames; where a surname is
    ambiguous the protocol links the person, which gives us the parliament id directly.
    """
    start = 0
    if anchor:
        m = re.search(r'id="%s"' % re.escape(anchor), html_text)
        if not m:
            return {}
        start = m.start()
    first = ROLL_CALL_HEADER.search(html_text, start)
    if not first:
        return {}
    # The lists must belong to this anchor: they follow within a few numbered paragraphs and no new vote
    # may be called in between.
    between = html_text[start + 10 : first.start()]
    if anchor and (len(re.findall(r'class="[^"]*randnummer', between)) > 3 or NEW_VOTE.search(html_to_text(between))):
        return {}
    end_match = re.compile(r"<hr|class=\"[^\"]*randnummer").search(html_text, first.end())
    segment = html_text[first.start() : end_match.start() if end_match else len(html_text)]
    segment = PERSON_LINK.sub(lambda m: f" @@{m.group(1)}@@ ", segment)
    text = html_to_text(segment)

    result: dict[str, list[tuple[str, int | None]]] = {"for": [], "against": []}
    headers = list(ROLL_CALL_HEADER.finditer(text))
    for i, header in enumerate(headers):
        block = text[header.end() : headers[i + 1].start() if i + 1 < len(headers) else len(text)]
        key = "for" if header.group(1).lower() == "ja" else "against"
        for token in re.split(r"[,;]|\.\s*$", block):
            token = token.strip(" .")
            if not token:
                continue
            link = re.fullmatch(r"@@(\d+)@@", token)
            result[key].append((token, int(link.group(1))) if link else (token, None))
    return result


# -- immunity -----------------------------------------------------------------------------

STGB_OFFENSES = {
    "111": "Üble Nachrede", "115": "Beleidigung", "146": "Betrug", "153": "Untreue", "165": "Geldwäscherei",
    "201": "Vergewaltigung", "283": "Verhetzung", "288": "Falsche Beweisaussage", "293": "Fälschung eines Beweismittels",
    "295": "Unterdrückung eines Beweismittels", "297": "Verleumdung", "302": "Missbrauch der Amtsgewalt",
    "304": "Bestechlichkeit", "305": "Vorteilsannahme", "306": "Vorteilsannahme zur Beeinflussung",
    "307": "Bestechung", "308": "Verbotene Intervention", "310": "Verletzung des Amtsgeheimnisses",
}


def parse_immunity_report(text: str) -> dict[str, Any]:
    authority = re.search(r"Ersuchen (?:der|des) (.+?),\s*(?:do\.|GZ|Zl)", text)
    case_ref = re.search(r"\bGZ\.?\s*([^,]+?),", text)
    # "wegen des Verdachtes einer strafbaren Handlung nach § 283 Abs. 1 Z 1 und Z 2, Abs. 2 StGB"
    offense = re.search(r"wegen[^§]{0,90}?(§§? ?\d+.{0,120}?\b(?:StGB|VerbotsG|FinStrG|[A-Z][A-Za-z]*G)\b)", text)
    offense_text = offense.group(1).strip() if offense else None
    if offense_text:
        section = re.search(r"§ ?(\d+)", offense_text)
        if section and section.group(1) in STGB_OFFENSES and "StGB" in offense_text:
            offense_text = f"{offense_text} ({STGB_OFFENSES[section.group(1)]})"
    decision = re.search(r"(?:wird|werde)\s+(nicht\s+)?zugestimmt|(nicht\s+)?zuzustimmen", text)
    recommendation = None
    if decision:
        recommendation = "refuse" if (decision.group(1) or decision.group(2)) else "consent"
    connection = None
    if re.search(r"\bkein Zusammenhang\b", text):
        connection = False
    elif re.search(r"\bein Zusammenhang\b", text):
        connection = True
    quote = re.search(r"Als Ergebnis seiner Beratungen.*?(?:zugestimmt|nicht zugestimmt)\s*\.", text)
    return {
        "authority": authority.group(1).strip() if authority else None,
        "case_ref": case_ref.group(1).strip() if case_ref else None,
        "offense": offense_text,
        "recommendation": recommendation,
        "political_connection": connection,
        "quote": quote.group(0) if quote else None,
    }