提交原始紀錄

完整差異

@@ -0,0 +1,156 @@\n+name: daily-briefing\n+\n+# Thrice-daily air & transport briefing (Asia/Taipei editions).\n+# UTC crons map to fixed Taipei editions — the mapping below must match\n+# CRON_TO_EDITION in pipeline/briefing.py (tests cross-check both):\n+#   15 23 * * *  ->  morning    (07:15 Taipei)\n+#   15 7  * * *  ->  afternoon  (15:15 Taipei)\n+#   15 15 * * *  ->  evening    (23:15 Taipei)\n+# A delayed run never changes the data window: the window is derived from\n+# the edition + Taipei date only (see pipeline/briefing.py).\n+\n+on:\n+  schedule:\n+    - cron: "15 23 * * *"\n+    - cron: "15 7 * * *"\n+    - cron: "15 15 * * *"\n+  workflow_dispatch:\n+    inputs:\n+      edition:\n+        description: "Briefing edition"\n+        required: true\n+        type: choice\n+        options:\n+          - morning\n+          - afternoon\n+          - evening\n+      date:\n+        description: "Taipei date YYYY-MM-DD (default: today in Taipei)"\n+        required: false\n+        type: string\n+\n+concurrency:\n+  group: briefing\n+  cancel-in-progress: false\n+\n+jobs:\n+  briefing:\n+    runs-on: ubuntu-latest\n+    permissions:\n+      contents: write # commits data/briefings back to the repo\n+      pages: write # configure-pages enablement\n+    steps:\n+      - uses: actions/checkout@v4\n+\n+      - name: Enable GitHub Pages (build_type=workflow)\n+        uses: actions/configure-pages@v5\n+        with:\n+          enablement: true\n+\n+      - uses: actions/setup-python@v5\n+        with:\n+          python-version: "3.12"\n+          cache: pip\n+\n+      - run: pip install -r requirements.txt\n+\n+      - name: Resolve edition (explicit cron mapping, never guessed from the clock)\n+        id: ed\n+        run: |\n+          if [ "${{ github.event_name }}" = "workflow_dispatch" ]; then\n+            echo "edition=${{ inputs.edition }}" >> "$GITHUB_OUTPUT"\n+            echo "date=${{ inputs.date }}" >> "$GITHUB_OUTPUT"\n+          else\n+            case "${{ github.event.schedule }}" in\n+              "15 23 * * *") echo "edition=morning" >> "$GITHUB_OUTPUT" ;;\n+              "15 7 * * *")  echo "edition=afternoon" >> "$GITHUB_OUTPUT" ;;\n+              "15 15 * * *") echo "edition=evening" >> "$GITHUB_OUTPUT" ;;\n+              *) echo "::error::unmapped schedule '${{ github.event.schedule }}'"; exit 1 ;;\n+            esac\n+            echo "date=" >> "$GITHUB_OUTPUT"\n+          fi\n+\n+      - name: Assemble briefing and build site\n+        env:\n+          # Existing free model keys, used ONLY for the optional single\n+          # batch intro call (BRIEFING_LLM_INTRO=true). Default: no call.\n+          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}\n+          NVIDIA_API_KEY: ${{ secrets.NVIDIA_API_KEY || secrets.API_1_NVIDIA_NEMOTRON_3_ULTRA_550B_A55B }}\n+          AVWIRE_PROVIDER_ORDER: ${{ vars.AVWIRE_PROVIDER_ORDER }}\n+          BRIEFING_LLM_INTRO: ${{ vars.BRIEFING_LLM_INTRO }}\n+          BRIEFING_MAX_ITEMS: ${{ vars.BRIEFING_MAX_ITEMS }}\n+          BRIEFING_MAX_ITEMS_PER_SECTION: ${{ vars.BRIEFING_MAX_ITEMS_PER_SECTION }}\n+          BRIEFING_EXTRA_LLM_CALLS_MAX: ${{ vars.BRIEFING_EXTRA_LLM_CALLS_MAX }}\n+        run: |\n+          if [ -n "${{ steps.ed.outputs.date }}" ]; then\n+            python pipeline/briefing.py --edition "${{ steps.ed.outputs.edition }}" --date "${{ steps.ed.outputs.date }}"\n+          else\n+            python pipeline/briefing.py --edition "${{ steps.ed.outputs.edition }}"\n+          fi\n+          python pipeline/build.py\n+\n+      - name: Scan build output and committed data for secrets/reasoning\n+        run: |\n+          scan() {\n+            set +e\n+            grep -rInE "$1" site/ data/\n+            rc=$?\n+            set -e\n+            if [ "$rc" -eq 0 ]; then\n+              echo "::error::$2 found in publishable output"\n+              exit 1\n+            elif [ "$rc" -ge 2 ]; then\n+              echo "::error::scan failed (grep rc=$rc)"  # fail closed\n+              exit 1\n+            fi\n+          }\n+          scan "nvapi-[A-Za-z0-9_-]{16,}|AIza[0-9A-Za-z_-]{20,}|sk-ant-[A-Za-z0-9_-]{16,}|Bearer [A-Za-z0-9_./+-]{20,}" "secret-like string"\n+          scan "reasoning_content|<think>" "reasoning trace"\n+          echo "scan: site/ and data/ clean"\n+\n+      # Upload BEFORE the data commit so the deploy never depends on git push.\n+      - uses: actions/upload-pages-artifact@v3\n+        with:\n+          path: site\n+\n+      - name: Commit briefing data (only when changed)\n+        continue-on-error: true # a push race must not fail the run or block deploy\n+        run: |\n+          git config user.name "avwire-bot"\n+          git config user.email "bot@users.noreply.github.com"\n+          git add data/briefings\n+          if git diff --cached --quiet; then\n+            echo "no changes to commit"\n+            exit 0\n+          fi\n+          git commit -m "briefing ${{ steps.ed.outputs.edition }} $(date -u +%FT%H:%M)"\n+          pushed=""\n+          for i in 1 2 3; do\n+            if ! git pull --rebase -X theirs origin "${GITHUB_REF_NAME}"; then\n+              git rebase --abort || true\n+              sleep 5\n+              continue\n+            fi\n+            if git push; then\n+              pushed=yes\n+              break\n+            fi\n+            sleep 5\n+          done\n+          if [ -z "$pushed" ]; then\n+            echo "::warning::briefing data could not be pushed after 3 attempts"\n+            exit 1\n+          fi\n+\n+  deploy:\n+    needs: briefing\n+    runs-on: ubuntu-latest\n+    permissions:\n+      pages: write\n+      id-token: write\n+    environment:\n+      name: github-pages\n+      url: ${{ steps.deployment.outputs.page_url }}\n+    steps:\n+      - uses: actions/deploy-pages@v4\n+        id: deployment\n@@ -0,0 +1,44 @@\n+{\n+  "morning": {\n+    "label_zh_tw": "晨間快報",\n+    "label_en": "Morning Briefing",\n+    "title_zh_tw": "AVWIRE 晨間航空與交通快報",\n+    "title_en": "AVWIRE Morning Air & Transport Briefing",\n+    "window_start_hour": 23,\n+    "window_end_hour": 7,\n+    "window_start_day_offset": -1,\n+    "cutoff_hour": 7,\n+    "scheduled_hour": 7,\n+    "scheduled_minute": 15,\n+    "route_suffix": "morning",\n+    "cron_utc": "15 23 * * *"\n+  },\n+  "afternoon": {\n+    "label_zh_tw": "午後快報",\n+    "label_en": "Afternoon Briefing",\n+    "title_zh_tw": "AVWIRE 午後航空與交通快報",\n+    "title_en": "AVWIRE Afternoon Air & Transport Briefing",\n+    "window_start_hour": 7,\n+    "window_end_hour": 15,\n+    "window_start_day_offset": 0,\n+    "cutoff_hour": 15,\n+    "scheduled_hour": 15,\n+    "scheduled_minute": 15,\n+    "route_suffix": "afternoon",\n+    "cron_utc": "15 7 * * *"\n+  },\n+  "evening": {\n+    "label_zh_tw": "晚間快報",\n+    "label_en": "Evening Briefing",\n+    "title_zh_tw": "AVWIRE 晚間航空與交通快報",\n+    "title_en": "AVWIRE Evening Air & Transport Briefing",\n+    "window_start_hour": 15,\n+    "window_end_hour": 23,\n+    "window_start_day_offset": 0,\n+    "cutoff_hour": 23,\n+    "scheduled_hour": 23,\n+    "scheduled_minute": 15,\n+    "route_suffix": "evening",\n+    "cron_utc": "15 15 * * *"\n+  }\n+}\n@@ -0,0 +1,143 @@\n+{\n+  "content_type": "daily_transport_briefing",\n+  "briefing_id": "2026-07-27-morning",\n+  "date": "2026-07-27",\n+  "edition": "morning",\n+  "edition_label": "晨間快報",\n+  "timezone": "Asia/Taipei",\n+  "window_start": "2026-07-26T23:00:00+08:00",\n+  "window_end": "2026-07-27T07:00:00+08:00",\n+  "cutoff_time": "2026-07-27T07:00:00+08:00",\n+  "scheduled_time": "2026-07-27T07:15:00+08:00",\n+  "generated_at": "2026-07-27T04:48:38.325454+08:00",\n+  "status": "published",\n+  "item_count": 2,\n+  "sections": {\n+    "aviation_incidents": [],\n+    "taiwan_aviation": [\n+      {\n+        "event_id": "ev-6b68235e9011a3ea",\n+        "item_type": "new",\n+        "previous_event_id": null,\n+        "category": "taiwan_aviation",\n+        "headline": "民航局公布中秋連假離島加班 1340架次7/27開訂",\n+        "summary": "交通部民用航空局今日宣布，針對9月25日至28日中秋教師節四天連假，規劃9月24日至29日六天疏運期，澎湖、金門、馬祖三離島航線共提供1340架次、111432個座位，於7月27日上午9時起開放訂位，民航局將視訂位狀況適時增班。",\n+        "headline_en": "CAA Taiwan Announces 1,340 Extra Flights for Mid-Autumn Holiday, Booking Opens July 27",\n+        "summary_en": "Taiwan's Civil Aeronautics Administration announced today that for the four-day Mid-Autumn Festival and Teachers' Day holiday from September 25 to 28, airlines will operate 1,340 flights with 111,432 seats to the outlying islands of Penghu, Kinmen, and Matsu. The transport period runs September 24 to 29. Booking opens July 27 at 9:00 AM, and the CAA will monitor demand to add flights as needed. The breakdown includes 658 flights to Penghu, 522 to Kinmen, and 160 to Matsu.",\n+        "event_time": null,\n+        "source_published_at": "2026-07-26T17:48:00+00:00",\n+        "source_updated_at": null,\n+        "location": "澎湖",\n+        "entities": [\n+          "交通部民用航空局",\n+          "CAA Taiwan",\n+          "澎湖",\n+          "金門",\n+          "馬祖",\n+          "臺灣",\n+          "115年7月23日",\n+          "115年9月25日至115年9月28日",\n+          "9月24日至9月29日",\n+          "7月27日"\n+        ],\n+        "severity": "routine",\n+        "taiwan_priority": true,\n+        "military": false,\n+        "sources": [\n+          {\n+            "name": "CAA Taiwan — 115年中秋及教師節連假離島管制航線下週一(7/27)上午9時開放訂位",\n+            "url": "https://www.caa.gov.tw/NewsPublish-Content.aspx?a=381&nid=2567&lang=1"\n+          }\n+        ],\n+        "risk_flags": [],\n+        "article_id": "a-20260726-1748-caa-taiwan-announces-1-340-extra-flights"\n+      }\n+    ],\n+    "international_aviation": [\n+      {\n+        "event_id": "ev-c03d9ba3b69dead0",\n+        "item_type": "new",\n+        "previous_event_id": null,\n+        "category": "aviation_industry",\n+        "headline": "FAA宣布世足賽反無人機行動成果 逾700架無人機被查扣",\n+        "summary": "美國運輸部長肖恩·杜菲宣布，聯邦、州和地方合作夥伴在2026年國際足聯世界盃期間實施了美國歷史上最全面的空域安全和無人機緩解行動，查獲超過700架無人機，確保數百萬球迷的安全。",\n+        "headline_en": "FAA Announces Successful Counter-Drone Efforts at FIFA World Cup 2026, Over 700 Drones Seized",\n+        "summary_en": "The Federal Aviation Administration has announced that the counter-drone efforts undertaken during the FIFA World Cup 2026 were successful in keeping millions of fans safe, with more than 700 drones seized during the tournament. According to Transportation Secretary Sean P. Duffy, federal, state, and local partners carried out the most comprehensive airspace security and drone mitigation effort in the history of the United States, demonstrating an unprecedented level of coordination to protect the event.",\n+        "event_time": null,\n+        "source_published_at": "2026-07-26T18:05:00+00:00",\n+        "source_updated_at": null,\n+        "location": "United States",\n+        "entities": [\n+          "FAA",\n+          "Department of Transportation",\n+          "FIFA World Cup 2026",\n+          "United States",\n+          "2026-07-22",\n+          "FIFA World Cup 2026"\n+        ],\n+        "severity": "routine",\n+        "taiwan_priority": false,\n+        "military": false,\n+        "sources": [\n+          {\n+            "name": "FAA — Successful Counter-Drone Efforts Helped Keep Millions of Fa…",\n+            "url": "https://www.faa.gov/newsroom/successful-counter-drone-efforts-helped-keep-millions-fans-safe-during-fifa-world-cup-2026"\n+          }\n+        ],\n+        "risk_flags": [],\n+        "article_id": "a-20260726-1805-faa-announces-successful-counter-drone-e"\n+      }\n+    ],\n+    "ground_and_maritime": []\n+  },\n+  "checked_sources": [\n+    {\n+      "name": "FAA",\n+      "ok": true,\n+      "last_fetch_utc": "2026-07-26T20:15Z"\n+    },\n+    {\n+      "name": "NTSB",\n+      "ok": true,\n+      "last_fetch_utc": "2026-07-26T20:15Z"\n+    },\n+    {\n+      "name": "ICAO",\n+      "ok": true,\n+      "last_fetch_utc": "2026-07-26T20:15Z"\n+    },\n+    {\n+      "name": "IATA",\n+      "ok": true,\n+      "last_fetch_utc": "2026-07-26T20:15Z"\n+    },\n+    {\n+      "name": "EASA",\n+      "ok": true,\n+      "last_fetch_utc": "2026-07-26T20:15Z"\n+    },\n+    {\n+      "name": "Eurocontrol",\n+      "ok": true,\n+      "last_fetch_utc": "2026-07-26T20:15Z"\n+    },\n+    {\n+      "name": "Reuters",\n+      "ok": true,\n+      "last_fetch_utc": "2026-07-26T20:15Z"\n+    },\n+    {\n+      "name": "The Aviation Herald",\n+      "ok": true,\n+      "last_fetch_utc": "2026-07-26T20:15Z"\n+    },\n+    {\n+      "name": "CAA Taiwan",\n+      "ok": true,\n+      "last_fetch_utc": "2026-07-26T20:15Z"\n+    }\n+  ],\n+  "warnings": [],\n+  "generation_mode": "deterministic",\n+  "generation_model": null\n+}\n@@ -0,0 +1,50 @@\n+{\n+  "events": {\n+    "ev-c03d9ba3b69dead0": {\n+      "fingerprint": "c03d9ba3b69dead0ca70699cf229d43479d3afdd63e563f2f7d14c13446f503c",\n+      "urls": [\n+        "https://www.faa.gov/newsroom/successful-counter-drone-efforts-helped-keep-millions-fans-safe-during-fifa-world-cup-2026"\n+      ],\n+      "section": "international_aviation",\n+      "cat": "reg",\n+      "first_seen": "2026-07-26T20:47:31.942978+00:00",\n+      "last_seen": "2026-07-26T20:47:31.942978+00:00",\n+      "first_briefing_id": "2026-07-27-morning",\n+      "article_ids": [\n+        "a-20260726-1805-faa-announces-successful-counter-drone-e"\n+      ],\n+      "fact_keys": [\n+        "faa announced successful counter-drone efforts during fifa world cup 2026 with more than 700 drones seized.",\n+        "more than 700 drones were seized.",\n+        "the effort helped keep millions of fans safe.",\n+        "transportation secretary sean p. duffy announced federal, state, and local partners carried out the most comprehensive airspace security and drone mitigation effort in u.s. history during fifa world cup 2026."\n+      ],\n+      "updates": {},\n+      "retain_until": "2026-08-25T20:47:31.942978+00:00"\n+    },\n+    "ev-6b68235e9011a3ea": {\n+      "fingerprint": "6b68235e9011a3ea4027e13dfae1b43b3c576095751b9c1751f6ab7fead8abf7",\n+      "urls": [\n+        "https://www.caa.gov.tw/NewsPublish-Content.aspx?a=381&nid=2567&lang=1"\n+      ],\n+      "section": "taiwan_aviation",\n+      "cat": "ops",\n+      "first_seen": "2026-07-26T20:47:31.942978+00:00",\n+      "last_seen": "2026-07-26T20:47:31.942978+00:00",\n+      "first_briefing_id": "2026-07-27-morning",\n+      "article_ids": [\n+        "a-20260726-1748-caa-taiwan-announces-1-340-extra-flights"\n+      ],\n+      "fact_keys": [\n+        "aviation transport period set for six days from september 24 to 29, 2026",\n+        "booking opens july 27, 2026 at 9:00 am",\n+        "caa taiwan announced holiday flight arrangements on july 23, 2026",\n+        "caa will monitor booking and add flights as needed",\n+        "mid-autumn and teachers' day holiday is four days from september 25 to 28, 2026",\n+        "total 1,340 flights and 111,432 seats for penghu, kinmen, and matsu routes"\n+      ],\n+      "updates": {},\n+      "retain_until": "2026-07-29T20:47:31.942978+00:00"\n+    }\n+  }\n+}\n@@ -0,0 +1,13 @@\n+{\n+  "briefings": [\n+    {\n+      "briefing_id": "2026-07-27-morning",\n+      "date": "2026-07-27",\n+      "edition": "morning",\n+      "status": "published",\n+      "cutoff_time": "2026-07-27T07:00:00+08:00",\n+      "generated_at": "2026-07-27T04:48:38.325454+08:00",\n+      "item_count": 2\n+    }\n+  ]\n+}\n@@ -0,0 +1,815 @@\n+"""briefing.py — thrice-daily air & transport briefing assembler.\n+\n+Selects already-verified articles (data/articles/) whose source time falls\n+inside a fixed 8-hour Asia/Taipei window, deduplicates events across\n+editions via a persistent event index, and writes one briefing JSON per\n+edition to data/briefings/. Assembly is deterministic by default: verified\n+headlines and summaries are reused as-is and no model is called. At most\n+one optional batch LLM call per edition (BRIEFING_LLM_INTRO=true) may add\n+an intro paragraph; any failure falls back to the deterministic output.\n+\n+Never fetches the network for content and never bypasses the editorial\n+gate: rejected groups never reach data/articles/, and manual_review items\n+stay in data/review.json until a human approves them, so reading only the\n+published article store enforces both exclusions structurally.\n+\n+Editions (config/briefing_editions.json is the single source of truth):\n+  morning    23:00 (previous day) <= source_time < 07:00, runs 07:15 TPE\n+  afternoon  07:00               <= source_time < 15:00, runs 15:15 TPE\n+  evening    15:00               <= source_time < 23:00, runs 23:15 TPE\n+Windows are half-open and fixed: a delayed run never widens or shifts them.\n+"""\n+from __future__ import annotations\n+\n+import argparse\n+import hashlib\n+import json\n+import os\n+import re\n+import sys\n+import traceback\n+from dataclasses import dataclass\n+from datetime import date, datetime, time, timedelta, timezone\n+from pathlib import Path\n+\n+sys.path.insert(0, str(Path(__file__).resolve().parent))\n+from common import (  # noqa: E402\n+    ARTICLES_DIR,\n+    DATA_DIR,\n+    load_json,\n+    norm_url,\n+    now_utc,\n+    parse_iso,\n+    save_json,\n+    squash_text,\n+)\n+\n+# Asia/Taipei has had a fixed +08:00 offset with no DST since 1980, so the\n+# static fallback (Windows without tzdata) is exact.\n+try:\n+    from zoneinfo import ZoneInfo\n+\n+    TPE = ZoneInfo("Asia/Taipei")\n+except Exception:  # pragma: no cover\n+    TPE = timezone(timedelta(hours=8), "UTC+8")\n+\n+ROOT = Path(__file__).resolve().parent.parent\n+EDITIONS_PATH = ROOT / "config" / "briefing_editions.json"\n+BRIEFINGS_DIR = DATA_DIR / "briefings"\n+INDEX_PATH = BRIEFINGS_DIR / "index.json"\n+EVENT_INDEX_PATH = BRIEFINGS_DIR / "event_index.json"\n+\n+# GitHub Actions cron (UTC) -> edition. briefing.yml must use the same\n+# mapping; test_briefing.py cross-checks both against the edition config.\n+CRON_TO_EDITION = {\n+    "15 23 * * *": "morning",\n+    "15 7 * * *": "afternoon",\n+    "15 15 * * *": "evening",\n+}\n+\n+SECTIONS = ("aviation_incidents", "taiwan_aviation",\n+            "international_aviation", "ground_and_maritime")\n+\n+# Event-index retention: incidents/investigations/policy keep 30 days for\n+# update matching; everything else 72 hours (spec minimum).\n+RETAIN_HOURS_DEFAULT = 72\n+RETAIN_HOURS_LONG = 30 * 24\n+\n+_SEVERITY_RANK = {"fatal": 3, "serious": 2, "significant": 1, "routine": 0}\n+\n+\n+def _int_setting(name: str, default: int, low: int, high: int) -> int:\n+    raw = os.environ.get(name)\n+    if raw in (None, ""):\n+        return default\n+    try:\n+        value = int(str(raw).strip())\n+    except (TypeError, ValueError):\n+        print(f"config: {name}={raw!r} is not an integer; using {default}")\n+        return default\n+    if not low <= value <= high:\n+        print(f"config: {name}={value} outside [{low}, {high}]; "\n+              f"using {default}")\n+        return default\n+    return value\n+\n+\n+MAX_ITEMS = _int_setting("BRIEFING_MAX_ITEMS", 15, 1, 50)\n+MAX_ITEMS_PER_SECTION = _int_setting("BRIEFING_MAX_ITEMS_PER_SECTION",\n+                                     6, 1, 20)\n+EXTRA_LLM_CALLS_MAX = _int_setting("BRIEFING_EXTRA_LLM_CALLS_MAX", 1, 0, 3)\n+\n+\n+def _llm_intro_enabled() -> bool:\n+    return (os.environ.get("BRIEFING_LLM_INTRO", "").strip().lower()\n+            in ("1", "true", "yes"))\n+\n+\n+def load_editions() -> dict:\n+    editions = load_json(EDITIONS_PATH, None)\n+    if not isinstance(editions, dict) or set(editions) != {\n+            "morning", "afternoon", "evening"}:\n+        raise SystemExit(f"briefing: bad or missing {EDITIONS_PATH}")\n+    return editions\n+\n+\n+# ── fixed window ─────────────────────────────────────────────────────────────\n+\n+@dataclass\n+class BriefingWindow:\n+    edition: str\n+    date: date  # Taipei calendar date the edition belongs to\n+    window_start: datetime  # aware, Asia/Taipei\n+    window_end: datetime\n+    cutoff_time: datetime\n+    scheduled_time: datetime\n+    timezone: str = "Asia/Taipei"\n+\n+    @property\n+    def briefing_id(self) -> str:\n+        return f"{self.date.isoformat()}-{self.edition}"\n+\n+    def contains(self, dt: datetime) -> bool:\n+        """Half-open [window_start, window_end) — compared in UTC."""\n+        if dt.tzinfo is None:\n+            raise ValueError("briefing window comparisons need aware datetimes")\n+        t = dt.astimezone(timezone.utc)\n+        return (self.window_start.astimezone(timezone.utc) <= t\n+                < self.window_end.astimezone(timezone.utc))\n+\n+\n+def get_briefing_window(edition: str, taipei_date: date) -> BriefingWindow:\n+    """Fixed 8-hour window for one edition on one Taipei date.\n+\n+    Derived only from the edition config and the date — never from the\n+    clock — so a delayed Actions run keeps the exact same window.\n+    """\n+    cfg = load_editions().get(edition)\n+    if cfg is None:\n+        raise ValueError(f"unknown edition {edition!r}")\n+    start_day = taipei_date + timedelta(days=int(cfg["window_start_day_offset"]))\n+    return BriefingWindow(\n+        edition=edition,\n+        date=taipei_date,\n+        window_start=datetime.combine(\n+            start_day, time(int(cfg["window_start_hour"])), tzinfo=TPE),\n+        window_end=datetime.combine(\n+            taipei_date, time(int(cfg["window_end_hour"])), tzinfo=TPE),\n+        cutoff_time=datetime.combine(\n+            taipei_date, time(int(cfg["cutoff_hour"])), tzinfo=TPE),\n+        scheduled_time=datetime.combine(\n+            taipei_date,\n+            time(int(cfg["scheduled_hour"]), int(cfg["scheduled_minute"])),\n+            tzinfo=TPE),\n+    )\n+\n+\n+# ── article classification (deterministic, from existing metadata) ──────────\n+\n+_TAIWAN_MARKERS = (\n+    "民航局", "臺灣", "台灣", "桃園機場", "桃園國際機場", "松山機場", "小港機場",\n+    "澎湖", "金門", "馬祖", "中華航空", "華航", "長榮航空", "星宇航空",\n+    "台灣虎航", "臺灣虎航", "立榮航空", "華信航空", "taiwan", "taoyuan",\n+    "caa taiwan", "china airlines", "eva air", "starlux", "tigerair taiwan",\n+    "uni air", "mandarin airlines",\n+)\n+_GROUND_MARKERS = (\n+    "台鐵", "臺鐵", "高鐵", "捷運", "輕軌", "鐵路", "國道客運", "公路重大",\n+    "railway", "high-speed rail", "metro line", "light rail", "train derail",\n+)\n+_MARITIME_MARKERS = (\n+    "港務", "航港", "海運", "商港", "貨輪", "客輪", "渡輪", "船舶",\n+    "maritime", "ferry", "cargo ship", "container ship", "port authority",\n+    "vessel",\n+)\n+_MILITARY_MARKERS = (\n+    "軍機", "國軍", "空軍", "軍用機", "military aircraft", "air force",\n+    "fighter jet",\n+)\n+_FATAL_MARKERS = (\n+    "罹難", "死亡", "喪生", "身亡", "fatal", "killed", "fatalities", "died",\n+)\n+_SERIOUS_MARKERS = (\n+    "墜毀", "墜機", "失事", "失聯", "劫機", "迫降", "crash", "crashed",\n+    "hull loss", "mayday", "hijack", "emergency landing",\n+)\n+\n+_IDENT_RE = re.compile(\n+    r"\b(?:[A-Z]{2}\d{2,4}|B-[A-Z0-9]{4,5}|N\d{2,5}[A-Z]{0,2}"\n+    r"|9M-[A-Z]{3}|JA\d{3,4}[A-Z]?)\b")\n+\n+\n+def _entity_values(article: dict) -> list[str]:\n+    ents = article.get("entities") or {}\n+    values: list[str] = []\n+    if isinstance(ents, dict):\n+        for group in ents.values():\n+            if isinstance(group, list):\n+                values.extend(str(v) for v in group if v)\n+    return values\n+\n+\n+def _text_blob(article: dict) -> str:\n+    parts = []\n+    for lang in ("zh", "en"):\n+        block = article.get(lang) or {}\n+        parts.append(str(block.get("title") or ""))\n+        parts.append(str(block.get("summary") or ""))\n+    for fact in article.get("facts") or []:\n+        if isinstance(fact, dict):\n+            parts.append(str(fact.get("claim") or ""))\n+    parts.extend(_entity_values(article))\n+    return squash_text(" ".join(parts))\n+\n+\n+def _has_marker(blob: str, markers: tuple[str, ...]) -> bool:\n+    return any(m.casefold() in blob for m in markers)\n+\n+\n+def classify_section(article: dict) -> str:\n+    blob = _text_blob(article)\n+    if _has_marker(blob, _GROUND_MARKERS) or _has_marker(blob, _MARITIME_MARKERS):\n+        return "ground_and_maritime"\n+    if article.get("cat") == "safety":\n+        return "aviation_incidents"\n+    if _has_marker(blob, _TAIWAN_MARKERS):\n+        return "taiwan_aviation"\n+    return "international_aviation"\n+\n+\n+def item_category(article: dict, section: str) -> str:\n+    if section == "aviation_incidents":\n+        return "aviation_incident"\n+    if section == "taiwan_aviation":\n+        return "taiwan_aviation"\n+    if section == "ground_and_maritime":\n+        blob = _text_blob(article)\n+        return ("maritime" if _has_marker(blob, _MARITIME_MARKERS)\n+                else "ground_transport")\n+    return "aviation_industry"\n+\n+\n+def classify_severity(article: dict) -> str:\n+    blob = _text_blob(article)\n+    if _has_marker(blob, _FATAL_MARKERS):\n+        return "fatal"\n+    if _has_marker(blob, _SERIOUS_MARKERS):\n+        return "serious"\n+    if article.get("cat") == "safety":\n+        return "significant"\n+    return "routine"\n+\n+\n+# ── event identity & dedup index ─────────────────────────────────────────────\n+\n+def article_source_time(article: dict) -> datetime | None:\n+    """Source-verified time when present, else our verified-publish time.\n+\n+    Half-open windows partition all time, so every article lands in exactly\n+    one edition either way; the fallback is our own honest timestamp, never\n+    an invented source time.\n+    """\n+    for key in ("sourcePublishedUtc", "publishedUtc"):\n+        raw = article.get(key)\n+        if raw:\n+            try:\n+                return parse_iso(str(raw))\n+            except ValueError:\n+                continue\n+    return None\n+\n+\n+def _canonical_urls(article: dict) -> list[str]:\n+    urls = []\n+    for src in article.get("sources") or []:\n+        url = (src or {}).get("url")\n+        if url:\n+            urls.append(norm_url(str(url)))\n+    return sorted(set(urls))\n+\n+\n+def _fact_keys(article: dict) -> list[str]:\n+    keys = set()\n+    for fact in article.get("facts") or []:\n+        if isinstance(fact, dict) and fact.get("claim"):\n+            keys.add(squash_text(fact["claim"]))\n+    return sorted(keys)\n+\n+\n+def event_fingerprint(article: dict, section: str) -> str:\n+    """Stable identity for one real-world event.\n+\n+    Built from category, core entities, locations, vehicle identifiers and\n+    the event date — never from edition, AI headline, generated_at or list\n+    position, so the id survives retitles and later editions. Canonical\n+    source URLs are deliberately NOT hashed (a reprint has a new URL but is\n+    the same event); they are kept in the index as alternate match keys.\n+    """\n+    ents = article.get("entities") or {}\n+\n+    def _vals(key):\n+        vals = ents.get(key) if isinstance(ents, dict) else None\n+        return sorted({squash_text(v) for v in vals or [] if v})[:6]\n+\n+    src_time = article_source_time(article)\n+    idents = sorted(set(_IDENT_RE.findall(\n+        " ".join([str((article.get("en") or {}).get("title") or "")]\n+                 + [str(f.get("claim") or "") for f in article.get("facts") or []\n+                    if isinstance(f, dict)]))))\n+    core = {\n+        "section": section,\n+        "organizations": _vals("organizations"),\n+        "locations": _vals("locations"),\n+        "idents": idents,\n+        "event_date": (src_time.astimezone(TPE).date().isoformat()\n+                       if src_time else None),\n+        "event_type": article.get("cat"),\n+    }\n+    return hashlib.sha256(\n+        json.dumps(core, sort_keys=True, ensure_ascii=False).encode("utf-8")\n+    ).hexdigest()\n+\n+\n+def load_event_index() -> dict:\n+    data = load_json(EVENT_INDEX_PATH, {})\n+    events = data.get("events")\n+    return {"events": events if isinstance(events, dict) else {}}\n+\n+\n+def prune_event_index(index: dict, now: datetime) -> None:\n+    events = index["events"]\n+    for event_id in list(events):\n+        raw = events[event_id].get("retain_until")\n+        try:\n+            if raw and parse_iso(raw) < now:\n+                del events[event_id]\n+        except ValueError:\n+            del events[event_id]\n+\n+\n+def _find_event(index: dict, event_id: str, urls: list[str]):\n+    events = index["events"]\n+    if event_id in events:\n+        return event_id, events[event_id]\n+    url_set = set(urls)\n+    for eid, event in events.items():\n+        if url_set & set(event.get("urls") or []):\n+            return eid, event\n+    return None, None\n+\n+\n+def _retain_until(section: str, cat, now: datetime) -> str:\n+    hours = (RETAIN_HOURS_LONG\n+             if section == "aviation_incidents" or cat == "reg"\n+             else RETAIN_HOURS_DEFAULT)\n+    return (now + timedelta(hours=hours)).isoformat()\n+\n+\n+# ── selection & assembly ─────────────────────────────────────────────────────\n+\n+def load_published_articles() -> tuple[list[dict], list[str]]:\n+    """All published articles + warnings for unreadable batch files."""\n+    articles, warnings = [], []\n+    if not ARTICLES_DIR.is_dir():\n+        return articles, warnings\n+    for path in sorted(ARTICLES_DIR.glob("*.json")):\n+        try:\n+            with open(path, encoding="utf-8") as fh:\n+                batch = json.load(fh)\n+        except (OSError, json.JSONDecodeError):\n+            warnings.append(f"article batch unreadable: {path.name}")\n+            continue\n+        for article in batch.get("articles") or []:\n+            if isinstance(article, dict):\n+                articles.append(article)\n+    return articles, warnings\n+\n+\n+def _eligible(article: dict) -> bool:\n+    zh = article.get("zh") or {}\n+    return bool(str(zh.get("title") or "").strip()\n+                and str(zh.get("summary") or "").strip()\n+                and _canonical_urls(article))\n+\n+\n+def _build_item(article: dict, section: str, event_id: str,\n+                item_type: str, previous_event_id, new_information):\n+    zh = article.get("zh") or {}\n+    en = article.get("en") or {}\n+    src_time = article_source_time(article)\n+    ents = article.get("entities") or {}\n+    locations = [str(v) for v in (ents.get("locations") or [])\n+                 if v] if isinstance(ents, dict) else []\n+    blob = _text_blob(article)\n+    item = {\n+        "event_id": event_id,\n+        "item_type": item_type,\n+        "previous_event_id": previous_event_id,\n+        "category": item_category(article, section),\n+        "headline": str(zh.get("title") or "").strip(),\n+        "summary": str(zh.get("summary") or "").strip(),\n+        "headline_en": str(en.get("title") or "").strip(),\n+        "summary_en": str(en.get("summary") or "").strip(),\n+        "event_time": None,\n+        "source_published_at": src_time.isoformat() if src_time else None,\n+        "source_updated_at": None,\n+        "location": locations[0] if locations else None,\n+        "entities": _entity_values(article)[:10],\n+        "severity": classify_severity(article),\n+        "taiwan_priority": _has_marker(blob, _TAIWAN_MARKERS),\n+        "military": _has_marker(blob, _MILITARY_MARKERS),\n+        "sources": [{"name": str((s or {}).get("name") or ""),\n+                     "url": str((s or {}).get("url") or "")}\n+                    for s in article.get("sources") or []\n+                    if (s or {}).get("url")],\n+        "risk_flags": list(article.get("riskFlags") or []),\n+        "article_id": article.get("id"),\n+    }\n+    if item_type == "update":\n+        item["new_information"] = list(new_information or [])\n+    return item\n+\n+\n+def select_items(articles: list[dict], window: BriefingWindow,\n+                 event_index: dict, now: datetime,\n+                 warnings: list[str]) -> list[tuple[str, dict]]:\n+    """(section, item) pairs for this window; mutates event_index."""\n+    events = event_index["events"]\n+    briefing_id = window.briefing_id\n+    in_window = []\n+    for article in articles:\n+        src_time = article_source_time(article)\n+        if src_time is None or not window.contains(src_time):\n+            continue\n+        if not _eligible(article):\n+            warnings.append(\n+                f"article {article.get('id')} skipped: missing verified "\n+                f"headline/summary or source URL")\n+            continue\n+        in_window.append((src_time, article))\n+    in_window.sort(key=lambda pair: pair[0], reverse=True)\n+\n+    picked: list[tuple[str, dict]] = []\n+    emitted: set[str] = set()\n+    for src_time, article in in_window:\n+        section = classify_section(article)\n+        fingerprint = event_fingerprint(article, section)\n+        event_id = "ev-" + fingerprint[:16]\n+        urls = _canonical_urls(article)\n+        facts = _fact_keys(article)\n+        matched_id, event = _find_event(event_index, event_id, urls)\n+        if matched_id:\n+            event_id = matched_id\n+\n+        if event_id in emitted:\n+            continue  # one item per event per edition\n+\n+        if event is None:\n+            events[event_id] = {\n+                "fingerprint": fingerprint,\n+                "urls": urls,\n+                "section": section,\n+                "cat": article.get("cat"),\n+                "first_seen": now.isoformat(),\n+                "last_seen": now.isoformat(),\n+                "first_briefing_id": briefing_id,\n+                "article_ids": [article.get("id")],\n+                "fact_keys": facts,\n+                "updates": {},\n+                "retain_until": _retain_until(section, article.get("cat"), now),\n+            }\n+            picked.append((section, _build_item(\n+                article, section, event_id, "new", None, None)))\n+            emitted.add(event_id)\n+            continue\n+\n+        if event.get("first_briefing_id") == briefing_id:\n+            # Re-run of the same edition: rebuild the same item.\n+            picked.append((section, _build_item(\n+                article, section, event_id, "new", None, None)))\n+            emitted.add(event_id)\n+            continue\n+\n+        prior_updates = event.get("updates") or {}\n+        if briefing_id in prior_updates:\n+            # Re-run: the update was already accepted for this edition.\n+            picked.append((section, _build_item(\n+                article, section, event_id, "update", event_id,\n+                prior_updates[briefing_id].get("new_information"))))\n+            emitted.add(event_id)\n+            continue\n+\n+        new_facts = [f for f in facts if f not in set(event.get("fact_keys") or [])]\n+        if not new_facts:\n+            # Same event, nothing substantive: retitles/reprints are not\n+            # updates. Remember the extra URLs so later reprints match too.\n+            event["urls"] = sorted(set(event.get("urls") or []) | set(urls))\n+            event["last_seen"] = now.isoformat()\n+            continue\n+\n+        original_claims = [\n+            str(f.get("claim")) for f in article.get("facts") or []\n+            if isinstance(f, dict) and f.get("claim")\n+            and squash_text(f["claim"]) in set(new_facts)]\n+        event["fact_keys"] = sorted(set(event.get("fact_keys") or [])\n+                                    | set(facts))\n+        event["urls"] = sorted(set(event.get("urls") or []) | set(urls))\n+        event["last_seen"] = now.isoformat()\n+        event.setdefault("article_ids", []).append(article.get("id"))\n+        event.setdefault("updates", {})[briefing_id] = {\n+            "article_id": article.get("id"),\n+            "new_information": original_claims,\n+        }\n+        picked.append((section, _build_item(\n+            article, section, event_id, "update", event_id, original_claims)))\n+        emitted.add(event_id)\n+    return picked\n+\n+\n+def arrange_sections(picked: list[tuple[str, dict]],\n+                     warnings: list[str]) -> dict:\n+    sections = {name: [] for name in SECTIONS}\n+    for section, item in picked:\n+        sections[section].append(item)\n+\n+    def sort_key(item):\n+        return (-_SEVERITY_RANK.get(item.get("severity"), 0),\n+                not item.get("taiwan_priority"),\n+                item.get("source_published_at") or "")\n+\n+    dropped = 0\n+    for name in SECTIONS:\n+        rows = sorted(sections[name], key=sort_key)\n+        if len(rows) > MAX_ITEMS_PER_SECTION:\n+            dropped += len(rows) - MAX_ITEMS_PER_SECTION\n+            rows = rows[:MAX_ITEMS_PER_SECTION]\n+        sections[name] = rows\n+\n+    total = sum(len(v) for v in sections.values())\n+    if total > MAX_ITEMS:\n+        overflow = total - MAX_ITEMS\n+        dropped += overflow\n+        # Trim from the least critical end of the section priority order.\n+        for name in reversed(SECTIONS):\n+            while overflow and sections[name]:\n+                sections[name].pop()\n+                overflow -= 1\n+    if dropped:\n+        warnings.append(f"{dropped} item(s) dropped by "\n+                        f"BRIEFING_MAX_ITEMS/_PER_SECTION caps")\n+    return sections\n+\n+\n+# ── optional single batch LLM intro ──────────────────────────────────────────\n+\n+INTRO_SCHEMA = {\n+    "type": "object",\n+    "properties": {\n+        "introZh": {"type": "string"},\n+        "introEn": {"type": "string"},\n+    },\n+    "required": ["introZh", "introEn"],\n+    "additionalProperties": False,\n+}\n+\n+INTRO_SYSTEM_PROMPT = (\n+    "You write a 1-2 sentence neutral intro for a transport-news briefing. "\n+    "Use ONLY the items given; add no outside facts, numbers, names or "\n+    "speculation. Return JSON {\"introZh\": \"...\", \"introEn\": \"...\"} — "\n+    "Traditional Chinese (Taiwan usage) and English. Each 30-160 characters."\n+)\n+\n+\n+def _intro_digest(sections: dict) -> str:\n+    lines = []\n+    for name in SECTIONS:\n+        for item in sections[name]:\n+            lines.append(f"- [{name}] {item['headline']} — {item['summary']}")\n+    return "\n".join(lines)\n+\n+\n+def maybe_llm_intro(briefing: dict, sections: dict,\n+                    warnings: list[str]) -> None:\n+    """At most EXTRA_LLM_CALLS_MAX (default 1) batch calls per edition.\n+\n+    Zero items or the flag off → zero calls. Any failure leaves the\n+    deterministic briefing untouched.\n+    """\n+    total = sum(len(v) for v in sections.values())\n+    if not _llm_intro_enabled() or total == 0 or EXTRA_LLM_CALLS_MAX <= 0:\n+        return\n+    try:\n+        from providers import build_providers\n+\n+        providers = [p for p in build_providers() if p.available()]\n+    except Exception as exc:  # provider layer unusable → deterministic\n+        warnings.append(f"intro skipped: providers unavailable ({exc})")\n+        return\n+    calls = 0\n+    for provider in providers:\n+        if calls >= EXTRA_LLM_CALLS_MAX:\n+            break\n+        calls += 1\n+        try:\n+            draft = provider.draft(INTRO_SYSTEM_PROMPT,\n+                                   _intro_digest(sections), INTRO_SCHEMA)\n+        except Exception as exc:\n+            warnings.append(f"intro call failed on {provider.name}: "\n+                            f"{type(exc).__name__}")\n+            continue\n+        if not isinstance(draft, dict):\n+            warnings.append(f"intro refused/empty on {provider.name}")\n+            continue\n+        intro_zh = str(draft.get("introZh") or "").strip()\n+        intro_en = str(draft.get("introEn") or "").strip()\n+        if not (10 <= len(intro_zh) <= 300 and 10 <= len(intro_en) <= 400):\n+            warnings.append("intro discarded: length outside bounds")\n+            continue\n+        briefing["intro_zh"] = intro_zh\n+        briefing["intro_en"] = intro_en\n+        briefing["generation_mode"] = "llm_assisted"\n+        briefing["generation_model"] = {"provider": provider.name,\n+                                        "model": provider.model}\n+        return\n+    # all attempts failed → deterministic output stands\n+\n+\n+# ── assembly, persistence ────────────────────────────────────────────────────\n+\n+def checked_sources_snapshot() -> tuple[list[dict], list[str]]:\n+    rows, warnings = [], []\n+    for src in load_json(DATA_DIR / "sources.json", []) or []:\n+        if not isinstance(src, dict):\n+            continue\n+        if src.get("kind") == "data":\n+            # Data platforms (e.g. the optional FlightAware stats API)\n+            # never feed news candidates; their status is not briefing\n+            # coverage.\n+            continue\n+        row = {"name": src.get("name"), "ok": bool(src.get("ok")),\n+               "last_fetch_utc": src.get("lastFetchUtc")}\n+        rows.append(row)\n+        if not row["ok"]:\n+            warnings.append(f"source fetch failing: {row['name']} — a fetch "\n+                            f"failure is a coverage gap, not an absence of "\n+                            f"events")\n+    return rows, warnings\n+\n+\n+def assemble_briefing(window: BriefingWindow, sections: dict,\n+                      checked_sources: list[dict], warnings: list[str],\n+                      generated_at: datetime, partial: bool) -> dict:\n+    cfg = load_editions()[window.edition]\n+    total = sum(len(v) for v in sections.values())\n+    return {\n+        "content_type": "daily_transport_briefing",\n+        "briefing_id": window.briefing_id,\n+        "date": window.date.isoformat(),\n+        "edition": window.edition,\n+        "edition_label": cfg["label_zh_tw"],\n+        "timezone": window.timezone,\n+        "window_start": window.window_start.isoformat(),\n+        "window_end": window.window_end.isoformat(),\n+        "cutoff_time": window.cutoff_time.isoformat(),\n+        "scheduled_time": window.scheduled_time.isoformat(),\n+        "generated_at": generated_at.astimezone(TPE).isoformat(),\n+        "status": "partial" if partial else "published",\n+        "item_count": total,\n+        "sections": sections,\n+        "checked_sources": checked_sources,\n+        "warnings": warnings,\n+        "generation_mode": "deterministic",\n+        "generation_model": None,\n+    }\n+\n+\n+def _briefing_changed(old, new) -> bool:\n+    if not isinstance(old, dict):\n+        return True\n+    ignore = ("generated_at",)\n+    a = {k: v for k, v in old.items() if k not in ignore}\n+    b = {k: v for k, v in new.items() if k not in ignore}\n+    return a != b\n+\n+\n+def upsert_index_row(row: dict) -> None:\n+    index = load_json(INDEX_PATH, {"briefings": []})\n+    rows = [r for r in index.get("briefings") or []\n+            if isinstance(r, dict)\n+            and r.get("briefing_id") != row["briefing_id"]]\n+    rows.append(row)\n+    rows.sort(key=lambda r: (str(r.get("date") or ""),\n+                             str(r.get("cutoff_time") or "")), reverse=True)\n+    new = {"briefings": rows}\n+    if new != index:\n+        save_json(INDEX_PATH, new)\n+\n+\n+def run_edition(edition: str, taipei_date: date) -> int:\n+    now = now_utc()\n+    window = get_briefing_window(edition, taipei_date)\n+    warnings: list[str] = []\n+\n+    articles, load_warnings = load_published_articles()\n+    warnings.extend(load_warnings)\n+    checked_sources, source_warnings = checked_sources_snapshot()\n+    warnings.extend(source_warnings)\n+\n+    event_index = load_event_index()\n+    prune_event_index(event_index, now)\n+    picked = select_items(articles, window, event_index, now, warnings)\n+    sections = arrange_sections(picked, warnings)\n+\n+    # partial = our own inputs were degraded, never "no events".\n+    partial = bool(load_warnings or source_warnings)\n+    briefing = assemble_briefing(window, sections, checked_sources,\n+                                 warnings, now, partial)\n+    maybe_llm_intro(briefing, sections, warnings)\n+\n+    path = BRIEFINGS_DIR / f"{window.briefing_id}.json"\n+    old = load_json(path, None)\n+    if _briefing_changed(old, briefing):\n+        save_json(path, briefing)\n+        save_json(EVENT_INDEX_PATH, event_index)\n+        changed = "written"\n+    else:\n+        changed = "unchanged"\n+\n+    upsert_index_row({\n+        "briefing_id": window.briefing_id,\n+        "date": window.date.isoformat(),\n+        "edition": edition,\n+        "status": briefing["status"],\n+        "cutoff_time": briefing["cutoff_time"],\n+        "generated_at": briefing["generated_at"],\n+        "item_count": briefing["item_count"],\n+    })\n+    print(f"briefing: {window.briefing_id} status={briefing['status']} "\n+          f"items={briefing['item_count']} mode={briefing['generation_mode']} "\n+          f"warnings={len(warnings)} ({changed})")\n+    return 0\n+\n+\n+def record_failure(edition: str, taipei_date: date, error: str) -> None:\n+    """A failed run never touches briefing files, and never downgrades an\n+    edition that already succeeded (a crashing re-run must not hide the\n+    previously published briefing)."""\n+    briefing_id = f"{taipei_date.isoformat()}-{edition}"\n+    for row in load_json(INDEX_PATH, {}).get("briefings") or []:\n+        if (isinstance(row, dict) and row.get("briefing_id") == briefing_id\n+                and row.get("status") in ("published", "partial")):\n+            print(f"briefing: {briefing_id} already succeeded earlier; "\n+                  f"failure not recorded over it")\n+            return\n+    upsert_index_row({\n+        "briefing_id": briefing_id,\n+        "date": taipei_date.isoformat(),\n+        "edition": edition,\n+        "status": "failed",\n+        "cutoff_time": None,\n+        "generated_at": now_utc().astimezone(TPE).isoformat(),\n+        "item_count": 0,\n+        "error": error[:300],\n+    })\n+\n+\n+def main(argv=None) -> int:\n+    parser = argparse.ArgumentParser(description=__doc__)\n+    parser.add_argument("--edition", choices=sorted(CRON_TO_EDITION.values()))\n+    parser.add_argument("--cron", help="GitHub Actions schedule string; "\n+                                       "mapped via CRON_TO_EDITION")\n+    parser.add_argument("--date", help="Taipei date YYYY-MM-DD "\n+                                       "(default: today in Asia/Taipei)")\n+    args = parser.parse_args(argv)\n+\n+    edition = args.edition\n+    if not edition and args.cron:\n+        edition = CRON_TO_EDITION.get(args.cron.strip())\n+        if not edition:\n+            print(f"briefing: unknown cron {args.cron!r}")\n+            return 1\n+    if not edition:\n+        print("briefing: --edition or --cron required")\n+        return 1\n+\n+    if args.date:\n+        try:\n+            taipei_date = date.fromisoformat(args.date.strip())\n+        except ValueError:\n+            print(f"briefing: bad --date {args.date!r} (want YYYY-MM-DD)")\n+            return 1\n+    else:\n+        taipei_date = now_utc().astimezone(TPE).date()\n+\n+    try:\n+        return run_edition(edition, taipei_date)\n+    except Exception as exc:  # never nuke previous briefings on a crash\n+        traceback.print_exc()\n+        record_failure(edition, taipei_date, f"{type(exc).__name__}: {exc}")\n+        print(f"briefing: FAILED {taipei_date}-{edition}; previous "\n+              f"briefings left untouched")\n+        return 0\n+\n+\n+if __name__ == "__main__":\n+    sys.exit(main())\n@@ -80,8 +80,21 @@\n         "aboutKicker": "Methodology", "aboutTitle": "方法論",\n         "footerAbout": "全自動航空新聞聚合站。GitHub Actions 每小時抓取各可信來源，自動撰寫並發布，每篇文末標註原始出處。本頁為設計原型，內容為示意樣本。",\n         "footerSources": "資料來源 Data Sources",\n-        "nav": ["最新", "事故資料庫", "來源", "方法論"],\n+        "nav": ["最新", "快報", "事故資料庫", "來源", "方法論"],\n         "cats": ["全部", "事故", "法規", "商業", "營運"],\n+        # daily briefings\n+        "brKicker": "Daily Briefing", "brIndexTitle": "快報",\n+        "brIndexSub": "每日三次（臺北時間 07:15／15:15／23:15）自動彙整固定八小時窗口內、已通過驗證發布的航空與交通事件。",\n+        "brLatest": "最新快報", "brWindow": "資料範圍", "brCutoff": "資料截止",\n+        "brGenerated": "系統整理完成", "brTo": "至",\n+        "brEmpty": "截至本期資料截止時間，系統在本次查核的指定來源中，未發現符合收錄門檻的新事件。",\n+        "brPartial": "本期部分來源資料未能完整取得，內容可能不完整；來源擷取失敗不代表沒有新事件。",\n+        "brChecked": "本期查核來源", "brWarnings": "資料覆蓋提示",\n+        "brUpdate": "更新", "brItems": "則", "brNoItems": "本分類本期無收錄項目",\n+        "brModeDet": "程式化組裝（未使用模型）", "brModeLlm": "模型輔助導言",\n+        "brSecs": ["航空事故與事件", "臺灣航空", "國際航空", "陸運與海運"],\n+        "brSevs": {"fatal": "致命", "serious": "嚴重",\n+                   "significant": "重要", "routine": "一般"},\n         "sevs": ["全部", "事故", "嚴重事件", "事件"],\n         "themeOpts": ["亮", "暗"],\n         # production-only keys (not in the design dictionary)\n@@ -119,8 +132,24 @@\n         "aboutKicker": "Methodology", "aboutTitle": "Methodology",\n         "footerAbout": "A fully automated aviation news aggregator. GitHub Actions fetches trusted sources hourly, writes and publishes automatically, and credits originals at the end of every article. Design prototype; sample content.",\n         "footerSources": "Data Sources",\n-        "nav": ["Latest", "Incident DB", "Sources", "Methodology"],\n+        "nav": ["Latest", "Briefings", "Incident DB", "Sources", "Methodology"],\n         "cats": ["All", "Safety", "Regulation", "Business", "Operations"],\n+        # daily briefings\n+        "brKicker": "Daily Briefing", "brIndexTitle": "Briefings",\n+        "brIndexSub": "Three times daily (07:15 / 15:15 / 23:15 Taipei time): verified aviation and transport events from a fixed eight-hour window.",\n+        "brLatest": "Latest briefing", "brWindow": "Data window",\n+        "brCutoff": "Data cutoff", "brGenerated": "Compiled", "brTo": "to",\n+        "brEmpty": "As of this edition's data cutoff, no new events meeting the inclusion bar were found in the sources checked for this edition.",\n+        "brPartial": "Some sources could not be fetched in full for this edition; coverage may be incomplete. A fetch failure does not mean no events occurred.",\n+        "brChecked": "Sources checked", "brWarnings": "Coverage notes",\n+        "brUpdate": "UPDATE", "brItems": "items",\n+        "brNoItems": "No items in this section this edition",\n+        "brModeDet": "Deterministic assembly (no model used)",\n+        "brModeLlm": "Model-assisted intro",\n+        "brSecs": ["Aviation incidents", "Taiwan aviation",\n+                   "International aviation", "Ground & maritime"],\n+        "brSevs": {"fatal": "Fatal", "serious": "Serious",\n+                   "significant": "Significant", "routine": "Routine"},\n         "sevs": ["All", "Accident", "Serious", "Incident"],\n         "themeOpts": ["Light", "Dark"],\n         # production-only keys (not in the design dictionary)\n@@ -931,6 +960,152 @@ def sources_rows_view(sources, lang: str, now):\n     } for s in sources]\n \n \n+# ── daily briefings ──────────────────────────────────────────────────────────\n+\n+BRIEFING_SECTIONS = ("aviation_incidents", "taiwan_aviation",\n+                     "international_aviation", "ground_and_maritime")\n+_BRIEF_EDITIONS = load_json(\n+    Path(__file__).resolve().parent.parent / "config"\n+    / "briefing_editions.json", {})\n+_EN_MONTHS = ("Jan", "Feb", "Mar", "Apr", "May", "Jun",\n+              "Jul", "Aug", "Sep", "Oct", "Nov", "Dec")\n+_BRIEF_SEV_CLS = {"fatal": "tag-accent", "serious": "tag-accent",\n+                  "significant": "tag-outline", "routine": "tag-neutral"}\n+\n+\n+def _tpe_dt(iso):\n+    try:\n+        return parse_iso(str(iso)).astimezone(TPE)\n+    except (ValueError, TypeError):\n+        return None\n+\n+\n+def _fmt_tpe_date(dt, lang: str) -> str:\n+    if lang == "zh":\n+        return f"{dt.year} 年 {dt.month} 月 {dt.day} 日"\n+    return f"{_EN_MONTHS[dt.month - 1]} {dt.day}, {dt.year}"\n+\n+\n+def _fmt_tpe(iso, lang: str, time_only=False) -> str:\n+    dt = _tpe_dt(iso)\n+    if dt is None:\n+        return "—"\n+    if time_only:\n+        return dt.strftime("%H:%M")\n+    return f"{_fmt_tpe_date(dt, lang)} {dt:%H:%M}"\n+\n+\n+def latest_briefing(rows):\n+    """Newest usable briefing by cutoff_time (never generated_at).\n+\n+    failed rows are ignored, so a failed run leaves the previous\n+    successful edition in place.\n+    """\n+    best = None\n+    for r in rows or []:\n+        if not isinstance(r, dict):\n+            continue\n+        if r.get("status") not in ("published", "partial"):\n+            continue\n+        cut = _tpe_dt(r.get("cutoff_time"))\n+        if cut is None:\n+            continue\n+        if best is None or cut > best[0]:\n+            best = (cut, r)\n+    return best[1] if best else None\n+\n+\n+def collect_briefings():\n+    """Readable published/partial briefings, newest cutoff first."""\n+    index = load_json(DATA_DIR / "briefings" / "index.json", {})\n+    briefs = []\n+    for row in index.get("briefings") or []:\n+        if not isinstance(row, dict):\n+            continue\n+        if row.get("status") not in ("published", "partial"):\n+            continue\n+        bid = str(row.get("briefing_id") or "")\n+        if not bid or not _ID_RE.match(bid):\n+            continue\n+        data = load_json(DATA_DIR / "briefings" / f"{bid}.json", None)\n+        if (isinstance(data, dict)\n+                and data.get("status") in ("published", "partial")):\n+            briefs.append(data)\n+    briefs.sort(key=lambda b: str(b.get("cutoff_time") or ""), reverse=True)\n+    return briefs\n+\n+\n+def _brief_title(b, lang: str) -> str:\n+    cfg = _BRIEF_EDITIONS.get(b.get("edition")) or {}\n+    key = "title_zh_tw" if lang == "zh" else "title_en"\n+    return str(cfg.get(key) or b.get("edition_label") or "AVWIRE Briefing")\n+\n+\n+def _brief_label(b, lang: str) -> str:\n+    cfg = _BRIEF_EDITIONS.get(b.get("edition")) or {}\n+    key = "label_zh_tw" if lang == "zh" else "label_en"\n+    return str(cfg.get(key) or b.get("edition_label") or "")\n+\n+\n+def brief_item_view(item, lang: str, published_ids):\n+    zh_head = str(item.get("headline") or "").strip()\n+    zh_sum = str(item.get("summary") or "").strip()\n+    head = zh_head if lang == "zh" else (\n+        str(item.get("headline_en") or "").strip() or zh_head)\n+    summary = zh_sum if lang == "zh" else (\n+        str(item.get("summary_en") or "").strip() or zh_sum)\n+    art_id = item.get("article_id")\n+    url = (page_url(lang, f"news/{art_id}/")\n+           if art_id in published_ids else None)\n+    return {\n+        "headline": head, "summary": summary,\n+        "time": _fmt_tpe(item.get("source_published_at"), lang,\n+                         time_only=True),\n+        "severity": str(item.get("severity") or "routine"),\n+        "sev_cls": _BRIEF_SEV_CLS.get(item.get("severity"), "tag-neutral"),\n+        "update": item.get("item_type") == "update",\n+        "url": url,\n+        "sources": [s for s in item.get("sources") or []\n+                    if isinstance(s, dict) and s.get("url")],\n+    }\n+\n+\n+def brief_view(b, lang: str, t, published_ids):\n+    sections = []\n+    raw_sections = b.get("sections") or {}\n+    for i, name in enumerate(BRIEFING_SECTIONS):\n+        items = [brief_item_view(it, lang, published_ids)\n+                 for it in raw_sections.get(name) or []\n+                 if isinstance(it, dict)]\n+        sections.append({"label": t["brSecs"][i], "items": items})\n+    gen_model = b.get("generation_model") or {}\n+    date_dt = _tpe_dt(b.get("cutoff_time"))\n+    return {\n+        "id": b.get("briefing_id"),\n+        "title": _brief_title(b, lang),\n+        "label": _brief_label(b, lang),\n+        "date_label": _fmt_tpe_date(date_dt, lang) if date_dt else "—",\n+        "window_start": _fmt_tpe(b.get("window_start"), lang),\n+        "window_end": _fmt_tpe(b.get("window_end"), lang),\n+        "cutoff": _fmt_tpe(b.get("cutoff_time"), lang),\n+        "generated": _fmt_tpe(b.get("generated_at"), lang, time_only=True),\n+        "partial": b.get("status") == "partial",\n+        "total": sum(len(s["items"]) for s in sections),\n+        "sections": sections,\n+        "intro": str(b.get("intro_zh" if lang == "zh" else "intro_en")\n+                     or "").strip(),\n+        "mode_label": (t["brModeLlm"] if b.get("generation_mode")\n+                       == "llm_assisted" else t["brModeDet"]),\n+        "model_label": (f"{gen_model.get('provider')}:{gen_model.get('model')}"\n+                        if gen_model else ""),\n+        "checked": [c for c in b.get("checked_sources") or []\n+                    if isinstance(c, dict)],\n+        "warnings": [str(w) for w in b.get("warnings") or []],\n+        "url": page_url(lang, f"briefings/{b.get('briefing_id')}/"),\n+        "cutoff_iso": str(b.get("cutoff_time") or ""),\n+    }\n+\n+\n def stats_views(stats, lang: str):\n     if not isinstance(stats, dict):\n         stats = {}\n@@ -998,7 +1173,8 @@ def stats_views(stats, lang: str):\n \n def base_ctx(lang, page, sub, *, title, description, ticker, build, hreflang=True):\n     t = L[lang]\n-    nav_defs = [("home", ""), ("incidents", "incidents/"),\n+    nav_defs = [("home", ""), ("briefings", "briefings/"),\n+                ("incidents", "incidents/"),\n                 ("sources", "sources/"), ("about", "about/")]\n     return {\n         "lang": lang, "t": t, "base": BASE_PATH, "favicon": FAVICON,\n@@ -1029,11 +1205,17 @@ def main() -> int:\n     now = now_utc()\n \n     articles = collect_articles()\n+    published_ids = {a["id"] for a in articles}\n     flashes = prep_flashes(load_json(DATA_DIR / "flashes.json", None),\n-                           {a["id"] for a in articles})\n+                           published_ids)\n     incidents = prep_incidents(load_json(DATA_DIR / "incidents.json", None))\n     sources = merged_sources(load_json(DATA_DIR / "sources.json", []))\n     stats_raw = load_json(DATA_DIR / "stats.json", {})\n+    briefings = collect_briefings()\n+    brief_rows = [r for r in (load_json(\n+        DATA_DIR / "briefings" / "index.json", {}).get("briefings") or [])\n+        if isinstance(r, dict)]\n+    latest_row = latest_briefing(brief_rows)\n \n     # clean output tree\n     if SITE_DIR.exists():\n@@ -1090,16 +1272,51 @@ def main() -> int:\n         ticker = TICKER_SEP.join(f"{f['time']}  {f['text']}" for f in fl)\n         sv = stats_views(stats_raw, lang)\n \n+        bviews = [brief_view(b, lang, t, published_ids) for b in briefings]\n+        latest_brief = None\n+        if latest_row:\n+            latest_brief = next(\n+                (bv for bv in bviews\n+                 if bv["id"] == latest_row.get("briefing_id")), None)\n+\n         ctx = base_ctx(lang, "home", "", title=t["siteName"],\n                        description=t["siteDesc"], ticker=ticker, build=build)\n         ctx.update(hero=hero, feed=feed, flashes=fl, agg=agg,\n                    source_status=source_status_view(sources, now),\n                    stats=sv["tiles"], otp=sv["otp"], fleet=sv["fleet"],\n                    delays=sv["delays"],\n-                   cat_seg=list(zip(cat_keys, t["cats"])))\n+                   cat_seg=list(zip(cat_keys, t["cats"])),\n+                   latest_brief=latest_brief)\n         render(env, "home.html", rel_path(lang, "index.html"), ctx)\n         pages += 1\n \n+        for bv in bviews:\n+            try:\n+                ctx = base_ctx(lang, "briefings", f"briefings/{bv['id']}/",\n+                               title=f"{t['siteName']} — {bv['title']}",\n+                               description=f"{bv['title']} · {bv['date_label']}",\n+                               ticker=ticker, build=build)\n+                ctx["b"] = bv\n+                render(env, "briefing.html",\n+                       rel_path(lang, f"briefings/{bv['id']}/index.html"), ctx)\n+                pages += 1\n+            except Exception as exc:  # one bad briefing must not kill the build\n+                failed += 1\n+                print(f"build: briefing {bv['id']} ({lang}) failed: {exc}")\n+\n+        groups: dict[str, list] = {}\n+        for bv in bviews:  # bviews sorted newest cutoff first -> dates desc\n+            groups.setdefault(bv["date_label"], []).append(bv)\n+        ctx = base_ctx(lang, "briefings", "briefings/",\n+                       title=f"{t['siteName']} — {t['brIndexTitle']}",\n+                       description=t["brIndexSub"], ticker=ticker, build=build)\n+        ctx.update(groups=[\n+            {"date": d, "rows": sorted(rows, key=lambda r: r["cutoff_iso"])}\n+            for d, rows in groups.items()])\n+        render(env, "briefings.html",\n+               rel_path(lang, "briefings/index.html"), ctx)\n+        pages += 1\n+\n         for a, v in zip(articles, views):\n             try:\n                 ctx = base_ctx(lang, "home", f"news/{a['id']}/",\n@@ -0,0 +1,71 @@\n+{% extends "base.html" %}\n+{% block content %}\n+  <div style="max-width:860px;margin:0 auto;padding:28px 20px 64px;">\n+    <a href="{{ base }}/{% if lang == 'en' %}en/{% endif %}briefings/" class="btn btn-ghost" style="margin-bottom:18px;">← {{ t.brIndexTitle }}</a>\n+    <span class="card-kicker">{{ t.brKicker }}</span>\n+    <h1 style="font-size:32px;line-height:1.2;margin:10px 0 6px;">{{ b.title }}</h1>\n+    <div style="font-size:17px;font-weight:700;margin-bottom:14px;">{{ b.date_label }}</div>\n+    <div class="text-muted" style="font-size:13px;line-height:2;border-top:2px solid var(--color-divider);border-bottom:2px solid var(--color-divider);padding:10px 0;margin-bottom:6px;">\n+      <div><span style="font-weight:800;">{{ t.brWindow }}</span>：{{ b.window_start }} {{ t.brTo }} {{ b.window_end }}（UTC+8）</div>\n+      <div><span style="font-weight:800;">{{ t.brCutoff }}</span>：{{ b.cutoff }}</div>\n+    </div>\n+    <div class="text-muted" style="font-size:12px;margin-bottom:20px;display:flex;gap:14px;flex-wrap:wrap;">\n+      <span>{{ t.brGenerated }}：{{ b.generated }}</span>\n+      <span>{{ b.mode_label }}{% if b.model_label %} · {{ b.model_label }}{% endif %}</span>\n+    </div>\n+    {% if b.partial %}\n+    <div style="border:2px solid var(--color-divider);border-left:6px solid var(--color-accent);padding:12px 16px;margin-bottom:20px;">\n+      <p class="text-muted" style="font-size:13px;line-height:1.8;margin:0;">{{ t.brPartial }}</p>\n+    </div>\n+    {% endif %}\n+    {% if b.intro %}\n+    <p style="font-size:15.5px;line-height:1.9;margin:0 0 20px;">{{ b.intro }}</p>\n+    {% endif %}\n+    {% if b.total == 0 %}\n+    <div style="border:2px solid var(--color-divider);padding:18px 20px;margin:8px 0 20px;">\n+      <p class="text-muted" style="font-size:14.5px;line-height:1.9;margin:0;">{{ t.brEmpty }}</p>\n+    </div>\n+    {% endif %}\n+    {% for sec in b.sections %}\n+    {% if sec['items'] %}\n+    <h4 style="margin:26px 0 4px;">{{ sec.label }}</h4>\n+    {% for it in sec['items'] %}\n+    <div style="padding:14px 0;border-bottom:1px solid var(--color-divider);">\n+      <div style="display:flex;gap:10px;align-items:baseline;flex-wrap:wrap;">\n+        <span class="tag {{ it.sev_cls }}">{{ t.brSevs[it.severity] }}</span>\n+        {% if it.update %}<span class="tag tag-outline">{{ t.brUpdate }}</span>{% endif %}\n+        <span class="text-muted" style="font-size:12px;font-weight:800;">{{ it.time }}</span>\n+      </div>\n+      {% if it.url %}\n+      <a href="{{ it.url }}" style="display:block;font-size:16.5px;font-weight:700;line-height:1.5;margin:8px 0 6px;">{{ it.headline }}</a>\n+      {% else %}\n+      <div style="font-size:16.5px;font-weight:700;line-height:1.5;margin:8px 0 6px;">{{ it.headline }}</div>\n+      {% endif %}\n+      <p class="text-muted" style="font-size:14px;line-height:1.85;margin:0 0 8px;">{{ it.summary }}</p>\n+      <div style="display:flex;gap:12px;flex-wrap:wrap;font-size:12px;">\n+        {% for s in it.sources %}\n+        <a href="{{ s.url }}" target="_blank" rel="noopener" class="text-muted">{{ s.name }}</a>\n+        {% endfor %}\n+      </div>\n+    </div>\n+    {% endfor %}\n+    {% endif %}\n+    {% endfor %}\n+    <div style="border:2px solid var(--color-divider);padding:16px 20px 12px;margin-top:32px;">\n+      <h6 style="margin:0 0 10px;">{{ t.brChecked }}</h6>\n+      <div style="display:flex;gap:10px 18px;flex-wrap:wrap;font-size:12.5px;">\n+        {% for c in b.checked %}\n+        <span style="display:inline-flex;align-items:center;gap:6px;">\n+          <span class="dot{% if not c.ok %} err{% endif %}"></span>{{ c.name }}\n+        </span>\n+        {% endfor %}\n+      </div>\n+      {% if b.warnings %}\n+      <h6 style="margin:14px 0 6px;">{{ t.brWarnings }}</h6>\n+      {% for w in b.warnings %}\n+      <p class="text-muted" style="font-size:12px;line-height:1.7;margin:0 0 4px;">· {{ w }}</p>\n+      {% endfor %}\n+      {% endif %}\n+    </div>\n+  </div>\n+{% endblock %}\n@@ -0,0 +1,21 @@\n+{% extends "base.html" %}\n+{% block content %}\n+  <div style="max-width:860px;margin:0 auto;padding:28px 20px 64px;">\n+    <span class="card-kicker">{{ t.brKicker }}</span>\n+    <h2 style="margin:8px 0 10px;">{{ t.brIndexTitle }}</h2>\n+    <p class="text-muted" style="font-size:14px;line-height:1.9;margin:0 0 8px;">{{ t.brIndexSub }}</p>\n+    {% for g in groups %}\n+    <h4 style="margin:28px 0 6px;">{{ g.date }}</h4>\n+    {% for r in g.rows %}\n+    <a href="{{ r.url }}" style="display:flex;gap:14px;align-items:baseline;padding:12px 0;border-bottom:1px solid var(--color-divider);color:inherit;text-decoration:none;flex-wrap:wrap;">\n+      <span style="font-weight:800;">{{ r.label }}</span>\n+      <span class="text-muted" style="font-size:12.5px;">{{ t.brCutoff }} {{ r.cutoff }}</span>\n+      {% if r.partial %}<span class="tag tag-outline">partial</span>{% endif %}\n+      <span class="text-muted" style="margin-left:auto;font-size:12.5px;">{{ r.total }} {{ t.brItems }}</span>\n+    </a>\n+    {% endfor %}\n+    {% else %}\n+    <p class="text-muted" style="font-size:14px;margin-top:20px;">{{ t.awaiting }}</p>\n+    {% endfor %}\n+  </div>\n+{% endblock %}\n@@ -2,6 +2,13 @@\n {% block content %}\n   <div class="home-grid">\n     <div class="rail-left">\n+      {% if latest_brief %}\n+      <a href="{{ latest_brief.url }}" style="display:block;border:2px solid var(--color-divider);border-left:6px solid var(--color-accent);padding:12px 14px;margin-bottom:20px;color:inherit;text-decoration:none;">\n+        <span class="card-kicker" style="font-size:10px;">{{ t.brLatest }}</span>\n+        <div style="font-weight:800;font-size:15px;margin:6px 0 2px;">{{ latest_brief.label }}</div>\n+        <div class="text-muted" style="font-size:12px;">{{ latest_brief.date_label }} · {{ latest_brief.total }} {{ t.brItems }}</div>\n+      </a>\n+      {% endif %}\n       <h6 style="display:flex;justify-content:space-between;align-items:baseline;margin:0 0 12px;">{{ t.flash }} <span style="color:var(--color-accent);font-size:10px;">● LIVE</span></h6>\n       {% for f in flashes %}\n       {% if f.url %}\n@@ -0,0 +1,514 @@\n+"""Tests for pipeline/briefing.py (thrice-daily transport briefing).\n+\n+No pytest required — run from the repo root:\n+\n+    py tests\\test_briefing.py\n+\n+Everything runs offline against a temp AVWIRE_DATA_DIR with fixture\n+articles; model providers are replaced by a fake module that counts\n+calls, so no real RSS, API, NVIDIA or Gemini endpoint is ever touched.\n+"""\n+from __future__ import annotations\n+\n+import json\n+import os\n+import re\n+import shutil\n+import sys\n+import tempfile\n+import types\n+from datetime import date, datetime, timedelta, timezone\n+from pathlib import Path\n+\n+REPO = Path(__file__).resolve().parent.parent\n+TMP = Path(tempfile.mkdtemp(prefix="avwire-briefing-"))\n+os.environ["AVWIRE_DATA_DIR"] = str(TMP)\n+\n+sys.path.insert(0, str(REPO / "pipeline"))\n+import briefing  # noqa: E402\n+from briefing import (  # noqa: E402\n+    CRON_TO_EDITION,\n+    TPE,\n+    event_fingerprint,\n+    get_briefing_window,\n+)\n+\n+CHECKS = 0\n+FAILED = 0\n+\n+\n+def check(name: str, cond) -> None:\n+    global CHECKS, FAILED\n+    CHECKS += 1\n+    if cond:\n+        print(f"PASS {name}")\n+    else:\n+        FAILED += 1\n+        print(f"FAIL {name}")\n+\n+\n+# ── fixtures ─────────────────────────────────────────────────────────────────\n+\n+def reset() -> None:\n+    for sub in ("articles", "briefings"):\n+        shutil.rmtree(TMP / sub, ignore_errors=True)\n+    for f in ("sources.json", "review.json"):\n+        try:\n+            (TMP / f).unlink()\n+        except FileNotFoundError:\n+            pass\n+\n+\n+def tpe(y, mo, d, h, mi=0):\n+    return datetime(y, mo, d, h, mi, tzinfo=TPE)\n+\n+\n+def utc_stamp(dt) -> str:\n+    return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%MZ")\n+\n+\n+def make_article(art_id, title, src_dt, *, cat="ops", orgs=("CAA Taiwan",),\n+                 locs=("臺灣",), claims=("fact one", "fact two"),\n+                 urls=("https://example.gov/a",), summary="摘要內容"):\n+    return {\n+        "id": art_id,\n+        "publishedUtc": utc_stamp(src_dt),\n+        "sourcePublishedUtc": utc_stamp(src_dt),\n+        "cat": cat,\n+        "primarySource": "Test",\n+        "image": None,\n+        "zh": {"title": title, "summary": summary, "body": [summary]},\n+        "en": {"title": f"EN {title}", "summary": "en summary",\n+               "body": ["en body"]},\n+        "sources": [{"name": "Test source", "url": u} for u in urls],\n+        "writer": "nvidia:test-model",\n+        "facts": [{"factId": f"F{i+1}", "claim": c, "sourceQuote": c}\n+                  for i, c in enumerate(claims)],\n+        "riskFlags": [],\n+        "eventStatus": "confirmed",\n+        "entities": {"organizations": list(orgs), "locations": list(locs),\n+                     "event_dates": []},\n+    }\n+\n+\n+def write_articles(articles, fname="batch.json") -> None:\n+    path = TMP / "articles" / fname\n+    path.parent.mkdir(parents=True, exist_ok=True)\n+    path.write_text(json.dumps({"articles": articles}, ensure_ascii=False),\n+                    encoding="utf-8")\n+\n+\n+def write_sources(ok=True) -> None:\n+    (TMP / "sources.json").write_text(json.dumps([\n+        {"name": "FAA", "ok": ok, "lastFetchUtc": "2026-07-27T00:00Z"},\n+        {"name": "CAA Taiwan", "ok": True,\n+         "lastFetchUtc": "2026-07-27T00:00Z"},\n+    ]), encoding="utf-8")\n+\n+\n+def load_brief(briefing_id):\n+    return json.loads((TMP / "briefings" / f"{briefing_id}.json")\n+                      .read_text(encoding="utf-8"))\n+\n+\n+def load_index():\n+    return json.loads((TMP / "briefings" / "index.json")\n+                      .read_text(encoding="utf-8"))\n+\n+\n+def all_items(brief):\n+    return [it for sec in brief["sections"].values() for it in sec]\n+\n+\n+class FakeProvider:\n+    calls = 0\n+\n+    def __init__(self, name="fake", model="fake-model", result=None,\n+                 error=None):\n+        self.name, self.model = name, model\n+        self.label = f"{name}:{model}"\n+        self._result, self._error = result, error\n+\n+    def available(self):\n+        return True\n+\n+    def draft(self, system_prompt, user_prompt, schema):\n+        FakeProvider.calls += 1\n+        if self._error:\n+            raise self._error\n+        return self._result\n+\n+\n+def install_fake_providers(providers):\n+    fake = types.ModuleType("providers")\n+    fake.build_providers = lambda: list(providers)\n+    sys.modules["providers"] = fake\n+\n+\n+# ── 1-6: fixed half-open windows ────────────────────────────────────────────\n+\n+D = date(2026, 7, 27)\n+morning = get_briefing_window("morning", D)\n+afternoon = get_briefing_window("afternoon", D)\n+evening = get_briefing_window("evening", D)\n+\n+check("morning window is prev-day 23:00 -> 07:00 TPE",\n+      morning.window_start == tpe(2026, 7, 26, 23)\n+      and morning.window_end == tpe(2026, 7, 27, 7))\n+check("afternoon window is 07:00 -> 15:00 TPE",\n+      afternoon.window_start == tpe(2026, 7, 27, 7)\n+      and afternoon.window_end == tpe(2026, 7, 27, 15))\n+check("evening window is 15:00 -> 23:00 TPE",\n+      evening.window_start == tpe(2026, 7, 27, 15)\n+      and evening.window_end == tpe(2026, 7, 27, 23))\n+check("windows are timezone-aware (+08:00)",\n+      morning.window_start.utcoffset() == timedelta(hours=8)\n+      and morning.cutoff_time.tzinfo is not None)\n+check("07:00:00 belongs to afternoon, not morning",\n+      not morning.contains(tpe(2026, 7, 27, 7))\n+      and afternoon.contains(tpe(2026, 7, 27, 7)))\n+check("15:00:00 belongs to evening, not afternoon",\n+      not afternoon.contains(tpe(2026, 7, 27, 15))\n+      and evening.contains(tpe(2026, 7, 27, 15)))\n+check("23:00:00 belongs to next-day morning, not evening",\n+      not evening.contains(tpe(2026, 7, 27, 23))\n+      and get_briefing_window("morning", date(2026, 7, 28))\n+      .contains(tpe(2026, 7, 27, 23)))\n+check("boundary compare works across timezones (UTC input)",\n+      afternoon.contains(datetime(2026, 7, 26, 23, 0,\n+                                  tzinfo=timezone.utc)))\n+\n+# 7: a delayed run cannot move the window — it is a pure function of\n+# (edition, taipei_date), and the id is stable too.\n+w1 = get_briefing_window("morning", D)\n+w2 = get_briefing_window("morning", D)\n+check("window derived from edition+date only (delay-proof)",\n+      w1 == w2 and w1.briefing_id == "2026-07-27-morning")\n+\n+# 8: UTC runner -> correct Taipei date at the scheduled instants.\n+check("23:15 UTC previous day maps to Taipei morning date",\n+      datetime(2026, 7, 26, 23, 15, tzinfo=timezone.utc)\n+      .astimezone(TPE).date() == D)\n+check("07:15 UTC maps to Taipei afternoon date",\n+      datetime(2026, 7, 27, 7, 15, tzinfo=timezone.utc)\n+      .astimezone(TPE).date() == D)\n+\n+# 9: cron -> edition mapping, cross-checked against config and workflow.\n+check("cron mapping matches spec",\n+      CRON_TO_EDITION == {"15 23 * * *": "morning",\n+                          "15 7 * * *": "afternoon",\n+                          "15 15 * * *": "evening"})\n+_editions_cfg = json.loads((REPO / "config" / "briefing_editions.json")\n+                           .read_text(encoding="utf-8"))\n+check("config cron_utc agrees with CRON_TO_EDITION",\n+      {cfg["cron_utc"]: name for name, cfg in _editions_cfg.items()}\n+      == CRON_TO_EDITION)\n+_wf = (REPO / ".github" / "workflows" / "briefing.yml").read_text(\n+    encoding="utf-8")\n+check("workflow declares exactly the three edition crons",\n+      all(f'cron: "{c}"' in _wf for c in CRON_TO_EDITION)\n+      and _wf.count("- cron:") == 3)\n+check("workflow maps each schedule string explicitly",\n+      all(re.search(re.escape(f'"{c}")') + r'\s+echo "edition=' + e + '"',\n+                    _wf)\n+          for c, e in CRON_TO_EDITION.items()))\n+check("workflow supports dispatch edition+date inputs",\n+      "workflow_dispatch" in _wf and "edition:" in _wf and "date:" in _wf\n+      and "concurrency" in _wf)\n+\n+# 11: no now-minus-24h briefing logic anywhere in the module.\n+_src = (REPO / "pipeline" / "briefing.py").read_text(encoding="utf-8")\n+check("no 24-hour rolling-window logic in briefing.py",\n+      "hours=24" not in _src\n+      and re.search(r"now\w*\(\)\s*-\s*timedelta", _src) is None)\n+\n+# ── 10: CLI dispatch (edition + date, and cron form) ─────────────────────────\n+\n+reset()\n+write_sources(ok=True)\n+art_morning = make_article("a-m1", "晨間事件A", tpe(2026, 7, 27, 6, 30))\n+write_articles([art_morning])\n+check("main --edition/--date runs", briefing.main(\n+    ["--edition", "morning", "--date", "2026-07-27"]) == 0)\n+bm = load_brief("2026-07-27-morning")\n+check("briefing schema basics", bm["content_type"]\n+      == "daily_transport_briefing" and bm["edition"] == "morning"\n+      and bm["timezone"] == "Asia/Taipei"\n+      and bm["window_start"] == "2026-07-26T23:00:00+08:00"\n+      and bm["window_end"] == "2026-07-27T07:00:00+08:00"\n+      and bm["cutoff_time"] == bm["window_end"])\n+check("morning picked the 06:30 article",\n+      bm["item_count"] == 1 and all_items(bm)[0]["item_type"] == "new")\n+check("deterministic mode with no model recorded",\n+      bm["generation_mode"] == "deterministic"\n+      and bm["generation_model"] is None)\n+check("main --cron maps to edition", briefing.main(\n+    ["--cron", "15 7 * * *", "--date", "2026-07-27"]) == 0\n+    and (TMP / "briefings" / "2026-07-27-afternoon.json").exists())\n+\n+# ── 12-14: only the published article store feeds briefings ─────────────────\n+\n+reset()\n+write_sources(ok=True)\n+(TMP / "review.json").write_text(json.dumps({"pending": [{\n+    "id": "r1", "status": "manual_review",\n+    "zh": {"title": "覆核中的墜機事件", "summary": "不得出現"},\n+}]}, ensure_ascii=False), encoding="utf-8")\n+write_articles([make_article("a-pub", "已發布事件", tpe(2026, 7, 27, 6))])\n+briefing.main(["--edition", "morning", "--date", "2026-07-27"])\n+bm = load_brief("2026-07-27-morning")\n+_txt = json.dumps(bm, ensure_ascii=False)\n+check("pending manual_review items never appear",\n+      "覆核中的墜機事件" not in _txt and bm["item_count"] == 1)\n+check("selection reads only data/articles (rejects can't exist there)",\n+      all_items(bm)[0]["article_id"] == "a-pub")\n+check("items carry >=1 source and verified zh headline+summary",\n+      all_items(bm)[0]["sources"] and all_items(bm)[0]["headline"])\n+\n+# incomplete article (no source URL) is skipped, not published\n+reset()\n+write_sources(ok=True)\n+bad = make_article("a-bad", "無來源事件", tpe(2026, 7, 27, 6), urls=())\n+write_articles([bad, make_article("a-ok", "有來源事件",\n+                                  tpe(2026, 7, 27, 5))])\n+briefing.main(["--edition", "morning", "--date", "2026-07-27"])\n+bm = load_brief("2026-07-27-morning")\n+check("article without source URL is skipped with a warning",\n+      bm["item_count"] == 1\n+      and any("a-bad" in w for w in bm["warnings"]))\n+\n+# ── 15-18: cross-edition dedup, updates, id stability ────────────────────────\n+\n+reset()\n+write_sources(ok=True)\n+base_kw = dict(cat="safety", orgs=("Delta Air Lines",), locs=("高雄小港",),\n+               claims=("Delta A350 landed at Kaohsiung",))\n+write_articles([make_article("a-ev1", "達美A350抵達小港",\n+                             tpe(2026, 7, 27, 6, 10), **base_kw)],\n+               "m.json")\n+briefing.main(["--edition", "morning", "--date", "2026-07-27"])\n+ev_id = all_items(load_brief("2026-07-27-morning"))[0]["event_id"]\n+\n+# afternoon: same facts, retitled, reprinted on a new URL -> not an update\n+write_articles([make_article("a-ev2", "換個標題的同一事件",\n+                             tpe(2026, 7, 27, 8, 5),\n+                             urls=("https://other-site.example/copy",),\n+                             **base_kw)], "a.json")\n+briefing.main(["--edition", "afternoon", "--date", "2026-07-27"])\n+ba = load_brief("2026-07-27-afternoon")\n+check("same event without new facts never repeats in a later edition",\n+      ba["item_count"] == 0)\n+\n+# evening: same event with a genuinely new fact -> update item\n+write_articles([make_article(\n+    "a-ev3", "達美A350小港起飛離場", tpe(2026, 7, 27, 16, 40),\n+    cat="safety", orgs=("Delta Air Lines",), locs=("高雄小港",),\n+    claims=("Delta A350 landed at Kaohsiung",\n+            "Aircraft departed Kaohsiung on July 27"),\n+    urls=("https://example.gov/followup",))], "e.json")\n+briefing.main(["--edition", "evening", "--date", "2026-07-27"])\n+be = load_brief("2026-07-27-evening")\n+up = all_items(be)[0] if all_items(be) else {}\n+check("substantive new fact surfaces as an update item",\n+      be["item_count"] == 1 and up.get("item_type") == "update")\n+check("update binds to the original event id and lists new info",\n+      up.get("previous_event_id") == ev_id\n+      and up.get("new_information")\n+      == ["Aircraft departed Kaohsiung on July 27"])\n+check("event_id ignores edition, headline and article id",\n+      up.get("event_id") == ev_id)\n+a1 = make_article("x1", "標題甲", tpe(2026, 7, 27, 6), **base_kw)\n+a2 = make_article("x2", "完全不同的標題乙", tpe(2026, 7, 27, 6), **base_kw)\n+check("fingerprint is identical for retitled duplicates",\n+      event_fingerprint(a1, "aviation_incidents")\n+      == event_fingerprint(a2, "aviation_incidents"))\n+_fp_core = _src.split("core = {")[1].split("return hashlib")[0]\n+check("fingerprint core hashes no edition/headline/generated_at/position",\n+      all(bad not in _fp_core for bad in\n+          ("edition", "headline", "generated_at", "briefing_id", "title")))\n+\n+# safety article classified into incidents with severity ordering\n+check("safety article lands in aviation_incidents",\n+      len(be["sections"]["aviation_incidents"]) == 1)\n+\n+# ── 19-22: model-call discipline ─────────────────────────────────────────────\n+\n+# zero candidates -> zero calls even with the intro flag on\n+reset()\n+write_sources(ok=True)\n+write_articles([])\n+FakeProvider.calls = 0\n+install_fake_providers([FakeProvider(\n+    result={"introZh": "十" * 30, "introEn": "e" * 30})])\n+os.environ["BRIEFING_LLM_INTRO"] = "true"\n+briefing.main(["--edition", "morning", "--date", "2026-07-27"])\n+check("empty edition performs zero model calls", FakeProvider.calls == 0)\n+b_empty = load_brief("2026-07-27-morning")\n+check("empty edition is deterministic with empty sections",\n+      b_empty["generation_mode"] == "deterministic"\n+      and b_empty["item_count"] == 0\n+      and all(v == [] for v in b_empty["sections"].values()))\n+\n+# items with verified summaries -> no per-item rewriting; intro is one batch\n+reset()\n+write_sources(ok=True)\n+write_articles([make_article(f"a-{i}", f"事件{i}",\n+                             tpe(2026, 7, 27, 5, i),\n+                             orgs=(f"Org{i}",), locs=(f"City{i}",),\n+                             claims=(f"claim {i}",),\n+                             urls=(f"https://example.gov/{i}",))\n+                for i in range(4)])\n+FakeProvider.calls = 0\n+install_fake_providers([FakeProvider(\n+    result={"introZh": "本" * 30, "introEn": "e" * 30})])\n+briefing.main(["--edition", "morning", "--date", "2026-07-27"])\n+bm = load_brief("2026-07-27-morning")\n+check("4 items cost at most one batch call", FakeProvider.calls == 1)\n+check("llm_assisted mode records the real provider+model",\n+      bm["generation_mode"] == "llm_assisted"\n+      and bm["generation_model"] == {"provider": "fake",\n+                                     "model": "fake-model"}\n+      and bm["intro_zh"] == "本" * 30)\n+check("item headlines/summaries reused verbatim (no rewriting)",\n+      sorted(it["headline"] for it in all_items(bm))\n+      == [f"事件{i}" for i in range(4)])\n+\n+# model failure -> deterministic fallback, edition never fails\n+reset()\n+write_sources(ok=True)\n+write_articles([make_article("a-f", "事件F", tpe(2026, 7, 27, 6))])\n+FakeProvider.calls = 0\n+install_fake_providers([FakeProvider(error=RuntimeError("boom")),\n+                        FakeProvider(error=RuntimeError("boom2"))])\n+briefing.main(["--edition", "morning", "--date", "2026-07-27"])\n+bm = load_brief("2026-07-27-morning")\n+check("failed batch call is capped at BRIEFING_EXTRA_LLM_CALLS_MAX=1",\n+      FakeProvider.calls == 1)\n+check("model failure falls back to deterministic, edition still published",\n+      bm["status"] in ("published", "partial")\n+      and bm["generation_mode"] == "deterministic"\n+      and bm["item_count"] == 1)\n+del os.environ["BRIEFING_LLM_INTRO"]\n+\n+# ── 23-24: empty copy vs source failure ──────────────────────────────────────\n+\n+_build_src = (REPO / "pipeline" / "build.py").read_text(encoding="utf-8")\n+check("conservative empty-edition copy is the specified sentence",\n+      "截至本期資料截止時間，系統在本次查核的指定來源中，未發現符合收錄門檻的新事件。"\n+      in _build_src)\n+check("no 'nothing happened worldwide' style claims in templates",\n+      "全球沒有" not in _build_src and "均無異常" not in _build_src)\n+\n+reset()\n+write_sources(ok=False)  # FAA failing\n+write_articles([])\n+briefing.main(["--edition", "morning", "--date", "2026-07-27"])\n+bm = load_brief("2026-07-27-morning")\n+check("source fetch failure -> partial with a coverage warning, "\n+      "never 'no events'",\n+      bm["status"] == "partial"\n+      and any("coverage gap" in w for w in bm["warnings"])\n+      and any(not c["ok"] for c in bm["checked_sources"]))\n+\n+# ── 25: failed runs never clobber a successful edition ───────────────────────\n+\n+reset()\n+write_sources(ok=True)\n+write_articles([make_article("a-keep", "既有事件", tpe(2026, 7, 27, 8))])\n+briefing.main(["--edition", "afternoon", "--date", "2026-07-27"])\n+good = load_brief("2026-07-27-afternoon")\n+_orig_loader = briefing.load_published_articles\n+briefing.load_published_articles = lambda: (_ for _ in ()).throw(\n+    RuntimeError("disk gone"))\n+check("crashing re-run exits 0 (deploy of old content still possible)",\n+      briefing.main(["--edition", "afternoon", "--date", "2026-07-27"]) == 0)\n+briefing.load_published_articles = _orig_loader\n+check("briefing file survives the crash untouched",\n+      load_brief("2026-07-27-afternoon") == good)\n+check("index keeps the successful status for that edition",\n+      [r["status"] for r in load_index()["briefings"]\n+       if r["briefing_id"] == "2026-07-27-afternoon"] == ["published"])\n+\n+briefing.load_published_articles = lambda: (_ for _ in ()).throw(\n+    RuntimeError("still broken"))\n+briefing.main(["--edition", "evening", "--date", "2026-07-27"])\n+briefing.load_published_articles = _orig_loader\n+check("a failed edition is recorded as failed without a briefing file",\n+      [r["status"] for r in load_index()["briefings"]\n+       if r["briefing_id"] == "2026-07-27-evening"] == ["failed"]\n+      and not (TMP / "briefings" / "2026-07-27-evening.json").exists())\n+\n+# ── 26: homepage picks the latest by cutoff_time, skipping failed ────────────\n+\n+import build  # noqa: E402\n+\n+rows = [\n+    {"briefing_id": "2026-07-27-morning", "status": "published",\n+     "cutoff_time": "2026-07-27T07:00:00+08:00",\n+     "generated_at": "2026-07-27T23:59:00+08:00"},  # late regen: irrelevant\n+    {"briefing_id": "2026-07-27-afternoon", "status": "partial",\n+     "cutoff_time": "2026-07-27T15:00:00+08:00",\n+     "generated_at": "2026-07-27T15:18:00+08:00"},\n+    {"briefing_id": "2026-07-27-evening", "status": "failed",\n+     "cutoff_time": "2026-07-27T23:00:00+08:00",\n+     "generated_at": "2026-07-27T23:18:00+08:00"},\n+]\n+latest = build.latest_briefing(rows)\n+check("latest briefing = max cutoff_time among published/partial",\n+      latest["briefing_id"] == "2026-07-27-afternoon")\n+check("failed never becomes the latest briefing",\n+      build.latest_briefing([rows[2]]) is None)\n+\n+# ── 27-29: no secrets, no Perplexity, no paid APIs ───────────────────────────\n+\n+reset()\n+write_sources(ok=True)\n+write_articles([make_article("a-s", "事件S", tpe(2026, 7, 27, 6))])\n+os.environ["NVIDIA_API_KEY"] = "nvapi-TESTFAKEKEY1234567890"\n+briefing.main(["--edition", "morning", "--date", "2026-07-27"])\n+del os.environ["NVIDIA_API_KEY"]\n+_out = (TMP / "briefings" / "2026-07-27-morning.json").read_text(\n+    encoding="utf-8")\n+check("briefing output contains no secrets or reasoning traces",\n+      "nvapi-" not in _out and "reasoning_content" not in _out\n+      and "<think>" not in _out)\n+\n+_repo_hits = []\n+for path in REPO.rglob("*"):\n+    if (path.suffix not in (".py", ".yml", ".yaml", ".json", ".md", ".txt")\n+            or ".git" in path.parts or "tests" in path.parts\n+            or "site" in path.parts):\n+        continue\n+    text = path.read_text(encoding="utf-8", errors="ignore")\n+    for marker in ("PERPLEXITY_API_KEY", "api.perplexity.ai",\n+                   "NEWSAPI", "SERPAPI", "TAVILY_API", "EXA_API",\n+                   "customsearch.googleapis"):\n+        if marker in text:\n+            _repo_hits.append(f"{path.name}:{marker}")\n+check("repo adds no Perplexity/paid-search keys or endpoints",\n+      _repo_hits == [])\n+_req = (REPO / "requirements.txt").read_text(encoding="utf-8").lower()\n+check("requirements.txt adds no paid news/search dependency",\n+      all(m not in _req for m in ("newsapi", "bing", "serpapi", "tavily",\n+                                  "exa", "perplexity")))\n+\n+# ── caps ─────────────────────────────────────────────────────────────────────\n+\n+reset()\n+write_sources(ok=True)\n+write_articles([make_article(f"a-cap{i}", f"容量測試{i}",\n+                             tpe(2026, 7, 27, 3, i),\n+                             orgs=(f"CapOrg{i}",), locs=(f"CapCity{i}",),\n+                             claims=(f"cap claim {i}",),\n+                             urls=(f"https://example.gov/cap{i}",))\n+                for i in range(9)])\n+briefing.main(["--edition", "morning", "--date", "2026-07-27"])\n+bm = load_brief("2026-07-27-morning")\n+check("per-section cap (6) enforced with an explicit warning",\n+      bm["item_count"] == 6\n+      and any("dropped by" in w for w in bm["warnings"]))\n+\n+print(f"\n{CHECKS} checks passed, {FAILED} failed"\n+      if not FAILED else f"\n{CHECKS - FAILED}/{CHECKS} passed, "\n+      f"{FAILED} FAILED")\n+sys.exit(1 if FAILED else 0)