修正供應鏈風險資料處理：分析狀態、去重、排程與保存一致性

.env.example

@@ -37,8 +37,13 @@ ERP_DEMO_MODE=false
 # then provision user_organizations and organization_entitlements before startup.
 # ERP_ORGANIZATION_ID=your-organization-id
 
-# Optional service identity for the 24-hour supply-chain news refresh.
+# Supply-chain scheduler: background execution is opt-in.
 # The account must have risk.workspace.write; for the local demo use planner.
+ERP_SCHEDULER_ENABLED=0
+ERP_SCHEDULER_INTERVAL_SECONDS=86400
+ERP_SCHEDULER_INITIAL_DELAY_SECONDS=10
+ERP_SCHEDULER_MAX_ATTEMPTS=3
+ERP_SCHEDULER_RETRY_SECONDS=30
 ERP_SCHEDULER_ACTOR=
 
 # Optional: seed Agent Dashboard with synthetic demo records. Keep disabled for real data.

.gitignore

@@ -232,3 +232,8 @@ line bot/ngrok.exe
 # Local AI-assistant and planning state must never enter the public repository.
 .claude/
 .planning/
+
+# Isolated batch-one runtime state
+.isolated/
+*.db.news.lock
+*.db.scheduler.lock

app.py

@@ -5,12 +5,14 @@
 所有商業邏輯均位於 backend/，所有 UI 頁面均位於 frontend/
 """
 
+import os
 import streamlit as st
 
 # ── 載入 .env（模型設定 LLM_MODEL / 各供應商金鑰，issue #25）────────────
 try:
     from dotenv import load_dotenv
-    load_dotenv()
+    if os.getenv("ERP_ISOLATED_TEST") != "1":
+        load_dotenv()
 except Exception:
     pass
 
@@ -30,6 +32,9 @@
 
 # ── 頁面設定 ────────────────────────────────────────────────────────
 st.set_page_config(page_title="進銷存安全系統", page_icon="🛡️", layout="wide")
+if os.getenv("ERP_ISOLATED_TEST") == "1":
+    news_mode = "真實新聞快照" if os.getenv("ERP_NEWS_CAPTURE") else "固定新聞"
+    st.warning(f"隔離測試環境｜{news_mode}與模擬 AI｜不發送通知｜背景排程關閉")
 
 # ── 全域 CSS ────────────────────────────────────────────────────────
 st.markdown("""

backend/agent_orchestrator.py

