diff --git a/.github/workflows/rss_source_pipeline.yml b/.github/workflows/rss_source_pipeline.yml index 7bd252b..5148e33 100644 --- a/.github/workflows/rss_source_pipeline.yml +++ b/.github/workflows/rss_source_pipeline.yml @@ -4,17 +4,17 @@ on: workflow_dispatch: inputs: feeds_path: - description: "RSS feed config CSV path." + description: "Must remain the canonical feed configuration path." required: false default: "config/free_rss_feeds.csv" type: string aliases_path: - description: "Symbol alias CSV path." + description: "Must remain the canonical alias configuration path." required: false default: "config/core_us_equity_aliases.csv" type: string watchlist_path: - description: "Watchlist CSV path." + description: "Must remain the canonical watchlist path." required: false default: "data/live/political_watchlist.csv" type: string @@ -28,9 +28,7 @@ on: required: false default: "false" type: choice - options: - - "false" - - "true" + options: ["false", "true"] schedule: - cron: "15 12 * * 6" @@ -42,11 +40,54 @@ concurrency: cancel-in-progress: false jobs: + validate-workflow-boundary: + if: >- + github.repository == 'QuantStrategyLab/PoliticalEventTrackingResearch' && + github.ref == 'refs/heads/main' && + github.workflow_ref == 'QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/rss_source_pipeline.yml@refs/heads/main' + runs-on: ubuntu-latest + steps: + - name: Validate canonical workflow inputs + env: + EVENT_NAME: ${{ github.event_name }} + FEEDS_PATH: ${{ github.event.inputs.feeds_path || 'config/free_rss_feeds.csv' }} + ALIASES_PATH: ${{ github.event.inputs.aliases_path || 'config/core_us_equity_aliases.csv' }} + WATCHLIST_PATH: ${{ github.event.inputs.watchlist_path || 'data/live/political_watchlist.csv' }} + COMMIT_OUTPUTS: ${{ github.event.inputs.commit_outputs || 'false' }} + MAX_ITEMS_PER_FEED: ${{ github.event.inputs.max_items_per_feed || '50' }} + run: | + set -euo pipefail + python3 - <<'PY' + import os + expected = ( + "config/free_rss_feeds.csv", + "config/core_us_equity_aliases.csv", + "data/live/political_watchlist.csv", + ) + actual = tuple(os.environ[key] for key in ("FEEDS_PATH", "ALIASES_PATH", "WATCHLIST_PATH")) + if actual != expected: + raise SystemExit("workflow_input_path_invalid") + if os.environ["COMMIT_OUTPUTS"] not in {"false", "true"}: + raise SystemExit("workflow_option_invalid") + if os.environ["COMMIT_OUTPUTS"] == "true" and os.environ["MAX_ITEMS_PER_FEED"] != "50": + raise SystemExit("production_fetch_override_invalid") + PY + build-rss-source-events: + needs: validate-workflow-boundary + if: needs.validate-workflow-boundary.result == 'success' runs-on: ubuntu-latest timeout-minutes: 30 steps: - uses: actions/checkout@v6 + with: + repository: QuantStrategyLab/PoliticalEventTrackingResearch + ref: ${{ github.sha }} + persist-credentials: false + - name: Verify reviewed main checkout + env: + EXPECTED_SHA: ${{ github.sha }} + run: test "$(git rev-parse HEAD)" = "${EXPECTED_SHA}" - uses: actions/setup-python@v6 with: python-version: "3.11" @@ -54,50 +95,53 @@ jobs: run: python -m pip install -e . - name: Fetch RSS sources and extract mentions env: - FEEDS_PATH: ${{ github.event.inputs.feeds_path || 'config/free_rss_feeds.csv' }} - ALIASES_PATH: ${{ github.event.inputs.aliases_path || 'config/core_us_equity_aliases.csv' }} - WATCHLIST_PATH: ${{ github.event.inputs.watchlist_path || 'data/live/political_watchlist.csv' }} MAX_ITEMS_PER_FEED: ${{ github.event.inputs.max_items_per_feed || '50' }} run: | set -euo pipefail mkdir -p data/output/rss_source_pipeline + fetch_exit=0 python scripts/fetch_rss_sources.py \ - --feeds "${FEEDS_PATH}" \ + --feeds config/free_rss_feeds.csv \ --output data/output/rss_source_pipeline/source_items.csv \ --max-items-per-feed "${MAX_ITEMS_PER_FEED}" \ --continue-on-feed-error \ - --status-output data/output/rss_source_pipeline/source_fetch_status.json - python scripts/extract_source_mentions.py \ - --raw-items data/output/rss_source_pipeline/source_items.csv \ - --aliases "${ALIASES_PATH}" \ - --output data/output/rss_source_pipeline/source_events.csv - python scripts/build_tracker.py \ - --watchlist "${WATCHLIST_PATH}" \ - --events data/output/rss_source_pipeline/source_events.csv \ - --output data/output/rss_source_pipeline/source_tracker.csv + --status-output data/output/rss_source_pipeline/source_fetch_status.json || fetch_exit=$? + if [ -f data/output/rss_source_pipeline/source_items.csv ]; then + python scripts/extract_source_mentions.py \ + --raw-items data/output/rss_source_pipeline/source_items.csv \ + --aliases config/core_us_equity_aliases.csv \ + --output data/output/rss_source_pipeline/source_events.csv + python scripts/build_tracker.py \ + --watchlist data/live/political_watchlist.csv \ + --events data/output/rss_source_pipeline/source_events.csv \ + --output data/output/rss_source_pipeline/source_tracker.csv + fi + printf '%s\n' "${fetch_exit}" > data/output/rss_source_pipeline/fetch_exit.txt + printf 'fetch_exit=%s\n' "${fetch_exit}" + - 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: Validate feed completeness + run: python scripts/validate_fetch_status.py --status data/output/rss_source_pipeline/source_fetch_status.json --fetch-exit data/output/rss_source_pipeline/fetch_exit.txt - name: Publish live CSV outputs to repository env: COMMIT_OUTPUTS: ${{ github.event_name == 'schedule' && 'true' || github.event.inputs.commit_outputs || 'false' }} + GITHUB_TOKEN: ${{ github.token }} + EXPECTED_SHA: ${{ github.sha }} run: | set -euo pipefail - if [ "${COMMIT_OUTPUTS}" != "true" ]; then - echo "Live output commit disabled." - exit 0 - fi + if [ "${COMMIT_OUTPUTS}" != "true" ]; then echo "Live output commit disabled."; exit 0; fi + test "$(git rev-parse HEAD)" = "${EXPECTED_SHA}" mkdir -p data/live cp data/output/rss_source_pipeline/source_items.csv data/live/source_items.csv cp data/output/rss_source_pipeline/source_events.csv data/live/source_events.csv cp data/output/rss_source_pipeline/source_events.csv data/live/political_events.csv cp data/output/rss_source_pipeline/source_tracker.csv data/live/source_tracker.csv cp data/output/rss_source_pipeline/source_fetch_status.json data/live/source_fetch_status.json - python scripts/write_live_manifest.py \ - --base-dir . \ - --output data/live/source_manifest.json \ - data/live/source_fetch_status.json \ - data/live/source_items.csv \ - data/live/source_events.csv \ - data/live/political_events.csv \ - data/live/source_tracker.csv + python scripts/write_live_manifest.py --base-dir . --output data/live/source_manifest.json data/live/source_fetch_status.json data/live/source_items.csv data/live/source_events.csv data/live/political_events.csv data/live/source_tracker.csv git config user.name "github-actions[bot]" git config user.email "41898282+github-actions[bot]@users.noreply.github.com" git add data/live/source_items.csv data/live/source_events.csv data/live/political_events.csv data/live/source_tracker.csv data/live/source_fetch_status.json data/live/source_manifest.json @@ -105,11 +149,10 @@ jobs: echo "No live RSS output changes to commit." else git commit -m "Update live RSS source events [skip ci]" - git push + git config --local http.https://github.com/.extraheader "AUTHORIZATION: bearer ${GITHUB_TOKEN}" + cleanup_git_auth() { git config --local --unset-all http.https://github.com/.extraheader >/dev/null 2>&1 || true; } + trap cleanup_git_auth EXIT + git push origin HEAD:refs/heads/main + cleanup_git_auth + trap - EXIT fi - - 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 diff --git a/scripts/validate_fetch_status.py b/scripts/validate_fetch_status.py new file mode 100644 index 0000000..35fa2bf --- /dev/null +++ b/scripts/validate_fetch_status.py @@ -0,0 +1,30 @@ +#!/usr/bin/env python3 +"""Fail closed unless the existing RSS fetch status is complete.""" +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from political_event_tracking_research.rss_source_fetch import FetchStatusError, validate_fetch_status # noqa: E402 + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--status", required=True, type=Path) + parser.add_argument("--fetch-exit", required=True, type=Path) + args = parser.parse_args() + try: + fetch_exit = int(args.fetch_exit.read_text(encoding="utf-8").strip()) + complete = validate_fetch_status(json.loads(args.status.read_text(encoding="utf-8")), fetch_exit=fetch_exit) + except (FetchStatusError, OSError, UnicodeError, json.JSONDecodeError, TypeError, ValueError): + raise SystemExit("fetch_status_invalid") from None + if not complete: + raise SystemExit("fetch_incomplete") + + +if __name__ == "__main__": + main() diff --git a/src/political_event_tracking_research/rss_source_fetch.py b/src/political_event_tracking_research/rss_source_fetch.py index 98e6b13..5595351 100644 --- a/src/political_event_tracking_research/rss_source_fetch.py +++ b/src/political_event_tracking_research/rss_source_fetch.py @@ -48,6 +48,39 @@ def to_json(self) -> dict[str, object]: } +class FetchStatusError(ValueError): + def __init__(self, code: str) -> None: + self.code = code + super().__init__(code) + + +def validate_fetch_status(payload: object, *, fetch_exit: int = 0) -> bool: + if type(fetch_exit) is not int or fetch_exit < 0: + raise FetchStatusError("fetch_exit_invalid") + if fetch_exit != 0: + raise FetchStatusError("fetch_failed") + required = {"generated_at", "feed_count", "successful_feed_count", "failed_feed_count", "item_count", "feeds"} + entry_keys = {"feed_id", "feed_url", "ok", "item_count", "error"} + if not isinstance(payload, dict) or set(payload) != required or not isinstance(payload["feeds"], list): + raise FetchStatusError("fetch_status_shape_invalid") + counters = ("feed_count", "successful_feed_count", "failed_feed_count", "item_count") + if type(payload["generated_at"]) is not str or any(type(payload[key]) is not int or payload[key] < 0 for key in counters): + raise FetchStatusError("fetch_status_counter_invalid") + feeds = payload["feeds"] + if len(feeds) != payload["feed_count"]: + raise FetchStatusError("fetch_status_counter_invalid") + successful = 0 + total_items = 0 + for feed in feeds: + if not isinstance(feed, dict) or set(feed) != entry_keys or type(feed["feed_id"]) is not str or type(feed["feed_url"]) is not str or type(feed["ok"]) is not bool or type(feed["item_count"]) is not int or feed["item_count"] < 0 or type(feed["error"]) is not str: + raise FetchStatusError("fetch_status_entry_invalid") + successful += int(feed["ok"]) + total_items += feed["item_count"] + if successful != payload["successful_feed_count"] or payload["feed_count"] - successful != payload["failed_feed_count"] or total_items != payload["item_count"]: + raise FetchStatusError("fetch_status_counter_invalid") + return payload["failed_feed_count"] == 0 + + def load_feed_config(path: str | Path) -> list[FeedConfig]: feeds: list[FeedConfig] = [] for row in read_csv_rows(path): diff --git a/src/political_event_tracking_research/workflow_boundary.py b/src/political_event_tracking_research/workflow_boundary.py new file mode 100644 index 0000000..58599fc --- /dev/null +++ b/src/political_event_tracking_research/workflow_boundary.py @@ -0,0 +1,33 @@ +"""Pure guards for the existing RSS workflow boundary.""" +from __future__ import annotations + +from pathlib import PurePosixPath + +CANONICAL_FEEDS = "config/free_rss_feeds.csv" +CANONICAL_ALIASES = "config/core_us_equity_aliases.csv" +CANONICAL_WATCHLIST = "data/live/political_watchlist.csv" +CANONICAL_MAX_ITEMS_PER_FEED = "50" + + +class PathBoundaryError(ValueError): + def __init__(self, code: str) -> None: + self.code = code + super().__init__(code) + + +def validate_source_paths(feeds_path: object, aliases_path: object, watchlist_path: object) -> None: + expected = (CANONICAL_FEEDS, CANONICAL_ALIASES, CANONICAL_WATCHLIST) + actual = (feeds_path, aliases_path, watchlist_path) + if any(type(value) is not str or value != expected_value for value, expected_value in zip(actual, expected, strict=True)): + raise PathBoundaryError("workflow_input_path_invalid") + for value in actual: + path = PurePosixPath(value) + if path.is_absolute() or value != str(path) or any(part in {"", ".", ".."} for part in path.parts) or "\\" in value: + raise PathBoundaryError("workflow_input_path_invalid") + + +def validate_workflow_options(commit_outputs: object, max_items_per_feed: object) -> None: + if type(commit_outputs) is not str or commit_outputs not in {"false", "true"} or type(max_items_per_feed) is not str: + raise PathBoundaryError("workflow_option_invalid") + if commit_outputs == "true" and max_items_per_feed != CANONICAL_MAX_ITEMS_PER_FEED: + raise PathBoundaryError("production_fetch_override_invalid") diff --git a/tests/test_weekly_workflow_w0.py b/tests/test_weekly_workflow_w0.py new file mode 100644 index 0000000..37f4aaf --- /dev/null +++ b/tests/test_weekly_workflow_w0.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +from political_event_tracking_research.rss_source_fetch import FetchStatusError, validate_fetch_status +from political_event_tracking_research.workflow_boundary import PathBoundaryError, validate_source_paths, validate_workflow_options + + +def status(**overrides: object) -> dict[str, object]: + payload: dict[str, object] = { + "generated_at": "2026-07-16T00:00:00Z", + "feed_count": 2, + "successful_feed_count": 2, + "failed_feed_count": 0, + "item_count": 1, + "feeds": [ + {"feed_id": "a", "feed_url": "https://a.example/feed", "ok": True, "item_count": 1, "error": ""}, + {"feed_id": "b", "feed_url": "https://b.example/feed", "ok": True, "item_count": 0, "error": ""}, + ], + } + payload.update(overrides) + return payload + + +def test_fetch_status_validation_returns_complete_state() -> None: + assert validate_fetch_status(status()) is True + assert validate_fetch_status(status(failed_feed_count=1, successful_feed_count=1, feeds=[ + {"feed_id": "a", "feed_url": "https://a.example/feed", "ok": True, "item_count": 1, "error": ""}, + {"feed_id": "b", "feed_url": "https://b.example/feed", "ok": False, "item_count": 0, "error": "blocked"}, + ])) is False + with pytest.raises(FetchStatusError): + validate_fetch_status(status(), fetch_exit=7) + + +@pytest.mark.parametrize("change", [{"complete": False}, {"failed_feed_count": 1}, {"unknown": 1}, {"feeds": []}]) +def test_fetch_status_shape_or_incompleteness_fails_closed(change: dict[str, object]) -> None: + with pytest.raises(FetchStatusError): + validate_fetch_status({**status(), **change}) + + +@pytest.mark.parametrize("value", ["config/other.csv", "../config/free_rss_feeds.csv", "/tmp/feeds.csv", "config\\feeds.csv", ""]) +def test_manual_source_paths_are_canonical_only(value: str) -> None: + with pytest.raises(PathBoundaryError): + validate_source_paths(value, "config/core_us_equity_aliases.csv", "data/live/political_watchlist.csv") + + +def test_production_commit_rejects_fetch_override_before_fetch() -> None: + validate_workflow_options("false", "7") + validate_workflow_options("true", "50") + with pytest.raises(PathBoundaryError): + validate_workflow_options("true", "7") + + +def test_workflow_orders_debug_upload_gate_and_live_push() -> None: + workflow = Path(__file__).parents[1].joinpath(".github/workflows/rss_source_pipeline.yml").read_text(encoding="utf-8") + assert workflow.index("Validate canonical workflow inputs") < workflow.index("actions/checkout@") + assert workflow.index("Upload RSS source artifact") < workflow.index("Validate feed completeness") + assert workflow.index("Validate feed completeness") < workflow.index("Publish live CSV outputs") + assert "git push origin HEAD:refs/heads/main" in workflow + assert "git rev-parse HEAD" in workflow + assert "fetch_exit" in workflow + assert "--fetch-exit" in workflow + assert "COMMIT_OUTPUTS" in workflow + assert "weekly" not in workflow.lower()