完整差異
@@ -51,6 +51,8 @@ jobs:\n node --check static/traffic-dashboard.js\n python tests/test_traffic_dashboard.py\n python tests/test_scope.py\n+ python tests/test_fetch_health.py\n+ python tests/test_evaair.py\n python tests/test_companion.py\n python tests/test_flightwatch.py\n python tests/test_airport_codes.py\n@@ -0,0 +1,24 @@\n+Google News signed lookup protocol context adapted from:\n+https://github.com/SSujitX/google-news-url-decoder\n+\n+MIT License\n+\n+Copyright (c) 2024 Sujit Biswas\n+\n+Permission is hereby granted, free of charge, to any person obtaining a copy\n+of this software and associated documentation files (the "Software"), to deal\n+in the Software without restriction, including without limitation the rights\n+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell\n+copies of the Software, and to permit persons to whom the Software is\n+furnished to do so, subject to the following conditions:\n+\n+The above copyright notice and this permission notice shall be included in all\n+copies or substantial portions of the Software.\n+\n+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\n+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,\n+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE\n+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER\n+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\n+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE\n+SOFTWARE.\n@@ -1,6 +1,19 @@\n # SKYTICAL Open-source Scouting\n \n-更新日期：2026-10-18\n+更新日期：2026-10-23\n+\n+## 日本突發飛安搜尋與來源零筆提示（2026-10-23）\n+\n+本次透過 GitHub API 查證維護狀態、授權與 README，並實測日本語 RSS：\n+\n+| 專案 | 最近推送 | Stars | 語言／授權 | 比較結果 |\n+| --- | --- | --- | --- | --- |\n+| [GNews](https://github.com/ranahaani/GNews) | 2026-10-09 | 1,018 | Python／MIT | 可參考語言、國家與查詢設定；SKYTICAL 已有 RSS 與來源署名流程，不需再加封裝。 |\n+| [google-news-scraper](https://github.com/lewisdonovan/google-news-scraper) | 2026-10-15 | 384 | JavaScript／MIT | v3 改用 RSS，部分功能仍需 Puppeteer；增加 Node／瀏覽器流程不符本次需求。 |\n+| [feedparser](https://github.com/kurtmckee/feedparser) | 2026-10-06 | 2,448 | Python／BSD 類授權，上游 API 標示 NOASSERTION | 已有依賴，保留現行 RSS 正規化；零筆健康判斷屬 SKYTICAL 作業狀態，放在既有抓取流程。 |\n+| [google-news-url-decoder](https://github.com/SSujitX/google-news-url-decoder) | 2026-10-18 | 302 | Python／MIT | 實測確認站內轉址後需簽章查詢才能取得原文。僅在既有 resolver 修補必要協定，不導入套件、批次系統或非同步架構；協定 context 的授權聲明保存在 `docs/licenses/google-news-url-decoder.txt`。 |\n+\n+保留現有 requests、feedparser、BeautifulSoup 與來源／正文核驗。新增日本語搜尋與少量日文航空用語，所有解析失敗的 Google 連結仍不進入待稿；重要媒體連續三次成功抓取卻無可用新聞時才提示，低頻官方公告不套用此警示。原文解析相容修補屬現有功能 bugfix，沒有引入新套件或替換子系統。\n \n ## 撰稿模型搜尋評估（2026-10-18）\n \n@@ -75,7 +75,11 @@ Array, one entry per key in `common.SOURCES`, in registry order:\n "cover": {"zh": "...", "en": "..."},\n "public": true, // false = editorial input only; hidden from public source-status UI\n "lastFetchUtc": "2026-07-26T05:02Z", // last SUCCESSFUL fetch (persisted across runs)\n- "ok": true\n+ "ok": true, // HTTP/parsing success, not coverage proof\n+ "itemCount": 0, // usable items; null on failure\n+ "emptyRuns": 3, // monitored feeds only\n+ "emptyWarning": true, // 3 consecutive successful empty fetches\n+ "lastNonEmptyUtc": null // last successful fetch with usable items\n }\n ]\n ```\n@@ -2357,6 +2357,8 @@ def merged_sources(raw):\n "cover": s.get("cover") or reg["cover"],\n "lastFetchUtc": s.get("lastFetchUtc"),\n "ok": bool(s.get("ok", False)),\n+ "emptyWarning": bool(s.get("emptyWarning", False)),\n+ "emptyRuns": s.get("emptyRuns", 0),\n })\n return out\n \n@@ -2385,6 +2387,10 @@ def sources_rows_view(sources, lang: str, now):\n "kind": _bi(KIND.get(s["kind"]), lang, s["kind"]),\n "cover": _bi(s["cover"], lang),\n "last": ago(s["lastFetchUtc"], now),\n+ "emptyWarning": s.get("emptyWarning", False),\n+ "emptyNote": (f"連續 {s.get('emptyRuns', 0)} 次未取得新聞"\n+ if lang == "zh" else\n+ f"No usable news in {s.get('emptyRuns', 0)} consecutive fetches"),\n } for s in public_display_sources(sources)]\n \n \n@@ -156,6 +156,12 @@ def group_has_material(group: dict) -> bool:\n "華信航空", "華信",\n )\n \n+# Local-language safety discovery complements fixed publisher feeds.\n+JAPAN_SAFETY_QUERY = (\n+ '(航空機 OR 貨物機 OR 旅客機 OR 空港 OR 滑走路 OR ヘリコプター) '\n+ '(火災 OR 炎 OR 緊急着陸 OR 胴体着陸 OR 事故 OR 閉鎖) when:3d'\n+)\n+\n # Google News discovery query for Asian-airline fleet and identity changes.\n # The fixed publisher feeds are strong on regulators and large western\n # outlets but routinely miss Chinese/Japanese/Korean carrier developments.\n@@ -301,6 +307,7 @@ def group_has_material(group: dict) -> bool:\n },\n },\n "reuters": {\n+ "monitor_empty": True,\n "name": "Reuters",\n "kind": "media",\n "fmt": "RSS",\n@@ -318,6 +325,7 @@ def group_has_material(group: dict) -> bool:\n },\n },\n "avherald": {\n+ "monitor_empty": True,\n "name": "The Aviation Herald",\n "kind": "media",\n "fmt": "RSS",\n@@ -335,6 +343,7 @@ def group_has_material(group: dict) -> bool:\n },\n },\n "aerospaceglobalnews": {\n+ "monitor_empty": True,\n "name": "Aerospace Global News",\n "kind": "media",\n "fmt": "HTML",\n@@ -412,6 +421,23 @@ def group_has_material(group: dict) -> bool:\n "en": "Hainan Airlines and HNA filings — fleet, network and operations",\n },\n },\n+ "japansafety": {\n+ "name": "日本突發飛安新聞索引",\n+ "kind": "media",\n+ "fmt": "RSS",\n+ "url": "https://news.google.com",\n+ "endpoint": "https://news.google.com/rss/search?" + urlencode({\n+ "q": JAPAN_SAFETY_QUERY, "hl": "ja", "gl": "JP", "ceid": "JP:ja",\n+ }),\n+ "type": "rss",\n+ "publisher_passthrough": True,\n+ "scope_filter": True,\n+ "monitor_empty": True,\n+ "cover": {\n+ "zh": "日本機場起火、緊急降落與跑道封閉等飛安事件",\n+ "en": "Breaking aviation safety events reported in Japan",\n+ },\n+ },\n "asiaairlinefleet": {\n "name": "亞洲航空機隊新聞索引",\n "kind": "media",\n@@ -437,6 +463,7 @@ def group_has_material(group: dict) -> bool:\n },\n },\n "simpleflying": {\n+ "monitor_empty": True,\n "name": "Simple Flying",\n "kind": "media",\n "fmt": "RSS",\n@@ -449,6 +476,7 @@ def group_has_material(group: dict) -> bool:\n },\n },\n "aerotime": {\n+ "monitor_empty": True,\n "name": "AeroTime",\n "kind": "media",\n "fmt": "RSS",\n@@ -587,6 +615,9 @@ def _load_manufacturer_source_catalog() -> dict:\n # subject. Keep these terms specific enough that generic politics/defence\n # stories (budgets, speeches, presidential inspections, land exercises) fail.\n _TRANSPORT_ZH_TERMS = (\n+ # Japanese reports use different characters for airport/cargo aircraft.\n+ "航空機", "貨物機", "旅客機", "空港", "滑走路", "緊急着陸",\n+ "胴体着陸", "ヘリコプター",\n # civil and military aviation\n "航空", "民航", "飛安", "航班", "航機", "飛機", "小型機", "客機",\n "貨機", "機場", "飛航", "空域", "空管", "航管", "塔台",\n@@ -11,6 +11,7 @@\n from __future__ import annotations\n \n import calendar\n+import json\n import os\n import re\n import time\n@@ -55,6 +56,7 @@ def _matches_keywords(item: dict, keywords) -> bool:\n RETRY_DELAY = 2 # seconds between the two attempts\n RESOLVE_TIMEOUT = 6 # seconds for google-news link resolution\n MAX_RESOLVE_FAILURES = 3 # stop resolving after this many misses in a row\n+EMPTY_WARNING_RUNS = 3\n MAX_RSS_SCAN_ITEMS = 200 # broad feeds are filtered after parsing\n SUMMARY_MAX = 500\n GOOGLE_NEWS_HOST = "news.google.com"\n@@ -186,6 +188,12 @@ def _conditional_get(session: requests.Session, key: str, url: str,\n cache: dict) -> requests.Response:\n """GET with If-None-Match / If-Modified-Since from the validator cache."""\n entry = cache.get(key) or {}\n+ previous = load_json(RAW_DIR / f"{key}.json", {})\n+ if (SOURCES.get(key, {}).get("monitor_empty")\n+ and isinstance(previous, dict) and previous.get("items") == []):\n+ # Reparse empty snapshots after resolver/parser recovery even when\n+ # the upstream feed itself has not changed.\n+ entry = {}\n headers = {}\n if entry.get("url") == url: # ignore validators from an older endpoint\n if entry.get("etag"):\n@@ -247,20 +255,70 @@ def _clean_gnews_title(title: str, entry) -> str:\n \n \n def _resolve_gnews_link(session: requests.Session, link: str) -> str | None:\n- """Follow one redirect off news.google.com; None when it stays on Google."""\n+ """Resolve a publisher URL without admitting Google discovery pages.\n+\n+ Modern Google News links need a signed lookup after an internal redirect.\n+ Protocol context adapted from SSujitX/google-news-url-decoder (MIT);\n+ see docs/licenses/google-news-url-decoder.txt.\n+ """\n try:\n- resp = session.get(link, timeout=RESOLVE_TIMEOUT, allow_redirects=False)\n- except requests.RequestException:\n+ for _ in range(3):\n+ resp = session.get(link, timeout=RESOLVE_TIMEOUT, allow_redirects=False)\n+ location = resp.headers.get("Location")\n+ if resp.status_code in (301, 302, 303, 307, 308) and location:\n+ target = urljoin(link, location)\n+ if _publisher_url(target):\n+ return target\n+ if urlsplit(target).hostname != GOOGLE_NEWS_HOST:\n+ return None\n+ link = target\n+ continue\n+ if resp.status_code != 200:\n+ return None\n+ node = BeautifulSoup(resp.text, "lxml").select_one(\n+ "[data-n-a-id][data-n-a-ts][data-n-a-sg]")\n+ if node is None:\n+ return None\n+ context = [\n+ ["X", "X", ["X", "X"], None, None, 1, 1, "US:en", None,\n+ 1, None, None, None, None, None, 0, 1],\n+ "X", "X", 1, [1, 1, 1], 1, 1, None, 0, 0, None, 0,\n+ ]\n+ request = ["garturlreq", context, node["data-n-a-id"],\n+ int(node["data-n-a-ts"]), node["data-n-a-sg"]]\n+ decoded = session.post(\n+ "https://news.google.com/_/DotsSplashUi/data/batchexecute",\n+ data={"f.req": json.dumps([[\n+ ["Fbv4je", json.dumps(request), None, "0"]]])},\n+ timeout=RESOLVE_TIMEOUT, allow_redirects=False,\n+ )\n+ if decoded.status_code != 200:\n+ return None\n+ # The RPC uses an anti-XSSI prefix and optional length lines.\n+ for line in decoded.text.splitlines():\n+ if not line.startswith("[["):\n+ continue\n+ for row in json.loads(line):\n+ if (isinstance(row, list) and len(row) >= 3\n+ and row[0] == "wrb.fr" and row[1] == "Fbv4je"):\n+ payload = json.loads(row[2])\n+ if (isinstance(payload, list) and len(payload) >= 2\n+ and payload[0] == "garturlres"\n+ and _publisher_url(payload[1])):\n+ return payload[1]\n+ return None\n+ except (requests.RequestException, ValueError, TypeError, KeyError):\n return None\n- location = resp.headers.get("Location")\n- if resp.status_code in (301, 302, 303, 307, 308) and location:\n- location = urljoin(link, location)\n- host = urlsplit(location).netloc.lower()\n- if host and not (host == "google.com" or host.endswith(".google.com")):\n- return location\n return None\n \n \n+def _publisher_url(url: str) -> bool:\n+ parsed = urlsplit(url)\n+ host = parsed.hostname or ""\n+ return (parsed.scheme in {"http", "https"} and bool(host)\n+ and not (host == "google.com" or host.endswith(".google.com")))\n+\n+\n def _rss_items(session: requests.Session, key: str, entries: list,\n endpoint: str, max_items: int = MAX_ITEMS_PER_SOURCE) -> list[dict]:\n """Normalize feed entries to raw items (published as datetime or None)."""\n@@ -684,7 +742,18 @@ def _write_sources(last_success: dict[str, str | None]) -> None:\n output = []\n for key, spec in SOURCES.items():\n fetched = last_success.get(key)\n- carried = (previous.get(key) or {}).get("lastFetchUtc")\n+ old = previous.get(key) or {}\n+ carried = old.get("lastFetchUtc")\n+ snapshot = load_json(RAW_DIR / f"{key}.json", {})\n+ items = snapshot.get("items") if isinstance(snapshot, dict) else None\n+ count = len(items) if fetched and isinstance(items, list) else None\n+ streak = (int(old.get("emptyRuns") or 0) + 1\n+ if spec.get("monitor_empty") and count == 0 else 0)\n+ warning = streak >= EMPTY_WARNING_RUNS\n+ if warning:\n+ print(f"::warning title=News source has no usable items::{key}: "\n+ f"{streak} consecutive successful fetches returned 0 usable "\n+ "items; check discovery, parsing and original-link resolution")\n output.append({\n "key": key,\n "name": spec["name"],\n@@ -695,6 +764,10 @@ def _write_sources(last_success: dict[str, str | None]) -> None:\n "public": bool(spec.get("public", True)),\n "lastFetchUtc": fetched or carried,\n "ok": bool(fetched),\n+ "itemCount": count,\n+ "emptyRuns": streak,\n+ "emptyWarning": warning,\n+ "lastNonEmptyUtc": fetched if count else old.get("lastNonEmptyUtc"),\n })\n save_json(sources_path, output)\n ok_count = sum(1 for entry in output if entry["ok"])\n@@ -94,6 +94,11 @@\n "www.aerotime.aero", "aerotime.aero",\n "www.aerospaceglobalnews.com", "aerospaceglobalnews.com",\n "www.flightglobal.com", "flightglobal.com",\n+ # Japanese safety discovery: selected editorial publishers only;\n+ # robots and final redirect allowlisting remain mandatory.\n+ "www.chunichi.co.jp", "chunichi.co.jp",\n+ "newsdigest.jp", "www.fnn.jp", "newsdig.tbs.co.jp",\n+ "news.web.nhk", "www3.nhk.or.jp",\n }\n # NEVER add simpleflying.com or other Valnet properties: their robots.txt\n # terms explicitly prohibit automated text/data mining and AI use of page\n@@ -19,7 +19,9 @@ <h2 style="margin:8px 0 6px;">{{ t.srcTitle }}</h2>\n <td>60 min</td>\n <td class="text-muted" style="white-space:nowrap;">{{ s.last }}</td>\n <td>\n- {% if s.ok %}\n+ {% if s.emptyWarning %}\n+ <span class="text-muted" style="font-size:12px;">{{ s.emptyNote }}</span>\n+ {% elif s.ok %}\n <span style="display:inline-flex;align-items:center;gap:6px;font-size:12px;"><span class="dot"></span>OK</span>\n {% else %}\n <span class="text-muted" style="display:inline-flex;align-items:center;gap:6px;font-size:12px;"><span class="dot err"></span>ERR</span>\n@@ -53,7 +53,7 @@ def test_google_app_shell_without_direct_redirect_is_dropped(self):\n class Session:\n def get(self, url, **kwargs):\n return type("Response", (), {\n- "status_code": 200, "headers": {}})()\n+ "status_code": 200, "headers": {}, "text": ""})()\n \n items = fetch._rss_items(\n Session(), "google", [{\n@@ -0,0 +1,166 @@\n+"""Offline regression tests for safety discovery and empty-source warnings."""\n+import contextlib\n+import io\n+import json\n+import sys\n+import tempfile\n+import unittest\n+from datetime import datetime, timezone\n+from pathlib import Path\n+from types import SimpleNamespace\n+from unittest.mock import Mock, patch\n+from urllib.parse import parse_qs, urlsplit\n+\n+from jinja2 import DictLoader, Environment\n+\n+ROOT = Path(__file__).resolve().parents[1]\n+sys.path.insert(0, str(ROOT / "pipeline"))\n+import build\n+import common\n+import dedupe\n+import fetch\n+import fulltext\n+\n+\n+def response(status=200, text="", location=None):\n+ return SimpleNamespace(status_code=status, text=text,\n+ headers={"Location": location} if location else {})\n+\n+\n+class DiscoveryTests(unittest.TestCase):\n+ def test_japanese_safety_headlines_and_attribution(self):\n+ source = common.SOURCES["japansafety"]\n+ query = parse_qs(urlsplit(source["endpoint"]).query)\n+ self.assertEqual(query["ceid"], ["JP:ja"])\n+ self.assertIn("緊急着陸", query["q"][0])\n+ self.assertTrue(source["scope_filter"])\n+ for title in ("中部空港 着陸した貨物機から炎 滑走路を一時閉鎖",\n+ "ヘリコプターが緊急着陸", "旅客機が胴体着陸"):\n+ self.assertTrue(common.is_transport_headline({"title": title}))\n+ self.assertFalse(common.is_transport_headline({\n+ "title": "名古屋市の住宅で火災", "summary": "空港の近く。"}))\n+ self.assertEqual(dedupe._source_label(source, {"publisher": "中日新聞Web"}),\n+ "中日新聞Web")\n+ self.assertIn("www.chunichi.co.jp", fulltext.ALLOWED_HOSTS)\n+ self.assertNotIn("news.google.com", fulltext.ALLOWED_HOSTS)\n+\n+ def test_signed_resolution_after_internal_redirect(self):\n+ session = Mock()\n+ session.get.side_effect = [\n+ response(302, location="https://news.google.com/rss/articles/id?hl=ja"),\n+ response(text='<div data-n-a-id="id" data-n-a-ts="123" '\n+ 'data-n-a-sg="signature"></div>'),\n+ ]\n+ session.post.return_value = response(text=\n+ ")]}'\n\n123\n" + json.dumps([[\n+ "wrb.fr", "Fbv4je", json.dumps([\n+ "garturlres", "https://www.chunichi.co.jp/article/123"])\n+ ]]) + "\n")\n+ self.assertEqual(fetch._resolve_gnews_link(\n+ session, "https://news.google.com/rss/articles/id"),\n+ "https://www.chunichi.co.jp/article/123")\n+ self.assertFalse(session.post.call_args.kwargs["allow_redirects"])\n+ payload = json.loads(session.post.call_args.kwargs["data"]["f.req"])\n+ self.assertEqual(json.loads(payload[0][0][1])[-3:],\n+ ["id", 123, "signature"])\n+\n+ def test_failed_or_unsafe_resolution_never_enters_raw_items(self):\n+ for target in ("https://news.google.com/read/id", "javascript:alert(1)"):\n+ session = Mock()\n+ session.get.return_value = response(\n+ text='<div data-n-a-id="id" data-n-a-ts="123" '\n+ 'data-n-a-sg="sig"></div>')\n+ session.post.return_value = response(text=json.dumps([[\n+ "wrb.fr", "Fbv4je", json.dumps(["garturlres", target])]]))\n+ self.assertEqual(fetch._rss_items(session, "japansafety", [{\n+ "title": "中部空港 貨物機から炎", "link":\n+ "https://news.google.com/rss/articles/id"}],\n+ common.SOURCES["japansafety"]["endpoint"]), [])\n+\n+ def test_redirect_loop_and_malformed_rpc_are_bounded(self):\n+ session = Mock()\n+ session.get.return_value = response(\n+ 302, location="https://news.google.com/rss/articles/id")\n+ self.assertIsNone(fetch._resolve_gnews_link(\n+ session, "https://news.google.com/rss/articles/id"))\n+ self.assertEqual(session.get.call_count, 3)\n+ session.get.return_value = response(\n+ text='<div data-n-a-id="id" data-n-a-ts="123" data-n-a-sg="s"></div>')\n+ session.post.return_value = response(text="[[broken")\n+ self.assertIsNone(fetch._resolve_gnews_link(\n+ session, "https://news.google.com/rss/articles/id"))\n+\n+\n+class SourceHealthTests(unittest.TestCase):\n+ def test_empty_snapshot_is_reparsed_instead_of_cached_304(self):\n+ with tempfile.TemporaryDirectory() as folder:\n+ directory = Path(folder)\n+ (directory / "avherald.json").write_text('{"items": []}')\n+ url = common.SOURCES["avherald"]["endpoint"]\n+ cache = {"avherald": {"url": url, "etag": "old"}}\n+ with patch.object(fetch, "RAW_DIR", directory), \\n+ patch.object(fetch, "_request", return_value=response()) as request:\n+ fetch._conditional_get(object(), "avherald", url, cache)\n+ self.assertEqual(request.call_args.args[2], {})\n+\n+ def test_empty_streak_warning_recovery_and_failure(self):\n+ with tempfile.TemporaryDirectory() as folder:\n+ directory = Path(folder)\n+ raw = directory / "raw"\n+ raw.mkdir()\n+ with patch.object(fetch, "DATA_DIR", directory), \\n+ patch.object(fetch, "RAW_DIR", raw), \\n+ patch.object(fetch, "SOURCES", {"aerotime": common.SOURCES["aerotime"]}):\n+ def run(stamp, items, success=True):\n+ (raw / "aerotime.json").write_text(json.dumps({"items": items}))\n+ stream = io.StringIO()\n+ with contextlib.redirect_stdout(stream):\n+ fetch._write_sources({"aerotime": stamp if success else None})\n+ row = json.loads((directory / "sources.json").read_text(encoding="utf-8"))\n+ return row[0], stream.getvalue()\n+\n+ for n in range(1, 4):\n+ row, log = run(f"2026-10-23T0{n}:00Z", [])\n+ self.assertEqual(row["emptyRuns"], n)\n+ self.assertEqual(row["emptyWarning"], n == 3)\n+ self.assertEqual("::warning" in log, n == 3)\n+ self.assertTrue(row["ok"])\n+ merged = build.merged_sources([row])\n+ env = Environment(loader=DictLoader({\n+ "base.html": "{% block content %}{% endblock %}",\n+ "sources.html": (ROOT / "templates/sources.html").read_text(encoding="utf-8"),\n+ }), autoescape=True)\n+ for language, expected in [("zh", "連續 3 次"), ("en", "3 consecutive")]:\n+ view = build.sources_rows_view(\n+ merged, language, datetime.now(timezone.utc))\n+ alert = next(s for s in view if s["name"] == "AeroTime")\n+ self.assertTrue(alert["emptyWarning"])\n+ rendered = env.get_template("sources.html").render(rows=[alert], t={})\n+ self.assertIn(expected, rendered)\n+ self.assertNotIn(">OK</span>", rendered)\n+\n+ row, _ = run("2026-10-23T04:00Z", [{"title": "Aircraft update"}])\n+ self.assertEqual(row["emptyRuns"], 0)\n+ self.assertFalse(row["emptyWarning"])\n+ self.assertEqual(row["lastNonEmptyUtc"], "2026-10-23T04:00Z")\n+ row, _ = run("2026-10-23T05:00Z", [], success=False)\n+ self.assertFalse(row["ok"])\n+ self.assertIsNone(row["itemCount"])\n+ self.assertEqual(row["emptyRuns"], 0)\n+ self.assertEqual(row["lastFetchUtc"], "2026-10-23T04:00Z")\n+\n+ def test_low_cadence_official_sources_do_not_warn(self):\n+ with tempfile.TemporaryDirectory() as folder:\n+ directory = Path(folder)\n+ (directory / "faa.json").write_text('{"items": []}')\n+ with patch.object(fetch, "DATA_DIR", directory), \\n+ patch.object(fetch, "RAW_DIR", directory), \\n+ patch.object(fetch, "SOURCES", {"faa": common.SOURCES["faa"]}):\n+ for _ in range(4):\n+ fetch._write_sources({"faa": "2026-10-23T01:00Z"})\n+ row = json.loads((directory / "sources.json").read_text(encoding="utf-8"))[0]\n+ self.assertFalse(row["emptyWarning"])\n+\n+\n+if __name__ == "__main__":\n+ unittest.main()
