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>
83 lines
3.0 KiB
Python
83 lines
3.0 KiB
Python
import hashlib
|
|
import html
|
|
import re
|
|
import unicodedata
|
|
|
|
from bs4 import BeautifulSoup
|
|
from slugify import slugify
|
|
|
|
ACADEMIC_TITLES = re.compile(
|
|
r"\b(Mag\.?(\s?\(FH\))?|Dr\.?|DI|Dipl\.-Ing\.?|Ing\.?|MMag\.?|Prof\.?|Univ\.-Prof\.?|BA|MA|BSc|MSc|MBA|LL\.?M\.?|"
|
|
r"PhD|MAS|MPA|Bakk\.?|DDr\.?|KR|Abg\.?|Mmag\.?|MEd|BEd|Dipl\.-Päd\.?)(?=\s|,|$)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def normalize(text: str) -> str:
|
|
"""Lowercase, strip accents and collapse whitespace (matches SQL lower(unaccent()))."""
|
|
text = unicodedata.normalize("NFKD", text)
|
|
text = "".join(c for c in text if not unicodedata.combining(c))
|
|
return re.sub(r"\s+", " ", text).strip().lower()
|
|
|
|
|
|
def strip_titles(name: str) -> str:
|
|
name = ACADEMIC_TITLES.sub("", name)
|
|
name = re.sub(r"\s*,\s*", ", ", name)
|
|
return re.sub(r"\s+", " ", name).strip(" ,")
|
|
|
|
|
|
def make_slug(*parts: str) -> str:
|
|
return slugify(" ".join(p for p in parts if p), lowercase=True, replacements=[["ß", "ss"]])
|
|
|
|
|
|
def html_to_text(fragment: str | None) -> str:
|
|
if not fragment:
|
|
return ""
|
|
return re.sub(r"\s+", " ", BeautifulSoup(html.unescape(fragment), "html.parser").get_text(" ")).strip()
|
|
|
|
|
|
def fingerprint(*parts: object) -> str:
|
|
raw = "|".join(str(p) for p in parts)
|
|
return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:40]
|
|
|
|
|
|
LEGAL_FORMS = re.compile(
|
|
r"\b(gesellschaft\s+m\.?\s?b\.?\s?h\.?|ges\.?\s?m\.?\s?b\.?\s?h\.?|gmbh|g\.m\.b\.h\.?|aktiengesellschaft|ag|"
|
|
r"kommanditgesellschaft|kg|og|e\.?\s?u\.?|eg|e\.?\s?gen\.?|reg\.?\s?gen\.?\s?m\.?\s?b\.?\s?h\.?|se|ltd\.?|inc\.?|"
|
|
r"llc|s\.?a\.?|b\.?v\.?|n\.?v\.?|& co\.?|co\.?|und co\.?|gesmbh|mbh|kgaa|plc)\b",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def normalize_org_name(name: str) -> str:
|
|
"""'HYPO TIROL BANK AG' == 'Hypo Tirol Bank Aktiengesellschaft' -> 'hypo tirol bank'.
|
|
|
|
Lowercase, strip accents, quotes, punctuation and legal-form suffixes, so the same company is
|
|
recognised across sources (side-job lists, procurement notices, donation lists, ad reports).
|
|
"""
|
|
text = normalize(name)
|
|
text = re.sub(r"[\"'„“”»«]", "", text)
|
|
text = LEGAL_FORMS.sub(" ", text)
|
|
text = re.sub(r"[^\w\s-]", " ", text)
|
|
return re.sub(r"\s+", " ", text).strip(" -")
|
|
|
|
|
|
def firmenbuch_number(value: str | None) -> str | None:
|
|
"""'FN 306527 a', 'FN306527a', '306527a' -> '306527a' (Austrian company register number, normalised)."""
|
|
m = re.fullmatch(r"\s*(?:FN\s*)?(\d{1,6})\s?([a-zA-Z])\s*", value or "", re.IGNORECASE)
|
|
return f"{m.group(1)}{m.group(2).lower()}" if m else None
|
|
|
|
|
|
def format_eur_de(amount: object) -> str:
|
|
"""1234567.8 -> '1.234.567,80' (German number format; formats only the number, never the sentence)."""
|
|
try:
|
|
value = float(amount) # type: ignore[arg-type]
|
|
except (TypeError, ValueError):
|
|
return str(amount)
|
|
return f"{value:,.2f}".replace(",", "X").replace(".", ",").replace("X", ".")
|
|
|
|
|
|
def excerpt(text: str, limit: int = 400) -> str:
|
|
text = re.sub(r"\s+", " ", text).strip()
|
|
return text if len(text) <= limit else text[: limit - 1].rsplit(" ", 1)[0] + "…"
|