Add concrete weekly producer artifact

.github/workflows/rss_source_pipeline.yml

@@ -23,6 +23,16 @@ on:
         required: false
         default: "50"
         type: string
+      period_start:
+        description: "Manual completed ISO-week Monday date; required for manual runs."
+        required: false
+        default: ""
+        type: string
+      as_of:
+        description: "Manual completed ISO-week Sunday date; required for manual runs."
+        required: false
+        default: ""
+        type: string
       commit_outputs:
         description: "Commit generated live CSV outputs back to data/live."
         required: false
@@ -32,7 +42,7 @@ on:
           - "false"
           - "true"
   schedule:
-    - cron: "15 12 * * 6"
+    - cron: "15 12 * * 1"
 
 permissions:
   contents: write
@@ -107,9 +117,42 @@ jobs:
             git commit -m "Update live RSS source events [skip ci]"
             git push
           fi
+      - name: Build completed weekly producer artifact
+        env:
+          PERIOD_START: ${{ github.event.inputs.period_start || '' }}
+          AS_OF: ${{ github.event.inputs.as_of || '' }}
+          WATCHLIST_PATH: ${{ github.event.inputs.watchlist_path || 'data/live/political_watchlist.csv' }}
+        run: |
+          set -euo pipefail
+          test "${GITHUB_RUN_ATTEMPT}" = "1"
+          mkdir -p data/output/political-event-weekly-v1
+          ARGS=(
+            --generated-at "$(date -u +%Y-%m-%dT%H:%M:%S.000000Z)"
+            --workflow-ref "${GITHUB_WORKFLOW_REF}"
+            --source-run-id "${GITHUB_RUN_ID}"
+            --producer-ref "${GITHUB_SHA}"
+            --source-events data/output/rss_source_pipeline/source_events.csv
+            --watchlist "${WATCHLIST_PATH}"
+            --feed-status data/output/rss_source_pipeline/source_fetch_status.json
+            --output-dir data/output/political-event-weekly-v1
+          )
+          if [ "${GITHUB_EVENT_NAME}" = "schedule" ]; then
+            ARGS+=(--scheduled-today "$(date -u +%F)" --run-mode scheduled)
+          else
+            test -n "${PERIOD_START}" && test -n "${AS_OF}"
+            ARGS+=(--period-start "${PERIOD_START}" --as-of "${AS_OF}" --run-mode manual)
+          fi
+          python scripts/build_weekly_artifact.py "${ARGS[@]}"
       - name: Upload RSS source artifact
         uses: actions/upload-artifact@v7
         with:
           name: rss-source-pipeline
           path: data/output/rss_source_pipeline/
           if-no-files-found: error
+      - name: Upload political weekly artifact
+        uses: actions/upload-artifact@v7
+        with:
+          name: political-event-weekly-v1
+          path: data/output/political-event-weekly-v1/
+          if-no-files-found: error
+          retention-days: 30

docs/weekly_producer_artifact_contract.md

@@ -0,0 +1,40 @@
+# Political Event Weekly Producer Artifact
+
+`political-event-weekly-v1` is the producer-owned, complete-UTC-ISO-week artifact
+emitted by the RSS source pipeline.
+
+## Fixed contract
+
+- Scheduled execution is Monday `12:15 UTC` and covers the immediately preceding
+  complete Monday–Sunday UTC ISO week.
+- Manual execution must provide matching `period_start` (Monday) and `as_of`
+  (Sunday). No latest-source or wall-clock fallback is allowed for period identity.
+- `generated_at` is the actual UTC build time and is separate from period identity.
+- `GITHUB_RUN_ATTEMPT` must be exactly `1`; a rerun is fail-closed rather than a
+  new version of the same snapshot.
+- Artifact retention is exactly 30 days.
+
+The dedicated artifact contains exactly:
+
+```text
+period_lock.json
+political_events.csv
+political_watchlist.csv
+political_event_weekly.json
+weekly_manifest.json
+```
+
+The producer filters `political_events.csv` to the locked inclusive date range
+before writing it; malformed `event_date` fails closed. The original source
+events/watchlist input snapshot digest remains in `period_lock.json`, while the
+filtered event bytes and row count are bound by the artifact manifest.
+
+The manifest binds the first four files by name, byte length and SHA-256. It also
+records CSV headers/row counts, period/as-of/generated-at, producer SHA,
+workflow ref, run id/attempt, source snapshot digest/provenance, and complete
+feed counters. Any partial/failed/stale/missing feed, source mismatch, unsafe
+wire, or readback mismatch prevents dedicated artifact upload.
+
+This contract is producer-side only. QAR consumer acquisition/readback is a
+separate later slice; no legacy compatibility, identity store, Pages, publisher,
+or workflow permission expansion is implied.

