diff --git a/config.py b/config.py index 217bd33..87ed2c8 100644 --- a/config.py +++ b/config.py @@ -414,7 +414,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '') # ========================================== # 系統版本與路徑 # ========================================== -SYSTEM_VERSION = "V10.789" +SYSTEM_VERSION = "V10.790" LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log') public_url = PUBLIC_URL # 用於模板顯示 diff --git a/docs/AI_INTELLIGENCE_MODULE_SOT.md b/docs/AI_INTELLIGENCE_MODULE_SOT.md index 660375e..266b61a 100644 --- a/docs/AI_INTELLIGENCE_MODULE_SOT.md +++ b/docs/AI_INTELLIGENCE_MODULE_SOT.md @@ -344,7 +344,7 @@ SQL漏斗(~300筆) > **重要**: 此表由 `import_service.py` 使用 `df.to_sql()` 動態建立。 > 欄位名稱**完全繼承自 PChome 後台匯出的業績 Excel 原始欄位**，加上程式碼追加的 `snapshot_date`。 > V10.605 起，匯入器會掃描所有 worksheet 與前 15 列表頭，優先選擇含「日期 / 商品名稱 / 總業績或銷售金額」的明細工作表；格式或日期真的不合格的檔案會移到 Google Drive `匯入失敗`，避免每 30 分鐘重複告警。 -> V10.776 起，`pchome_sales_acquisition_service.py` 依序探測 Google Drive、顯式授權 HTTPS、顯式授權 IMAP 與受控落地目錄；所有非 Drive 檔案先做副檔名/魔術碼/大小檢查與 SHA-256 冪等，再由 `import_service.py` 在同一 DB transaction 內 replace、寫入及驗證 `daily_sales_snapshot` 與 `realtime_sales_monthly`。任一步失敗必須整批 rollback，不得留下半同步狀態。每次 run 需把 `trace_id`、OpenTelemetry-compatible `span_id`、`run_id`、`work_item_id`、來源、政策、check、execution、post-verifier 與 terminal 寫入 `pchome_sales_acquisition_receipts`；舊版人工暫停旗標不得停止此 medium-risk controlled-apply lane。正式完成仍要求最新業績 <= 1 天，不能以 source/test/UI 綠燈取代 production freshness closure。 +> V10.776 起，`pchome_sales_acquisition_service.py` 依序探測 Google Drive、顯式授權 HTTPS、顯式授權 IMAP 與受控落地目錄；所有非 Drive 檔案先做副檔名/魔術碼/大小檢查與 SHA-256 冪等，再由 `import_service.py` 在同一 DB transaction 內 replace、寫入及驗證 `daily_sales_snapshot` 與 `realtime_sales_monthly`。任一步失敗必須整批 rollback，不得留下半同步狀態。每次 run 需把 `trace_id`、OpenTelemetry-compatible `span_id`、`run_id`、`work_item_id`、來源、政策、check、execution、post-verifier 與 terminal 寫入 `pchome_sales_acquisition_receipts`；舊版人工暫停旗標不得停止此 medium-risk controlled-apply lane。V10.790 起再以 `pchome_sales_source_reconciliation_v1` 跨 run 累積 provider 空跑/失敗次數，disabled 來源不得冒充 `no_candidate`，ready-empty 走固定重試、provider failure 走有界指數退避，所有已啟用 fallback 仍依序探測；相同狀態 fingerprint 在 cooldown 內抑制重複通知，決策與 `next_retry_at` 一併寫入原 receipt。正式完成仍要求最新業績 <= 1 天，不能以 source/test/UI 綠燈取代 production freshness closure。 #### 已確認的關鍵欄位（PChome 後台報表欄位名稱） diff --git a/docs/guides/pchome_sales_import_automation.md b/docs/guides/pchome_sales_import_automation.md index 4b23125..d2fed2d 100644 --- a/docs/guides/pchome_sales_import_automation.md +++ b/docs/guides/pchome_sales_import_automation.md @@ -73,3 +73,17 @@ PChome 後台業績匯出 - `daily_sales_snapshot` 與 `realtime_sales_monthly` 最新日期落後不超過 1 天。 - `/daily_sales` 與 `/growth_analysis` 圖表 smoke 通過。 - 失敗檔只告警一次，並被移到 `匯入失敗`。 + +## 來源 Reconciliation 合約 + +`PChomeSalesAcquisitionService` 每輪必須把四個授權來源的設定狀態、探測結果、 +連續空跑/失敗次數、fallback chain、下一次重試時間與通知去重決策寫入同一張 +`pchome_sales_acquisition_receipts`。正式合約名稱為 +`pchome_sales_source_reconciliation_v1`。 + +- ready 但無檔：維持 30 分鐘固定重試，並繼續探測所有已啟用 fallback。 +- provider 失敗：使用 30、60、120、240 分鐘有界指數退避。 +- disabled 來源：必須顯示為 `disabled`，不得誤報為已探測但沒有候選。 +- 相同 stale/error fingerprint：6 小時內抑制重複通知；狀態改變立即通知。 +- 低階 Drive importer 由 acquisition controller 呼叫時不直接發 stale 告警，避免雙重通知。 +- receipt 與 API 只能輸出 public-safe 狀態，不得回傳 URL、帳號、token 或郵件內容。 diff --git a/scheduler.py b/scheduler.py index 3c6963a..a42fdfa 100644 --- a/scheduler.py +++ b/scheduler.py @@ -2060,7 +2060,40 @@ def run_auto_import_task(): "decision_ready": result.get('decision_ready'), "requires_upstream_acquisition": result.get('requires_upstream_acquisition', True), "safe_next_action": result.get('safe_next_action'), + "next_retry_at": result.get('next_retry_at'), + "notification_policy": result.get('notification_policy'), } + notification_policy = result.get('notification_policy') or {} + if notification_policy.get('send'): + try: + now_str = datetime.now(TAIPEI_TZ).strftime('%Y-%m-%d %H:%M') + next_retry = result.get('next_retry_at') or '下一輪排程' + message = ( + f"[EwoooC] PChome 業績來源需自動重試 ({now_str})\n" + f"狀態：{run_status}\n" + f"最新業績：{result.get('latest_sales_date') or '尚未建立'}\n" + f"落後：{result.get('data_lag_days')} 天\n" + f"下一輪：{next_retry}\n" + f"Trace：{str(result.get('trace_id') or '')[:12]}" + ) + notifier = NotificationManager() + notifier._send_line_messages([message]) + notifier._send_telegram_messages([message]) + notification_sent = True + logging.info( + "[Scheduler] [AutoImport] source-state notification sent | fingerprint=%s", + notification_policy.get('fingerprint'), + ) + except Exception as notify_error: + logging.error( + "[Scheduler] [AutoImport] source-state notification failed | Error: %s", + notify_error, + ) + else: + logging.info( + "[Scheduler] [AutoImport] duplicate source-state notification suppressed | reason=%s", + notification_policy.get('reason') or 'policy_suppressed', + ) elif result.get('success'): logging.info(f"[Scheduler] [AutoImport] ✅ 自動匯入完成 | Message: {result.get('message')}") stats = { @@ -2164,27 +2197,37 @@ def run_auto_import_task(): "receipt_id": result.get('receipt_id'), "receipt_persisted": result.get('receipt_persisted'), "safe_next_action": result.get('safe_next_action'), + "next_retry_at": result.get('next_retry_at'), + "notification_policy": result.get('notification_policy'), } # V-New: 匯入失敗時也發送通知 - try: - logging.info("[Scheduler] [AutoImport] 📢 準備發送自動匯入失敗通知...") - now_str = datetime.now(TAIPEI_TZ).strftime('%Y-%m-%d %H:%M') - message = ( - f"⚠️ 當日業績自動匯入失敗 ({now_str})\n" - f"{'='*30}\n" - f"❌ 匯入狀態：失敗\n" - f"📌 錯誤訊息：{result.get('message', 'Unknown error')}\n" - f"🔎 Trace：{str(result.get('trace_id') or '')[:12]}\n" - f"{'='*30}\n" - f"系統將依安全重試策略再次取得授權來源" + notification_policy = result.get('notification_policy') or {} + if notification_policy and not notification_policy.get('send'): + logging.info( + "[Scheduler] [AutoImport] duplicate failure notification suppressed | reason=%s", + notification_policy.get('reason') or 'policy_suppressed', ) - notifier = NotificationManager() - notifier._send_line_messages([message]) - notifier._send_telegram_messages([message]) - logging.info("[Scheduler] [AutoImport] ✅ 匯入失敗通知已發送") - except Exception as e: - logging.error(f"[Scheduler] [AutoImport] ❌ 發送失敗通知時發生錯誤 | Error: {e}") + else: + try: + logging.info("[Scheduler] [AutoImport] 📢 準備發送自動匯入失敗通知...") + now_str = datetime.now(TAIPEI_TZ).strftime('%Y-%m-%d %H:%M') + message = ( + f"⚠️ 當日業績自動匯入失敗 ({now_str})\n" + f"{'='*30}\n" + f"❌ 匯入狀態：失敗\n" + f"📌 錯誤訊息：{result.get('message', 'Unknown error')}\n" + f"🔎 Trace：{str(result.get('trace_id') or '')[:12]}\n" + f"{'='*30}\n" + f"系統將依安全重試策略再次取得授權來源" + ) + notifier = NotificationManager() + notifier._send_line_messages([message]) + notifier._send_telegram_messages([message]) + notification_sent = True + logging.info("[Scheduler] [AutoImport] ✅ 匯入失敗通知已發送") + except Exception as e: + logging.error(f"[Scheduler] [AutoImport] ❌ 發送失敗通知時發生錯誤 | Error: {e}") _save_stats('auto_import_task', stats) diff --git a/services/import_service.py b/services/import_service.py index 481b328..7e85d9d 100644 --- a/services/import_service.py +++ b/services/import_service.py @@ -1175,7 +1175,7 @@ class ImportService: logger.error(traceback.format_exc()) return False - def auto_import_from_drive(self) -> Dict[str, Any]: + def auto_import_from_drive(self, *, send_stale_alert: bool = True) -> Dict[str, Any]: """ 從 Google Drive 自動匯入檔案 @@ -1243,7 +1243,7 @@ class ImportService: days_since = (datetime.now(TAIPEI_TZ).date() - normalized_last_date).days data_lag_days = days_since latest_sales_date = str(normalized_last_date) - if days_since >= 3: + if days_since >= 3 and send_stale_alert: _send_data_stale_alert( report_type="upstream_drive", last_date=str(normalized_last_date), diff --git a/services/pchome_sales_acquisition_service.py b/services/pchome_sales_acquisition_service.py index 1d3025a..0d2d181 100644 --- a/services/pchome_sales_acquisition_service.py +++ b/services/pchome_sales_acquisition_service.py @@ -23,6 +23,7 @@ from services.pchome_sales_acquisition_providers import ( env_bool, env_int, ) +from services.pchome_sales_source_reconciliation import build_source_reconciliation logger = logging.getLogger(__name__) @@ -80,7 +81,20 @@ class PChomeSalesAcquisitionService: finally: session.close() - def readiness(self) -> Dict[str, object]: + def _recent_receipts(self, limit: int = 48) -> List[Dict[str, object]]: + session = Session() + try: + receipts = session.query(SalesAcquisitionReceipt).order_by( + SalesAcquisitionReceipt.started_at.desc() + ).limit(max(1, min(limit, 336))).all() + return [receipt.to_public_dict() for receipt in receipts] + except Exception: + logger.warning("Unable to read sales acquisition history", exc_info=True) + return [] + finally: + session.close() + + def _source_inventory(self) -> List[Dict[str, object]]: try: drive = drive_service.check_auth_readiness(refresh_expired=False) drive_ready = bool(drive.get("ready")) @@ -104,7 +118,7 @@ class PChomeSalesAcquisitionService: ) local_path = os.getenv("PCHOME_SALES_LOCAL_DROP_DIR", "").strip() local_ready = bool(local_path and Path(local_path).is_dir()) - sources = [ + return [ {"id": "google_drive", "enabled": True, "ready": drive_ready, "status": drive_status}, { "id": "authorized_https", @@ -125,8 +139,13 @@ class PChomeSalesAcquisitionService: "status": "ready" if local_ready else "disabled" if not local_path else "path_missing", }, ] + + def readiness(self) -> Dict[str, object]: + sources = self._source_inventory() ready_count = sum(1 for source in sources if source["ready"]) freshness = self._freshness() + last_receipt = self._latest_receipt() + last_stages = (last_receipt or {}).get("stages") or {} return { "success": True, "automation_mode": "controlled_apply", @@ -141,7 +160,8 @@ class PChomeSalesAcquisitionService: }, "freshness": freshness, "runtime_closure": "closed" if freshness["decision_ready"] else "blocked_by_upstream_freshness", - "last_receipt": self._latest_receipt(), + "source_reconciliation": last_stages.get("source_reconciliation"), + "last_receipt": last_receipt, } def _persist_receipt( @@ -206,6 +226,9 @@ class PChomeSalesAcquisitionService: span_id = uuid.uuid4().hex[:16] run_id = uuid.uuid4().hex before = self._freshness() + source_inventory = self._source_inventory() + source_by_id = {str(source["id"]): source for source in source_inventory} + prior_receipts = self._recent_receipts() stages: Dict[str, object] = { "sensor": {"status": "started", "trigger": trigger, "trace_id": trace_id, "span_id": span_id}, "normalize": {"asset_id": "pchome.daily_sales.authorized_report"}, @@ -224,9 +247,11 @@ class PChomeSalesAcquisitionService: error_kind = None try: - drive_result = import_service.auto_import_from_drive() + drive_result = import_service.auto_import_from_drive(send_stale_alert=False) stages["providers"].append({ "source": "google_drive", + "enabled": True, + "ready": bool(source_by_id.get("google_drive", {}).get("ready")), "status": drive_result.get("status") or ("completed" if drive_result.get("success") else "failed"), "file_count": drive_result.get("file_count", 0), "imported_count": drive_result.get("imported_count", 0), @@ -258,17 +283,49 @@ class PChomeSalesAcquisitionService: ("authorized_imap", self.providers.imap_candidates), ("controlled_local_drop", self.providers.local_candidates), ): + source_state = source_by_id.get(source_name) or {} + if not source_state.get("enabled"): + stages["providers"].append({ + "source": source_name, + "enabled": False, + "ready": False, + "status": "disabled", + "candidate_count": 0, + }) + continue + if not source_state.get("ready"): + failed_results.append({ + "source_type": source_name, + "error_kind": "provider_not_ready", + "message": "授權來源設定尚未達到自動執行條件。", + }) + stages["providers"].append({ + "source": source_name, + "enabled": True, + "ready": False, + "status": "not_ready", + "error_kind": "provider_not_ready", + }) + continue try: provider_candidates = provider() candidates.extend(provider_candidates) stages["providers"].append({ "source": source_name, + "enabled": True, + "ready": True, "status": "candidate_found" if provider_candidates else "no_candidate", "candidate_count": len(provider_candidates), }) except ProviderError as exc: failed_results.append({"source_type": source_name, "error_kind": exc.kind, "message": exc.public_message}) - stages["providers"].append({"source": source_name, "status": "failed", "error_kind": exc.kind}) + stages["providers"].append({ + "source": source_name, + "enabled": True, + "ready": True, + "status": "failed", + "error_kind": exc.kind, + }) max_files = env_int("PCHOME_SALES_MAX_FILES_PER_RUN", 5, 1, 25) for candidate in candidates[:max_files]: @@ -342,6 +399,36 @@ class PChomeSalesAcquisitionService: except Exception: logger.warning("Unable to clean sales acquisition temp file", exc_info=True) + for provider_stage in stages["providers"]: + provider_source = provider_stage.get("source") + imported_for_source = [ + item for item in successful_results + if item.get("source_type") == provider_source + ] + duplicates_for_source = [ + item for item in duplicate_results + if item.get("source_type") == provider_source + ] + failures_for_source = [ + item for item in failed_results + if item.get("source_type") == provider_source + ] + if imported_for_source: + provider_stage["status"] = "completed" + provider_stage["imported_count"] = len(imported_for_source) + provider_stage["rows_imported"] = sum( + int(item.get("rows") or 0) for item in imported_for_source + ) + elif duplicates_for_source: + provider_stage["status"] = "duplicate" + provider_stage["duplicate_count"] = len(duplicates_for_source) + elif failures_for_source and provider_stage.get("status") != "disabled": + provider_stage["status"] = "failed" + provider_stage["error_kind"] = ( + provider_stage.get("error_kind") + or failures_for_source[0].get("error_kind") + ) + verifier = self._freshness() rows_imported = sum(int(item.get("rows") or 0) for item in successful_results) import_job_id = next((item.get("job_id") for item in reversed(successful_results) if item.get("job_id")), None) @@ -409,6 +496,24 @@ class PChomeSalesAcquisitionService: decision = "no_authorized_candidate" message = "所有已啟用授權來源目前都沒有新業績檔，正式資料未變更。" + source_reconciliation = build_source_reconciliation( + sources=source_inventory, + provider_attempts=stages["providers"], + prior_receipts=prior_receipts, + freshness=verifier, + now=datetime.now(TAIPEI_TZ), + retry_interval_minutes=env_int( + "PCHOME_SALES_RETRY_INTERVAL_MINUTES", 30, 1, 1440 + ), + max_failure_backoff_minutes=env_int( + "PCHOME_SALES_MAX_FAILURE_BACKOFF_MINUTES", 240, 30, 1440 + ), + alert_cooldown_minutes=env_int( + "PCHOME_SALES_ALERT_COOLDOWN_MINUTES", 360, 30, 1440 + ), + ) + stages["source_reconciliation"] = source_reconciliation + stages["execution"] = { "status": status, "successful_files": len(successful_results), @@ -458,6 +563,13 @@ class PChomeSalesAcquisitionService: decision = "runtime_verified_receipt_write_failed" message = "業績資料已通過驗證，但治理 receipt 寫入失敗，閉環尚未完成。" + if error_kind == "source_finalize_failed": + safe_next_action = "retry_source_finalization" + elif status == "failed_rolled_back": + safe_next_action = "replay_transaction_after_rollback" + else: + safe_next_action = source_reconciliation.get("safe_next_action") or "none" + return { "success": status in {"completed", "completed_no_write"}, "status": status, @@ -481,15 +593,10 @@ class PChomeSalesAcquisitionService: "decision_ready": verifier.get("decision_ready"), "requires_upstream_acquisition": not bool(verifier.get("decision_ready")), "manual_review_required": False, - "safe_next_action": ( - "scheduler_retry_authorized_sources" - if status == "blocked_with_safe_next_action" - else "repair_provider_or_replay_transaction" - if status in {"failed_rolled_back", "degraded_no_write"} - else "retry_source_finalization" - if error_kind == "source_finalize_failed" - else "none" - ), + "safe_next_action": safe_next_action, + "next_retry_at": source_reconciliation.get("next_retry_at"), + "notification_policy": source_reconciliation.get("notification_policy"), + "source_reconciliation": source_reconciliation, "providers": stages["providers"], "verifier": verifier, } diff --git a/services/pchome_sales_source_reconciliation.py b/services/pchome_sales_source_reconciliation.py new file mode 100644 index 0000000..c51c98d --- /dev/null +++ b/services/pchome_sales_source_reconciliation.py @@ -0,0 +1,326 @@ +"""Cross-run source health, retry, fallback, and alert decisions for PChome sales.""" + +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timedelta +from typing import Dict, Iterable, List, Mapping, Optional + + +SOURCE_PRIORITY = ( + "google_drive", + "authorized_https", + "authorized_imap", + "controlled_local_drop", +) + +_EMPTY_STATUSES = { + "no_candidate", + "no_pending_file", + "upstream_missing", + "upstream_stale", +} +_FAILED_STATUSES = {"failed", "not_ready", "probe_failed"} + + +def _provider_category( + source: Mapping[str, object], + attempt: Optional[Mapping[str, object]], +) -> str: + if not bool(source.get("enabled")): + return "disabled" + if not attempt: + return "not_probed" + + status = str(attempt.get("status") or "unknown") + if int(attempt.get("imported_count") or 0) > 0: + return "imported" + if int(attempt.get("duplicate_count") or 0) > 0 or status == "duplicate": + return "duplicate" + if int(attempt.get("candidate_count") or 0) > 0: + return "candidate_found" + if attempt.get("error_kind") or status in _FAILED_STATUSES: + return "failed" + if status in _EMPTY_STATUSES: + return "empty" + if not bool(source.get("ready")): + return "not_ready" + return "ready" + + +def _history_attempts( + receipts: Iterable[Mapping[str, object]], + source_id: str, +) -> List[Dict[str, object]]: + attempts: List[Dict[str, object]] = [] + for receipt in receipts: + stages = receipt.get("stages") or {} + if not isinstance(stages, Mapping): + continue + providers = stages.get("providers") or [] + if not isinstance(providers, list): + continue + for provider in providers: + if isinstance(provider, Mapping) and provider.get("source") == source_id: + attempts.append({ + **dict(provider), + "receipt_status": receipt.get("status"), + "attempted_at": receipt.get("completed_at") or receipt.get("started_at"), + }) + break + return attempts + + +def _consecutive_count(categories: Iterable[str], target: str) -> int: + count = 0 + for category in categories: + if category != target: + break + count += 1 + return count + + +def _parse_datetime(value: object) -> Optional[datetime]: + if not value: + return None + try: + parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except (TypeError, ValueError): + return None + return parsed + + +def _lag_bucket(lag_days: object) -> str: + if lag_days is None: + return "missing" + try: + lag = int(lag_days) + except (TypeError, ValueError): + return "missing" + if lag <= 1: + return "fresh" + if lag <= 3: + return "warning" + if lag <= 7: + return "stale" + return "critical" + + +def _previous_alert(receipts: Iterable[Mapping[str, object]]) -> Dict[str, object]: + for receipt in receipts: + stages = receipt.get("stages") or {} + if not isinstance(stages, Mapping): + continue + reconciliation = stages.get("source_reconciliation") or {} + if not isinstance(reconciliation, Mapping): + continue + alert = reconciliation.get("notification_policy") or {} + if isinstance(alert, Mapping) and alert.get("fingerprint"): + return { + **dict(alert), + "completed_at": receipt.get("completed_at") or receipt.get("started_at"), + } + return {} + + +def build_source_reconciliation( + *, + sources: Iterable[Mapping[str, object]], + provider_attempts: Iterable[Mapping[str, object]], + prior_receipts: Iterable[Mapping[str, object]], + freshness: Mapping[str, object], + now: datetime, + retry_interval_minutes: int = 30, + max_failure_backoff_minutes: int = 240, + alert_cooldown_minutes: int = 360, +) -> Dict[str, object]: + """Build a public-safe decision without exposing provider configuration values.""" + source_list = [dict(source) for source in sources] + attempt_by_source = { + str(item.get("source")): dict(item) + for item in provider_attempts + if item.get("source") + } + receipt_list = list(prior_receipts) + source_states: List[Dict[str, object]] = [] + + for source in source_list: + source_id = str(source.get("id") or "unknown") + attempt = attempt_by_source.get(source_id) + category = _provider_category(source, attempt) + history = _history_attempts(receipt_list, source_id) + historical_categories = [ + _provider_category(source, historical_attempt) + for historical_attempt in history + ] + current_and_history = [category, *historical_categories] + consecutive_empty = _consecutive_count(current_and_history, "empty") + consecutive_failures = _consecutive_count(current_and_history, "failed") + + retry_minutes = None + next_retry_at = None + if bool(source.get("enabled")) and category in {"duplicate", "empty", "failed", "not_ready"}: + retry_minutes = max(1, retry_interval_minutes) + if category in {"failed", "not_ready"}: + failure_run = max(consecutive_failures, 1) + retry_minutes = min( + retry_minutes * (2 ** min(failure_run - 1, 3)), + max_failure_backoff_minutes, + ) + next_retry_at = (now + timedelta(minutes=retry_minutes)).isoformat() + + action = { + "disabled": "activate_when_authorized_configuration_is_ready", + "not_probed": "probe_on_next_scheduler_run", + "not_ready": "reconcile_provider_then_retry", + "failed": "retry_with_bounded_backoff_and_probe_fallback", + "duplicate": "retry_on_schedule_and_probe_fallback", + "empty": "retry_on_schedule_and_probe_fallback", + "candidate_found": "validate_and_import_candidate", + "imported": "verify_freshness_and_monitor", + "ready": "monitor_on_schedule", + }[category] + source_states.append({ + "id": source_id, + "enabled": bool(source.get("enabled")), + "ready": bool(source.get("ready")), + "configuration_status": source.get("status") or "unknown", + "runtime_state": category, + "error_kind": attempt.get("error_kind") if attempt else None, + "candidate_count": int((attempt or {}).get("candidate_count") or 0), + "imported_count": int((attempt or {}).get("imported_count") or 0), + "duplicate_count": int((attempt or {}).get("duplicate_count") or 0), + "last_attempt_at": ( + now.isoformat() + if attempt and category != "disabled" + else history[0].get("attempted_at") + if history + else None + ), + "last_success_at": ( + now.isoformat() + if category == "imported" + else next( + ( + historical_attempt.get("attempted_at") + for historical_attempt in history + if _provider_category(source, historical_attempt) == "imported" + ), + None, + ) + ), + "consecutive_empty_runs": consecutive_empty, + "consecutive_failure_runs": consecutive_failures, + "retry_minutes": retry_minutes, + "next_retry_at": next_retry_at, + "action": action, + }) + + enabled_states = [item for item in source_states if item["enabled"]] + ready_states = [item for item in source_states if item["ready"]] + due_times = [ + str(item["next_retry_at"]) + for item in source_states + if item.get("next_retry_at") + ] + decision_ready = bool(freshness.get("decision_ready")) + any_candidate = any( + item["runtime_state"] in {"candidate_found", "imported"} + for item in source_states + ) + + if decision_ready: + status = "closed" + automation_decision = "monitor_scheduled_sources" + safe_next_action = "none" + elif any_candidate: + status = "partial" + automation_decision = "verify_candidate_import_and_freshness" + safe_next_action = "post_import_freshness_verification" + elif ready_states: + status = "degraded" + automation_decision = "retry_ready_sources_and_probe_enabled_fallbacks" + safe_next_action = "scheduled_retry_and_enabled_fallback_probe" + elif enabled_states: + status = "blocked" + automation_decision = "repair_enabled_sources_with_bounded_retry" + safe_next_action = "reconcile_enabled_source_health_then_retry" + else: + status = "blocked" + automation_decision = "reconcile_authorized_source_configuration" + safe_next_action = "reconcile_authorized_source_configuration" + + lag_bucket = _lag_bucket(freshness.get("data_lag_days")) + severity = ( + "critical" + if lag_bucket in {"critical", "missing"} or not ready_states + else "warning" + if lag_bucket in {"warning", "stale"} + else "info" + ) + fingerprint_payload = { + "lag_bucket": lag_bucket, + "source_states": [ + [item["id"], item["runtime_state"], item.get("error_kind")] + for item in source_states + ], + } + fingerprint = hashlib.sha256( + json.dumps(fingerprint_payload, sort_keys=True, ensure_ascii=True).encode("utf-8") + ).hexdigest()[:20] + previous_alert = _previous_alert(receipt_list) + previous_at = _parse_datetime(previous_alert.get("completed_at")) + elapsed_minutes = None + if previous_at: + comparable_now = now + if comparable_now.tzinfo and not previous_at.tzinfo: + previous_at = previous_at.replace(tzinfo=comparable_now.tzinfo) + elif previous_at.tzinfo and not comparable_now.tzinfo: + comparable_now = comparable_now.replace(tzinfo=previous_at.tzinfo) + elapsed_minutes = max(0, int((comparable_now - previous_at).total_seconds() // 60)) + state_changed = previous_alert.get("fingerprint") != fingerprint + send_notification = severity in {"warning", "critical"} and ( + state_changed + or elapsed_minutes is None + or elapsed_minutes >= max(1, alert_cooldown_minutes) + ) + + notification_policy = { + "event": "pchome_sales_source_freshness", + "severity": severity, + "fingerprint": fingerprint, + "send": send_notification, + "reason": ( + "state_changed" + if state_changed + else "cooldown_elapsed" + if send_notification + else "duplicate_within_cooldown" + ), + "cooldown_minutes": max(1, alert_cooldown_minutes), + } + + return { + "contract": "pchome_sales_source_reconciliation_v1", + "status": status, + "automation_decision": automation_decision, + "manual_review_required": False, + "freshness": dict(freshness), + "source_priority": [ + source_id + for source_id in SOURCE_PRIORITY + if any(item["id"] == source_id for item in source_states) + ], + "fallback_chain": [item["id"] for item in ready_states], + "single_source_dependency": len(ready_states) == 1, + "coverage": { + "ready": len(ready_states), + "enabled": len(enabled_states), + "total": len(source_states), + }, + "sources": source_states, + "next_retry_at": min(due_times) if due_times else None, + "safe_next_action": safe_next_action, + "notification_policy": notification_policy, + } diff --git a/templates/auto_import_index.html b/templates/auto_import_index.html index 5097bd6..51cb806 100644 --- a/templates/auto_import_index.html +++ b/templates/auto_import_index.html @@ -155,7 +155,19 @@ element.hidden = false; } - function sourceStatusLabel(source) { + function sourceStatusLabel(source, runtimeState) { + const runtimeLabels = { + imported: '已匯入', + duplicate: '無需更新', + candidate_found: '處理中', + empty: '待新檔', + failed: '重試中', + not_ready: '修復中', + not_probed: '待探測', + ready: '可用', + disabled: '未啟用' + }; + if (runtimeState && runtimeLabels[runtimeState]) return runtimeLabels[runtimeState]; if (source.ready) return '可用'; if (!source.enabled) return '未啟用'; return source.status === 'incomplete' ? '設定未完整' : source.status === 'path_missing' ? '路徑不存在' : '不可用'; @@ -176,14 +188,23 @@ const freshness = payload.freshness || {}; const coverage = payload.asset_coverage || {}; const receipt = payload.last_receipt || null; + const reconciliation = payload.source_reconciliation || {}; + const runtimeSources = new Map((reconciliation.sources || []).map(source => [source.id, source])); + const nextRetry = reconciliation.next_retry_at; const lag = freshness.data_lag_days; document.getElementById('freshnessValue').textContent = freshness.latest_sales_date || '無資料'; document.getElementById('freshnessMeta').textContent = lag === null || lag === undefined ? '尚未建立基準' : lag === 0 ? '今天已更新' : `落後 ${lag} 天`; document.getElementById('closureValue').textContent = freshness.decision_ready ? '可決策' : '受阻'; document.getElementById('closureValue').dataset.state = freshness.decision_ready ? 'ready' : 'blocked'; - document.getElementById('closureMeta').textContent = freshness.decision_ready ? '新鮮度守門通過' : '等待最新業績'; + document.getElementById('closureMeta').textContent = freshness.decision_ready + ? '新鮮度守門通過' + : nextRetry + ? `已排程 ${formatTime(nextRetry)}` + : '等待最新業績'; document.getElementById('sourceCoverageValue').textContent = `${coverage.ready || 0}/${coverage.total || 0}`; - document.getElementById('sourceCoverageMeta').textContent = `${coverage.percent || 0}% 來源可用`; + document.getElementById('sourceCoverageMeta').textContent = reconciliation.single_source_dependency + ? '僅 1 個來源可用' + : `${coverage.percent || 0}% 來源可用`; document.getElementById('lastRunValue').textContent = receipt ? formatTime(receipt.completed_at) : '尚無執行'; document.getElementById('lastRunMeta').textContent = receipt ? receiptStatusLabel(receipt.status) : '尚無執行紀錄'; document.getElementById('pageRuntimeSummary').textContent = freshness.decision_ready @@ -191,13 +212,17 @@ : `正式業績${lag === null || lag === undefined ? '尚未建立' : `已落後 ${lag} 天`}`; const list = document.getElementById('sourceList'); - list.innerHTML = (payload.sources || []).map(source => ` + list.innerHTML = (payload.sources || []).map(source => { + const runtime = runtimeSources.get(source.id) || {}; + const state = runtime.runtime_state || (source.ready ? 'ready' : source.enabled ? 'not_ready' : 'disabled'); + return `
- + ${escapeHtml(SOURCE_LABELS[source.id] || source.id)} - ${escapeHtml(sourceStatusLabel(source))} + ${escapeHtml(sourceStatusLabel(source, state))}
- `).join(''); + `; + }).join(''); } async function loadReadiness() { diff --git a/tests/test_pchome_sales_acquisition_service.py b/tests/test_pchome_sales_acquisition_service.py index 7a94ef2..25a0b58 100644 --- a/tests/test_pchome_sales_acquisition_service.py +++ b/tests/test_pchome_sales_acquisition_service.py @@ -177,12 +177,19 @@ def test_no_candidate_with_missing_freshness_is_blocked_and_receipted(monkeypatc _disable_remote_sources(monkeypatch) monkeypatch.delenv("PCHOME_SALES_LOCAL_DROP_DIR", raising=False) monkeypatch.setattr(import_service, "drive_service", EmptyDrive()) + monkeypatch.setattr(acquisition, "drive_service", ReadyDrive()) result = acquisition.PChomeSalesAcquisitionService().run(trigger="test") assert result["success"] is False assert result["status"] == "blocked_with_safe_next_action" - assert result["safe_next_action"] == "scheduler_retry_authorized_sources" + assert result["safe_next_action"] == "scheduled_retry_and_enabled_fallback_probe" + assert result["next_retry_at"] + assert result["source_reconciliation"]["contract"] == "pchome_sales_source_reconciliation_v1" + assert result["source_reconciliation"]["manual_review_required"] is False + assert result["providers"][1]["status"] == "disabled" + assert result["providers"][2]["status"] == "disabled" + assert result["providers"][3]["status"] == "disabled" assert result["receipt_persisted"] is True with import_service.engine.connect() as conn: receipt = conn.execute(text( @@ -303,6 +310,7 @@ def test_auto_import_api_uses_governed_acquisition_service(monkeypatch): def test_scheduler_continues_when_legacy_hitl_pause_is_set(monkeypatch): import scheduler import services.agent_actions as agent_actions + import services.notification_manager as notification_manager import services.pchome_sales_acquisition_service as acquisition calls = [] @@ -322,8 +330,19 @@ def test_scheduler_continues_when_legacy_hitl_pause_is_set(monkeypatch): "data_lag_days": 1, "decision_ready": True, "safe_next_action": "scheduler_retry_authorized_sources", + "next_retry_at": "2026-07-14T19:00:00+08:00", + "notification_policy": { + "send": False, + "reason": "duplicate_within_cooldown", + "fingerprint": "same-state", + }, }, ) + monkeypatch.setattr( + notification_manager, + "NotificationManager", + lambda: (_ for _ in ()).throw(AssertionError("suppressed notification must not send")), + ) monkeypatch.setattr(scheduler, "_save_stats", lambda name, stats: saved.update({"name": name, "stats": stats})) scheduler.run_auto_import_task() @@ -332,3 +351,5 @@ def test_scheduler_continues_when_legacy_hitl_pause_is_set(monkeypatch): assert saved["name"] == "auto_import_task" assert saved["stats"]["status"] == "Degraded" assert saved["stats"]["receipt_persisted"] is True + assert saved["stats"]["next_retry_at"] == "2026-07-14T19:00:00+08:00" + assert saved["stats"]["notification_policy"]["send"] is False diff --git a/tests/test_pchome_sales_source_reconciliation.py b/tests/test_pchome_sales_source_reconciliation.py new file mode 100644 index 0000000..59bcc71 --- /dev/null +++ b/tests/test_pchome_sales_source_reconciliation.py @@ -0,0 +1,126 @@ +from datetime import datetime, timedelta, timezone + +from services.pchome_sales_source_reconciliation import build_source_reconciliation + + +NOW = datetime(2026, 7, 14, 18, 30, tzinfo=timezone(timedelta(hours=8))) + + +def _source(source_id, *, enabled=True, ready=True, status="ready"): + return { + "id": source_id, + "enabled": enabled, + "ready": ready, + "status": status, + } + + +def _receipt(provider, *, completed_at, reconciliation=None): + stages = {"providers": [provider]} + if reconciliation: + stages["source_reconciliation"] = reconciliation + return { + "status": "blocked_with_safe_next_action", + "completed_at": completed_at, + "stages": stages, + } + + +def test_stale_single_source_schedules_retry_and_reports_fallback_gap(): + payload = build_source_reconciliation( + sources=[ + _source("google_drive"), + _source("authorized_https", enabled=False, ready=False, status="disabled"), + ], + provider_attempts=[ + {"source": "google_drive", "status": "upstream_stale", "file_count": 0}, + {"source": "authorized_https", "status": "disabled"}, + ], + prior_receipts=[], + freshness={"latest_sales_date": "2026-06-24", "data_lag_days": 20, "decision_ready": False}, + now=NOW, + ) + + assert payload["contract"] == "pchome_sales_source_reconciliation_v1" + assert payload["status"] == "degraded" + assert payload["fallback_chain"] == ["google_drive"] + assert payload["single_source_dependency"] is True + assert payload["next_retry_at"] == "2026-07-14T19:00:00+08:00" + assert payload["safe_next_action"] == "scheduled_retry_and_enabled_fallback_probe" + assert payload["notification_policy"]["severity"] == "critical" + assert payload["notification_policy"]["send"] is True + assert payload["sources"][0]["consecutive_empty_runs"] == 1 + assert payload["sources"][1]["runtime_state"] == "disabled" + + +def test_same_state_within_cooldown_suppresses_duplicate_notification(): + first = build_source_reconciliation( + sources=[_source("google_drive")], + provider_attempts=[{"source": "google_drive", "status": "upstream_stale"}], + prior_receipts=[], + freshness={"data_lag_days": 20, "decision_ready": False}, + now=NOW - timedelta(minutes=10), + ) + history = [_receipt( + {"source": "google_drive", "status": "upstream_stale"}, + completed_at=(NOW - timedelta(minutes=10)).isoformat(), + reconciliation=first, + )] + + second = build_source_reconciliation( + sources=[_source("google_drive")], + provider_attempts=[{"source": "google_drive", "status": "upstream_stale"}], + prior_receipts=history, + freshness={"data_lag_days": 20, "decision_ready": False}, + now=NOW, + ) + + assert second["notification_policy"]["fingerprint"] == first["notification_policy"]["fingerprint"] + assert second["notification_policy"]["send"] is False + assert second["notification_policy"]["reason"] == "duplicate_within_cooldown" + assert second["sources"][0]["consecutive_empty_runs"] == 2 + + +def test_provider_failures_use_bounded_exponential_backoff(): + failed = {"source": "authorized_https", "status": "failed", "error_kind": "http_fetch_failed"} + history = [ + _receipt(failed, completed_at=(NOW - timedelta(minutes=30)).isoformat()), + _receipt(failed, completed_at=(NOW - timedelta(minutes=60)).isoformat()), + ] + + payload = build_source_reconciliation( + sources=[_source("authorized_https")], + provider_attempts=[failed], + prior_receipts=history, + freshness={"data_lag_days": 5, "decision_ready": False}, + now=NOW, + retry_interval_minutes=30, + max_failure_backoff_minutes=240, + ) + + source = payload["sources"][0] + assert source["runtime_state"] == "failed" + assert source["consecutive_failure_runs"] == 3 + assert source["retry_minutes"] == 120 + assert source["next_retry_at"] == "2026-07-14T20:30:00+08:00" + assert source["action"] == "retry_with_bounded_backoff_and_probe_fallback" + + +def test_duplicate_source_remains_on_scheduled_retry_without_writing(): + payload = build_source_reconciliation( + sources=[_source("google_drive")], + provider_attempts=[{ + "source": "google_drive", + "status": "duplicate", + "duplicate_count": 1, + }], + prior_receipts=[], + freshness={"data_lag_days": 10, "decision_ready": False}, + now=NOW, + ) + + source = payload["sources"][0] + assert source["runtime_state"] == "duplicate" + assert source["retry_minutes"] == 30 + assert source["next_retry_at"] == "2026-07-14T19:00:00+08:00" + assert payload["safe_next_action"] == "scheduled_retry_and_enabled_fallback_probe"