@@ -201,6 +201,11 @@ def _llm(messages, model=None, tools=None, temperature=0.2, json_mode=False,
     用量記帳：每次成功呼叫記一列 llm_usage_logs（tokens + 成本），
     usage_tag 標記用途（route / agent:<id> / aggregate / smalltalk）供歸因。
     """
+    if os.getenv("ERP_ISOLATED_TEST") == "1":
+        from types import SimpleNamespace
+        from .isolated_runtime import fixture_completion
+        msg = SimpleNamespace(content=fixture_completion(messages, usage_tag), tool_calls=None)
+        return SimpleNamespace(choices=[SimpleNamespace(message=msg)])
     kw = {"messages": messages, "temperature": temperature}
     if tools:
         kw["tools"] = tools

backend/database.py

@@ -170,6 +170,9 @@ def init_db():
         ai_summary TEXT,
         updated_at TEXT
     )''')
+    from .news_store import migrate as migrate_news
+    migrate_news(conn)
+
     c.execute('''CREATE TABLE IF NOT EXISTS esg_targets (id INTEGER PRIMARY KEY AUTOINCREMENT, target_year INTEGER, scope INTEGER, baseline_kg_co2 REAL, target_kg_co2 REAL, note TEXT)''')
     # 永續 ESG：風險管理係數（地區/事件類型/供應商類別 → 風險分數 0–100、權重）
     c.execute('''CREATE TABLE IF NOT EXISTS esg_risk_factors (id INTEGER PRIMARY KEY AUTOINCREMENT, risk_type TEXT, risk_key TEXT, risk_score REAL, weight REAL, note TEXT, updated_at TEXT, UNIQUE(risk_type, risk_key))''')

backend/isolated_runtime.py

@@ -0,0 +1,89 @@
+"""Deterministic local fixtures and a network guard for the isolated launcher."""
+import ipaddress
+import os
+import json
+import re
+import socket
+from pathlib import Path
+
+
+def news_capture():
+    """Optional real-source snapshot selected by the local review launcher."""
+    path = os.getenv("ERP_NEWS_CAPTURE", "")
+    return json.loads(Path(path).read_text(encoding="utf-8")) if path else None
+
+
+def block_external_network():
+    if getattr(socket, "_erp_isolated", False):
+        return
+    original_connect = socket.socket.connect
+    original_connect_ex = socket.socket.connect_ex
+    original_getaddrinfo = socket.getaddrinfo
+
+    def allowed(host):
+        if host == "localhost":
+            return True
+        try:
+            return ipaddress.ip_address(host).is_loopback
+        except ValueError:
+            return False
+
+    def check(address):
+        if isinstance(address, tuple) and not allowed(address[0]):
+            raise OSError("Isolated ERP: external network and notifications are disabled")
+
+    def connect(sock, address):
+        check(address)
+        return original_connect(sock, address)
+
+    def connect_ex(sock, address):
+        check(address)
+        return original_connect_ex(sock, address)
+
+    def getaddrinfo(host, *args, **kwargs):
+        if host is not None and not allowed(host):
+            raise OSError("Isolated ERP: external DNS is disabled")
+        return original_getaddrinfo(host, *args, **kwargs)
+
+    socket.socket.connect = connect
+    socket.socket.connect_ex = connect_ex
+    socket.getaddrinfo = getaddrinfo
+    socket._erp_isolated = True
+
+
+def fixture_news(country):
+    capture = news_capture()
+    if capture:
+        from .region_matching import normalize
+        return [dict(a) for a in capture["articles"] if normalize(a.get("country")) == normalize(country)]
+    rows = [
+        ("zero", "港口恢復營運 [ZERO]", "確認目前無延遲。"),
+        ("delay", "港口罷工 [DELAY]", "固定測試事件：延遲五天。"),
+        ("unknown", "交期尚未確認 [UNKNOWN]", "目前沒有可靠的延遲天數。"),
+        ("invalid", "模型輸出格式錯誤案例 [INVALID]", "保留此原文供人工檢查。"),
+    ]
+    if os.getenv("ERP_ISOLATED_SCENARIO", "mixed") == "success":
+        rows = [row for row in rows if row[0] != "invalid"]
+    return [dict(country=country, region=None, title=f"{country} {title}", summary=summary,
+                 url=f"https://fixture.invalid/{country}/{key}", source="固定測試資料",
+                 published_at="2026-09-13 08:00", relevance_tag="supply_chain")
+            for key, title, summary in rows]
+
+
+def fixture_completion(prompt, tag):
+    if tag == "analysis:news_batch":
+        results = []
+        for idx, body in re.findall(r"【新聞編號 (\d+)】\n(.*?)(?=【新聞編號|$)", str(prompt), re.S):
+            country = next((c for c in ("台灣", "日本", "美國", "南韓", "中國", "越南", "墨西哥", "德國", "新加坡") if body.startswith(c)), "台灣")
+            delay = 0 if "[ZERO]" in body else None if "[UNKNOWN]" in body else "invalid" if "[INVALID]" in body else 5
+            results.append({"news_id": int(idx), "相關性": "YES", "國家": country, "地區": "不明",
+                            "事件類型": "交通", "預計延遲": delay, "繁體中文簡要": "固定模擬分析；非即時新聞。"})
+        return json.dumps({"results": results}, ensure_ascii=False)
+    if tag == "analysis:heatmap":
+        return json.dumps({"摘要": "固定測試摘要：台灣北區確認為 0%／0 天，日本為 65%／5 天。",
+            "更新": [{"地區": "台灣 北區", "風險": 0}, {"地區": "日本", "風險": 65}],
+            "事件": [{"類型": "交通", "地區": "北區", "國家": "台灣", "延遲天數": 0, "描述": "固定零值測試"},
+                     {"類型": "罷工", "地區": "日本", "國家": "日本", "延遲天數": 5, "描述": "固定延遲測試"}]}, ensure_ascii=False)
+    if tag == "analysis:po_alternative":
+        return '{"results": []}'
+    return "隔離測試環境：這是固定模擬回應，未呼叫外部模型或發送通知。"

backend/job_lock.py

@@ -0,0 +1,36 @@
+"""Process-wide advisory file locks, released by the OS even on process death."""
+import os
+from contextlib import contextmanager
+from pathlib import Path
+
+
+@contextmanager
+def exclusive_job_lock(db_path, name):
+    path = Path(str(db_path) + f".{name}.lock")
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("a+b") as handle:
+        handle.seek(0, 2)
+        if handle.tell() == 0:
+            handle.write(b"0")
+            handle.flush()
+        handle.seek(0)
+        acquired = False
+        try:
+            if os.name == "nt":
+                import msvcrt
+                msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
+            else:
+                import fcntl
+                fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+            acquired = True
+        except (OSError, BlockingIOError):
+            pass
+        try:
+            yield acquired
+        finally:
+            if acquired:
+                handle.seek(0)
+                if os.name == "nt":
+                    msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
+                else:
+                    fcntl.flock(handle.fileno(), fcntl.LOCK_UN)

backend/llm_client.py

@@ -19,6 +19,8 @@
 
 def llm_available() -> bool:
     """是否已設定任何可用的模型供應商（.env 驅動）。"""
+    if os.getenv("ERP_ISOLATED_TEST") == "1":
+        return True
     return bool(os.getenv("LLM_MODEL") or os.getenv("OPENAI_API_KEY")
                 or os.getenv("GEMINI_API_KEY"))
 
@@ -30,6 +32,9 @@ def complete_text(prompt, system: str | None = None, temperature: float = 0.2,
     單次文字補全。prompt 可為字串或 messages list。
     回傳純文字（失敗拋例外，由呼叫端決定 fallback 行為）。
     """
+    if os.getenv("ERP_ISOLATED_TEST") == "1":
+        from .isolated_runtime import fixture_completion
+        return fixture_completion(prompt, tag)
     from backend.agent_orchestrator import _llm, _content
     from backend.prompts import PROMPT_DEFENSE_BASELINE
 

backend/news_store.py

@@ -0,0 +1,98 @@
+"""Raw news retention, durable analysis state and pre-analysis deduplication."""
+import hashlib
+import re
+import sqlite3
+import unicodedata
+from datetime import datetime, timezone
+from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
+
+
+def now():
+    return datetime.now(timezone.utc).isoformat()
+
+
+def identity_keys(item):
+    url = str(item.get("url") or "").strip()
+    if url:
+        parts = urlsplit(url)
+        query = sorted((k, v) for k, v in parse_qsl(parts.query, keep_blank_values=True)
+                       if not k.lower().startswith("utm_") and k.lower() not in {"fbclid", "gclid"})
+        url = urlunsplit((parts.scheme.lower(), parts.netloc.lower(), parts.path.rstrip("/"), urlencode(query), ""))
+    title = re.sub(r"\s+", " ", unicodedata.normalize("NFKC", str(item.get("title") or ""))).strip().casefold()
+    # Different syndication URLs with the same headline/source/publication day are one story.
+    content = "|".join((title, str(item.get("source") or "").strip().casefold(), str(item.get("published_at") or "")[:10])) if title else ""
+    digest = lambda value: hashlib.sha256(value.encode("utf-8")).hexdigest() if value else None
+    return digest(url), digest(content)
+
+
+def migrate(conn):
+    columns = {r[1] for r in conn.execute("PRAGMA table_info(supply_chain_news)")}
+    additions = {"analysis_status": "TEXT NOT NULL DEFAULT 'legacy_unverified'",
+                 "analysis_error": "TEXT", "analysis_summary": "TEXT", "analysis_country": "TEXT",
+                 "analysis_region": "TEXT", "analyzed_at": "TEXT", "url_key": "TEXT", "content_key": "TEXT"}
+    for name, declaration in additions.items():
+        if name not in columns:
+            conn.execute(f"ALTER TABLE supply_chain_news ADD COLUMN {name} {declaration}")
+    if "estimated_delay" not in {r[1] for r in conn.execute("PRAGMA table_info(risk_heatmap)")}:
+        conn.execute("ALTER TABLE risk_heatmap ADD COLUMN estimated_delay INTEGER")
+    # Keep historical duplicates and IDs (events may reference them); index only the first owner.
+    conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS news_url_unique ON supply_chain_news(url_key) WHERE url_key IS NOT NULL")
+    conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS news_content_unique ON supply_chain_news(content_key) WHERE content_key IS NOT NULL")
+    if "url_key" not in columns:
+        rows = conn.execute("SELECT id,title,url,source,published_at FROM supply_chain_news ORDER BY id").fetchall()
+        for row in rows:
+            uk, ck = identity_keys(dict(zip(("id", "title", "url", "source", "published_at"), row)))
+            for field, value in (("url_key", uk), ("content_key", ck)):
+                if value and not conn.execute(f"SELECT 1 FROM supply_chain_news WHERE {field}=?", (value,)).fetchone():
+                    conn.execute(f"UPDATE supply_chain_news SET {field}=? WHERE id=?", (value, row[0]))
+    conn.execute("""CREATE TABLE IF NOT EXISTS scheduled_jobs (
+        job_key TEXT PRIMARY KEY, status TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0,
+        started_at TEXT, finished_at TEXT, error TEXT, result_json TEXT)""")
+
+
+def find_existing(conn, item):
+    uk, ck = identity_keys(item)
+    return conn.execute("SELECT id,analysis_status FROM supply_chain_news WHERE url_key=? OR content_key=? ORDER BY id LIMIT 1", (uk, ck)).fetchone()
+
+
+def store_raw(conn, item):
+    existing = find_existing(conn, item)
+    if existing:
+        return existing[0], False
+    uk, ck = identity_keys(item)
+    if not (uk or ck):
+        raise ValueError("News needs a title or URL")
+    values = [item.get(k) for k in ("country", "region", "title", "summary", "url", "source", "published_at", "relevance_tag")]
+    try:
+        cur = conn.execute("""INSERT INTO supply_chain_news
+            (country,region,title,summary,url,source,published_at,relevance_tag,fetched_at,
+             analysis_status,is_relevant,estimated_delay,url_key,content_key)
+            VALUES (?,?,?,?,?,?,?,?,?,'pending',NULL,NULL,?,?)""", (*values, now(), uk, ck))
+        return cur.lastrowid, True
+    except sqlite3.IntegrityError:
+        existing = find_existing(conn, item)
+        if existing:
+            return existing[0], False
+        raise
+
+
+def store_analysis(conn, news_id, result):
+    from .risk_validation import number, text, EVENT_TYPES
+    success = result.get("analysis_status") == "succeeded"
+    delay = number(result.get("estimated_delay"), maximum=365, integer=True, nullable=True) if success else None
+    if success:
+        if type(result.get("is_relevant")) is not bool:
+            raise ValueError("Invalid analysis relevance")
+        for key in ("country", "region", "chinese_summary"):
+            text(result.get(key))
+        if result.get("event_type") not in EVENT_TYPES:
+            raise ValueError("Invalid event type")
+        if not result["is_relevant"] and delay not in (None, 0):
+            raise ValueError("Irrelevant news cannot have delay")
+    conn.execute("""UPDATE supply_chain_news SET analysis_status=?,analysis_error=?,
+        analysis_summary=?,analysis_country=?,analysis_region=?,category=?,is_relevant=?,
+        estimated_delay=?,analyzed_at=? WHERE id=? AND analysis_status!='succeeded'""",
+        ("succeeded" if success else "failed", None if success else result.get("analysis_error", "invalid_output"),
+         result.get("chinese_summary") if success else None, result.get("country") if success else None,
+         result.get("region") if success else None, result.get("event_type") if success else None,
+         int(result["is_relevant"]) if success else None, delay, now(), news_id))

backend/prompts.py

@@ -44,7 +44,7 @@
     {{"地區": "<合法區域名稱>", "風險": <0-100 整數>}}
   ],
   "事件": [
-    {{"類型": "<戰爭|氣候|罷工|政策|交通|其他>", "地區": "<合法區域名稱>", "國家": "<國家名>", "延遲天數": <整數>, "描述": "<一句話>"}}
+    {{"類型": "<戰爭|氣候|罷工|政策|交通|其他>", "地區": "<合法區域名稱>", "國家": "<國家名>", "延遲天數": <0-365 整數或 null，未知用 null，確認無延遲用 0>, "描述": "<一句話>"}}
   ]
 }}"""
 
@@ -64,7 +64,7 @@
 【輸出格式】只輸出以下 JSON 物件（頂層必須是物件、不要其他文字），results 共 {impact_count} 筆：
 {{
   "results": [
-    {{"po_id": "<採購單號>", "延遲天數": <整數>, "建議": "<含具體國家/地區名的替代建議>"}}
+    {{"po_id": "<採購單號>", "延遲天數": <0-365 整數或 null，未知用 null，確認無延遲用 0>, "建議": "<含具體國家/地區名的替代建議>"}}
   ]
 }}"""
 
@@ -110,7 +110,7 @@
       "地區": "地區/城市名（務必繁體中文）",
       "事件類型": "戰爭, 氣候, 罷工, 政策, 交通, 其他",
       "繁體中文簡要": "這則新聞的 150 字內繁體中文簡要分析",
-      "預計延遲": <數字，不相關則填 0>
+      "預計延遲": <0-365 整數或 null；0 為確認無延遲，null 為未知，不可用慣例猜測缺漏資料>
     }}
   ]
 }}