Backups, freshness monitoring, persistent logs, procurement cleanup

- backup service: nightly pg_dump with retention, recorded as a run
- GET /api/v1/status/sources: 503 when a scheduled source or the backup
  missed its last two slots (for Uptime Kuma)
- worker logs to ${DATA_DIR}/logs, so they survive redeploys
- procurement: reject implausible award dates (1900, contract end dates),
  skip placeholder contractors, credit framework agreements to the BBG

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
simonandClaude Opus 5.5 committed 2026-11-13 00:46:18 +01:00
1 parent 79639e5319
commit 405c21a8c5
12 files changed
+251 -18

No files matched your search

+4
View File
@@ -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
+2
View File
@@ -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
+18
View File
@@ -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)
+10 -2
View File
@@ -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",
]
+23
View File
@@ -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 "
+36 -5
View File
@@ -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,
+25 -1
View File
@@ -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/<hostname>.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] = {}
+59
View File
@@ -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)]
+33
View File
@@ -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 <host> -U politrack -d politrack --clean <file>
set -u
KEEP="${BACKUP_KEEP_DAYS:-14}"
export PGHOST=db PGUSER=politrack PGDATABASE=politrack
record() { # record <status> <json stats>
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
+21
View File
@@ -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 (<service>.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"]
+15 -9
View File
@@ -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 <file>`. 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
+5 -1
View File
@@ -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