AIPlayground/politics-tracking
3
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Backfill: former MEPs, older elections, EC 2014-2019, social history paging

Browse Source 
- europarl.meps also loads the Austrian MEPs of the 2019-2024 term; roles end
  with the last membership, so former MEPs don't show as current
- bmi.elections: National Council 2017 and EU 2019 preference votes
- eu.commission_meetings: the 2014-2019 Commission
- social.bluesky / social.mastodon: -p backfill_pages=N pages older posts

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
simonandClaude Opus 5.5 committed 2026-11-13 00:34:24 +01:00
1 parent 4bd4e5b2db
commit 79639e5319
3 files changed
+49 -16
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/politrack/sources/elections.py
+15 -4
View File
@@ -41,7 +41,7 @@ class Election:
held_on: date
body: str # slug of the elected body
path: str # folder on bmi.gv.at
result: str
result: str | None # None: only in a format we can't read (EU 2019: legacy .xls)
preferences: tuple[tuple[str, str], ...] # (layout, file): layouts "bund", "land", "eu"
@@ -56,6 +56,12 @@ ELECTIONS = (
Election("euw2024", "Europawahl 2024", "2024 European Parliament election", date(2024, 6, 9),
"europaeisches-parlament", "europawahlen/europawahl_2024/files", "endgueltiges_ergebnis_europawahl_26062024.xlsx",
(("eu", "gesamtaufstellung_erzielte_vorzugsstimmen_eu_wahl_2024_parteien.xlsx"),)),
Election("nrw2017", "Nationalratswahl 2017", "2017 National Council election", date(2017, 10, 15), "nationalrat",
"nationalratswahlen/nationalratswahl_2017/files", "nrw17_endgueltiges_gesamtergebnis.xlsx",
(("bund", "nrw17_vorzugsstimmen_bundesweit.xlsx"),)),
Election("euw2019", "Europawahl 2019", "2019 European Parliament election", date(2019, 5, 26),
"europaeisches-parlament", "europawahlen/europawahl_2019/files/endgueltiges_ergebnis", None,
(("eu", "euw_2019_vorzugsstimmen_gesamthomepage.xlsx"),)),
)
BY_KEY = {e.key: e for e in ELECTIONS}
@@ -130,8 +136,12 @@ def parse_preferences(layout: str, workbook: list[tuple[str, list[tuple]]]) -> l
if not cells:
continue
first = row[0]
if layout == "eu" and first == "LP":
total_col = next((i for i, c in enumerate(row) if c == "Gesamt"), None)
if layout == "eu" and any(isinstance(c, str) and re.match(r"Bewerber|Kandidat", c) for c in row):
# header: "LP | Bewerber(in) | BGLD ... | Gesamt" (2024) or "_ | Kandidat(in) | ... | Gesamt" (2019).
# 2024 repeats a re-sorted copy of the table further right: its header must not win.
if total_col is None:
total_col = next((i for i, c in enumerate(row) if isinstance(c, str) and c.strip() == "Gesamt"),
None)
continue
if isinstance(first, str) and len(cells) == 1:
if (m := LIST_HEAD.match(first)):
@@ -173,7 +183,8 @@ class ElectionResults(Adapter):
if election.key not in wanted:
continue
base = f"{BMI}/{election.path}"
yield Request(url=f"{base}/{election.result}", kind="result", meta={"election": election.key})
if election.result:
yield Request(url=f"{base}/{election.result}", kind="result", meta={"election": election.key})
for layout, file in election.preferences:
yield Request(url=f"{base}/{file}", kind="preferences", meta={"election": election.key, "layout": layout})
backend/politrack/sources/eu.py
+17 -7
View File
@@ -42,7 +42,9 @@ EP_SPEC = SourceSpec(
tos_notes="EP Open Data API v2 (limit 500 requests / 5 min / endpoint).",
)
EC_EXPORT = "https://ec.europa.eu/transparency-initiative/meetings/data/meetings/dataxlsx"
EC_TERMS = ("meetingscommissionrepresentatives2429", "meetingscommissionrepresentatives1924")
EC_TERMS = ("meetingscommissionrepresentatives2429", "meetingscommissionrepresentatives1924",
"meetingscommissionrepresentatives1419")
EP_PAST_TERMS = (9,) # Austrian MEPs of earlier terms: their votes and declarations, roles with end dates
EP_API = "https://data.europarl.europa.eu/api/v2"
# Austrian Commissioners in the terms covered by EC_TERMS (fallback if the DB has no Commission roles yet).
AUSTRIAN_COMMISSIONERS = ("Johannes Hahn", "Magnus Brunner")
@@ -123,10 +125,15 @@ class EuropeanParliament(Adapter):
description = "Current Austrian MEPs (mandates) and their declarations"
schedule = "30 4 * * 2"
creates_persons = True
parser_version = 2 # roles of former MEPs end with their last membership
def start(self, ctx: RunContext) -> Iterator[Request]:
yield Request(url=f"{EP_API}/meps/show-current", kind="meps",
params={"country-of-representation": "AT", "format": "application/ld+json", "limit": 100})
for term in EP_PAST_TERMS:
yield Request(url=f"{EP_API}/meps", kind="meps",
params={"parliamentary-term": term, "country-of-representation": "AT",
"format": "application/ld+json", "limit": 100})
def parse(self, ctx: RunContext, response: Response) -> Iterator[Any]:
if not response.body.strip(): # the API answers some MEPs' declaration requests with an empty body
@@ -157,19 +164,22 @@ class EuropeanParliament(Adapter):
if mep.get("img"):
yield PersonRecord(ref=ref, full_name=meta["name"], photo_url=mep["img"],
birth_date=date.fromisoformat(mep["bday"][:10]) if mep.get("bday") else None)
starts = [m.get("memberDuring", {}).get("startDate") for m in mep.get("hasMembership") or []
if "EU_PARLIAMENT" in str(m.get("membershipClassification", ""))]
starts = sorted(s for s in starts if s)
periods = sorted(((m.get("memberDuring") or {}).get("startDate"), (m.get("memberDuring") or {}).get("endDate"))
for m in mep.get("hasMembership") or []
if "EU_PARLIAMENT" in str(m.get("membershipClassification", ""))
and (m.get("memberDuring") or {}).get("startDate"))
body = OrgRef(slug="europaeisches-parlament")
start = date.fromisoformat(starts[-1]) if starts else None
start = date.fromisoformat(periods[-1][0][:10]) if periods else None
# former MEPs: the latest membership has ended; current ones have an open end
end = date.fromisoformat(periods[-1][1][:10]) if periods and periods[-1][1] else None
url = f"https://www.europarl.europa.eu/meps/de/{meta['mep_id']}"
yield Claim(
claim_type="role", person=ref, org=body, valid_from=start,
claim_type="role", person=ref, org=body, valid_from=start, valid_to=end,
summary_de=f"Mitglied des Europäischen Parlaments (Fraktion {meta.get('group') or '?'})",
summary_en=f"Member of the European Parliament ({meta.get('group') or '?'})",
data={"ep_group": meta.get("group")},
evidence=[Evidence(url=url, doc_type="api_record", title=f"Europäisches Parlament: {meta['name']}")],
facet=Role(role="mp", level="eu", body=body, start=start),
facet=Role(role="mp", level="eu", body=body, start=start, end=end),
)
def parse_declarations(self, response: Response, docs: list[dict]) -> Iterator[Claim]:
backend/politrack/sources/social.py
+17 -5
View File
@@ -99,16 +99,22 @@ class BlueskyPosts(Adapter):
queue = "social"
def start(self, ctx: RunContext) -> Iterator[Request]:
pages = int(ctx.params.get("backfill_pages", 0)) # older pages per account (history backfill)
for acc in accounts(ctx, "bluesky", r"bsky\.app/profile/"):
m = re.search(r"bsky\.app/profile/([^/?#]+)", acc["url"])
if m:
yield Request(url="https://public.api.bsky.app/xrpc/app.bsky.feed.getAuthorFeed",
params={"actor": m.group(1), "limit": 50, "filter": "posts_no_replies"}, kind="feed",
meta={**acc["owner"], "handle": m.group(1)})
meta={**acc["owner"], "handle": m.group(1), **({"pages_left": pages} if pages else {})})
def parse(self, ctx: RunContext, response: Response) -> Iterator[Claim]:
def parse(self, ctx: RunContext, response: Response) -> Iterator[Any]:
handle = response.meta["handle"]
for item in response.json().get("feed", []):
data = response.json()
if response.meta.get("pages_left") and data.get("cursor") and data.get("feed"):
yield Request(url=response.request.url, kind="feed",
params={**(response.request.params or {}), "cursor": data["cursor"]},
meta={**response.meta, "pages_left": response.meta["pages_left"] - 1})
for item in data.get("feed", []):
post = item.get("post") or {}
if item.get("reason"): # repost of someone else's post
continue
@@ -131,12 +137,13 @@ class MastodonPosts(Adapter):
queue = "social"
def start(self, ctx: RunContext) -> Iterator[Request]:
pages = int(ctx.params.get("backfill_pages", 0)) # older pages per account (history backfill)
for acc in accounts(ctx, "mastodon"): # instances have arbitrary hosts: platform only
m = re.match(r"^https?://([^/]+)/@([^/@]+)", acc["url"])
if m:
host, user = m.groups()
yield Request(url=f"https://{host}/api/v1/accounts/lookup", params={"acct": user}, kind="lookup",
meta={**acc["owner"], "host": host})
meta={**acc["owner"], "host": host, **({"pages_left": pages} if pages else {})})
def parse(self, ctx: RunContext, response: Response) -> Iterator[Any]:
meta = response.meta
@@ -147,7 +154,12 @@ class MastodonPosts(Adapter):
params={"limit": 40, "exclude_replies": "true", "exclude_reblogs": "true"},
kind="statuses", meta=meta)
return
for status in response.json():
statuses = response.json()
if meta.get("pages_left") and statuses:
yield Request(url=response.request.url, kind="statuses",
params={**(response.request.params or {}), "max_id": statuses[-1]["id"]},
meta={**meta, "pages_left": meta["pages_left"] - 1})
for status in statuses:
if status.get("visibility") not in (None, "public"):
continue
yield post_claim(owner=meta, platform="mastodon", url=status.get("url") or status.get("uri"),