scripts/build_weekly_artifact.py

@@ -0,0 +1,91 @@
+#!/usr/bin/env python3
+"""Build the dedicated producer-owned political-event weekly artifact."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from datetime import date, datetime, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from political_event_tracking_research.weekly_artifact import (  # noqa: E402
+    WeeklyArtifactError,
+    build_weekly_artifact,
+    completed_week_period,
+)
+
+
+def _date(value: str) -> date:
+    try:
+        return date.fromisoformat(value)
+    except ValueError:
+        raise argparse.ArgumentTypeError("invalid ISO date") from None
+
+
+def _generated_at(value: str) -> datetime:
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        raise argparse.ArgumentTypeError("invalid UTC timestamp") from None
+    if parsed.tzinfo != timezone.utc:
+        raise argparse.ArgumentTypeError("generated_at must be UTC")
+    return parsed
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--period-start", type=_date)
+    parser.add_argument("--as-of", type=_date)
+    parser.add_argument("--scheduled-today", type=_date)
+    parser.add_argument("--generated-at", required=True, type=_generated_at)
+    parser.add_argument("--workflow-ref", required=True)
+    parser.add_argument("--source-run-id", required=True)
+    parser.add_argument("--producer-ref", required=True)
+    parser.add_argument("--source-events", required=True, type=Path)
+    parser.add_argument("--watchlist", required=True, type=Path)
+    parser.add_argument("--feed-status", required=True, type=Path)
+    parser.add_argument("--output-dir", required=True, type=Path)
+    parser.add_argument("--run-mode", choices=("scheduled", "manual"), required=True)
+    args = parser.parse_args()
+
+    if args.scheduled_today is not None:
+        if args.run_mode != "scheduled" or args.period_start is not None or args.as_of is not None:
+            parser.error("scheduled mode requires only --scheduled-today")
+        period_start, period_end = completed_week_period(args.scheduled_today)
+        as_of = period_end - date.resolution
+    elif args.run_mode == "manual" and args.period_start is not None and args.as_of is not None:
+        period_start, as_of = args.period_start, args.as_of
+    else:
+        parser.error("manual mode requires --period-start and --as-of")
+
+    try:
+        feed_status = json.loads(args.feed_status.read_text(encoding="utf-8"))
+        files = build_weekly_artifact(
+            period_start=period_start,
+            as_of=as_of,
+            generated_at=args.generated_at,
+            workflow_ref=args.workflow_ref,
+            source_run_id=args.source_run_id,
+            producer_ref=args.producer_ref,
+            source_events=args.source_events.read_bytes(),
+            watchlist=args.watchlist.read_bytes(),
+            feed_status=feed_status,
+            source_provenance="official_rss_source_pipeline_v1",
+            run_mode=args.run_mode,
+        )
+        args.output_dir.mkdir(parents=True, exist_ok=True)
+        if any(args.output_dir.iterdir()):
+            raise WeeklyArtifactError("artifact_output_not_empty")
+        for name, content in files.items():
+            (args.output_dir / name).write_bytes(content)
+    except WeeklyArtifactError as exc:
+        raise SystemExit(exc.code) from None
+    except (OSError, UnicodeError, json.JSONDecodeError, TypeError, ValueError):
+        raise SystemExit("weekly_artifact_input_invalid") from None
+
+
+if __name__ == "__main__":
+    main()

src/political_event_tracking_research/rss_source_fetch.py

@@ -178,6 +178,7 @@ def write_fetch_status(path: str | Path, statuses: list[FeedFetchStatus], *, ite
         "feed_count": len(statuses),
         "successful_feed_count": sum(1 for item in statuses if item.ok),
         "failed_feed_count": sum(1 for item in statuses if not item.ok),
+        "complete": bool(statuses) and all(item.ok for item in statuses),
         "item_count": item_count,
         "feeds": [item.to_json() for item in statuses],
     }