提交原始紀錄

完整差異

@@ -53,6 +53,7 @@ jobs:\n           python pipeline/fetch.py\n           python pipeline/dedupe.py\n           python pipeline/write.py\n+          python pipeline/images.py\n           python pipeline/build.py\n \n       - name: Scan build output and committed data for secrets/reasoning\n@@ -0,0 +1,16 @@\n+{\n+  "articles": {\n+    "a-20260726-1805-faa-announces-successful-counter-drone-e": {\n+      "status": "none",\n+      "attempts": 1,\n+      "checked_utc": "2026-07-26T20:56:22.386600+00:00",\n+      "next_retry_utc": "2026-07-27T02:56:22.386600+00:00"\n+    },\n+    "a-20260726-1748-caa-taiwan-announces-1-340-extra-flights": {\n+      "status": "none",\n+      "attempts": 1,\n+      "checked_utc": "2026-07-26T20:56:22.386600+00:00",\n+      "next_retry_utc": "2026-07-27T02:56:22.386600+00:00"\n+    }\n+  }\n+}\n@@ -245,3 +245,32 @@ site/\n ```\n \n All internal links are prefixed with `common.BASE_PATH` (default `/avwire`).\n+\n+## data/briefings/  (written by briefing.py, read by build.py)\n+\n+```\n+data/briefings/\n+├─ index.json            {"briefings": [{briefing_id, date, edition,\n+│                          status published|partial|failed, cutoff_time,\n+│                          generated_at, item_count}]}\n+├─ <date>-<edition>.json one briefing per edition (schema in briefing.py:\n+│                          fixed half-open Asia/Taipei window, sections,\n+│                          checked_sources, warnings, generation_mode)\n+└─ event_index.json      cross-edition dedup: {"events": {event_id:\n+                           {fingerprint, urls, fact_keys, updates, ...}}}\n+```\n+\n+Briefings select only from data/articles/ (already verified/published);\n+failed runs never modify briefing files and never downgrade a\n+published/partial index row. build.py renders /briefings/ and\n+/briefings/<id>/ for published|partial only.\n+\n+## data/images.json + article "image"  (written by images.py)\n+\n+images.py (after write.py) may set article.image =\n+{url, link, credit, license, provider,\n+ kind airframe_photo|file_photo, matched} — an external embed from\n+Planespotters.net (same airframe, by registration) or Wikimedia Commons\n+(free-licensed file photo), never a claimed event photo. images.json\n+caches per-article match/none results. build.normalize_image() is the\n+only reader; templates always show kind + credit + license + backlink.\n@@ -66,6 +66,9 @@\n         "refreshNote": "本文隨來源每小時更新", "sourcesBox": "資料來源 Sources",\n         "genNote": "本文由自動化系統彙整生成，內容以原始來源為準。",\n         "aiModel": "AI 撰稿模型",\n+        "photoKind": {"airframe_photo": "同機資料照片",\n+                      "file_photo": "資料照片（非事件現場照片）"},\n+        "photoSource": "照片來源",\n         "incKicker": "Incident Database", "incTitle": "事故資料庫",\n         "incSub": "自動彙整全球民航事故與事件，資料來自各國調查機關與專業追報來源，每小時更新。",\n         "incCountLabel": "筆紀錄",\n@@ -118,6 +121,9 @@\n         "refreshNote": "Refreshes hourly with sources", "sourcesBox": "Sources",\n         "genNote": "This article was compiled automatically; the original sources prevail.",\n         "aiModel": "Drafted by",\n+        "photoKind": {"airframe_photo": "File photo of this airframe",\n+                      "file_photo": "File photo (not from this event)"},\n+        "photoSource": "Photo source",\n         "incKicker": "Incident Database", "incTitle": "Incident database",\n         "incSub": "Automatically aggregated civil-aviation accidents and incidents, from state investigators and specialist trackers. Updated hourly.",\n         "incCountLabel": "records",\n@@ -606,6 +612,38 @@ def _safe_web_url(url: str) -> bool:\n         return False\n \n \n+def normalize_image(raw_img):\n+    """Article image -> {url, link, credit, license, provider, kind} or None.\n+\n+    Accepts the images.py match dict or a plain URL string (legacy).\n+    Only http(s) URLs survive; a credited external embed keeps its\n+    attribution and backlink so CC/photographer terms are honored.\n+    """\n+    if isinstance(raw_img, dict):\n+        url = str(raw_img.get("url") or "")\n+        if not url or not _safe_web_url(url):\n+            return None\n+        link = str(raw_img.get("link") or "")\n+        return {\n+            "url": url,\n+            "link": link if link and _safe_web_url(link) else None,\n+            "credit": str(raw_img.get("credit") or "") or None,\n+            "license": str(raw_img.get("license") or "") or None,\n+            "provider": str(raw_img.get("provider") or "") or None,\n+            "kind": (raw_img.get("kind")\n+                     if raw_img.get("kind") in ("airframe_photo",\n+                                                "file_photo")\n+                     else "file_photo"),\n+        }\n+    if raw_img:\n+        url = str(raw_img)\n+        if not _safe_web_url(url):\n+            return None\n+        return {"url": url, "link": None, "credit": None, "license": None,\n+                "provider": None, "kind": "file_photo"}\n+    return None\n+\n+\n def rel_path(lang: str, sub: str) -> str:\n     return ("en/" + sub) if lang == "en" else sub\n \n@@ -699,9 +737,7 @@ def side(own, other):\n         return None\n \n     cat = raw.get("cat") if isinstance(raw.get("cat"), str) else "ops"\n-    image = str(raw["image"]) if raw.get("image") else None\n-    if image and not _safe_web_url(image):\n-        image = None\n+    image = normalize_image(raw.get("image"))\n     # Display the SOURCE's newest publication time when the write stage\n     # recorded one; the generation time (dt) is used for ordering only, so\n     # a days-old official release is never presented as breaking news.\n@@ -802,9 +838,7 @@ def collect_agg_items(now):\n             if k in seen:\n                 continue\n             seen.add(k)\n-            image = str(it["image"]) if it.get("image") else None\n-            if image and not _safe_web_url(image):\n-                image = None\n+            image = normalize_image(it.get("image"))\n             items.append({\n                 "dt": dt, "title": title, "url": url,\n                 "summary": str(it.get("summary") or "").strip(),\n@@ -0,0 +1,345 @@\n+"""images.py — attach a real, matching photo to published articles.\n+\n+Runs after write.py. For each recent article without an image it tries,\n+in order:\n+\n+  1. planespotters.net public photos API (no key) by aircraft\n+     registration found in the article — a photo of the SAME airframe\n+     (kind "airframe_photo"). Free with photographer credit + backlink,\n+     which the site renders under the image.\n+  2. Wikimedia Commons API (no key) with a conservative query built only\n+     from entities the article actually verified (airline + aircraft\n+     type, else airport, else airline). Only freely-licensed results\n+     (CC*/public domain) are accepted, credited and linked (kind\n+     "file_photo" — labelled as a file photo, never as an event photo).\n+\n+The photo is embedded by URL from the origin service (no files are\n+committed); a match is stored on the article JSON and cached in\n+data/images.json. No confident match -> the article simply keeps no\n+image. Any network failure is logged and skipped: this stage may never\n+fail the pipeline, never calls a model, and never requires an API key.\n+"""\n+from __future__ import annotations\n+\n+import re\n+import sys\n+from datetime import timedelta\n+from html import unescape\n+from pathlib import Path\n+\n+import requests\n+\n+sys.path.insert(0, str(Path(__file__).resolve().parent))\n+from common import (  # noqa: E402\n+    ARTICLES_DIR,\n+    DATA_DIR,\n+    USER_AGENT,\n+    load_json,\n+    now_utc,\n+    parse_iso,\n+    save_json,\n+)\n+\n+ROOT = Path(__file__).resolve().parent.parent\n+CACHE_PATH = DATA_DIR / "images.json"\n+TIMEOUT = (10, 30)\n+HEADERS = {"User-Agent": USER_AGENT}\n+\n+# Politeness caps per run; articles are few and the cache is permanent.\n+MAX_LOOKUPS_PER_RUN = 6\n+MAX_ATTEMPTS = 3          # per article before we stop retrying "none"\n+RETRY_AFTER_HOURS = 6\n+MAX_ARTICLE_AGE_DAYS = 14  # never backfill ancient batches\n+\n+PLANESPOTTERS_URL = "https://api.planespotters.net/pub/photos/reg/{reg}"\n+COMMONS_API = "https://commons.wikimedia.org/w/api.php"\n+\n+_FREE_LICENSE_RE = re.compile(r"\b(cc[ -]|cc0|public domain|pd-)", re.I)\n+_TAG_RE = re.compile(r"<[^>]+>")\n+# Registrations, kept deliberately conservative to avoid false hits in\n+# prose (e.g. plain "N95" is NOT matched): TW B-#####, US N-regs with a\n+# letter suffix or 4-5 digits, JA####, and common hyphenated prefixes.\n+_REG_RE = re.compile(\n+    r"\b(B-\d{4,5}|N\d{2,4}[A-Z]{1,2}|N\d{4,5}|JA\d{3,4}[A-Z]?"\n+    r"|(?:HL|VH|HS|9V|9M|D|F|G|PH|EI|OE|VT|A6|A7|TC)-[A-Z]{3,4})\b")\n+\n+_airlines = [\n+    a for a in (load_json(ROOT / "config" / "airline_icao_codes.json", {})\n+                .get("airlines") or [])\n+    if isinstance(a, dict) and a.get("airline_name_en")\n+]\n+_types = {\n+    str(k): str(v)\n+    for k, v in (load_json(ROOT / "config" / "aircraft_types.json", {})\n+                 .get("types") or {}).items()\n+}\n+_tw_airports = [\n+    a for a in (load_json(ROOT / "config" / "tw_civil_airports.json", {})\n+                .get("airports") or [])\n+    if isinstance(a, dict)\n+]\n+\n+\n+def _article_text(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+        parts.extend(str(p) for p in block.get("body") or [])\n+    for fact in article.get("facts") or []:\n+        if isinstance(fact, dict):\n+            parts.append(str(fact.get("claim") or ""))\n+            parts.append(str(fact.get("sourceQuote") or ""))\n+    ents = article.get("entities") or {}\n+    if isinstance(ents, dict):\n+        for group in ents.values():\n+            if isinstance(group, list):\n+                parts.extend(str(v) for v in group if v)\n+    return " ".join(parts)\n+\n+\n+def find_registration(article: dict) -> str | None:\n+    match = _REG_RE.search(_article_text(article))\n+    return match.group(1) if match else None\n+\n+\n+def find_airline(article: dict) -> str | None:\n+    text = _article_text(article)\n+    low = text.casefold()\n+    for a in _airlines:\n+        en = str(a.get("airline_name_en") or "")\n+        zh = str(a.get("airline_name_zh_tw") or "")\n+        if (en and en.casefold() in low) or (zh and zh in text):\n+            return en or zh\n+    return None\n+\n+\n+def find_aircraft_type(article: dict) -> str | None:\n+    """Display name for a type the article mentions.\n+\n+    Accepts the full name ("Airbus A350-900"), the bare model\n+    ("A350-900"), the ICAO code ("A359"), or the family ("A350") — the\n+    family match returns the family-level name so the photo query never\n+    claims a sub-variant the article did not verify.\n+    """\n+    text = _article_text(article)\n+    low = text.casefold()\n+    for code, name in _types.items():\n+        maker = name.split()[0]\n+        bare = name.split()[-1]\n+        if name.casefold() in low or (len(bare) >= 4\n+                                      and bare.casefold() in low):\n+            return name\n+        if len(code) >= 4 and re.search(rf"\b{re.escape(code)}\b", text):\n+            return name\n+        family = bare.split("-")[0]\n+        if (len(family) >= 4 and family.casefold() != bare.casefold()\n+                and re.search(rf"\b{re.escape(family)}\b", text, re.I)):\n+            return f"{maker} {family}"\n+    return None\n+\n+\n+def find_airport(article: dict) -> str | None:\n+    text = _article_text(article)\n+    low = text.casefold()\n+    for ap in _tw_airports:\n+        for key in ("name_zh", "name_en"):\n+            val = str(ap.get(key) or "")\n+            if val and (val in text or val.casefold() in low):\n+                return str(ap.get("name_en") or val)\n+    return None\n+\n+\n+def _strip_html(value: str) -> str:\n+    return re.sub(r"\s+", " ", _TAG_RE.sub(" ", unescape(value))).strip()\n+\n+\n+# ── providers ────────────────────────────────────────────────────────────────\n+\n+def lookup_planespotters(reg: str) -> dict | None:\n+    """Photo of the exact airframe by registration; None when unavailable."""\n+    resp = requests.get(PLANESPOTTERS_URL.format(reg=reg),\n+                        headers=HEADERS, timeout=TIMEOUT)\n+    if resp.status_code != 200:\n+        return None\n+    photos = (resp.json() or {}).get("photos") or []\n+    for photo in photos:\n+        thumb = ((photo.get("thumbnail_large") or photo.get("thumbnail"))\n+                 or {}).get("src")\n+        link = photo.get("link")\n+        if not thumb or not link:\n+            continue\n+        return {\n+            "url": str(thumb),\n+            "link": str(link),\n+            "credit": str(photo.get("photographer") or "").strip() or None,\n+            "license": None,  # per-photographer; the backlink is the record\n+            "provider": "Planespotters.net",\n+            "kind": "airframe_photo",\n+            "matched": reg,\n+        }\n+    return None\n+\n+\n+def lookup_commons(query: str, require_tokens: list[str]) -> dict | None:\n+    """First freely-licensed Commons bitmap matching the query.\n+\n+    require_tokens: at least one must appear in the file title, so a\n+    generic search can never attach an unrelated photo.\n+    """\n+    resp = requests.get(COMMONS_API, headers=HEADERS, timeout=TIMEOUT,\n+                        params={\n+                            "action": "query", "format": "json",\n+                            "generator": "search", "gsrnamespace": 6,\n+                            "gsrlimit": 8,\n+                            "gsrsearch": f"filetype:bitmap {query}",\n+                            "prop": "imageinfo",\n+                            "iiprop": "url|extmetadata",\n+                            "iiurlwidth": 1280,\n+                        })\n+    if resp.status_code != 200:\n+        return None\n+    pages = ((resp.json() or {}).get("query") or {}).get("pages") or {}\n+    tokens = [t.casefold() for t in require_tokens if t]\n+    for page in sorted(pages.values(), key=lambda p: p.get("index", 99)):\n+        title = str(page.get("title") or "")\n+        if tokens and not any(t in title.casefold() for t in tokens):\n+            continue\n+        for info in page.get("imageinfo") or []:\n+            meta = info.get("extmetadata") or {}\n+            license_name = _strip_html(\n+                str((meta.get("LicenseShortName") or {}).get("value") or ""))\n+            if not _FREE_LICENSE_RE.search(license_name):\n+                continue\n+            thumb = info.get("thumburl") or info.get("url")\n+            if not thumb:\n+                continue\n+            artist = _strip_html(\n+                str((meta.get("Artist") or {}).get("value") or ""))[:80]\n+            return {\n+                "url": str(thumb),\n+                "link": str(info.get("descriptionshorturl")\n+                            or info.get("descriptionurl") or thumb),\n+                "credit": artist or None,\n+                "license": license_name or None,\n+                "provider": "Wikimedia Commons",\n+                "kind": "file_photo",\n+                "matched": query,\n+            }\n+    return None\n+\n+\n+def resolve_image(article: dict) -> dict | None:\n+    """Best honest match for one article, or None. Network errors bubble."""\n+    reg = find_registration(article)\n+    if reg:\n+        photo = lookup_planespotters(reg)\n+        if photo:\n+            return photo\n+    airline = find_airline(article)\n+    actype = find_aircraft_type(article)\n+    airport = find_airport(article)\n+    if airline and actype:\n+        # family token ("A350") so Commons titles with any sub-variant match\n+        type_token = actype.split()[-1].split("-")[0]\n+        return lookup_commons(f'{airline} {actype}', [type_token, airline])\n+    if actype:\n+        type_token = actype.split()[-1].split("-")[0]\n+        return lookup_commons(f'{actype} aircraft', [type_token])\n+    if airline:\n+        return lookup_commons(f'{airline} aircraft', [airline])\n+    if airport:\n+        return lookup_commons(f'{airport}', [airport])\n+    return None  # nothing visual verified in this article -> no image\n+\n+\n+# ── batch update ─────────────────────────────────────────────────────────────\n+\n+def _recent_batches():\n+    cutoff = now_utc() - timedelta(days=MAX_ARTICLE_AGE_DAYS)\n+    for path in sorted(ARTICLES_DIR.glob("*.json"), reverse=True):\n+        batch = load_json(path, None)\n+        if not isinstance(batch, dict):\n+            continue\n+        articles = [a for a in batch.get("articles") or []\n+                    if isinstance(a, dict)]\n+        fresh = False\n+        for a in articles:\n+            try:\n+                if parse_iso(str(a.get("publishedUtc"))) >= cutoff:\n+                    fresh = True\n+            except (ValueError, TypeError):\n+                continue\n+        if fresh:\n+            yield path, batch, articles\n+\n+\n+def main() -> int:\n+    if not ARTICLES_DIR.is_dir():\n+        print("images: no articles yet")\n+        return 0\n+    cache = load_json(CACHE_PATH, {})\n+    entries = cache.get("articles")\n+    if not isinstance(entries, dict):\n+        entries = {}\n+    cache = {"articles": entries}\n+    now = now_utc()\n+    lookups = attached = 0\n+\n+    for path, batch, articles in _recent_batches():\n+        changed = False\n+        for article in articles:\n+            art_id = str(article.get("id") or "")\n+            if not art_id or article.get("image"):\n+                continue\n+            entry = entries.get(art_id) or {}\n+            if entry.get("status") == "none":\n+                if entry.get("attempts", 0) >= MAX_ATTEMPTS:\n+                    continue\n+                try:\n+                    if (entry.get("next_retry_utc")\n+                            and parse_iso(entry["next_retry_utc"]) > now):\n+                        continue\n+                except ValueError:\n+                    pass\n+            if entry.get("status") == "matched" and entry.get("image"):\n+                article["image"] = entry["image"]\n+                changed = True\n+                attached += 1\n+                continue\n+            if lookups >= MAX_LOOKUPS_PER_RUN:\n+                continue\n+            lookups += 1\n+            try:\n+                image = resolve_image(article)\n+            except Exception as exc:  # enhancement only: never fail the run\n+                print(f"images: lookup failed for {art_id}: "\n+                      f"{type(exc).__name__}: {exc}")\n+                continue\n+            if image:\n+                article["image"] = image\n+                changed = True\n+                attached += 1\n+                entries[art_id] = {"status": "matched", "image": image,\n+                                   "checked_utc": now.isoformat()}\n+            else:\n+                attempts = entry.get("attempts", 0) + 1\n+                entries[art_id] = {\n+                    "status": "none", "attempts": attempts,\n+                    "checked_utc": now.isoformat(),\n+                    "next_retry_utc":\n+                        (now + timedelta(hours=RETRY_AFTER_HOURS))\n+                        .isoformat(),\n+                }\n+        if changed:\n+            save_json(path, batch)\n+\n+    if cache != load_json(CACHE_PATH, {}):\n+        save_json(CACHE_PATH, cache)\n+    print(f"images: {lookups} lookup(s), {attached} article(s) got a photo")\n+    return 0\n+\n+\n+if __name__ == "__main__":\n+    sys.exit(main())\n@@ -11,6 +11,19 @@ <h1 style="font-size:38px;line-height:1.18;margin:14px 0 12px;">{{ a.title }}</h\n         <span style="margin-left:auto;">{{ t.refreshNote }}</span>\n       </div>\n     </div>\n+    {% if a.image %}\n+    <figure style="margin:0 0 22px;">\n+      <img class="grayscale" src="{{ a.image.url }}" alt="{{ a.title }}" loading="lazy" style="width:100%;max-height:420px;object-fit:cover;border:1px solid var(--color-divider);">\n+      <figcaption class="text-muted" style="font-size:11.5px;margin-top:6px;display:flex;gap:10px;flex-wrap:wrap;">\n+        <span>{{ t.photoKind[a.image.kind] }}</span>\n+        {% if a.image.credit %}<span>© {{ a.image.credit }}</span>{% endif %}\n+        {% if a.image.license %}<span>{{ a.image.license }}</span>{% endif %}\n+        {% if a.image.link %}\n+        <a href="{{ a.image.link }}" target="_blank" rel="noopener">{{ a.image.provider or t.photoSource }}</a>\n+        {% elif a.image.provider %}<span>{{ a.image.provider }}</span>{% endif %}\n+      </figcaption>\n+    </figure>\n+    {% endif %}\n     {% for p in a.body %}\n     <p style="font-size:16px;line-height:1.95;margin:0 0 18px;">{{ p }}</p>\n     {% endfor %}\n@@ -38,7 +38,8 @@ <h6 style="margin:28px 0 12px;">{{ t.sourceStatus }}</h6>\n       {% if hero %}\n       <a href="{{ hero.url }}"{% if hero.external %} target="_blank" rel="noopener"{% endif %} style="display:block;color:inherit;text-decoration:none;cursor:pointer;">\n         {% if hero.image %}\n-        <img class="grayscale hero-img" src="{{ hero.image }}" alt="">\n+        <img class="grayscale hero-img" src="{{ hero.image.url }}" alt="">\n+        <div class="text-muted" style="font-size:10.5px;margin-top:4px;">{{ t.photoKind[hero.image.kind] }}{% if hero.image.credit %} · © {{ hero.image.credit }}{% endif %}{% if hero.image.license %} · {{ hero.image.license }}{% endif %}{% if hero.image.provider %} · {{ hero.image.provider }}{% endif %}</div>\n         {% else %}\n         <div class="grayscale hero-ph">\n           <span class="text-muted" style="font-family:monospace;font-size:11px;">{{ t.heroImg }}</span>\n@@ -0,0 +1,272 @@\n+"""Tests for pipeline/images.py (auto photo matching, offline).\n+\n+No pytest required — run from the repo root:\n+\n+    py tests\\test_images.py\n+\n+All HTTP is mocked; no real Planespotters/Wikimedia call is ever made.\n+"""\n+from __future__ import annotations\n+\n+import json\n+import os\n+import shutil\n+import sys\n+import tempfile\n+import types\n+from datetime import timedelta, timezone\n+from pathlib import Path\n+\n+REPO = Path(__file__).resolve().parent.parent\n+TMP = Path(tempfile.mkdtemp(prefix="avwire-images-"))\n+os.environ["AVWIRE_DATA_DIR"] = str(TMP)\n+\n+sys.path.insert(0, str(REPO / "pipeline"))\n+import images  # noqa: E402\n+import build  # noqa: E402\n+from common import now_utc  # noqa: E402\n+\n+CHECKS = 0\n+FAILED = 0\n+\n+\n+def check(name, cond):\n+    global CHECKS, FAILED\n+    CHECKS += 1\n+    print(("PASS " if cond else "FAIL ") + name)\n+    if not cond:\n+        FAILED += 1\n+\n+\n+# ── HTTP mock ────────────────────────────────────────────────────────────────\n+\n+class FakeResp:\n+    def __init__(self, status=200, data=None):\n+        self.status_code = status\n+        self._data = data or {}\n+\n+    def json(self):\n+        return self._data\n+\n+\n+class FakeHttp:\n+    def __init__(self):\n+        self.calls = []\n+        self.planespotters = FakeResp(200, {"photos": []})\n+        self.commons = FakeResp(200, {})\n+\n+    def get(self, url, **kwargs):\n+        self.calls.append((url, kwargs.get("params")))\n+        if "planespotters" in url:\n+            return self.planespotters\n+        return self.commons\n+\n+\n+fake = FakeHttp()\n+images.requests = types.SimpleNamespace(get=fake.get)\n+\n+PS_HIT = FakeResp(200, {"photos": [{\n+    "thumbnail_large": {"src": "https://cdn.planespotters.net/x_large.jpg"},\n+    "link": "https://www.planespotters.net/photo/123",\n+    "photographer": "Jane Spotter",\n+}]})\n+COMMONS_HIT = FakeResp(200, {"query": {"pages": {\n+    "1": {"index": 2, "title": "File:Random building.jpg", "imageinfo": [{\n+        "thumburl": "https://upload.wikimedia.org/no.jpg",\n+        "extmetadata": {"LicenseShortName": {"value": "CC BY-SA 4.0"},\n+                        "Artist": {"value": "X"}}}]},\n+    "2": {"index": 1, "title": "File:Delta Air Lines Airbus A350-941.jpg",\n+          "imageinfo": [{\n+              "thumburl": "https://upload.wikimedia.org/delta-a350.jpg",\n+              "descriptionshorturl": "https://commons.wikimedia.org/w/1",\n+              "extmetadata": {\n+                  "LicenseShortName": {"value": "CC BY-SA 4.0"},\n+                  "Artist": {"value": "<a href='#'>Some Photographer</a>"},\n+              }}]},\n+    "3": {"index": 0, "title": "File:Delta A350 copyrighted.jpg",\n+          "imageinfo": [{\n+              "thumburl": "https://upload.wikimedia.org/nonfree.jpg",\n+              "extmetadata": {\n+                  "LicenseShortName": {"value": "All rights reserved"},\n+                  "Artist": {"value": "Y"}}}]},\n+}}})\n+\n+\n+def make_article(art_id, body_en, published=None, image=None):\n+    published = published or now_utc()\n+    return {\n+        "id": art_id,\n+        "publishedUtc": published.astimezone(timezone.utc)\n+        .strftime("%Y-%m-%dT%H:%MZ"),\n+        "cat": "ops", "primarySource": "T", "image": image,\n+        "zh": {"title": "標題", "summary": "摘要", "body": ["內文"]},\n+        "en": {"title": body_en, "summary": body_en, "body": [body_en]},\n+        "sources": [{"name": "S", "url": "https://example.gov/x"}],\n+        "writer": "nvidia:test", "facts": [], "riskFlags": [],\n+        "eventStatus": "confirmed",\n+        "entities": {"organizations": [], "locations": [],\n+                     "event_dates": []},\n+    }\n+\n+\n+def write_batch(articles, fname="b.json"):\n+    p = TMP / "articles" / fname\n+    p.parent.mkdir(parents=True, exist_ok=True)\n+    p.write_text(json.dumps({"articles": articles}, ensure_ascii=False),\n+                 encoding="utf-8")\n+    return p\n+\n+\n+def read_batch(p):\n+    return json.loads(p.read_text(encoding="utf-8"))["articles"]\n+\n+\n+def reset():\n+    shutil.rmtree(TMP / "articles", ignore_errors=True)\n+    try:\n+        (TMP / "images.json").unlink()\n+    except FileNotFoundError:\n+        pass\n+    fake.calls.clear()\n+    fake.planespotters = FakeResp(200, {"photos": []})\n+    fake.commons = FakeResp(200, {})\n+\n+\n+# ── entity extraction guards ─────────────────────────────────────────────────\n+\n+check("registration found in article text",\n+      images.find_registration(make_article(\n+          "a", "Delta A350 registration N512DN arrived")) == "N512DN")\n+check("plain N95 is never treated as a registration",\n+      images.find_registration(make_article(\n+          "a", "Airport staff wore N95 masks")) is None)\n+check("airline dictionary match (Delta Air Lines)",\n+      images.find_airline(make_article(\n+          "a", "A Delta Air Lines flight arrived")) == "Delta Air Lines")\n+check("aircraft family match never over-claims a sub-variant",\n+      images.find_aircraft_type(make_article(\n+          "a", "The Airbus A350 landed")) == "Airbus A350")\n+check("ICAO type code A359 maps to its display name",\n+      images.find_aircraft_type(make_article(\n+          "a", "DAL flight, type A359, arrived")) == "Airbus A350-900")\n+check("no visual entity -> resolve_image returns None without HTTP",\n+      (fake.calls.clear() is None\n+       and images.resolve_image(make_article("a", "FAA policy update"))\n+       is None and fake.calls == []))\n+\n+# ── tier 1: planespotters by registration ────────────────────────────────────\n+\n+reset()\n+fake.planespotters = PS_HIT\n+p = write_batch([make_article("a-reg", "Delta A350 N512DN arrived")])\n+images.main()\n+art = read_batch(p)[0]\n+check("registration match embeds the same-airframe photo",\n+      art["image"]["url"] == "https://cdn.planespotters.net/x_large.jpg"\n+      and art["image"]["kind"] == "airframe_photo"\n+      and art["image"]["provider"] == "Planespotters.net")\n+check("photographer credit and backlink preserved",\n+      art["image"]["credit"] == "Jane Spotter"\n+      and art["image"]["link"] == "https://www.planespotters.net/photo/123")\n+\n+# ── tier 2: commons with license + token filters ────────────────────────────\n+\n+reset()\n+fake.commons = COMMONS_HIT\n+p = write_batch([make_article("a-com", "Delta Air Lines Airbus A350 news")])\n+images.main()\n+art = read_batch(p)[0]\n+check("commons picks the free-licensed, token-matching file",\n+      art["image"]["url"] == "https://upload.wikimedia.org/delta-a350.jpg"\n+      and art["image"]["license"] == "CC BY-SA 4.0"\n+      and art["image"]["kind"] == "file_photo")\n+check("artist HTML stripped in credit",\n+      art["image"]["credit"] == "Some Photographer")\n+check("non-free and token-mismatched files rejected",\n+      "nonfree" not in json.dumps(art["image"])\n+      and "Random building" not in json.dumps(art["image"]))\n+\n+# ── cache behaviour ──────────────────────────────────────────────────────────\n+\n+reset()\n+fake.commons = COMMONS_HIT\n+p = write_batch([make_article("a-cache", "Delta Air Lines Airbus A350")])\n+images.main()\n+n_first = len(fake.calls)\n+arts = read_batch(p)\n+for a in arts:  # simulate the next hourly run on a fresh article copy\n+    a["image"] = None\n+write_batch(arts)\n+fake.calls.clear()\n+images.main()\n+check("matched result is reused from cache with zero HTTP calls",\n+      n_first >= 1 and fake.calls == []\n+      and read_batch(p)[0]["image"]["kind"] == "file_photo")\n+\n+reset()\n+p = write_batch([make_article("a-none", "Delta Air Lines Airbus A350")])\n+fake.commons = FakeResp(200, {})  # no results\n+images.main()\n+fake.calls.clear()\n+images.main()\n+check("negative result is not retried before the retry window",\n+      fake.calls == [] and read_batch(p)[0]["image"] is None)\n+\n+reset()\n+existing = {"url": "https://example.org/keep.jpg", "kind": "file_photo"}\n+p = write_batch([make_article("a-keep", "Delta Air Lines Airbus A350",\n+                              image=dict(existing))])\n+images.main()\n+check("existing images are never overwritten or re-looked-up",\n+      read_batch(p)[0]["image"] == existing and fake.calls == [])\n+\n+# ── network failure never breaks the pipeline ────────────────────────────────\n+\n+reset()\n+write_batch([make_article("a-err", "Delta Air Lines Airbus A350")])\n+\n+\n+def _boom(url, **kw):\n+    raise OSError("network down")\n+\n+\n+images.requests = types.SimpleNamespace(get=_boom)\n+check("network failure -> exit 0, article simply keeps no image",\n+      images.main() == 0)\n+images.requests = types.SimpleNamespace(get=fake.get)\n+\n+# ── build-side normalization & honest labels ─────────────────────────────────\n+\n+img = build.normalize_image({"url": "https://upload.wikimedia.org/a.jpg",\n+                             "link": "https://commons.wikimedia.org/w/1",\n+                             "credit": "P", "license": "CC BY 4.0",\n+                             "provider": "Wikimedia Commons",\n+                             "kind": "file_photo"})\n+check("normalize_image keeps attribution fields",\n+      img["credit"] == "P" and img["license"] == "CC BY 4.0"\n+      and img["kind"] == "file_photo")\n+check("normalize_image rejects non-http urls",\n+      build.normalize_image({"url": "javascript:alert(1)"}) is None\n+      and build.normalize_image("data:text/html,x") is None)\n+check("legacy string image still accepted",\n+      build.normalize_image("https://x.example/y.jpg")["kind"]\n+      == "file_photo")\n+check("unknown kind coerced to file_photo (never fake 'event photo')",\n+      build.normalize_image({"url": "https://x.example/y.jpg",\n+                             "kind": "event_photo"})["kind"]\n+      == "file_photo")\n+check("honest file-photo labels present in both languages",\n+      "資料照片（非事件現場照片）" in\n+      (REPO / "pipeline" / "build.py").read_text(encoding="utf-8")\n+      and "File photo (not from this event)" in\n+      (REPO / "pipeline" / "build.py").read_text(encoding="utf-8"))\n+check("images.py needs no API key and calls no paid endpoint",\n+      all(m not in (REPO / "pipeline" / "images.py")\n+          .read_text(encoding="utf-8")\n+          for m in ("API_KEY", "apikey", "newsapi", "bing", "serpapi",\n+                    "perplexity")))\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)