feat: add weekly source contract foundation

src/political_event_tracking_research/weekly_contract.py

@@ -0,0 +1,167 @@
+"""Pure producer-owned contract for complete official PERT weekly inputs."""
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+import unicodedata
+from collections.abc import Mapping
+from dataclasses import dataclass
+from datetime import date, datetime, timedelta, timezone
+from pathlib import PurePosixPath
+
+SCHEMA_VERSION = "1"
+CONTRACT_VERSION = "political_event_weekly.v1"
+CADENCE = "weekly"
+MAX_SAFE_JSON_INTEGER = 2**53 - 1
+_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
+_TIMESTAMP_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?Z$")
+_SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
+_SHA1_RE = re.compile(r"^[0-9a-f]{40}$")
+_KEYS = frozenset({
+    "schema_version", "contract_version", "cadence", "as_of", "period_start", "period_end_exclusive",
+    "generated_at", "run_mode", "producer_ref", "source_provenance", "source_artifacts", "feed_status",
+})
+_FEED_KEYS = frozenset({"feed_count", "successful_feed_count", "failed_feed_count", "stale_feed_count", "missing_feed_count", "complete"})
+_ARTIFACT_KEYS = frozenset({"path", "sha256", "row_count"})
+
+
+class WeeklyContractError(ValueError):
+    def __init__(self, code: str) -> None:
+        self.code = code
+        super().__init__(code)
+
+
+@dataclass(frozen=True, slots=True)
+class WeeklySourceArtifact:
+    path: str
+    sha256: str
+    row_count: int
+
+
+@dataclass(frozen=True, slots=True)
+class WeeklyFeedStatus:
+    feed_count: int
+    successful_feed_count: int
+    failed_feed_count: int
+    stale_feed_count: int
+    missing_feed_count: int
+    complete: bool
+
+
+@dataclass(frozen=True, slots=True)
+class WeeklySourceContract:
+    as_of: date
+    period_start: date
+    period_end_exclusive: date
+    generated_at: datetime
+    run_mode: str
+    producer_ref: str
+    source_provenance: str
+    source_artifacts: tuple[WeeklySourceArtifact, ...]
+    feed_status: WeeklyFeedStatus
+
+
+def _invalid(code: str) -> WeeklyContractError:
+    return WeeklyContractError(code)
+
+
+def _date(value: object) -> date:
+    if type(value) is not str or not _DATE_RE.fullmatch(value):
+        raise _invalid("date_invalid")
+    try:
+        return date.fromisoformat(value)
+    except ValueError:
+        raise _invalid("date_invalid") from None
+
+
+def _generated_at(value: object) -> datetime:
+    if type(value) is not str or not _TIMESTAMP_RE.fullmatch(value):
+        raise _invalid("generated_at_invalid")
+    try:
+        parsed = datetime.fromisoformat(value[:-1] + "+00:00")
+    except ValueError:
+        raise _invalid("generated_at_invalid") from None
+    if parsed.tzinfo != timezone.utc:
+        raise _invalid("generated_at_invalid")
+    return parsed
+
+
+def _artifact(value: object) -> WeeklySourceArtifact:
+    if not isinstance(value, Mapping) or set(value) != _ARTIFACT_KEYS:
+        raise _invalid("source_artifact_invalid")
+    path = value["path"]
+    canonical = PurePosixPath(path) if type(path) is str else None
+    if (
+        type(path) is not str or not path or not path.isascii() or unicodedata.normalize("NFC", path) != path
+        or canonical is None or canonical.is_absolute() or path != str(canonical) or path.endswith("/")
+        or "//" in path or any(part in {"", ".", ".."} for part in canonical.parts) or "\\" in path
+    ):
+        raise _invalid("source_artifact_invalid")
+    digest = value["sha256"]
+    row_count = value["row_count"]
+    if type(digest) is not str or not _SHA256_RE.fullmatch(digest) or type(row_count) is not int or not 0 <= row_count <= MAX_SAFE_JSON_INTEGER:
+        raise _invalid("source_artifact_invalid")
+    return WeeklySourceArtifact(path, digest, row_count)
+
+
+def _feed_status(value: object) -> WeeklyFeedStatus:
+    if not isinstance(value, Mapping) or set(value) != _FEED_KEYS or type(value.get("complete")) is not bool:
+        raise _invalid("feed_status_invalid")
+    values = [value[key] for key in _FEED_KEYS if key != "complete"]
+    if any(type(item) is not int or not 0 <= item <= MAX_SAFE_JSON_INTEGER for item in values):
+        raise _invalid("feed_status_invalid")
+    status = WeeklyFeedStatus(*(value[key] for key in ("feed_count", "successful_feed_count", "failed_feed_count", "stale_feed_count", "missing_feed_count")), value["complete"])
+    if status.feed_count <= 0 or status.successful_feed_count != status.feed_count or any((status.failed_feed_count, status.stale_feed_count, status.missing_feed_count)) or not status.complete:
+        raise _invalid("feed_status_incomplete")
+    return status
+
+
+def parse_weekly_contract(value: Mapping[str, object]) -> WeeklySourceContract:
+    if not isinstance(value, Mapping) or set(value) != _KEYS:
+        raise _invalid("contract_shape_invalid")
+    if value["schema_version"] != SCHEMA_VERSION or value["contract_version"] != CONTRACT_VERSION or value["cadence"] != CADENCE:
+        raise _invalid("contract_version_invalid")
+    if type(value["run_mode"]) is not str or value["run_mode"] not in {"scheduled", "manual"}:
+        raise _invalid("run_mode_invalid")
+    producer_ref = value["producer_ref"]
+    provenance = value["source_provenance"]
+    if type(producer_ref) is not str or not _SHA1_RE.fullmatch(producer_ref) or type(provenance) is not str or not provenance:
+        raise _invalid("provenance_invalid")
+    as_of = _date(value["as_of"])
+    start = _date(value["period_start"])
+    end = _date(value["period_end_exclusive"])
+    if start.weekday() != 0 or end != start + timedelta(days=7) or as_of != end - timedelta(days=1):
+        raise _invalid("period_invalid")
+    generated_at = _generated_at(value["generated_at"])
+    if generated_at < datetime.combine(end, datetime.min.time(), timezone.utc):
+        raise _invalid("generated_at_before_period_end")
+    artifacts_value = value["source_artifacts"]
+    if not isinstance(artifacts_value, list) or not artifacts_value:
+        raise _invalid("source_artifact_invalid")
+    artifacts = tuple(sorted((_artifact(item) for item in artifacts_value), key=lambda item: item.path))
+    if len({item.path for item in artifacts}) != len(artifacts):
+        raise _invalid("source_artifact_duplicate")
+    return WeeklySourceContract(as_of, start, end, generated_at, value["run_mode"], producer_ref, provenance, artifacts, _feed_status(value["feed_status"]))
+
+
+def serialize_weekly_contract(contract: WeeklySourceContract) -> bytes:
+    if not isinstance(contract, WeeklySourceContract):
+        raise _invalid("contract_type_invalid")
+    try:
+        payload = {
+            "schema_version": SCHEMA_VERSION, "contract_version": CONTRACT_VERSION, "cadence": CADENCE,
+            "as_of": contract.as_of.isoformat(), "period_start": contract.period_start.isoformat(),
+            "period_end_exclusive": contract.period_end_exclusive.isoformat(),
+            "generated_at": contract.generated_at.isoformat(timespec="microseconds").replace("+00:00", "Z"),
+            "run_mode": contract.run_mode, "producer_ref": contract.producer_ref, "source_provenance": contract.source_provenance,
+            "source_artifacts": [{"path": item.path, "sha256": item.sha256, "row_count": item.row_count} for item in contract.source_artifacts],
+            "feed_status": {"feed_count": contract.feed_status.feed_count, "successful_feed_count": contract.feed_status.successful_feed_count, "failed_feed_count": contract.feed_status.failed_feed_count, "stale_feed_count": contract.feed_status.stale_feed_count, "missing_feed_count": contract.feed_status.missing_feed_count, "complete": contract.feed_status.complete},
+        }
+    except (AttributeError, TypeError, ValueError, OverflowError):
+        raise _invalid("contract_invalid") from None
+    parse_weekly_contract(payload)
+    try:
+        return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    except (TypeError, UnicodeError, ValueError):
+        raise _invalid("serialization_invalid") from None

