simonandClaude Opus 5.5
583ccba31c
Initial import of Politrack, a self-hosted service that tracks Austrian politicians from federal to municipal level - roles, votes, money flows, statements, meetings, social media and legal proceedings - with every claim linked to its sources. - backend: FastAPI + Postgres/pgvector; adapter framework (fetch -> parse -> load -> enrich) with robots.txt/TDM-aware fetching, a raw-response archive for reparsing, and a claim/evidence model with corroboration - sources: Parliament, Wikidata, EU Parliament/Commission, Council of Ministers, Rechnungshof party reports and donations, procurement, RTR media transparency, political ad libraries, WKStA, news feeds, Wikipedia edits, Bluesky, Mastodon, public Telegram channels, YouTube and X (official API) - web: Next.js frontend in German and English with typed API client - mcp: MCP server for AI agents and RAG - docs: architecture, data model, sources, legal notes, Unraid deployment and contributor guides; Docker Compose setup for Unraid Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
75 lines
4.2 KiB
Markdown
75 lines
4.2 KiB
Markdown
# Writing an enricher
|
||
|
||
An **enricher** derives new facts from data that is already in the database: linking headlines to
|
||
people, detecting trips and meetings, classifying legal matters, tagging topics, corroborating claims,
|
||
computing embeddings. Adapters *collect*; enrichers *connect*.
|
||
|
||
```python
|
||
from collections.abc import Iterator
|
||
|
||
from politrack.core import Enricher, RunContext, register
|
||
from politrack.core.records import Claim, Evidence, PersonRef
|
||
from politrack.core.spec import DERIVED
|
||
|
||
|
||
@register
|
||
class AmbassadorAppointments(Enricher):
|
||
spec = DERIVED # derived claims are attributed to "politrack" + the original evidence
|
||
name = "enrich.ambassadors"
|
||
description = "Ambassador appointments mentioned in Council of Ministers minutes"
|
||
schedule = "20 6 * * *" # cron (UTC); runs on the `enrich` queue
|
||
|
||
def run(self, ctx: RunContext) -> Iterator[Claim]:
|
||
rows = ctx.conn.execute(
|
||
"""SELECT d.id, d.url, d.title, d.published_at, d.full_text FROM document d
|
||
JOIN source s ON s.id = d.source_id
|
||
WHERE s.slug = 'bundesregierung' AND d.full_text ILIKE '%%Botschafter%%'
|
||
AND NOT (d.meta ? 'ambassadors_at')"""
|
||
).fetchall()
|
||
for doc in rows:
|
||
for name in find_appointees(doc["full_text"]): # a pure, unit-tested helper
|
||
yield Claim(
|
||
claim_type="career", person=PersonRef(name=name),
|
||
summary_de=f"{name} als Botschafter:in vorgeschlagen",
|
||
occurred_at=doc["published_at"],
|
||
evidence=[Evidence(url=doc["url"], title=doc["title"], source="bundesregierung",
|
||
quote=..., doc_type="protocol")],
|
||
)
|
||
ctx.conn.execute("UPDATE document SET meta = meta || jsonb_build_object('ambassadors_at', now()) "
|
||
"WHERE id = %s", (doc["id"],))
|
||
ctx.stats["documents"] += 1
|
||
```
|
||
|
||
## Rules
|
||
|
||
* **Yield records, don't write facts yourself.** Records go through the same Loader as adapter output:
|
||
entity resolution, claim identity, evidence, corroboration status. Direct SQL is fine for *housekeeping*
|
||
(marking documents as processed, maintenance like closing roles at death – see `enrich/maintenance.py`).
|
||
* **Cite the original.** Evidence of a derived claim points at the original document and names its
|
||
`source` slug, so the user's source filter and the corroboration logic work on the real publisher.
|
||
* **Be incremental.** Remember what you processed (a marker in `document.meta`, or `ctx.state`), so a run
|
||
costs proportional to new data.
|
||
* **Be conservative with people.** Enrichers never create persons (`creates_persons = False`); unknown
|
||
names are skipped and counted (`unresolved_person`). Prefer precision over recall – a missed link is
|
||
better than a wrong one.
|
||
* **Sensitive output needs review.** Anything about legal matters: `sensitive=True, review=True`.
|
||
* **Put logic in pure helpers** (regexes, classifiers) and unit-test them in `tests/test_extraction.py`.
|
||
|
||
The runner commits every 200 records, so a crash loses little work.
|
||
|
||
## Existing enrichers
|
||
|
||
| name | what it does |
|
||
|---|---|
|
||
| `enrich.linker` | finds politicians (aliases, "Bundeskanzler Stocker" office titles) in headlines/teasers → `media_mention` claims |
|
||
| `enrich.events` | trips, summits, bilateral meetings in official news → `Event` + participation, `travel`/`meeting` claims |
|
||
| `enrich.legal` | investigation/indictment/verdict vocabulary in headlines → `legal_proceeding` claims (review queue) |
|
||
| `enrich.topics` | keyword topics (mirrors Parliament's subject list) → `topic_assignment` |
|
||
| `enrich.roles` | gives open-ended offices the holder's party; closes roles at death |
|
||
| `enrich.rollcall` | faction majority positions derived from individual roll-call votes |
|
||
| `enrich.corroborate` | merges near-duplicate claims, recomputes reported/corroborated/disputed |
|
||
| `enrich.embed` | chunks + embeddings for semantic search/RAG (needs Ollama) |
|
||
| `config.social_accounts` | imports `config/social_accounts.yaml` (curated accounts, party channels) |
|
||
|
||
Run them all with `politrack enrich`, or one with `politrack run enrich.linker`.
|