simonandClaude Opus 5.5
405c21a8c5
- 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>
93 lines
4.0 KiB
Python
93 lines
4.0 KiB
Python
"""Sources, topics, statistics, moderation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from psycopg import Connection
|
|
|
|
|
|
def list_sources(conn: Connection) -> list[dict]:
|
|
return conn.execute(
|
|
"""
|
|
SELECT s.slug, s.name, s.kind, s.homepage, s.license, s.access_method, s.default_trust, s.tos_notes,
|
|
(SELECT count(*) FROM document d WHERE d.source_id = s.id) AS documents
|
|
FROM source s ORDER BY s.kind, s.default_trust DESC, s.name
|
|
"""
|
|
).fetchall()
|
|
|
|
|
|
def list_topics(conn: Connection) -> list[dict]:
|
|
return conn.execute("SELECT slug, name_de, name_en FROM topic ORDER BY name_de").fetchall()
|
|
|
|
|
|
def pending_claims(conn: Connection, limit: int = 50) -> list[dict]:
|
|
return conn.execute(
|
|
"""
|
|
SELECT c.id, c.claim_type, c.summary_de, c.occurred_at, p.full_name,
|
|
(SELECT jsonb_agg(jsonb_build_object('url', d.url, 'title', d.title))
|
|
FROM claim_evidence e JOIN document d ON d.id = e.document_id WHERE e.claim_id = c.id) AS evidence
|
|
FROM claim c LEFT JOIN person p ON p.id = c.person_id
|
|
WHERE c.review_status = 'pending' AND c.merged_into IS NULL
|
|
ORDER BY c.occurred_at DESC NULLS LAST LIMIT %s
|
|
""",
|
|
(limit,),
|
|
).fetchall()
|
|
|
|
|
|
def set_review(conn: Connection, claim_id: int, decision: str, reviewer: str) -> None:
|
|
if decision not in ("published", "rejected"):
|
|
raise ValueError("decision must be 'published' or 'rejected'")
|
|
conn.execute(
|
|
"UPDATE claim SET review_status = %s, reviewed_by = %s, reviewed_at = now() WHERE id = %s",
|
|
(decision, reviewer, claim_id),
|
|
)
|
|
|
|
|
|
def stats(conn: Connection) -> dict:
|
|
return conn.execute(
|
|
"""
|
|
SELECT (SELECT count(*) FROM person) AS persons,
|
|
(SELECT count(*) FROM person_status WHERE status = 'active') AS active_persons,
|
|
(SELECT count(*) FROM organization WHERE kind = 'party') AS parties,
|
|
(SELECT count(*) FROM claim WHERE merged_into IS NULL AND review_status = 'published') AS claims,
|
|
(SELECT count(*) FROM claim WHERE merged_into IS NULL AND status = 'corroborated') AS corroborated,
|
|
(SELECT count(*) FROM claim WHERE review_status = 'pending') AS pending_review,
|
|
(SELECT count(*) FROM document) AS documents,
|
|
(SELECT count(*) FROM vote_event) AS vote_events,
|
|
(SELECT count(*) FROM vote_event WHERE roll_call) AS roll_call_votes,
|
|
(SELECT count(*) FROM event) AS events,
|
|
(SELECT count(*) FROM money_flow) AS money_flows,
|
|
(SELECT max(finished_at) FROM run WHERE status IN ('ok', 'partial')) AS last_update
|
|
"""
|
|
).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 "
|
|
"ORDER BY id DESC LIMIT %s",
|
|
(limit,),
|
|
).fetchall()
|