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>
286 lines
15 KiB
Python
286 lines
15 KiB
Python
"""Public procurement: contract award notices published as open "core data" (Kerndaten, §62/64 BVergG 2018).
|
||
|
||
Every contracting authority (or the platform it uses) publishes a "Kerndatenquelle" (KDQ): an XML index
|
||
<kdq><item id lastmod><url>…</url></item>…</kdq> listed as a dataset on data.gv.at. Each item is one
|
||
notice; since 2023/24 notices use the EU eForms UBL schema. We keep ContractAwardNotice documents
|
||
(who bought what from whom, for how much) and skip calls for tenders.
|
||
|
||
Correlations this enables: suppliers that are also party donors, sponsors or side-job employers of
|
||
politicians; contracts by ministries led by a given minister (organisation pages).
|
||
|
||
Volume: tens of thousands of notices; first run backfills from POLITRACK_PROCUREMENT_SINCE, later runs only
|
||
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
|
||
from typing import Any
|
||
|
||
from politrack.config import get_settings
|
||
from politrack.core import Adapter, Request, Response, RunContext, SourceSpec, register
|
||
from politrack.core.records import Claim, Evidence, Money, OrgRef
|
||
from politrack.text import firmenbuch_number, format_eur_de
|
||
|
||
HUB = "https://www.data.gv.at/api/hub/search/search"
|
||
SPEC = SourceSpec(
|
||
slug="vergabe", name="Öffentliche Aufträge (Kerndaten, data.gv.at)", kind="official",
|
||
homepage="https://www.data.gv.at", license="Open data (free licence required by §64 BVergG 2018)", access="bulk",
|
||
trust=90, min_interval=1.0,
|
||
tos_notes="Contracting authorities must publish award core data as open, machine-readable data under a free "
|
||
"licence (BVergG 2018). We read each publisher's Kerndatenquelle (KDQ) XML and its eForms notices.",
|
||
)
|
||
NS = {
|
||
"cbc": "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2",
|
||
"cac": "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2",
|
||
"efac": "http://data.europa.eu/p27/eforms-ubl-extension-aggregate-components/1",
|
||
"efbc": "http://data.europa.eu/p27/eforms-ubl-extension-basic-components/1",
|
||
"efext": "http://data.europa.eu/p27/eforms-ubl-extensions/1",
|
||
"ext": "urn:oasis:names:specification:ubl:schema:xsd:CommonExtensionComponents-2",
|
||
}
|
||
CATALOG_QUERIES = ("Kerndaten", "Kerndatenquelle", "KDQ BVergG")
|
||
|
||
|
||
def local(tag: str) -> str:
|
||
return tag.rsplit("}", 1)[-1]
|
||
|
||
|
||
def text(node: ET.Element | None, path: str) -> str | None:
|
||
if node is None:
|
||
return None
|
||
found = node.find(path, NS)
|
||
return found.text.strip() if found is not None and found.text else None
|
||
|
||
|
||
def parse_kerndaten(root: ET.Element) -> dict[str, Any] | None:
|
||
"""Austrian core-data format (BVergG 2018, schema brz.gv.at/eproc/kd): KD_8_2_* = awarded contracts.
|
||
|
||
Still used by many publishers (e-shops, below-threshold awards). Contractor NATIONALID is usually the
|
||
company register number ("183286p"), occasionally the VAT id.
|
||
"""
|
||
if not local(root.tag).startswith("KD_8_2"):
|
||
return None
|
||
|
||
def t(node: ET.Element | None, path: str) -> str | None:
|
||
found = node.find(path) if node is not None else None
|
||
return found.text.strip() if found is not None and found.text else None
|
||
|
||
buyer = root.find("{*}CONTRACTING_BODY/{*}ADDRESS_CONTRACTING_BODY")
|
||
title = " ".join(p.text.strip() for p in root.iterfind("{*}OBJECT_CONTRACT/{*}TITLE/{*}P") if p.text)
|
||
cpv = root.find("{*}OBJECT_CONTRACT/{*}CPV_MAIN/{*}CPV_CODE")
|
||
contracts = []
|
||
for awarded in root.iterfind(".//{*}AWARD_CONTRACT/{*}AWARDED_CONTRACT"):
|
||
contracts.append({
|
||
"id": None,
|
||
"title": None,
|
||
"awarded": t(awarded, "{*}DATE_CONCLUSION_CONTRACT"),
|
||
"amount": t(awarded, "{*}VAL_TOTAL"),
|
||
"winners": [{"name": t(c, "{*}ADDRESS_CONTRACTOR/{*}OFFICIALNAME"),
|
||
"company_id": t(c, "{*}ADDRESS_CONTRACTOR/{*}NATIONALID")}
|
||
for c in awarded.iterfind("{*}CONTRACTOR")],
|
||
})
|
||
return {
|
||
"notice_id": t(root, "{*}EXTENSION[@KEY='Referenz']") or t(root, "{*}OBJECT_CONTRACT/{*}REFERENCE_NUMBER"),
|
||
"issued": None,
|
||
"title": title or None,
|
||
"cpv": cpv.get("CODE") if cpv is not None else None,
|
||
"total": None,
|
||
"buyers": [{"name": t(buyer, "{*}OFFICIALNAME"), "company_id": None}] if buyer is not None else [],
|
||
"contracts": contracts,
|
||
}
|
||
|
||
|
||
def parse_award_notice(xml: bytes) -> dict[str, Any] | None:
|
||
"""Extract buyer, winners, amounts, title, CPV and dates from an award notice: eForms
|
||
ContractAwardNotice, or the Austrian core-data format (parse_kerndaten)."""
|
||
root = ET.fromstring(xml)
|
||
if local(root.tag).startswith("KD_"):
|
||
return parse_kerndaten(root)
|
||
if local(root.tag) != "ContractAwardNotice":
|
||
return None
|
||
orgs: dict[str, dict[str, Any]] = {}
|
||
for org in root.iterfind(".//efac:Organizations/efac:Organization/efac:Company", NS):
|
||
org_id = text(org, "cac:PartyIdentification/cbc:ID")
|
||
if org_id:
|
||
orgs[org_id] = {"name": text(org, "cac:PartyName/cbc:Name"),
|
||
"company_id": text(org, "cac:PartyLegalEntity/cbc:CompanyID"),
|
||
"city": text(org, "cac:PostalAddress/cbc:CityName")}
|
||
buyer_ids = [e.text for e in root.iterfind("cac:ContractingParty/cac:Party/cac:PartyIdentification/cbc:ID", NS)
|
||
if e.text]
|
||
result = root.find(".//efac:NoticeResult", NS)
|
||
tender_amounts = {text(t, "cbc:ID"): text(t, "cac:LegalMonetaryTotal/cbc:PayableAmount")
|
||
for t in (result.iterfind("efac:LotTender", NS) if result is not None else [])}
|
||
tendering_parties = {text(tp, "cbc:ID"): [e.text for e in tp.iterfind("efac:Tenderer/cbc:ID", NS) if e.text]
|
||
for tp in (result.iterfind("efac:TenderingParty", NS) if result is not None else [])}
|
||
lot_tender_party = {text(t, "cbc:ID"): text(t, "efac:TenderingParty/cbc:ID")
|
||
for t in (result.iterfind("efac:LotTender", NS) if result is not None else [])}
|
||
contracts = []
|
||
for contract in (result.iterfind("efac:SettledContract", NS) if result is not None else []):
|
||
tender_id = text(contract, "efac:LotTender/cbc:ID")
|
||
winners = tendering_parties.get(lot_tender_party.get(tender_id), [])
|
||
contracts.append({
|
||
"id": text(contract, "cbc:ID"),
|
||
"title": text(contract, "cbc:Title"),
|
||
"awarded": text(contract, "cbc:AwardDate") or text(contract, "cbc:IssueDate"),
|
||
"amount": tender_amounts.get(tender_id),
|
||
"winners": [orgs.get(w, {"name": None}) | {"ref": w} for w in winners],
|
||
})
|
||
project_title = text(root, "cac:ProcurementProject/cbc:Name")
|
||
return {
|
||
"notice_id": text(root, "cbc:ID"),
|
||
"issued": text(root, "cbc:IssueDate"),
|
||
"title": project_title,
|
||
"cpv": text(root, "cac:ProcurementProject/cac:MainCommodityClassification/cbc:ItemClassificationCode"),
|
||
"total": text(root, ".//efac:NoticeResult/cbc:TotalAmount"),
|
||
"buyers": [orgs.get(b, {"name": None}) | {"ref": b} for b in buyer_ids],
|
||
"contracts": contracts,
|
||
}
|
||
|
||
|
||
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") or PLACEHOLDER_NAME.match(org["name"].strip()):
|
||
return None
|
||
company_id = (org.get("company_id") or "").replace(" ", "")
|
||
ids = {}
|
||
if company_id.upper().startswith("ATU"):
|
||
ids["uid"] = company_id.upper()
|
||
elif fn := firmenbuch_number(company_id):
|
||
ids["firmenbuch"] = fn
|
||
return OrgRef(name=org["name"], kind=kind, ids=ids)
|
||
|
||
|
||
def as_date(value: str | None) -> date | None:
|
||
if not value:
|
||
return None
|
||
try:
|
||
return date.fromisoformat(value[:10])
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
@register
|
||
class PublicContracts(Adapter):
|
||
spec = SPEC
|
||
name = "vergabe.awards"
|
||
description = "Contract awards of Austrian public bodies (eForms and BVergG core data via data.gv.at)"
|
||
schedule = "0 2 * * *"
|
||
queue = "bulk"
|
||
parser_version = 3 # 3: implausible award dates, placeholder contractors, framework-agreement buyers
|
||
|
||
def start(self, ctx: RunContext) -> Iterator[Request]:
|
||
for query in CATALOG_QUERIES:
|
||
yield Request(url=HUB, params={"q": query, "limit": 100, "page": 0}, kind="catalog", conditional=False)
|
||
|
||
def parse(self, ctx: RunContext, response: Response) -> Iterator[Any]:
|
||
kind = response.request.kind
|
||
if kind == "catalog":
|
||
result = response.json().get("result") or {}
|
||
for dataset in result.get("results") or []:
|
||
for dist in dataset.get("distributions") or []:
|
||
url = (dist.get("access_url") or dist.get("download_url") or [None])
|
||
url = url[0] if isinstance(url, list) else url
|
||
if url and url.startswith("http") and "data.gv.at/katalog" not in url:
|
||
if ctx.params.get("only_url") and ctx.params["only_url"] not in url:
|
||
continue # targeted runs: -p only_url=bbg.gv.at
|
||
title = (dataset.get("title") or {}).get("de") or ""
|
||
yield Request(url=url, kind="kdq", meta={"publisher": title.split("von", 1)[-1].strip(" '")})
|
||
if result.get("count", 0) > (response.request.params["page"] + 1) * 100:
|
||
yield Request(url=HUB, params={**response.request.params, "page": response.request.params["page"] + 1},
|
||
kind="catalog", conditional=False)
|
||
elif kind == "kdq":
|
||
yield from self.parse_kdq(ctx, response)
|
||
elif kind == "notice":
|
||
yield from self.parse_notice(ctx, response)
|
||
|
||
def parse_kdq(self, ctx: RunContext, response: Response) -> Iterator[Request]:
|
||
try:
|
||
root = ET.fromstring(response.body)
|
||
except ET.ParseError:
|
||
ctx.stats["kdq_not_xml"] += 1
|
||
return
|
||
state_key = f"kdq:{response.request.url}"
|
||
since = ctx.params.get("since") or ctx.state.get(state_key) or get_settings().procurement_since
|
||
limit = int(ctx.params.get("max_notices_per_source", 3000))
|
||
items = [(item.get("lastmod") or "", item.findtext("{*}url") or item.findtext("url"))
|
||
for item in root.iter() if local(item.tag) == "item"]
|
||
# Oldest first so an interrupted backfill resumes where it stopped (cursor = newest lastmod handled).
|
||
items = sorted(i for i in items if i[1] and i[0] > since)[:limit]
|
||
for lastmod, url in items:
|
||
# lastmod in meta (not in the URL) + conditional requests: unchanged notices cost a 304 at most.
|
||
# The cursor advances only when a notice has been parsed (see advance_cursor), so an interrupted
|
||
# run resumes with the first notice it did not get to.
|
||
yield Request(url=url.strip(), kind="notice", meta={"publisher": response.meta.get("publisher"),
|
||
"lastmod": lastmod, "state_key": state_key})
|
||
|
||
@staticmethod
|
||
def advance_cursor(ctx: RunContext, response: Response) -> None:
|
||
key, lastmod = response.meta.get("state_key"), response.meta.get("lastmod")
|
||
if key and lastmod and lastmod > (ctx.state.get(key) or ""):
|
||
ctx.state.set(key, lastmod)
|
||
|
||
def parse_notice(self, ctx: RunContext, response: Response) -> Iterator[Claim]:
|
||
self.advance_cursor(ctx, response) # notices arrive oldest-first per publisher
|
||
try:
|
||
notice = parse_award_notice(response.body)
|
||
except ET.ParseError:
|
||
ctx.stats["notice_not_xml"] += 1
|
||
return
|
||
if notice is None:
|
||
ctx.stats["not_an_award"] += 1
|
||
return
|
||
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 = 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"),
|
||
**({"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,
|
||
payee_name=supplier.name, paid_on=awarded,
|
||
reference=f"{notice['notice_id']}/{contract['id']}", category=notice["cpv"]),
|
||
)
|