diff --git a/.env.example b/.env.example index 278e9c9..9d2bb1d 100644 --- a/.env.example +++ b/.env.example @@ -31,6 +31,10 @@ POLITRACK_API_CORS_ORIGINS=["http://localhost:3000"] # Moderation endpoints (/api/v1/review, /api/v1/admin/runs). Leave empty to moderate via the CLI only. # POLITRACK_ADMIN_TOKEN= +# Nightly database dumps (backup service). Ideally on a different disk or share than DATA_DIR. +# BACKUP_DIR=/mnt/user/backups/politrack +# BACKUP_KEEP_DAYS=14 + # Raw-response archive inside the containers (mounted from ${DATA_DIR}/raw by docker-compose.yml). POLITRACK_RAW_DIR=/data/raw diff --git a/.gitattributes b/.gitattributes index c89a609..d6b350d 100644 --- a/.gitattributes +++ b/.gitattributes @@ -1,2 +1,4 @@ # Golden fixtures are compared byte for byte; never convert their line endings. backend/tests/fixtures/** -text +# Shell scripts run in Linux containers: LF even in a Windows checkout. +*.sh text eol=lf diff --git a/backend/politrack/api/main.py b/backend/politrack/api/main.py index c74d380..1f6e9a2 100644 --- a/backend/politrack/api/main.py +++ b/backend/politrack/api/main.py @@ -12,6 +12,7 @@ from typing import Annotated from fastapi import Depends, FastAPI, Header, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse from psycopg import Connection from politrack import queries @@ -66,6 +67,23 @@ def healthz(conn: DB) -> dict: return {"ok": True} +BACKUP_SCHEDULE = "30 2 * * *" # the `backup` service in docker-compose.yml logs each dump as a run + + +@app.get("/api/v1/status/sources", include_in_schema=False) +def source_status(conn: DB) -> JSONResponse: + """For monitoring (e.g. Uptime Kuma): HTTP 503 as soon as a scheduled source or the nightly backup missed its + last two slots. Lists every task with its last successful run.""" + from politrack.core import registry + + registry.load_builtin() + schedules = {name: cls.schedule for name, cls in {**registry.ADAPTERS, **registry.ENRICHERS}.items() if cls.schedule} + schedules["backup"] = BACKUP_SCHEDULE + rows = queries.source_freshness(conn, schedules) + stale = [r["adapter"] for r in rows if r["stale"]] + return JSONResponse({"ok": not stale, "stale": stale, "sources": rows}, status_code=503 if stale else 200) + + # -- reference data ----------------------------------------------------------------------- @app.get("/api/v1/stats", response_model=s.Stats) diff --git a/backend/politrack/queries/__init__.py b/backend/politrack/queries/__init__.py index 6dbd03a..2dba341 100644 --- a/backend/politrack/queries/__init__.py +++ b/backend/politrack/queries/__init__.py @@ -1,6 +1,14 @@ """Read queries shared by the REST API and the MCP server (one module per domain).""" -from politrack.queries.admin import list_sources, list_topics, pending_claims, recent_runs, set_review, stats +from politrack.queries.admin import ( + list_sources, + list_topics, + pending_claims, + recent_runs, + set_review, + source_freshness, + stats, +) from politrack.queries.claims import calendar, list_claims, timeline_histogram from politrack.queries.common import SourceFilter from politrack.queries.money import list_money, money_summary, social_feed, wiki_edits @@ -13,5 +21,5 @@ __all__ = [ "SourceFilter", "calendar", "former_overview", "get_organization", "get_party", "get_person", "list_claims", "list_money", "list_organizations", "list_parties", "list_persons", "list_sources", "list_topics", "list_votes", "money_summary", "pending_claims", "recent_runs", "person_network", "person_votes", "search", "set_review", "social_feed", - "stats", "timeline_histogram", "vote_detail", "wiki_edits", + "source_freshness", "stats", "timeline_histogram", "vote_detail", "wiki_edits", ] diff --git a/backend/politrack/queries/admin.py b/backend/politrack/queries/admin.py index dc5c8db..030816f 100644 --- a/backend/politrack/queries/admin.py +++ b/backend/politrack/queries/admin.py @@ -2,6 +2,8 @@ from __future__ import annotations +from datetime import datetime, timedelta, timezone + from psycopg import Connection @@ -61,6 +63,27 @@ def stats(conn: Connection) -> dict: ).fetchone() +def source_freshness(conn: Connection, schedules: dict[str, str], now: datetime | None = None) -> list[dict]: + """Last successful run per scheduled task. Stale = its two latest scheduled slots passed without a + successful run (plus an hour's grace for long runs), so one slow or failed run alone never alerts.""" + from croniter import croniter + + now = now or datetime.now(timezone.utc) + last_ok = {r["adapter"]: r["at"] for r in conn.execute( + "SELECT adapter, max(COALESCE(finished_at, started_at)) AS at FROM run " + "WHERE status IN ('ok', 'partial') GROUP BY adapter")} + rows = [] + for name, schedule in sorted(schedules.items()): + it = croniter(schedule, now) + due = it.get_prev(datetime) + before = it.get_prev(datetime) + deadline = before - timedelta(hours=1) + last = last_ok.get(name) + rows.append({"adapter": name, "schedule": schedule, "last_success": last.isoformat() if last else None, + "stale": last is None or last < deadline, "latest_slot": due.isoformat()}) + return rows + + def recent_runs(conn: Connection, limit: int = 50) -> list[dict]: return conn.execute( "SELECT adapter, mode, status, started_at, finished_at, stats, left(error, 2000) AS error FROM run " diff --git a/backend/politrack/sources/procurement.py b/backend/politrack/sources/procurement.py index 0c423c7..dd0af03 100644 --- a/backend/politrack/sources/procurement.py +++ b/backend/politrack/sources/procurement.py @@ -14,6 +14,7 @@ fetch items whose lastmod changed. Runs on the `bulk` queue. from __future__ import annotations +import re import xml.etree.ElementTree as ET from collections.abc import Iterator from datetime import date @@ -139,8 +140,28 @@ def parse_award_notice(xml: bytes) -> dict[str, Any] | None: } +PLACEHOLDER_NAME = re.compile(r"^[\W_]*$|^(?:n/?a|k\.?\s?a\.?|unbekannt|keine angabe|tbd|xxx+)$", re.IGNORECASE) +FRAMEWORK_BUYER = re.compile(r"^Auftraggeber (?:sind|ist)\b", re.IGNORECASE) + + +def plausible_date(value: str | None, latest: date) -> date | None: + """An award date, or None if it can't be one: spreadsheet placeholders (1899-12-30, 1900-01-01) and dates after + the notice was fetched (some publishers put the contract's end date in the conclusion field).""" + d = as_date(value) + return d if d and date(1995, 1, 1) <= d <= latest else None + + +def buyer_name(name: str) -> tuple[str, str | None]: + """Framework agreements name every eligible buyer in one sentence ("Auftraggeber sind die Republik Österreich + (Bund), die Bundesbeschaffung GmbH sowie alle weiteren ..."). They are concluded by the federal procurement + agency, so it is the buyer; the sentence is kept as the original wording.""" + if FRAMEWORK_BUYER.match(name) and "Bundesbeschaffung" in name: + return "Bundesbeschaffung GmbH", name + return name, None + + def org_ref(org: dict[str, Any], kind: str) -> OrgRef | None: - if not org.get("name"): + if not org.get("name") or PLACEHOLDER_NAME.match(org["name"].strip()): return None company_id = (org.get("company_id") or "").replace(" ", "") ids = {} @@ -167,7 +188,7 @@ class PublicContracts(Adapter): description = "Contract awards of Austrian public bodies (eForms and BVergG core data via data.gv.at)" schedule = "0 2 * * *" queue = "bulk" - parser_version = 2 + parser_version = 3 # 3: implausible award dates, placeholder contractors, framework-agreement buyers def start(self, ctx: RunContext) -> Iterator[Request]: for query in CATALOG_QUERIES: @@ -230,22 +251,32 @@ class PublicContracts(Adapter): if notice is None: ctx.stats["not_an_award"] += 1 return - buyer = next((org_ref(b, "public_body") for b in notice["buyers"] if b.get("name")), None) + named = next((b for b in notice["buyers"] if b.get("name")), None) + buyer_text = None + if named is not None: + name, buyer_text = buyer_name(named["name"]) + named = {**named, "name": name} + buyer = org_ref(named, "public_body") if named else None if buyer is None: return + latest = response.fetched_at.date() for contract in notice["contracts"]: for winner in contract["winners"]: supplier = org_ref(winner, "company") if supplier is None: + ctx.stats["contractor_unnamed"] += 1 continue amount = contract["amount"] or (notice["total"] if len(notice["contracts"]) == 1 else None) - awarded = as_date(contract["awarded"]) or as_date(notice["issued"]) + awarded = plausible_date(contract["awarded"], latest) or plausible_date(notice["issued"], latest) + if contract["awarded"] and awarded is None: + ctx.stats["implausible_award_date"] += 1 title = contract["title"] or notice["title"] or "Auftrag" yield Claim( claim_type="money_flow", org=buyer, counterpart_org=supplier, occurred_at=awarded, summary_de=f"Auftrag „{title}“ an {supplier.name}" + (f" – {format_eur_de(amount)} €" if amount else ""), summary_en=f"Contract \"{title}\" awarded to {supplier.name}" + (f" – EUR {amount}" if amount else ""), - data={"notice_id": notice["notice_id"], "cpv": notice["cpv"], "publisher": response.meta.get("publisher")}, + data={"notice_id": notice["notice_id"], "cpv": notice["cpv"], "publisher": response.meta.get("publisher"), + **({"buyer_text": buyer_text} if buyer_text else {})}, evidence=[Evidence(url=response.request.url, doc_type="register_entry", title=f"Vergabe: {title}", published_at=awarded, quote=f"{buyer.name} → {supplier.name}: {title}")], facet=Money(kind="contract", amount=amount, payer=buyer, payer_name=buyer.name, payee=supplier, diff --git a/backend/politrack/worker.py b/backend/politrack/worker.py index d334a9e..1013aa0 100644 --- a/backend/politrack/worker.py +++ b/backend/politrack/worker.py @@ -19,7 +19,11 @@ from __future__ import annotations import asyncio import logging +import os +import socket from datetime import datetime, timedelta, timezone +from logging.handlers import RotatingFileHandler +from pathlib import Path from croniter import croniter from procrastinate import App, PsycopgConnector @@ -31,9 +35,29 @@ from politrack.core import registry from politrack.core.runner import run from politrack.db import connection -logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") +LOG_FORMAT = "%(asctime)s %(levelname)s %(name)s: %(message)s" +logging.basicConfig(level=logging.INFO, format=LOG_FORMAT) log = logging.getLogger("politrack.watchdog") + +def _log_to_file() -> None: + """Container logs vanish when a container is recreated (every deploy); keep a rotating copy in + $POLITRACK_LOG_DIR/.log so failures can still be traced afterwards.""" + log_dir = os.environ.get("POLITRACK_LOG_DIR") + if not log_dir: + return + try: + handler = RotatingFileHandler(Path(log_dir) / f"{socket.gethostname()}.log", maxBytes=20_000_000, + backupCount=5, encoding="utf-8") + except OSError as exc: # e.g. the host folder belongs to root: logs stay on stdout only + log.warning("cannot write logs to %s: %s", log_dir, exc) + return + handler.setFormatter(logging.Formatter(LOG_FORMAT)) + logging.getLogger().addHandler(handler) + + +_log_to_file() + app = App(connector=PsycopgConnector(conninfo=get_settings().database_url)) TASKS: dict = {} SCHEDULES: dict[str, str] = {} diff --git a/backend/tests/test_parsers.py b/backend/tests/test_parsers.py index 3421dad..48005fa 100644 --- a/backend/tests/test_parsers.py +++ b/backend/tests/test_parsers.py @@ -158,3 +158,62 @@ def test_parlament_calendar_titles_are_html_decoded(): params={"since_year": 2026}) assert [e.title for e in events] == ["84. Sitzung des Nationalrates"] assert events[0].description == "Plenarsitzung NR" + + +def test_procurement_rejects_implausible_award_dates(): + from datetime import date + + from politrack.sources.procurement import plausible_date + + today = date(2026, 11, 13) + assert plausible_date("2025-02-17", today) == date(2025, 2, 17) + assert plausible_date("1900-12-30", today) is None # spreadsheet placeholder + assert plausible_date("2028-09-30", today) is None # the contract's end, not its award + assert plausible_date(None, today) is None + + +def test_procurement_placeholder_contractor_and_framework_buyer(): + from politrack.sources.procurement import buyer_name, org_ref + + assert org_ref({"name": "-"}, "company") is None + assert org_ref({"name": " n/a "}, "company") is None + assert org_ref({"name": "Swietelsky AG"}, "company").name == "Swietelsky AG" + sentence = ("Auftraggeber sind die Republik Österreich (Bund), die Bundesbeschaffung GmbH sowie alle weiteren " + "Auftraggeber gemäß der den Ausschreibungsunterlagen beiliegenden Liste") + assert buyer_name(sentence) == ("Bundesbeschaffung GmbH", sentence) + assert buyer_name("ASFINAG Alpenstraßen GmbH") == ("ASFINAG Alpenstraßen GmbH", None) + + +def test_source_freshness_alerts_after_two_missed_slots(): + from datetime import datetime, timezone + + from politrack.queries.admin import source_freshness + + class Conn: + def __init__(self, rows): + self.rows = rows + + def execute(self, *_): + return self.rows + + now = datetime(2026, 11, 13, 12, 0, tzinfo=timezone.utc) + daily = "15 3 * * *" + rows = [{"adapter": "fresh", "at": datetime(2026, 11, 13, 3, 40, tzinfo=timezone.utc)}, + {"adapter": "one_missed", "at": datetime(2026, 11, 12, 3, 40, tzinfo=timezone.utc)}, + {"adapter": "two_missed", "at": datetime(2026, 11, 11, 3, 40, tzinfo=timezone.utc)}] + result = {r["adapter"]: r["stale"] for r in source_freshness( + Conn(rows), {"fresh": daily, "one_missed": daily, "two_missed": daily, "never": daily}, now)} + assert result == {"fresh": False, "one_missed": False, "two_missed": True, "never": True} + + +def test_election_preferences_eu_layouts(): + from politrack.sources.elections import parse_preferences, split_name + + assert split_name("Babler, MSc, Andreas") == ("Babler", "Andreas") + assert split_name("Brunner, Dr. Magnus") == ("Brunner", "Dr. Magnus") + eu2019 = [("ÖVP", [(None, "Kandidat(in)", "BGLD", "Gesamt"), (1, "Karas, Dr. Othmar ", 3012, 103035)])] + eu2024 = [("ÖVP", [("LP", "Bewerber(in)", "BGLD", "Gesamt", None, "Darstellung", None, None), + (None, None, None, None, None, "(LP)", "Bewerber(in)", "Gesamt"), + (1, "Lopatka, Dr. Reinhold", 2268, 58013, None, 3, "Bernhuber, Alexander", 44641)])] + assert [(c["name"], c["votes"]) for c in parse_preferences("eu", eu2019)] == [("Dr. Othmar Karas", 103035)] + assert [(c["name"], c["votes"]) for c in parse_preferences("eu", eu2024)] == [("Dr. Reinhold Lopatka", 58013)] diff --git a/backup.sh b/backup.sh new file mode 100644 index 0000000..b37e781 --- /dev/null +++ b/backup.sh @@ -0,0 +1,33 @@ +#!/bin/sh +# Nightly database backup, run by the `backup` service in docker-compose.yml. +# Writes politrack-YYYY-MM-DD.dump (pg_dump custom format) to /backups at 02:30 UTC, deletes dumps older than +# $BACKUP_KEEP_DAYS days, and records each attempt as a run of "backup", so /api/v1/status/sources alerts +# when backups stop. Restore: pg_restore -h -U politrack -d politrack --clean +set -u +KEEP="${BACKUP_KEEP_DAYS:-14}" +export PGHOST=db PGUSER=politrack PGDATABASE=politrack + +record() { # record + psql -q -c "INSERT INTO run (adapter, mode, status, finished_at, stats) VALUES ('backup', 'backup', '$1', now(), '$2'::jsonb)" \ + || echo "could not record the backup run" +} + +while true; do + now=$(date -u +%s) + next=$(date -u -d "today 02:30" +%s) + [ "$next" -le "$now" ] && next=$(date -u -d "tomorrow 02:30" +%s) + echo "next backup at $(date -u -d "@$next")" + sleep $((next - now)) + + file="/backups/politrack-$(date -u +%F).dump" + if pg_dump -Fc -f "$file.tmp" && mv "$file.tmp" "$file"; then + size=$(stat -c %s "$file") + echo "backup ok: $file ($size bytes)" + record ok "{\"file\": \"$(basename "$file")\", \"bytes\": $size}" + else + rm -f "$file.tmp" + echo "backup FAILED" + record error '{}' + fi + find /backups -name 'politrack-*.dump' -mtime "+$KEEP" -delete +done diff --git a/docker-compose.yml b/docker-compose.yml index 376a3ef..d6ee993 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -9,9 +9,12 @@ x-backend: &backend environment: POLITRACK_DATABASE_URL: postgresql://politrack:${POSTGRES_PASSWORD}@db:5432/politrack POLITRACK_RAW_DIR: /data/raw + POLITRACK_LOG_DIR: /data/logs volumes: # Raw-response archive: lets `politrack reparse` rebuild data after parser fixes without refetching. - ${DATA_DIR:-./data}/raw:/data/raw + # Worker logs (.log, rotating), kept across redeploys. + - ${DATA_DIR:-./data}/logs:/data/logs depends_on: db: condition: service_healthy @@ -62,16 +65,34 @@ services: # another machine pointed at this database. worker-scrape: <<: *worker + hostname: worker-scrape # names the log file command: ["procrastinate", "--app=politrack.worker.app", "worker", "--queues=scrape,social", "--concurrency=2"] worker-bulk: <<: *worker + hostname: worker-bulk command: ["procrastinate", "--app=politrack.worker.app", "worker", "--queues=bulk", "--concurrency=1"] worker-enrich: <<: *worker + hostname: worker-enrich command: ["procrastinate", "--app=politrack.worker.app", "worker", "--queues=enrich", "--concurrency=1"] + # Nightly pg_dump at 02:30 UTC into ${BACKUP_DIR}, kept for ${BACKUP_KEEP_DAYS} days (see backup.sh). + backup: + image: pgvector/pgvector:pg16 + restart: unless-stopped + environment: + PGPASSWORD: ${POSTGRES_PASSWORD} + BACKUP_KEEP_DAYS: ${BACKUP_KEEP_DAYS:-14} + volumes: + - ./backup.sh:/backup.sh:ro + - ${BACKUP_DIR:-./data/backups}:/backups + entrypoint: ["sh", "/backup.sh"] + depends_on: + db: + condition: service_healthy + mcp: <<: *worker command: ["politrack", "mcp"] diff --git a/docs/deployment-unraid.md b/docs/deployment-unraid.md index c152259..936d86a 100644 --- a/docs/deployment-unraid.md +++ b/docs/deployment-unraid.md @@ -23,9 +23,9 @@ docker compose build docker compose up -d ``` - Docker creates `${DATA_DIR}/raw` as root, but the backend runs as a non-root user, and every fetch - then fails with "Permission denied". Hand it over once after the first `up`: - `chown -R "$(docker run --rm --entrypoint id politrack-backend -u):$(docker run --rm --entrypoint id politrack-backend -g)" "$DATA_DIR/raw"`. + Docker creates `${DATA_DIR}/raw` and `${DATA_DIR}/logs` as root, but the backend runs as a non-root user, and + every fetch then fails with "Permission denied". Hand them over once after the first `up`: + `chown -R "$(docker run --rm --entrypoint id politrack-backend -u):$(docker run --rm --entrypoint id politrack-backend -g)" "$DATA_DIR/raw" "$DATA_DIR/logs"`. `migrate` runs once (schema, seed data, source registry, job-queue tables), then `api`, `web`, `mcp` and three workers start: @@ -76,11 +76,11 @@ bge-m3 runs fine on CPU for this data volume. With an NVIDIA GPU, add a device r ## Backups -* Database: a nightly `pg_dump`, e.g. with the **User Scripts** plugin: - ```bash - docker exec politrack-db-1 pg_dump -U politrack -Fc politrack > /mnt/user/backups/politrack/$(date +%F).dump - find /mnt/user/backups/politrack -mtime +30 -delete - ``` +* Database: the `backup` service dumps it every night at 02:30 UTC to `BACKUP_DIR` (default `./data/backups`; + better a different share, e.g. `/mnt/user/backups/politrack`) and keeps `BACKUP_KEEP_DAYS` (14) days. Each dump is + recorded as a run of `backup`, so the status endpoint below alerts when backups stop. Restore with + `pg_restore -h localhost -p 5433 -U politrack -d politrack --clean `. For off-site copies, sync `BACKUP_DIR` + elsewhere. * Raw archive (`${DATA_DIR}/raw`): optional to back up – it saves refetching after parser fixes, and it can be pruned (`find raw -mtime +365 -delete`); missing blobs are simply refetched. @@ -96,7 +96,13 @@ bge-m3 runs fine on CPU for this data volume. With an NVIDIA GPU, add a device r * `GET /api/v1/admin/runs` with `Authorization: Bearer $POLITRACK_ADMIN_TOKEN`. * Fetch log: `SELECT adapter, status, count(*) FROM fetch_log WHERE fetched_at > now() - interval '1 day' GROUP BY 1, 2;` * Job queue: `SELECT task_name, status, count(*) FROM procrastinate_jobs GROUP BY 1, 2;` -* Uptime Kuma can poll `http://api:8000/healthz`. +* `GET /api/v1/status/sources` returns HTTP 503 as soon as a scheduled source (or the backup) missed its last two + slots, and lists every task with its last successful run: point an HTTP monitor (Uptime Kuma) at it. + `/healthz` only checks that the API reaches the database. +* The `watchdog` task (every 10 minutes) fails jobs of dead workers, closes abandoned runs and starts any scheduled + source whose last slot was missed (Procrastinate skips slots that are more than ten minutes overdue). +* Worker logs: `${DATA_DIR}/logs/worker-*.log` (rotating), kept across redeploys; `docker compose logs` only covers + the current containers. ## Troubleshooting diff --git a/docs/roadmap.md b/docs/roadmap.md index dda0dc7..019fcf2 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -34,7 +34,11 @@ Next.js frontend (de/en), Docker Compose for Unraid. - [x] X deletion sync (`social.x_deletions`): posts gone from X keep a marked record without their text - [x] EP roll-call votes of Austrian MEPs, including amendments and part votes (`europarl.votes`) - [x] Committee memberships of the configured legislative periods (list 250, `parlament.persons`) -- [x] Election results and preference votes: NR 2024, NR 2019, EU 2024 (`bmi.elections`) +- [x] Election results and preference votes: NR 2024, NR 2019, NR 2017, EU 2024, EU 2019 preference votes (`bmi.elections`) +- [x] Former Austrian MEPs of the 2019-2024 term; EC meetings of the 2014-2019 Commission; Bluesky/Mastodon history paging +- [x] Operations: watchdog for missed schedules and stalled jobs, nightly backups, `/api/v1/status/sources` for + monitoring, worker logs kept across redeploys; procurement data cleaned (award dates, placeholder contractors, + framework-agreement buyers) - [x] Draft legal pages (`/impressum`, `/datenschutz`), corrections contact, portrait proxy (`/api/img`) - [x] Pages: organisations, money explorer, social feed, Wikipedia edits, person network ("follow the money"), say-vs-vote on roll calls