完整差異
@@ -2,7 +2,7 @@ name: hourly-update\n \n on:\n schedule:\n- - cron: "0 * * * *" # every hour on the hour (UTC)\n+ - cron: "1 * * * *" # hourly; 16:01 UTC is 00:01 Asia/Taipei\n workflow_dispatch:\n push:\n branches: [main]\n@@ -65,6 +65,9 @@ jobs:\n python pipeline/dedupe.py\n python pipeline/write.py\n python pipeline/images.py\n+ # The saved Taipei date makes this one LLM-assisted refresh per day.\n+ # Missing/invalid model output falls back to published article titles.\n+ python pipeline/search_prompts.py --daily-if-due\n # a briefing run may have committed newer briefing JSONs since this\n # job checked out; deploys race (last wins), so refresh them before\n # building or this deploy can put a stale briefing page back live\n@@ -317,6 +317,15 @@ Planespotters.net (same airframe, by registration) or Wikimedia Commons\n caches per-article match/none results. build.normalize_image() is the\n only reader; templates always show kind + credit + license + backlink.\n \n+## data/search-prompts.json (written by search_prompts.py, read by build.py)\n+\n+One daily set of six bilingual search-box suggestions. `targetDateTpe` prevents\n+more than one generation per Taipei calendar day; `generationMode` is `llm`\n+only after every suggestion passes article-ID, length, safety, uniqueness and\n+title-anchor validation. `sourceArticleIds` records the published articles used.\n+When no provider is available or model output is invalid, deterministic prompts\n+are built from those article titles instead. No model reasoning is stored.\n+\n ## data/usage.json (written by usage.py, read by build.py)\n \n Cumulative LLM API spend: per-model-label {calls, inputTokens,\n@@ -20,6 +20,7 @@\n import time\n import unicodedata\n from datetime import timedelta, timezone\n+from functools import lru_cache\n from pathlib import Path\n from urllib.parse import urlsplit\n \n@@ -1001,6 +1002,24 @@ def collect_articles():\n return sorted(by_id.values(), key=lambda a: a["dt"], reverse=True)\n \n \n+@lru_cache(maxsize=2)\n+def header_search_placeholders(lang: str) -> tuple[str, ...]:\n+ """Load the daily LLM output, with the built-in prompts as fallback."""\n+ fallback = tuple(L[lang]["headerSearchPlaceholders"])\n+ payload = load_json(DATA_DIR / "search-prompts.json", {})\n+ prompts = ((payload.get("prompts") or {}).get(lang)\n+ if isinstance(payload, dict) else None)\n+ if not isinstance(prompts, list) or not 2 <= len(prompts) <= 8:\n+ return fallback\n+ cleaned = []\n+ for value in prompts:\n+ text = re.sub(r"\s+", " ", str(value or "")).strip()\n+ if not text or len(text) > 96 or re.search(r"[\x00-\x1f<>]", text):\n+ return fallback\n+ cleaned.append(text)\n+ return tuple(cleaned) if len(set(cleaned)) == len(cleaned) else fallback\n+\n+\n def notification_summary(value: str, lang: str) -> str:\n """Make legacy article summaries glanceable without changing source data."""\n text = re.sub(r"\s+", " ", str(value or "")).strip()\n@@ -2323,7 +2342,10 @@ def render_manual_workbench(env, build) -> int:\n # ── rendering ────────────────────────────────────────────────────────────────\n \n def base_ctx(lang, page, sub, *, title, description, ticker, build, hreflang=True):\n- t = L[lang]\n+ t = dict(L[lang])\n+ prompts = list(header_search_placeholders(lang))\n+ t["headerSearchPlaceholders"] = prompts\n+ t["headerSearchPlaceholder"] = prompts[0]\n nav_defs = [("home", ""), ("briefings", "briefings/"),\n ("radar", "radar/"),\n ("incidents", "incidents/"),\n@@ -0,0 +1,308 @@\n+"""Generate daily, article-grounded search-box prompts.\n+\n+The hourly workflow calls this once per run. A saved Taipei calendar date\n+keeps the LLM budget to one successful generation per day; the first run after\n+midnight refreshes the file. Invalid or unavailable model output always falls\n+back to deterministic prompts made from published article titles.\n+"""\n+from __future__ import annotations\n+\n+import argparse\n+import json\n+import os\n+import re\n+import sys\n+from datetime import datetime, timedelta, timezone\n+from pathlib import Path\n+\n+sys.path.insert(0, str(Path(__file__).resolve().parent))\n+from common import DATA_DIR, iso_minute, load_json, now_utc, parse_iso, save_json # noqa: E402\n+\n+TPE = timezone(timedelta(hours=8), "UTC+8")\n+OUTPUT_NAME = "search-prompts.json"\n+PROMPT_COUNT = 6\n+MAX_CANDIDATES = 12\n+MAX_ZH_CHARS = 26\n+MAX_EN_CHARS = 58\n+\n+PROMPT_SCHEMA = {\n+ "type": "object",\n+ "properties": {\n+ "prompts": {\n+ "type": "array",\n+ "items": {\n+ "type": "object",\n+ "properties": {\n+ "articleId": {"type": "string"},\n+ "zh": {"type": "string"},\n+ "en": {"type": "string"},\n+ },\n+ "required": ["articleId", "zh", "en"],\n+ "additionalProperties": False,\n+ },\n+ },\n+ },\n+ "required": ["prompts"],\n+ "additionalProperties": False,\n+}\n+\n+SYSTEM_PROMPT = """You create six short search-box suggestions for SKYTICAL.\n+Use ONLY the supplied published articles. Pick six distinct article IDs and\n+return JSON matching the schema. Each suggestion must clearly reuse a named\n+airline, flight number, airport, city, aircraft type, registration or other\n+searchable entity from that article's title. Do not add facts, dates, numbers,\n+causes or conclusions. Traditional Chinese must use Taiwan wording and be at\n+most 26 characters; English must be at most 58 characters. Do not use URLs,\n+HTML, markdown, relative-time words such as today/yesterday, or generic prompts.\n+The suggestions should feel friendly and natural, not like headlines."""\n+\n+_SPACE_RE = re.compile(r"\s+")\n+_ASCII_TOKEN_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9-]{1,}")\n+_CJK_RE = re.compile(r"[\u3400-\u9fff]+")\n+_UNSAFE_RE = re.compile(\n+ r"https?://|<|>|reasoning_content|\b(?:today|yesterday)\b|今天|昨天",\n+ re.IGNORECASE,\n+)\n+\n+\n+def _text(value) -> str:\n+ return _SPACE_RE.sub(" ", str(value or "")).strip()\n+\n+\n+def _article_time(raw: dict) -> datetime | None:\n+ try:\n+ return parse_iso(str(raw.get("publishedUtc") or ""))\n+ except (TypeError, ValueError):\n+ return None\n+\n+\n+def collect_candidates(articles_dir: Path, now: datetime) -> tuple[list[dict], str]:\n+ """Prefer today's Taipei stories, then month-to-date, then latest."""\n+ rows = []\n+ for path in sorted(articles_dir.glob("*.json")) if articles_dir.is_dir() else []:\n+ payload = load_json(path, {})\n+ article_rows = payload.get("articles") if isinstance(payload, dict) else []\n+ for raw in article_rows if isinstance(article_rows, list) else []:\n+ if not isinstance(raw, dict) or raw.get("archived") is True:\n+ continue\n+ article_id = _text(raw.get("id"))\n+ published = _article_time(raw)\n+ if not article_id or published is None or published > now + timedelta(hours=6):\n+ continue\n+ zh = raw.get("zh") if isinstance(raw.get("zh"), dict) else {}\n+ en = raw.get("en") if isinstance(raw.get("en"), dict) else {}\n+ zh_title = _text(zh.get("title") or en.get("title"))\n+ en_title = _text(en.get("title") or zh.get("title"))\n+ if not zh_title or not en_title:\n+ continue\n+ rows.append({\n+ "id": article_id,\n+ "publishedUtc": iso_minute(published),\n+ "publishedTpeDate": published.astimezone(TPE).strftime("%Y-%m-%d"),\n+ "zhTitle": zh_title,\n+ "zhSummary": _text(zh.get("summary"))[:240],\n+ "enTitle": en_title,\n+ "enSummary": _text(en.get("summary"))[:320],\n+ "_dt": published,\n+ })\n+\n+ by_id = {}\n+ for row in sorted(rows, key=lambda item: item["_dt"], reverse=True):\n+ by_id.setdefault(row["id"], row)\n+ ordered = list(by_id.values())\n+ today = now.astimezone(TPE).date()\n+ month_key = (today.year, today.month)\n+ today_rows = [row for row in ordered if row["_dt"].astimezone(TPE).date() == today]\n+ month_rows = [row for row in ordered\n+ if (row["_dt"].astimezone(TPE).year,\n+ row["_dt"].astimezone(TPE).month) == month_key]\n+\n+ selected = []\n+ for pool in (today_rows, month_rows, ordered):\n+ for row in pool:\n+ if row["id"] not in {item["id"] for item in selected}:\n+ selected.append(row)\n+ if len(selected) >= MAX_CANDIDATES:\n+ break\n+ if len(selected) >= MAX_CANDIDATES:\n+ break\n+\n+ if len(today_rows) >= PROMPT_COUNT:\n+ window = "today"\n+ elif month_rows:\n+ window = "month_to_date"\n+ else:\n+ window = "latest"\n+ for row in selected:\n+ row.pop("_dt", None)\n+ return selected, window\n+\n+\n+def _shares_anchor(prompt: str, title: str) -> bool:\n+ folded_prompt = prompt.casefold()\n+ tokens = {\n+ token.casefold() for token in _ASCII_TOKEN_RE.findall(title)\n+ if len(token) >= 3\n+ }\n+ if any(token in folded_prompt for token in tokens):\n+ return True\n+ for run in _CJK_RE.findall(title):\n+ if any(run[index:index + 2] in prompt\n+ for index in range(max(0, len(run) - 1))):\n+ return True\n+ return False\n+\n+\n+def validate_llm_prompts(draft, candidates: list[dict]) -> tuple[dict, list[str]] | None:\n+ rows = draft.get("prompts") if isinstance(draft, dict) else None\n+ if not isinstance(rows, list) or len(rows) != PROMPT_COUNT:\n+ return None\n+ source_by_id = {row["id"]: row for row in candidates}\n+ seen_ids, source_ids, zh_prompts, en_prompts = set(), [], [], []\n+ for row in rows:\n+ if not isinstance(row, dict):\n+ return None\n+ article_id = _text(row.get("articleId"))\n+ zh = _text(row.get("zh"))\n+ en = _text(row.get("en"))\n+ source = source_by_id.get(article_id)\n+ if source is None or article_id in seen_ids:\n+ return None\n+ if not (4 <= len(zh) <= MAX_ZH_CHARS and 8 <= len(en) <= MAX_EN_CHARS):\n+ return None\n+ if _UNSAFE_RE.search(zh) or _UNSAFE_RE.search(en):\n+ return None\n+ if not _shares_anchor(zh, source["zhTitle"]):\n+ return None\n+ if not _shares_anchor(en, source["enTitle"]):\n+ return None\n+ seen_ids.add(article_id)\n+ source_ids.append(article_id)\n+ zh_prompts.append(zh)\n+ en_prompts.append(en)\n+ if len(set(zh_prompts)) != PROMPT_COUNT or len(set(en_prompts)) != PROMPT_COUNT:\n+ return None\n+ return {"zh": zh_prompts, "en": en_prompts}, source_ids\n+\n+\n+def _truncate(text: str, limit: int) -> str:\n+ return text if len(text) <= limit else text[:limit - 1].rstrip() + "…"\n+\n+\n+def deterministic_prompts(candidates: list[dict]) -> tuple[dict, list[str]]:\n+ zh, en, source_ids = [], [], []\n+ for row in candidates:\n+ zh_prompt = _truncate(f"搜尋：{row['zhTitle']}", MAX_ZH_CHARS)\n+ en_prompt = _truncate(f"Explore: {row['enTitle']}", MAX_EN_CHARS)\n+ if zh_prompt in zh or en_prompt in en:\n+ continue\n+ zh.append(zh_prompt)\n+ en.append(en_prompt)\n+ source_ids.append(row["id"])\n+ if len(zh) >= PROMPT_COUNT:\n+ break\n+ generic_zh = [\n+ "搜尋班號，例如 CI100", "搜尋航空公司", "搜尋機場或城市",\n+ "搜尋機型，例如 A350", "探索本月航空新聞", "看看最新航空動態",\n+ ]\n+ generic_en = [\n+ "Search a flight number, e.g. CI100", "Search an airline",\n+ "Search an airport or city", "Search an aircraft type, e.g. A350",\n+ "Explore this month's aviation news", "See the latest aviation news",\n+ ]\n+ for zh_prompt, en_prompt in zip(generic_zh, generic_en):\n+ if len(zh) >= PROMPT_COUNT:\n+ break\n+ if zh_prompt in zh or en_prompt in en:\n+ continue\n+ zh.append(zh_prompt)\n+ en.append(en_prompt)\n+ return {"zh": zh, "en": en}, source_ids\n+\n+\n+def _provider_limit() -> int:\n+ try:\n+ return max(1, min(int(os.environ.get("SEARCH_PROMPTS_LLM_ATTEMPTS", "3")), 5))\n+ except ValueError:\n+ return 3\n+\n+\n+def update_daily_prompts(*, data_dir: Path = DATA_DIR, now: datetime | None = None,\n+ force: bool = False, providers=None,\n+ record_usage: bool = True) -> bool:\n+ now = now or now_utc()\n+ output_path = data_dir / OUTPUT_NAME\n+ target_date = now.astimezone(TPE).strftime("%Y-%m-%d")\n+ existing = load_json(output_path, {})\n+ if not force and isinstance(existing, dict) \\n+ and existing.get("targetDateTpe") == target_date:\n+ print(f"search-prompts: {target_date} already generated")\n+ return False\n+\n+ candidates, window = collect_candidates(data_dir / "articles", now)\n+ prompts, source_ids = deterministic_prompts(candidates)\n+ generation_mode, provider_label = "deterministic", None\n+ attempted = []\n+\n+ if len(candidates) >= PROMPT_COUNT:\n+ if providers is None:\n+ try:\n+ from providers import build_providers\n+ providers = build_providers()\n+ except Exception as exc:\n+ print(f"search-prompts: providers unavailable ({type(exc).__name__})")\n+ providers = []\n+ digest = json.dumps(candidates, ensure_ascii=False, separators=(",", ":"))\n+ for provider in list(providers or [])[:_provider_limit()]:\n+ attempted.append(provider)\n+ try:\n+ draft = provider.draft(SYSTEM_PROMPT, digest, PROMPT_SCHEMA)\n+ validated = validate_llm_prompts(draft, candidates)\n+ except Exception as exc:\n+ print(f"search-prompts: {getattr(provider, 'label', '?')} failed "\n+ f"({type(exc).__name__})")\n+ continue\n+ if validated is None:\n+ print(f"search-prompts: {getattr(provider, 'label', '?')} "\n+ "returned invalid prompts")\n+ continue\n+ prompts, source_ids = validated\n+ generation_mode = "llm"\n+ provider_label = str(getattr(provider, "label", "unknown"))[:160]\n+ break\n+\n+ if record_usage and attempted:\n+ try:\n+ import usage\n+ usage.record_providers(attempted)\n+ except Exception as exc:\n+ print(f"search-prompts: usage update failed ({type(exc).__name__})")\n+\n+ payload = {\n+ "version": 1,\n+ "generatedUtc": iso_minute(now),\n+ "targetDateTpe": target_date,\n+ "sourceWindow": window,\n+ "generationMode": generation_mode,\n+ "provider": provider_label,\n+ "sourceArticleIds": source_ids,\n+ "prompts": prompts,\n+ }\n+ save_json(output_path, payload)\n+ print(f"search-prompts: wrote {PROMPT_COUNT} prompts for {target_date} "\n+ f"({generation_mode}, {window})")\n+ return True\n+\n+\n+def main() -> int:\n+ parser = argparse.ArgumentParser()\n+ group = parser.add_mutually_exclusive_group()\n+ group.add_argument("--daily-if-due", action="store_true")\n+ group.add_argument("--force", action="store_true")\n+ args = parser.parse_args()\n+ update_daily_prompts(force=args.force)\n+ return 0\n+\n+\n+if __name__ == "__main__":\n+ raise SystemExit(main())\n@@ -89,22 +89,9 @@ def main() -> None:\n r"data-search-placeholders='([^']+)'", zh).group(1))\n en_placeholders = json.loads(re.search(\n r"data-search-placeholders='([^']+)'", en).group(1))\n- assert zh_placeholders == [\n- "搜尋班號，例如 CI100",\n- "搜尋航空公司",\n- "搜尋機場或城市",\n- "搜尋機型，例如 A350",\n- "隨意搜尋想看的 SKYTICAL 內容",\n- "今天想看什麼航空新聞？",\n- ]\n- assert en_placeholders == [\n- "Search a flight number, e.g. CI100",\n- "Search an airline",\n- "Search an airport or city",\n- "Search an aircraft type, e.g. A350",\n- "Explore anything on SKYTICAL",\n- "What aviation news are you looking for?",\n- ]\n+ assert zh_placeholders == list(build.header_search_placeholders("zh"))\n+ assert en_placeholders == list(build.header_search_placeholders("en"))\n+ assert len(zh_placeholders) == len(en_placeholders) == 6\n assert zh.count('id="news-search-form"') == 1\n assert 'class="header-search-form"' in home\n assert 'class="header-search-form"' in article\n@@ -0,0 +1,116 @@\n+"""Offline checks for daily article-grounded search suggestions."""\n+from __future__ import annotations\n+\n+import json\n+import sys\n+import tempfile\n+from datetime import datetime, timezone\n+from pathlib import Path\n+\n+ROOT = Path(__file__).resolve().parent.parent\n+sys.path.insert(0, str(ROOT / "pipeline"))\n+\n+import search_prompts # noqa: E402\n+\n+\n+class FakeProvider:\n+ label = "fake:test-model"\n+ http_calls = 1\n+ usage = {"inputTokens": 100, "outputTokens": 50, "usageEvents": 1}\n+\n+ def __init__(self):\n+ self.calls = 0\n+\n+ def draft(self, system_prompt, user_prompt, schema):\n+ self.calls += 1\n+ assert "ONLY the supplied published articles" in system_prompt\n+ assert schema == search_prompts.PROMPT_SCHEMA\n+ candidates = json.loads(user_prompt)\n+ return {"prompts": [\n+ {\n+ "articleId": row["id"],\n+ "zh": f"關注：{row['zhTitle']}",\n+ "en": f"Explore {row['enTitle']}",\n+ }\n+ for row in candidates[:6]\n+ ]}\n+\n+\n+def article(index: int) -> dict:\n+ return {\n+ "id": f"a-20260810-000{index}-test-{index}",\n+ "publishedUtc": f"2026-08-10T00:0{index}:00Z",\n+ "zh": {\n+ "title": f"航空{index} A32{index} 抵達桃園",\n+ "summary": f"航空{index}的 A32{index} 航機抵達桃園機場。",\n+ },\n+ "en": {\n+ "title": f"Airline {index} A32{index} arrives at Taoyuan",\n+ "summary": f"Airline {index}'s A32{index} arrived at Taoyuan.",\n+ },\n+ }\n+\n+\n+def main() -> None:\n+ now = datetime(2026, 8, 10, 6, 30, tzinfo=timezone.utc)\n+ with tempfile.TemporaryDirectory() as tmp:\n+ data_dir = Path(tmp)\n+ articles_dir = data_dir / "articles"\n+ articles_dir.mkdir()\n+ (articles_dir / "batch.json").write_text(\n+ json.dumps({"articles": [article(i) for i in range(8)]},\n+ ensure_ascii=False),\n+ encoding="utf-8",\n+ )\n+\n+ candidates, window = search_prompts.collect_candidates(articles_dir, now)\n+ assert len(candidates) == 8\n+ assert window == "today"\n+\n+ provider = FakeProvider()\n+ changed = search_prompts.update_daily_prompts(\n+ data_dir=data_dir, now=now, providers=[provider],\n+ record_usage=False)\n+ assert changed is True and provider.calls == 1\n+ payload = json.loads(\n+ (data_dir / "search-prompts.json").read_text(encoding="utf-8"))\n+ assert payload["targetDateTpe"] == "2026-08-10"\n+ assert payload["generationMode"] == "llm"\n+ assert payload["provider"] == "fake:test-model"\n+ assert len(payload["prompts"]["zh"]) == 6\n+ assert len(payload["sourceArticleIds"]) == 6\n+\n+ second_provider = FakeProvider()\n+ assert search_prompts.update_daily_prompts(\n+ data_dir=data_dir, now=now, providers=[second_provider],\n+ record_usage=False) is False\n+ assert second_provider.calls == 0\n+\n+ fallback_dir = data_dir / "fallback"\n+ fallback_articles = fallback_dir / "articles"\n+ fallback_articles.mkdir(parents=True)\n+ (fallback_articles / "batch.json").write_text(\n+ json.dumps({"articles": [article(i) for i in range(8)]},\n+ ensure_ascii=False),\n+ encoding="utf-8",\n+ )\n+ invalid = FakeProvider()\n+ invalid.draft = lambda *_args: {"prompts": []}\n+ search_prompts.update_daily_prompts(\n+ data_dir=fallback_dir, now=now, force=True, providers=[invalid],\n+ record_usage=False)\n+ fallback = json.loads(\n+ (fallback_dir / "search-prompts.json").read_text(encoding="utf-8"))\n+ assert fallback["generationMode"] == "deterministic"\n+ assert fallback["provider"] is None\n+ assert fallback["prompts"]["zh"][0].startswith("搜尋：航空7 A327")\n+\n+ workflow = (ROOT / ".github" / "workflows" / "hourly.yml").read_text(\n+ encoding="utf-8")\n+ assert 'cron: "1 * * * *"' in workflow\n+ assert "python pipeline/search_prompts.py --daily-if-due" in workflow\n+ print("test_search_prompts: OK")\n+\n+\n+if __name__ == "__main__":\n+ main()