tests/test_weekly_contract.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+from datetime import date
+
+import pytest
+
+from political_event_tracking_research.weekly_contract import (
+    WeeklyContractError,
+    parse_weekly_contract,
+    serialize_weekly_contract,
+)
+
+
+def payload(**overrides: object) -> dict[str, object]:
+    value: dict[str, object] = {
+        "schema_version": "1",
+        "contract_version": "political_event_weekly.v1",
+        "cadence": "weekly",
+        "as_of": "2026-07-12",
+        "period_start": "2026-07-06",
+        "period_end_exclusive": "2026-07-13",
+        "generated_at": "2026-07-13T00:00:00.123456Z",
+        "run_mode": "manual",
+        "producer_ref": "a" * 40,
+        "source_provenance": "official_political_event_tracking_research",
+        "source_artifacts": [
+            {"path": "data/live/political_events.csv", "sha256": "b" * 64, "row_count": 11},
+        ],
+        "feed_status": {
+            "feed_count": 9,
+            "successful_feed_count": 9,
+            "failed_feed_count": 0,
+            "stale_feed_count": 0,
+            "missing_feed_count": 0,
+            "complete": True,
+        },
+    }
+    value.update(overrides)
+    return value
+
+
+def test_valid_weekly_contract_round_trips_deterministically():
+    contract = parse_weekly_contract(payload())
+    encoded = serialize_weekly_contract(contract)
+    assert encoded == serialize_weekly_contract(parse_weekly_contract(__import__("json").loads(encoded)))
+    assert contract.as_of == date(2026, 7, 12)
+
+
+@pytest.mark.parametrize("field,value", [
+    ("schema_version", "2"),
+    ("contract_version", "political_event_weekly.v2"),
+    ("cadence", "daily"),
+    ("as_of", "2026-07-13"),
+    ("period_start", "2026-07-07"),
+    ("period_end_exclusive", "2026-07-14"),
+    ("generated_at", "2026-07-12T23:59:59Z"),
+    ("generated_at", "2026-07-13T00:00:00+08:00"),
+])
+def test_period_and_time_contract_fail_closed(field, value):
+    with pytest.raises(WeeklyContractError):
+        parse_weekly_contract(payload(**{field: value}))
+
+
+def test_scheduled_and_manual_are_explicit():
+    assert parse_weekly_contract(payload(run_mode="scheduled")).run_mode == "scheduled"
+    with pytest.raises(WeeklyContractError):
+        parse_weekly_contract(payload(run_mode=""))
+    with pytest.raises(WeeklyContractError):
+        parse_weekly_contract({key: value for key, value in payload().items() if key != "as_of"})
+
+
+def test_real_feed_counters_round_trip_without_serializer_defaults():
+    status = {"feed_count": 12, "successful_feed_count": 12, "failed_feed_count": 0, "stale_feed_count": 0, "missing_feed_count": 0, "complete": True}
+    contract = parse_weekly_contract(payload(feed_status=status))
+    encoded = serialize_weekly_contract(contract)
+    assert b'"feed_count":12' in encoded
+    assert parse_weekly_contract(__import__("json").loads(encoded)).feed_status == contract.feed_status
+
+
+@pytest.mark.parametrize("status", [
+    {"feed_count": 9, "successful_feed_count": 8, "failed_feed_count": 1, "stale_feed_count": 0, "missing_feed_count": 0, "complete": True},
+    {"feed_count": 9, "successful_feed_count": 9, "failed_feed_count": 0, "stale_feed_count": 1, "missing_feed_count": 0, "complete": True},
+    {"feed_count": 9, "successful_feed_count": 9, "failed_feed_count": 0, "stale_feed_count": 0, "missing_feed_count": 0, "complete": False},
+])
+def test_partial_stale_or_incomplete_feed_status_fails_closed(status):
+    with pytest.raises(WeeklyContractError, match="feed_status"):
+        parse_weekly_contract(payload(feed_status=status))
+
+
+def test_artifacts_are_sorted_and_duplicate_or_unsafe_inputs_fail_closed():
+    items = [
+        {"path": "z.csv", "sha256": "c" * 64, "row_count": 1},
+        {"path": "a.csv", "sha256": "d" * 64, "row_count": 2},
+    ]
+    contract = parse_weekly_contract(payload(source_artifacts=items))
+    assert [item.path for item in contract.source_artifacts] == ["a.csv", "z.csv"]
+    with pytest.raises(WeeklyContractError):
+        parse_weekly_contract(payload(source_artifacts=[items[0], items[0]]))
+    with pytest.raises(WeeklyContractError):
+        parse_weekly_contract(payload(source_artifacts=[{"path": "../secret", "sha256": "c" * 64, "row_count": 1}]))
+    for alias in ("a//b.csv", "a/b.csv/", "./a.csv", "a/./b.csv", "a\\b.csv", "é.csv"):
+        with pytest.raises(WeeklyContractError):
+            parse_weekly_contract(payload(source_artifacts=[{"path": alias, "sha256": "c" * 64, "row_count": 1}]))
+
+
+@pytest.mark.parametrize("field", ["row_count", "feed_count", "successful_feed_count"])
+def test_wire_integers_use_safe_json_range_and_reject_bool(field):
+    artifact = {"path": "data/live/political_events.csv", "sha256": "b" * 64, "row_count": 11}
+    status = payload()["feed_status"]
+    if field == "row_count":
+        with pytest.raises(WeeklyContractError):
+            parse_weekly_contract(payload(source_artifacts=[{**artifact, "row_count": 2**53}]))
+        with pytest.raises(WeeklyContractError):
+            parse_weekly_contract(payload(source_artifacts=[{**artifact, "row_count": True}]))
+    else:
+        with pytest.raises(WeeklyContractError):
+            parse_weekly_contract(payload(feed_status={**status, field: 2**53}))
+        with pytest.raises(WeeklyContractError):
+            parse_weekly_contract(payload(feed_status={**status, field: True}))
+
+
+def test_unknown_fields_and_wire_type_confusion_fail_closed():
+    with pytest.raises(WeeklyContractError):
+        parse_weekly_contract(payload(extra=True))
+    with pytest.raises(WeeklyContractError):
+        parse_weekly_contract(payload(schema_version=1))
+    with pytest.raises(WeeklyContractError):
+        parse_weekly_contract(payload(feed_status={**payload()["feed_status"], "feed_count": True}))