diff --git a/.env.example b/.env.example index 751bbd9..96cb697 100644 --- a/.env.example +++ b/.env.example @@ -600,6 +600,28 @@ GOOGLE_DRIVE_CREDENTIALS_FILE=/app/config/google_credentials.json GOOGLE_DRIVE_TOKEN_FILE=/app/config/google_token.json GOOGLE_DRIVE_LEGACY_PICKLE_FILE=/app/config/google_token.pickle +# Authorized PChome sales acquisition. Providers are opt-in and never expose +# credentials through readiness APIs or receipts. +PCHOME_SALES_HTTP_ENABLED=false +PCHOME_SALES_HTTP_URL= +PCHOME_SALES_HTTP_ALLOWED_HOSTS= +PCHOME_SALES_HTTP_ALLOWED_PORTS=443 +PCHOME_SALES_HTTP_BEARER_TOKEN= +PCHOME_SALES_HTTP_ALLOW_PRIVATE=false +PCHOME_SALES_HTTP_TIMEOUT_SECONDS=30 +PCHOME_SALES_IMAP_ENABLED=false +PCHOME_SALES_IMAP_HOST= +PCHOME_SALES_IMAP_PORT=993 +PCHOME_SALES_IMAP_USER= +PCHOME_SALES_IMAP_PASSWORD= +PCHOME_SALES_IMAP_MAILBOX=INBOX +PCHOME_SALES_IMAP_SUBJECT=即時業績 +PCHOME_SALES_IMAP_MAX_MESSAGES=20 +PCHOME_SALES_LOCAL_DROP_DIR= +PCHOME_SALES_TEMP_DIR=/app/data/temp/pchome-sales +PCHOME_SALES_MAX_FILE_BYTES=26214400 +PCHOME_SALES_MAX_FILES_PER_RUN=5 + EXTERNAL_OFFER_SYNC_LIMIT=500 MOMO_TARGETED_SEARCH_LIMIT_PER_TERM=3 MOMO_TARGETED_SEARCH_MAX_PRODUCTS=8 diff --git a/config.py b/config.py index 8adb219..083b399 100644 --- a/config.py +++ b/config.py @@ -402,7 +402,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '') # ========================================== # 系統版本與路徑 # ========================================== -SYSTEM_VERSION = "V10.775" +SYSTEM_VERSION = "V10.776" LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log') public_url = PUBLIC_URL # 用於模板顯示 diff --git a/database/import_models.py b/database/import_models.py index 3f1e607..6ce3644 100644 --- a/database/import_models.py +++ b/database/import_models.py @@ -110,3 +110,64 @@ class ImportConfig(Base): 'created_at': self.created_at.isoformat() if self.created_at else None, 'updated_at': self.updated_at.isoformat() if self.updated_at else None, } + + +class SalesAcquisitionReceipt(Base): + """Durable controlled-apply receipt for PChome sales acquisition runs.""" + __tablename__ = 'pchome_sales_acquisition_receipts' + + receipt_id = Column(String(64), primary_key=True) + trace_id = Column(String(64), nullable=False, index=True) + span_id = Column(String(16), nullable=False, index=True) + run_id = Column(String(64), nullable=False, index=True) + work_item_id = Column(String(100), nullable=False, index=True) + status = Column(String(40), nullable=False, index=True) + source_type = Column(String(40), nullable=False, index=True) + source_fingerprint = Column(String(64), index=True) + source_ref_hash = Column(String(64)) + source_file_name = Column(String(500)) + risk_level = Column(String(20), nullable=False, default='medium') + policy_decision = Column(String(40), nullable=False) + decision = Column(String(80), nullable=False) + import_job_id = Column(Integer, index=True) + rows_imported = Column(Integer, default=0) + before_freshness_json = Column(Text) + stage_receipts_json = Column(Text) + verifier_json = Column(Text) + error_kind = Column(String(100)) + public_message = Column(Text) + started_at = Column(DateTime, default=taipei_now, nullable=False) + completed_at = Column(DateTime) + + def to_public_dict(self): + def parse_json(value): + if not value: + return None + try: + import json + return json.loads(value) + except Exception: + return None + + return { + 'receipt_id': self.receipt_id, + 'trace_id': self.trace_id, + 'span_id': self.span_id, + 'run_id': self.run_id, + 'work_item_id': self.work_item_id, + 'status': self.status, + 'source_type': self.source_type, + 'source_file_name': self.source_file_name, + 'risk_level': self.risk_level, + 'policy_decision': self.policy_decision, + 'decision': self.decision, + 'import_job_id': self.import_job_id, + 'rows_imported': self.rows_imported or 0, + 'before_freshness': parse_json(self.before_freshness_json), + 'stages': parse_json(self.stage_receipts_json), + 'verifier': parse_json(self.verifier_json), + 'error_kind': self.error_kind, + 'message': self.public_message, + 'started_at': self.started_at.isoformat() if self.started_at else None, + 'completed_at': self.completed_at.isoformat() if self.completed_at else None, + } diff --git a/docs/AI_INTELLIGENCE_MODULE_SOT.md b/docs/AI_INTELLIGENCE_MODULE_SOT.md index c27dc72..c13c86d 100644 --- a/docs/AI_INTELLIGENCE_MODULE_SOT.md +++ b/docs/AI_INTELLIGENCE_MODULE_SOT.md @@ -339,6 +339,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。 #### 已確認的關鍵欄位（PChome 後台報表欄位名稱） diff --git a/docs/guides/pchome_ai_automation_priority_backlog.md b/docs/guides/pchome_ai_automation_priority_backlog.md index 045720f..1e9e493 100644 --- a/docs/guides/pchome_ai_automation_priority_backlog.md +++ b/docs/guides/pchome_ai_automation_priority_backlog.md @@ -1,6 +1,6 @@ # PChome AI 自動化優先工作清單 -> 最後更新: 2026-07-10 Asia/Taipei +> 最後更新: 2026-07-11 Asia/Taipei > 範圍: PChome growth / MOMO mapping AI 自動化主線 > 真相來源: Gitea `dev` / `main`、正式環境 `/health`、正式 DB readback、machine-verifiable artifacts @@ -16,12 +16,12 @@ | 順序 | 工作主線 | 狀態 | 下一個完成條件 | |---|---|---|---| -| P0-1 | 授權業績資料自動取得與匯入 | 進行中 | 最新業績 <= 1 天，source receipt 與 post-verifier 同 run 完成 | +| P0-1 | 授權業績資料自動取得與匯入 | 程式已完成，待正式來源與新鮮資料閉環 | 最新業績 <= 1 天，source receipt 與 post-verifier 同 run 完成 | | P0-2 | 全商品身份圖譜與營收加權對應 | 進行中 | 覆蓋 80% 營收商品的 >= 90% | | P0-3 | 合法多電商資料取得 | 進行中 | 每個啟用來源具 freshness SLO、normalize、matcher、verifier | -| P0-4 | 多平台營收決策引擎 | 程式已完成，待推版回讀 | 正式環境以至少 2 個已啟用平台產生可重播決策 | +| P0-4 | 多平台營收決策引擎 | 已部署，受多來源資料缺口阻擋 | 正式環境以至少 2 個已啟用平台產生可重播決策 | | P0-5 | 調價、促銷、內容、曝光實驗與歸因 | 未開始 | 同 run 完成 apply、轉換驗證、增量營收歸因與 rollback | -| P1-1 | 專業 UI/UX 作戰 cockpit | 程式已完成，待推版與視覺 QA | desktop/mobile 首屏無文字牆、無假圖、只顯示一個主要下一步 | +| P1-1 | 專業 UI/UX 作戰 cockpit | V10.776 source-ready，待正式推版與視覺 QA | desktop/mobile 首屏無文字牆、無假圖、只顯示一個主要下一步 | | P1-2 | Ollama-first MCP / RAG / PixelRAG runtime | 進行中 | 正式 flags 啟用，PixelRAG receipt 可 replay，Qwen3-VL 驗證通過 | | P1-3 | 資安與治理自動化 | 進行中 | low/medium/high controlled apply 自動閉環，critical 才 break-glass | @@ -32,12 +32,23 @@ PixelRAG blocked-page 分類、外部報價正規化、正式 DB 安全邊界。 新增 Yahoo / ETMall / friDay / Rakuten PixelRAG alias、多平台 evidence 與 actionable 分層、 業績 freshness gate、全 catalog 覆蓋率、漏跑排程追趕，以及首屏來源摘要與過期資料阻擋。 +2026-07-11 P0-1 source receipt：新增 Drive → 授權 HTTPS → 授權 IMAP → 受控落地目錄的 +自動取得順序、SHA-256 冪等、同交易雙表 replace/verifier/rollback、 +`trace_id/span_id/run_id/work_item_id` durable receipt、舊人工暫停旗標 sunset 與首屏來源 cockpit。 +程式完成不等於 production closure；正式退出條件仍是取得日期 <= 1 天的授權報表並完成 runtime readback。 + +### 2026-07-11 外部專業基準落地 + +- OWASP SSRF Prevention Cheat Sheet：授權 HTTPS provider 採 HTTPS scheme、host/port allowlist、公開 IP 驗證、禁止 redirect，並固定連到已驗證 IP、保留原 hostname TLS 驗證，降低 DNS rebinding / TOCTOU 風險。 +- OpenTelemetry Trace Context：receipt 使用 32 位小寫 hex `trace_id` 與 16 位小寫 hex `span_id`，保留後續跨服務 log/trace correlation 契約。 +- NIST AI RMF：acquisition run 以 policy/asset mapping、風險決策、量測/verifier 與 retry/rollback/learning writeback 對應 Govern、Map、Measure、Manage；UI 或 source-ready 不得取代 production runtime closure。 + ## 不可遺漏的插入需求 以下是使用者在主線推進期間插入、且必須保留在完整工作項目裡的要求。後續不得再把這些需求漏掉或變成口頭提醒。 1. 正式環境才是最新版本真相；不得用本機檔案、舊筆記、舊分支蓋過 production truth。 -2. 版本不得搞錯；目前正式環境 `/health` 版本推進目標為 `V10.745`，完成條件必須包含明確版本 bump、部署與 readback。 +2. 版本不得搞錯；2026-07-11 開工 live truth 為正式環境 `V10.775`，本輪 source 目標為 `V10.776`；完成條件必須包含明確版本 bump、部署與 readback，後續仍以當下 `/health` 為準。 3. GitHub 全面 freeze；不得使用 GitHub、`gh`、GitHub API、GitHub Actions、PR、issue、mirror 或 read-only GitHub 流程。 4. 實作結果必須推到 Gitea `dev` 與 `main`；若改到 runtime 行為，還必須部署到正式環境並回讀。 5. 主方向是 AI 自動化，不是人工審核；人工欄位只能當 evidence / ledger / UI truth，不得阻擋低爆炸半徑、可驗證、可回滾的 controlled apply。 @@ -69,7 +80,7 @@ PixelRAG blocked-page 分類、外部報價正規化、正式 DB 安全邊界。 已完成: -- 正式環境版本真相 guard：`/health` 是最高真相，本輪推進目標為 `V10.745`。 +- 正式環境版本真相 guard：`/health` 是最高真相；2026-07-11 開工基線為 `V10.775`，本輪 source 目標為 `V10.776`。 - Gitea `dev` / `main` source truth 對齊。 - Retry exception controlled apply executor 已落地，目標表為 `pchome_product_matches`。 - 正式 DB 已 controlled apply 4 個 selectors: diff --git a/governance/pchome_growth_program_review.json b/governance/pchome_growth_program_review.json index 9498f10..cb80f5a 100644 --- a/governance/pchome_growth_program_review.json +++ b/governance/pchome_growth_program_review.json @@ -1,6 +1,6 @@ { "version": "pchome_growth_program_review_v1", - "generated_at": "2026-07-10T21:54:36+08:00", + "generated_at": "2026-07-11T12:00:00+08:00", "supersedes_status_only_completion": true, "goal": "以新鮮 PChome 業績、可信跨平台同款資料與可歸因動作持續提升業績", "production_baseline": { @@ -73,12 +73,17 @@ "objective": "Automatically acquire and import the latest PChome sales report", "completed": [ "Google Drive polling and import", + "authorized HTTPS, IMAP and controlled local-drop provider implementation", + "SHA-256 source idempotency", + "atomic two-table replace with rollback", + "same-run source, policy, execution and freshness receipt", "post-import table verification", - "staleness detection" + "staleness detection", + "compact source-readiness cockpit" ], "remaining": [ - "authorized PChome backend, scheduled export, or mailbox-to-Drive acquisition connector", - "same-run source receipt and freshness verifier" + "configure at least one production authorized upstream export provider", + "receive and import a report dated within one day in production" ], "production_acceptance": "latest sales age <= 1 day and decision_ready=true" }, @@ -197,10 +202,14 @@ "removed fake sales sparkline", "desktop and 390px mobile visual verification without horizontal overflow", "repaired the circular accent-token fallback that made primary buttons unreadable", - "production V10.775 static asset readback" + "production V10.775 static asset readback", + "sales automation first-viewport cockpit", + "four-source readiness and runtime-closure readback", + "manual upload removed from the primary workflow", + "six-column verified-run history" ], "remaining": [ - "production visible readback", + "production V10.776 visible readback", "per-product cross-platform offer drawer and experiment outcome view" ], "production_acceptance": "first viewport shows freshness, revenue trend, coverage, risk and one executable next action without overflow" @@ -239,11 +248,13 @@ "completed": [ "source and runtime review artifacts", "read-only visual evidence boundary", - "production DB lifecycle red lines" + "production DB lifecycle red lines", + "OpenTelemetry-compatible trace_id and span_id plus run_id and work_item_id envelope for sales acquisition", + "medium-risk controlled apply without legacy HITL terminal", + "durable acquisition receipt and atomic rollback verifier" ], "remaining": [ "replace preview-only gate chains with executable work items", - "shared trace_id, run_id and work_item_id envelope", "SBOM and dependency policy enforcement", "automated drift work item creation", "security control coverage and production closure metrics" diff --git a/migrations/046_pchome_sales_acquisition_receipts.sql b/migrations/046_pchome_sales_acquisition_receipts.sql new file mode 100644 index 0000000..3e8e2c0 --- /dev/null +++ b/migrations/046_pchome_sales_acquisition_receipts.sql @@ -0,0 +1,53 @@ +-- ============================================================================= +-- Migration 046: PChome authorized sales acquisition controlled-apply receipts +-- Additive only. No product, price, sales, or historical rows are rewritten. +-- ============================================================================= + +CREATE TABLE IF NOT EXISTS pchome_sales_acquisition_receipts ( + receipt_id VARCHAR(64) PRIMARY KEY, + trace_id VARCHAR(64) NOT NULL, + span_id VARCHAR(16) NOT NULL, + run_id VARCHAR(64) NOT NULL, + work_item_id VARCHAR(100) NOT NULL, + status VARCHAR(40) NOT NULL, + source_type VARCHAR(40) NOT NULL, + source_fingerprint VARCHAR(64), + source_ref_hash VARCHAR(64), + source_file_name VARCHAR(500), + risk_level VARCHAR(20) NOT NULL DEFAULT 'medium', + policy_decision VARCHAR(40) NOT NULL, + decision VARCHAR(80) NOT NULL, + import_job_id INTEGER, + rows_imported INTEGER NOT NULL DEFAULT 0, + before_freshness_json TEXT, + stage_receipts_json TEXT, + verifier_json TEXT, + error_kind VARCHAR(100), + public_message TEXT, + started_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + completed_at TIMESTAMP +); + +ALTER TABLE pchome_sales_acquisition_receipts + ADD COLUMN IF NOT EXISTS span_id VARCHAR(16); +UPDATE pchome_sales_acquisition_receipts +SET span_id = SUBSTRING(MD5(receipt_id), 1, 16) +WHERE span_id IS NULL; +ALTER TABLE pchome_sales_acquisition_receipts + ALTER COLUMN span_id SET NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_pchome_sales_acq_trace ON pchome_sales_acquisition_receipts (trace_id); +CREATE INDEX IF NOT EXISTS idx_pchome_sales_acq_span ON pchome_sales_acquisition_receipts (span_id); +CREATE INDEX IF NOT EXISTS idx_pchome_sales_acq_run ON pchome_sales_acquisition_receipts (run_id); +CREATE INDEX IF NOT EXISTS idx_pchome_sales_acq_work_item ON pchome_sales_acquisition_receipts (work_item_id); +CREATE INDEX IF NOT EXISTS idx_pchome_sales_acq_status ON pchome_sales_acquisition_receipts (status); +CREATE INDEX IF NOT EXISTS idx_pchome_sales_acq_source ON pchome_sales_acquisition_receipts (source_type); +CREATE INDEX IF NOT EXISTS idx_pchome_sales_acq_fingerprint ON pchome_sales_acquisition_receipts (source_fingerprint); +CREATE INDEX IF NOT EXISTS idx_pchome_sales_acq_job ON pchome_sales_acquisition_receipts (import_job_id); + +GRANT ALL PRIVILEGES ON pchome_sales_acquisition_receipts TO momo; + +DO $$ +BEGIN + RAISE NOTICE 'Migration 046 complete: PChome sales acquisition receipts are ready'; +END $$; diff --git a/requirements.txt b/requirements.txt index 7aced09..4370b99 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,6 +4,7 @@ gunicorn>=20.1 pandas>=1.5 pytz>=2026.2 openpyxl>=3.1.5 +xlrd>=2.0.1 SQLAlchemy>=1.4 psycopg2-binary>=2.9 schedule>=1.2.2 diff --git a/routes/auto_import_routes.py b/routes/auto_import_routes.py index 2b3b5de..873d2a6 100644 --- a/routes/auto_import_routes.py +++ b/routes/auto_import_routes.py @@ -8,6 +8,7 @@ from flask import Blueprint, render_template, jsonify, request import logging from services.import_service import humanize_import_error, import_service from services.google_drive_service import drive_service +from services.pchome_sales_acquisition_service import pchome_sales_acquisition_service # 建立 Blueprint auto_import_bp = Blueprint('auto_import', __name__) @@ -193,9 +194,9 @@ def list_drive_files(): @auto_import_bp.route('/api/manual_import', methods=['POST']) def manual_import(): - """手動觸發匯入""" + """Backward-compatible command endpoint for the automated acquisition run.""" try: - result = import_service.auto_import_from_drive() + result = pchome_sales_acquisition_service.run(trigger='operator_command') return jsonify(result) @@ -207,6 +208,33 @@ def manual_import(): }), 500 +@auto_import_bp.route('/api/auto_import/readiness', methods=['GET']) +def auto_import_readiness(): + """Return public-safe provider, freshness, and runtime-closure state.""" + try: + return jsonify(pchome_sales_acquisition_service.readiness()) + except Exception: + logger.exception("取得業績來源自動化狀態失敗") + return jsonify({ + 'success': False, + 'message': '暫時無法取得業績來源自動化狀態。' + }), 500 + + +@auto_import_bp.route('/api/auto_import/run', methods=['POST']) +def run_auto_import(): + """Run the governed acquisition, controlled apply, and verifier chain.""" + try: + return jsonify(pchome_sales_acquisition_service.run(trigger='operator_command')) + except Exception as e: + logger.exception("執行業績來源自動化失敗") + return jsonify({ + 'success': False, + 'status': 'failed_rolled_back', + 'message': f'自動取得與匯入失敗。{humanize_import_error(e)}' + }), 500 + + @auto_import_bp.route('/api/reset_stuck_jobs', methods=['POST']) def reset_stuck_jobs(): """重置逾時未完成的匯入任務""" diff --git a/scheduler.py b/scheduler.py index 383cdeb..3c6963a 100644 --- a/scheduler.py +++ b/scheduler.py @@ -2016,38 +2016,61 @@ def verify_import_data_sync(expected_rows: int = None, date_range: dict = None) def run_auto_import_task(): """ - V-New: 自動從 Google Drive 匯入當日業績 - 每半小時檢查一次 Google Drive 是否有新的 Excel 檔案 + 每半小時從授權來源取得當日業績，執行交易式匯入與獨立驗證。 """ - # ADR-012 Phase 4: HITL 暫停檢查 notification_sent = False try: from services.agent_actions import is_task_paused if is_task_paused("run_auto_import_task"): - logging.info("[Scheduler] [AutoImport] ⏸️ 任務被 HITL 暫停中，本次跳過") - return + logging.warning( + "[Scheduler] [AutoImport] legacy HITL pause is superseded for this medium-risk " + "controlled-apply lane; continuing with bounded execution" + ) except Exception as pause_check_error: logging.debug( - f"[Scheduler] [AutoImport] HITL 暫停檢查失敗但繼續排程 | Error: {pause_check_error}", + f"[Scheduler] [AutoImport] legacy pause readback failed; continuing | Error: {pause_check_error}", exc_info=True, ) try: - from services.import_service import import_service from services.notification_manager import NotificationManager + from services.pchome_sales_acquisition_service import pchome_sales_acquisition_service - logging.info("[Scheduler] [AutoImport] 🚀 啟動 Google Drive 自動匯入任務") + logging.info("[Scheduler] [AutoImport] 啟動授權來源自動取得與交易式匯入") - # 執行自動匯入 - result = import_service.auto_import_from_drive() + result = pchome_sales_acquisition_service.run(trigger="scheduler") + run_status = result.get('status') - if result.get('success'): + if run_status in {'blocked_with_safe_next_action', 'degraded_no_write', 'partial'}: + logging.warning( + "[Scheduler] [AutoImport] 自動取得尚未完整閉環 | status=%s | trace_id=%s | latest=%s | lag=%s", + run_status, + result.get('trace_id'), + result.get('latest_sales_date'), + result.get('data_lag_days'), + ) + stats = { + "status": "Degraded", + "upstream_status": run_status, + "trace_id": result.get('trace_id'), + "receipt_id": result.get('receipt_id'), + "receipt_persisted": result.get('receipt_persisted'), + "latest_sales_date": result.get('latest_sales_date'), + "data_lag_days": result.get('data_lag_days'), + "decision_ready": result.get('decision_ready'), + "requires_upstream_acquisition": result.get('requires_upstream_acquisition', True), + "safe_next_action": result.get('safe_next_action'), + } + elif result.get('success'): logging.info(f"[Scheduler] [AutoImport] ✅ 自動匯入完成 | Message: {result.get('message')}") stats = { "file_count": result.get('file_count', 0), "imported_count": result.get('imported_count', 0), - "status": "Degraded" if result.get('status') in {'upstream_stale', 'upstream_missing'} else "Success", - "upstream_status": result.get('status'), + "status": "Success", + "upstream_status": run_status, + "trace_id": result.get('trace_id'), + "receipt_id": result.get('receipt_id'), + "receipt_persisted": result.get('receipt_persisted'), "latest_sales_date": result.get('latest_sales_date'), "data_lag_days": result.get('data_lag_days'), "decision_ready": result.get('decision_ready'), @@ -2087,6 +2110,7 @@ def run_auto_import_task(): f"📊 當日業績自動匯入通知 ({now_str})\n" f"{'='*30}\n" f"✅ 匯入狀態：成功\n" + f"🔎 Trace：{result.get('trace_id', '')[:12]}\n" f"📁 處理檔案數：{result.get('file_count', 0)} 個\n" f"📝 共匯入記錄：{result.get('total_rows', 0)} 筆\n" f"📅 數據日期期間：{date_range_str}\n" @@ -2134,7 +2158,12 @@ def run_auto_import_task(): logging.error(f"[Scheduler] [AutoImport] ❌ 自動匯入失敗 | Message: {result.get('message')}") stats = { "status": "Failed", - "error": result.get('message', 'Unknown error') + "error": result.get('message', 'Unknown error'), + "upstream_status": run_status, + "trace_id": result.get('trace_id'), + "receipt_id": result.get('receipt_id'), + "receipt_persisted": result.get('receipt_persisted'), + "safe_next_action": result.get('safe_next_action'), } # V-New: 匯入失敗時也發送通知 @@ -2146,8 +2175,9 @@ def run_auto_import_task(): 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"請檢查 Google Drive 設定或手動匯入" + f"系統將依安全重試策略再次取得授權來源" ) notifier = NotificationManager() notifier._send_line_messages([message]) diff --git a/services/import_service.py b/services/import_service.py index 23dae7c..481b328 100644 --- a/services/import_service.py +++ b/services/import_service.py @@ -10,6 +10,7 @@ import json import os import re from datetime import date, datetime, timedelta +from pathlib import Path from typing import Any, Dict, Optional import pandas as pd @@ -299,7 +300,8 @@ def _read_daily_sales_excel(file_path: str) -> tuple[pd.DataFrame, dict]: """ candidates: list[dict] = [] - with pd.ExcelFile(file_path, engine="openpyxl") as excel: + excel_engine = "xlrd" if Path(file_path).suffix.lower() == ".xls" else "openpyxl" + with pd.ExcelFile(file_path, engine=excel_engine) as excel: for sheet_name in excel.sheet_names: try: preview = pd.read_excel( @@ -399,6 +401,116 @@ def _should_quarantine_failed_import(error_message: str) -> bool: ] return any(marker in error_message for marker in permanent_error_markers) + +def _prepare_monthly_dataframe(df: pd.DataFrame, conn, job_id: int) -> tuple[pd.DataFrame, list]: + """Normalize the sales dataframe to the existing monthly-table contract.""" + monthly_table = "realtime_sales_monthly" + df_monthly = df.drop(columns=["snapshot_date"], errors="ignore").copy() + df_monthly = df_monthly.rename( + columns={ + column: column.replace("%", "_pct").replace("(", "_").replace(")", "_") + for column in df_monthly.columns + } + ) + + target_columns = _table_columns(conn, monthly_table) + date_column = _find_daily_sales_date_column(df_monthly.columns) + if date_column and date_column != "日期" and "日期" in target_columns and "日期" not in df_monthly.columns: + df_monthly = df_monthly.rename(columns={date_column: "日期"}) + + extra_columns = set(df_monthly.columns) - target_columns + if extra_columns: + logger.warning("任務 %s 移除月報目標表未定義欄位: %s", job_id, sorted(extra_columns)) + df_monthly = df_monthly.drop(columns=sorted(extra_columns), errors="ignore") + + if "日期" not in df_monthly.columns: + raise ValueError("業績分析同步缺少日期欄位，已中止整批匯入") + if df_monthly.empty or not len(df_monthly.columns): + raise ValueError("業績分析同步沒有可寫入欄位，已中止整批匯入") + + parsed_monthly_dates = pd.to_datetime(df_monthly["日期"], errors="coerce") + if parsed_monthly_dates.isna().any(): + invalid_rows = int(parsed_monthly_dates.isna().sum()) + raise ValueError(f"業績分析同步有 {invalid_rows} 筆無效日期，已中止整批匯入") + df_monthly["日期"] = parsed_monthly_dates.dt.strftime("%Y-%m-%d") + return df_monthly, sorted(df_monthly["日期"].unique().tolist()) + + +def _replace_sales_tables_atomic(df: pd.DataFrame, job_id: int) -> Dict[str, Any]: + """Replace snapshot and monthly rows in one transaction and verify both writes.""" + snapshot_dates = _normalise_date_values_for_sql(df["snapshot_date"].dropna().unique()) + if not snapshot_dates: + raise ValueError("當日業績沒有可驗證的資料日期，已中止整批匯入") + + with engine.begin() as conn: + snapshot_placeholders, snapshot_params = _build_in_clause("snapshot_date", snapshot_dates) + snapshot_expr = _date_filter_expr("snapshot_date") + deleted_snapshot = conn.execute( + text( + "DELETE FROM daily_sales_snapshot " + f"WHERE {snapshot_expr} IN ({snapshot_placeholders})" + ), + snapshot_params, + ).rowcount + + df.to_sql( + "daily_sales_snapshot", + conn, + if_exists="append", + index=False, + method="multi", + chunksize=1000, + ) + snapshot_count = conn.execute( + text( + "SELECT COUNT(*) FROM daily_sales_snapshot " + f"WHERE {snapshot_expr} IN ({snapshot_placeholders})" + ), + snapshot_params, + ).scalar() + if snapshot_count != len(df): + raise RuntimeError( + f"當日業績寫入驗證失敗: 預期 {len(df)} 筆, 實際 {snapshot_count} 筆" + ) + + df_monthly, monthly_dates = _prepare_monthly_dataframe(df, conn, job_id) + monthly_placeholders, monthly_params = _build_in_clause("monthly_date", monthly_dates) + deleted_monthly = conn.execute( + text( + 'DELETE FROM realtime_sales_monthly WHERE "日期" ' + f"IN ({monthly_placeholders})" + ), + monthly_params, + ).rowcount + df_monthly.to_sql( + "realtime_sales_monthly", + conn, + if_exists="append", + index=False, + method="multi", + chunksize=1000, + ) + monthly_count = conn.execute( + text( + 'SELECT COUNT(*) FROM realtime_sales_monthly WHERE "日期" ' + f"IN ({monthly_placeholders})" + ), + monthly_params, + ).scalar() + if monthly_count != len(df_monthly): + raise RuntimeError( + f"業績分析寫入驗證失敗: 預期 {len(df_monthly)} 筆, 實際 {monthly_count} 筆" + ) + + return { + "snapshot_rows": snapshot_count, + "monthly_rows": monthly_count, + "deleted_snapshot_rows": deleted_snapshot, + "deleted_monthly_rows": deleted_monthly, + "verified": True, + "atomic": True, + } + # 資料庫設定 - 使用 config.py 中的設定，支援 PostgreSQL 和 SQLite def _create_engine_with_pool(db_path): """建立帶有連線池配置的資料庫引擎""" @@ -512,8 +624,8 @@ class ImportService: Args: job_type: 任務類型（daily_sales 或 vendor_stockout） - drive_file_id: Google Drive 檔案 ID - drive_file_name: 檔案名稱 + drive_file_id: Google Drive ID or governed source fingerprint + drive_file_name: Source file name drive_file_size: 檔案大小 Returns: @@ -720,6 +832,92 @@ class ImportService: finally: session.close() + def import_local_daily_sales( + self, + file_path: str, + *, + source_type: str, + source_fingerprint: str, + source_file_name: Optional[str] = None, + ) -> Dict[str, Any]: + """Import a provider-fetched sales file with fingerprint idempotency.""" + if not os.path.isfile(file_path) or os.path.islink(file_path): + return { + 'success': False, + 'status': 'rejected', + 'error_kind': 'invalid_local_file', + 'message': '來源檔案不存在或不符合安全檔案規則。', + } + + safe_source = re.sub(r'[^a-z0-9_-]+', '-', str(source_type).lower()).strip('-')[:40] + if not safe_source or not re.fullmatch(r'[0-9a-f]{64}', str(source_fingerprint or '')): + return { + 'success': False, + 'status': 'rejected', + 'error_kind': 'invalid_source_identity', + 'message': '來源識別或內容指紋無效，未執行寫入。', + } + + source_identity = f'{safe_source}:sha256:{source_fingerprint}' + session = Session() + try: + existing = session.query(ImportJob).filter_by( + job_type='daily_sales', + drive_file_id=source_identity, + status='completed', + ).order_by(ImportJob.id.desc()).first() + if existing: + return { + 'success': True, + 'status': 'duplicate_no_write', + 'duplicate': True, + 'job_id': existing.id, + 'job': _public_import_job_payload(existing), + 'message': '相同內容已完成匯入，本次以冪等方式略過寫入。', + } + finally: + session.close() + + file_name = os.path.basename(source_file_name or file_path)[:500] + job_id = self.create_import_job( + 'daily_sales', + source_identity, + file_name, + os.path.getsize(file_path), + ) + if not job_id: + return { + 'success': False, + 'status': 'failed', + 'error_kind': 'job_create_failed', + 'message': '無法建立匯入任務，正式資料未變更。', + } + + session = Session() + try: + job = session.query(ImportJob).filter_by(id=job_id).first() + if job: + job.local_file_path = file_path + session.commit() + finally: + session.close() + + self.update_job_status(job_id, 'downloading', 40, '授權來源檔案已取得') + success = self.process_daily_sales_import(job_id, file_path) + public_job = self.get_job_status(job_id) + return { + 'success': success, + 'status': 'completed' if success else 'failed_rolled_back', + 'duplicate': False, + 'job_id': job_id, + 'job': public_job, + 'message': ( + '業績檔已完成交易式匯入與雙表驗證。' + if success + else '業績檔未通過交易式匯入驗證，正式資料未變更。' + ), + } + def process_daily_sales_import(self, job_id: int, file_path: str) -> bool: """ 處理當日業績匯入 @@ -769,6 +967,15 @@ class ImportService: logger.error(error_msg) self.update_job_status(job_id, 'failed', 50, '日期驗證失敗', error_msg) return False + if parsed_dates.isna().any(): + invalid_rows = int(parsed_dates.isna().sum()) + error_msg = ( + f"Excel 日期防禦失敗：日期欄位「{date_col}」有 {invalid_rows} 筆無效日期，" + "為避免部分資料寫入已中止整批匯入" + ) + logger.error(error_msg) + self.update_job_status(job_id, 'failed', 50, '日期驗證失敗', error_msg) + return False df['snapshot_date'] = parsed_dates.dt.date logger.info(f"使用日期欄位: {date_col}") @@ -789,225 +996,88 @@ class ImportService: return False else: # 使用當前日期 - df['snapshot_date'] = datetime.now(TAIPEI_TZ).date() + fallback_date = datetime.now(TAIPEI_TZ).date() + df['snapshot_date'] = fallback_date + if '日期' not in df.columns: + df['日期'] = fallback_date.isoformat() logger.info("未找到日期欄位，使用當前日期（台北時區）") - # 寫入資料庫 - 使用全域的 engine（支援 PostgreSQL 和 SQLite） - # 使用模組頂部定義的 engine，確保連接到正確的資料庫 - - # 更新進度 total_rows = len(df) self.update_job_progress(job_id, total_rows=total_rows, processed_rows=0) - - # 取得此次匯入的日期範圍 - import_dates = df['snapshot_date'].unique() - logger.info(f"本次匯入包含 {len(import_dates)} 個日期的資料") - - # 刪除資料庫中相同日期的舊資料（覆蓋邏輯） - if len(import_dates) > 0: - # 過濾掉 None 值 - valid_dates = _normalise_date_values_for_sql(import_dates) - - if valid_dates: - date_placeholders, date_params = _build_in_clause("snapshot_date", valid_dates) - snapshot_date_expr = _date_filter_expr("snapshot_date") - - with engine.connect() as conn: - # 刪除相同日期的舊資料 - delete_query = text( - f"DELETE FROM {table_name} WHERE {snapshot_date_expr} IN ({date_placeholders})" - ) - result = conn.execute(delete_query, date_params) - deleted_count = result.rowcount - conn.commit() - - if deleted_count > 0: - logger.info(f"已刪除 {deleted_count} 筆舊資料（覆蓋模式）") - - # 寫入資料庫（帶驗證和重試機制） + self.update_job_status(job_id, 'importing', 60, '交易式寫入與雙表驗證...') max_retries = 2 - retry_count = 0 - write_success = False - - while retry_count <= max_retries and not write_success: - try: - if retry_count > 0: - logger.warning(f"任務 {job_id} 第 {retry_count} 次重試寫入...") - self.update_job_status(job_id, 'importing', 60, f'重試寫入中 ({retry_count}/{max_retries})...') - - df.to_sql( - table_name, - engine, - if_exists='append', - index=False, - method='multi', - chunksize=1000 - ) - - # V-Fix: 匯入後驗證 - 確認資料已正確寫入資料庫 - self.update_job_status(job_id, 'importing', 85, '驗證資料寫入...') - - # 取得本次匯入的日期 - import_dates = df['snapshot_date'].dropna().unique() - if len(import_dates) > 0: - # 查詢資料庫中這些日期的資料筆數 - raw_valid_dates = [d for d in import_dates if d is not None] - valid_dates = _normalise_date_values_for_sql(raw_valid_dates) - date_placeholders, date_params = _build_in_clause("verify_date", valid_dates) - snapshot_date_expr = _date_filter_expr("snapshot_date") - - with engine.connect() as conn: - verify_query = text( - f"SELECT COUNT(*) FROM {table_name} WHERE {snapshot_date_expr} IN ({date_placeholders})" - ) - result = conn.execute(verify_query, date_params) - db_count = result.scalar() - - # 驗證：資料庫筆數應該 >= 本次匯入筆數（可能有其他日期的舊資料） - expected_count = len(df[df['snapshot_date'].isin(raw_valid_dates)]) - - if db_count >= expected_count: - logger.info(f"任務 {job_id} 驗證成功: 預期 {expected_count} 筆, 資料庫有 {db_count} 筆") - write_success = True - else: - logger.warning(f"任務 {job_id} 驗證失敗: 預期 {expected_count} 筆, 資料庫只有 {db_count} 筆") - retry_count += 1 - else: - # 沒有有效日期，跳過驗證 - logger.warning(f"任務 {job_id} 無法驗證: 沒有有效的 snapshot_date") - write_success = True - - except Exception as write_error: - logger.error(f"任務 {job_id} 寫入失敗 (嘗試 {retry_count + 1}): {str(write_error)}") - retry_count += 1 - if retry_count > max_retries: - raise write_error - - if not write_success: - error_msg = f"資料寫入驗證失敗，已重試 {max_retries} 次" - self.update_job_status(job_id, 'failed', 85, '驗證失敗', error_msg) - logger.error(f"任務 {job_id} {error_msg}") - return False - - # === V-New 2026-01-15: 同步寫入 realtime_sales_monthly === - # 目的：讓當日業績 raw data 同時呈現在「業績分析儀表板」 - # 2026-01-30 修復：加強欄位驗證、同步狀態追蹤、失敗告警 - self.update_job_status(job_id, 'importing', 90, '同步至業績分析儀表板...') - - sync_success = False + write_receipt = None sync_error_msg = None - monthly_table = 'realtime_sales_monthly' - - try: - # 準備資料：移除 snapshot_date 欄位（realtime_sales_monthly 不需要此欄位） - df_monthly = df.drop(columns=['snapshot_date'], errors='ignore') - - # 2026-01-30 修正：強化欄位名稱轉換 - # 將特殊字符轉換為 PostgreSQL 安全格式 - column_mapping = {} - for col in df_monthly.columns: - new_col = col.replace('%', '_pct').replace('(', '_').replace(')', '_') - column_mapping[col] = new_col - df_monthly = df_monthly.rename(columns=column_mapping) - - # 記錄轉換的欄位 - converted_cols = [f"'{k}' -> '{v}'" for k, v in column_mapping.items() if k != v] - if converted_cols: - logger.info(f"任務 {job_id} 欄位名稱轉換: {', '.join(converted_cols)}") - logger.info(f"任務 {job_id} 欄位轉換完成，共 {len(df_monthly.columns)} 個欄位") - - # 2026-01-30 新增：驗證 DataFrame 欄位和目標表欄位是否一致 - with engine.connect() as conn: - target_columns = _table_columns(conn, monthly_table) - - df_columns = set(df_monthly.columns) - missing_in_table = df_columns - target_columns - missing_in_df = target_columns - df_columns - - if missing_in_table: - logger.warning(f"任務 {job_id} 欄位警告: DataFrame 有但表中沒有: {missing_in_table}") - # 移除表中沒有的欄位，避免 INSERT 失敗 - df_monthly = df_monthly.drop(columns=list(missing_in_table), errors='ignore') - logger.info(f"任務 {job_id} 已移除多餘欄位，剩餘 {len(df_monthly.columns)} 個欄位") - - if missing_in_df: - logger.warning(f"任務 {job_id} 欄位警告: 表中有但 DataFrame 沒有: {missing_in_df}") - - # 取得本次匯入的日期列表（使用原始「日期」欄位） - unique_dates = [] - if '日期' in df.columns: - unique_dates = df['日期'].dropna().unique().tolist() - logger.info(f"任務 {job_id} 準備同步 {len(unique_dates)} 個日期的資料") - - if len(unique_dates) > 0: - # 刪除 realtime_sales_monthly 中相同日期的舊資料（去重） - date_placeholders, date_params = _build_in_clause("monthly_date", unique_dates) - - with engine.connect() as conn: - delete_monthly_query = text( - f'DELETE FROM {monthly_table} WHERE "日期" IN ({date_placeholders})' - ) - result = conn.execute(delete_monthly_query, date_params) - deleted_monthly = result.rowcount - conn.commit() - - if deleted_monthly > 0: - logger.info(f"任務 {job_id} 已從 {monthly_table} 刪除 {deleted_monthly} 筆同日期舊資料") - - # 寫入 realtime_sales_monthly - df_monthly.to_sql( - monthly_table, - engine, - if_exists='append', - index=False, - method='multi', - chunksize=1000 - ) - - logger.info(f"任務 {job_id} 已同步 {len(df_monthly)} 筆資料至 {monthly_table}") - - # 驗證同步結果 - if len(unique_dates) > 0: - with engine.connect() as conn: - date_placeholders, date_params = _build_in_clause("monthly_verify_date", unique_dates) - verify_query = text( - f'SELECT COUNT(*) FROM {monthly_table} WHERE "日期" IN ({date_placeholders})' + for attempt in range(max_retries + 1): + try: + if attempt: + self.update_job_status( + job_id, + 'importing', + 60, + f'整批交易重試中 ({attempt}/{max_retries})...' ) - verify_count = conn.execute(verify_query, date_params).scalar() + write_receipt = _replace_sales_tables_atomic(df, job_id) + break + except Exception as write_error: + sync_error_msg = str(write_error) + logger.error( + "任務 %s 交易式寫入失敗 (嘗試 %s/%s): %s", + job_id, + attempt + 1, + max_retries + 1, + sync_error_msg, + exc_info=True, + ) - if verify_count >= len(df_monthly): - logger.info(f"任務 {job_id} 同步驗證成功: {monthly_table} 現有 {verify_count} 筆資料") - sync_success = True - else: - sync_error_msg = f"同步驗證失敗: 預期 {len(df_monthly)} 筆, 實際 {verify_count} 筆" - logger.error(f"任務 {job_id} {sync_error_msg}") - else: - sync_success = True # 沒有日期資料時視為成功 + sync_success = bool(write_receipt and write_receipt.get('verified')) + monthly_table = 'realtime_sales_monthly' + if not sync_success: + summary = { + 'imported_count': 0, + 'table_name': table_name, + 'synced_to': None, + 'sync_success': False, + 'sync_error': sync_error_msg, + 'verified': False, + 'atomic': True, + 'rolled_back': True, + 'source_sheet': excel_metadata.get("sheet_name"), + 'source_header_row': excel_metadata.get("header_row"), + 'message': '業績資料驗證失敗，整批交易已回滾，正式資料未變更。', + } + session = Session() + try: + job = session.query(ImportJob).filter_by(id=job_id).first() + if job: + job.import_summary = json.dumps(summary, ensure_ascii=False) + session.commit() + finally: + session.close() - except Exception as sync_error: - # 同步失敗，記錄完整錯誤 - import traceback - sync_error_msg = str(sync_error) - logger.error(f"任務 {job_id} 同步至 {monthly_table} 失敗: {sync_error_msg}") - logger.error(f"任務 {job_id} 同步錯誤堆疊:\n{traceback.format_exc()}") - - # 2026-01-30 新增：發送同步失敗告警 + self.update_job_progress( + job_id, + processed_rows=total_rows, + success_rows=0, + error_rows=total_rows, + ) + self.update_job_status( + job_id, + 'failed', + 95, + '交易驗證失敗，已回滾', + sync_error_msg or '交易式寫入驗證失敗', + ) try: from services.notification_manager import NotificationManager - notifier = NotificationManager() - alert_msg = ( - f"⚠️ 業績資料同步失敗告警\n" - f"{'='*30}\n" - f"任務 ID: {job_id}\n" - f"目標表: {monthly_table}\n" - f"錯誤: {sync_error_msg[:200]}\n" - f"{'='*30}\n" - f"daily_sales_snapshot 已匯入成功，但業績分析儀表板需要手動同步" - ) - notifier._send_telegram_messages([alert_msg]) - logger.info(f"任務 {job_id} 已發送同步失敗告警") + NotificationManager()._send_telegram_messages([ + f"業績匯入失敗（任務 {job_id}）：雙表交易已回滾，正式資料未變更。" + ]) except Exception as notify_error: - logger.error(f"任務 {job_id} 發送告警失敗: {notify_error}") + logger.error("任務 %s 發送回滾告警失敗: %s", job_id, notify_error) + return False + + self.update_job_status(job_id, 'importing', 90, '雙表交易驗證完成') # 計算日期範圍 @@ -1170,7 +1240,7 @@ class ImportService: if isinstance(last_date, date) else date.fromisoformat(str(last_date)[:10]) ) - days_since = (date.today() - normalized_last_date).days + 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: @@ -1212,6 +1282,7 @@ class ImportService: total_rows = 0 all_dates = [] # 收集所有匯入的日期 failed_files = [] + archive_failed_files = [] for file in files: file_id = file['id'] @@ -1260,10 +1331,20 @@ class ImportService: if drive_service.move_file(file_id, archive_folder): logger.info(f"已移動 Google Drive 檔案到「{archive_folder}」: {file_name}") + self.update_job_status(job_id, 'completed', 100, '完成') else: logger.warning(f"無法移動 Google Drive 檔案: {file_name}") - - self.update_job_status(job_id, 'completed', 100, '完成') + archive_failed_files.append({ + 'file': file_name, + 'job_id': job_id, + 'error': '業績已匯入，但 Google Drive 來源封存失敗', + }) + self.update_job_status( + job_id, + 'completed', + 100, + '匯入完成，來源封存待自動重試', + ) imported_count += 1 # 讀取 job summary 取得匯入筆數和日期範圍 @@ -1347,15 +1428,29 @@ class ImportService: f'找到 {len(files)} 個檔案，成功匯入 {imported_count} 個，' f'失敗 {len(failed_files)} 個。{public_error}' ) + elif archive_failed_files: + message = ( + f'成功匯入 {imported_count} 個檔案，但有 {len(archive_failed_files)} 個來源封存待重試' + ) else: message = f'成功匯入 {imported_count} 個檔案' return { - 'success': len(failed_files) == 0, + 'success': len(failed_files) == 0 and len(archive_failed_files) == 0, + 'status': ( + 'partial_source_finalize_failed' + if archive_failed_files + else 'completed' + if not failed_files + else 'failed' + ), 'message': message, 'file_count': len(files), 'imported_count': imported_count, 'failed_count': len(failed_files), + 'source_finalize_ok': len(archive_failed_files) == 0, + 'archive_failed_count': len(archive_failed_files), + 'archive_errors': archive_failed_files, 'errors': failed_files, 'total_rows': total_rows, 'date_range': date_range diff --git a/services/pchome_sales_acquisition_providers.py b/services/pchome_sales_acquisition_providers.py new file mode 100644 index 0000000..8e8483a --- /dev/null +++ b/services/pchome_sales_acquisition_providers.py @@ -0,0 +1,380 @@ +"""Authorized PChome sales report providers and file-boundary guards.""" + +from __future__ import annotations + +import hashlib +import http.client +import imaplib +import ipaddress +import os +import re +import shutil +import socket +import ssl +import tempfile +from dataclasses import dataclass +from email import policy +from email.header import decode_header, make_header +from email.parser import BytesParser +from pathlib import Path +from typing import Callable, List, Optional +from urllib.parse import unquote, urlparse + + +VALID_EXTENSIONS = {".xlsx", ".xls"} +XLSX_MAGIC = b"PK\x03\x04" +XLS_MAGIC = b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" + + +class ProviderError(RuntimeError): + def __init__(self, kind: str, public_message: str): + super().__init__(public_message) + self.kind = kind + self.public_message = public_message + + +class _PinnedHTTPSConnection(http.client.HTTPSConnection): + """Connect to the validated IP while preserving hostname TLS verification.""" + + def __init__(self, hostname: str, port: int, pinned_ip: str, timeout: int): + super().__init__(hostname, port=port, timeout=timeout, context=ssl.create_default_context()) + self._pinned_ip = pinned_ip + + def connect(self): + self.sock = socket.create_connection( + (self._pinned_ip, self.port), + self.timeout, + self.source_address, + ) + self.sock = self._context.wrap_socket(self.sock, server_hostname=self.host) + + +@dataclass +class AcquisitionCandidate: + source_type: str + file_path: str + file_name: str + fingerprint: str + source_ref_hash: str + finalize_success: Callable[[], None] + finalize_rejected: Callable[[], None] + cleanup: Callable[[], None] + + +def env_bool(name: str, default: bool = False) -> bool: + fallback = "true" if default else "false" + return os.getenv(name, fallback).strip().lower() in {"1", "true", "yes", "on"} + + +def env_int(name: str, default: int, minimum: int = 1, maximum: int = 1000) -> int: + try: + value = int(os.getenv(name, str(default))) + except (TypeError, ValueError): + value = default + return max(minimum, min(value, maximum)) + + +def _hash_text(value: str) -> str: + return hashlib.sha256(value.encode("utf-8", errors="ignore")).hexdigest() + + +def _sha256_file(file_path: str) -> str: + digest = hashlib.sha256() + with open(file_path, "rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _safe_file_name(value: str, default: str = "pchome-sales.xlsx") -> str: + candidate = os.path.basename(str(value or "").replace("\x00", "")).strip() + candidate = re.sub(r"[^0-9A-Za-z._()\-\u4e00-\u9fff]+", "_", candidate) + return (candidate or default)[:180] + + +def _decode_mail_header(value: Optional[str]) -> str: + if not value: + return "" + try: + return str(make_header(decode_header(value))) + except Exception: + return str(value) + + +def _make_temp_path(suffix: str) -> str: + target = Path(os.getenv("PCHOME_SALES_TEMP_DIR", "data/temp/pchome-sales")) + target.mkdir(parents=True, exist_ok=True) + fd, file_path = tempfile.mkstemp(prefix="sales-", suffix=suffix, dir=str(target)) + os.close(fd) + return file_path + + +def _validate_excel_file(file_path: str, file_name: str) -> None: + path = Path(file_path) + max_bytes = env_int("PCHOME_SALES_MAX_FILE_BYTES", 25 * 1024 * 1024, 1024, 200 * 1024 * 1024) + if not path.is_file() or path.is_symlink(): + raise ProviderError("unsafe_file", "來源檔案不存在或不符合安全檔案規則。") + if path.stat().st_size <= 0 or path.stat().st_size > max_bytes: + raise ProviderError("invalid_file_size", "來源檔案大小不符合匯入政策，未執行寫入。") + suffix = Path(file_name).suffix.lower() + if suffix not in VALID_EXTENSIONS: + raise ProviderError("unsupported_file_type", "來源不是允許的 Excel 業績檔，未執行寫入。") + with path.open("rb") as handle: + magic = handle.read(8) + signature_matches = ( + suffix == ".xlsx" and magic.startswith(XLSX_MAGIC) + ) or ( + suffix == ".xls" and magic == XLS_MAGIC + ) + if not signature_matches: + raise ProviderError("invalid_excel_signature", "來源檔案內容不是有效的 Excel 格式，未執行寫入。") + + +def _archive_file(source: Path, folder_name: str, fingerprint: str) -> None: + if not source.exists() or source.is_symlink(): + return + target_dir = source.parent / folder_name + target_dir.mkdir(parents=True, exist_ok=True) + target = target_dir / source.name + if target.exists(): + target = target_dir / f"{source.stem}-{fingerprint[:10]}{source.suffix}" + os.replace(str(source), str(target)) + + +class AuthorizedSalesProviders: + """Side-effect-bounded report providers in governed priority order.""" + + def validate_http_url(self, url: str) -> tuple[str, int, str]: + parsed = urlparse(url) + if parsed.scheme.lower() != "https" or not parsed.hostname or parsed.username or parsed.password: + raise ProviderError("http_url_rejected", "授權報表網址必須是無內嵌帳密的 HTTPS 網址。") + hostname = parsed.hostname.lower().rstrip(".") + try: + port = parsed.port or 443 + except ValueError as exc: + raise ProviderError("http_port_rejected", "授權報表網址的連接埠格式無效。") from exc + allowed_ports = { + int(item.strip()) + for item in os.getenv("PCHOME_SALES_HTTP_ALLOWED_PORTS", "443").split(",") + if item.strip().isdigit() + } + if port not in allowed_ports: + raise ProviderError("http_port_rejected", "授權報表連接埠未列入允許清單，未發出連線。") + allowed_hosts = { + item.strip().lower().rstrip(".") + for item in os.getenv("PCHOME_SALES_HTTP_ALLOWED_HOSTS", "").split(",") + if item.strip() + } + if hostname not in allowed_hosts: + raise ProviderError("http_host_not_allowed", "授權報表主機未列入允許清單，未發出連線。") + try: + addresses = { + item[4][0] + for item in socket.getaddrinfo(hostname, port, type=socket.SOCK_STREAM) + } + except OSError as exc: + raise ProviderError("http_dns_failed", "授權報表主機目前無法解析。") from exc + if not addresses: + raise ProviderError("http_dns_failed", "授權報表主機目前沒有可用位址。") + if not env_bool("PCHOME_SALES_HTTP_ALLOW_PRIVATE"): + for address in addresses: + ip = ipaddress.ip_address(address) + if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: + raise ProviderError("http_private_address_rejected", "授權報表主機解析到受限制網段，未發出連線。") + return hostname, port, sorted(addresses)[0] + + def http_candidates(self) -> List[AcquisitionCandidate]: + if not env_bool("PCHOME_SALES_HTTP_ENABLED"): + return [] + url = os.getenv("PCHOME_SALES_HTTP_URL", "").strip() + if not url: + raise ProviderError("http_not_configured", "授權 HTTPS 來源已啟用但網址尚未設定。") + hostname, port, pinned_ip = self.validate_http_url(url) + parsed = urlparse(url) + headers = {"Accept": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet, application/vnd.ms-excel"} + token = os.getenv("PCHOME_SALES_HTTP_BEARER_TOKEN", "").strip() + if token: + headers["Authorization"] = f"Bearer {token}" + timeout = env_int("PCHOME_SALES_HTTP_TIMEOUT_SECONDS", 30, 3, 120) + max_bytes = env_int("PCHOME_SALES_MAX_FILE_BYTES", 25 * 1024 * 1024, 1024, 200 * 1024 * 1024) + connection = _PinnedHTTPSConnection(hostname, port, pinned_ip, timeout) + path = parsed.path or "/" + if parsed.query: + path = f"{path}?{parsed.query}" + try: + connection.request("GET", path, headers=headers) + response = connection.getresponse() + if response.status != 200: + raise ProviderError("http_status_rejected", f"授權 HTTPS 報表來源回應 {response.status}，未執行匯入。") + content_length = response.getheader("Content-Length") + if content_length and int(content_length) > max_bytes: + raise ProviderError("http_file_too_large", "授權報表超過檔案大小上限，未執行匯入。") + disposition = response.getheader("Content-Disposition", "") + match = re.search(r"filename\*?=(?:UTF-8''|\")?([^\";]+)", disposition, re.I) + raw_name = unquote(match.group(1).strip()) if match else os.path.basename(parsed.path) + file_name = _safe_file_name(raw_name) + suffix = Path(file_name).suffix.lower() + if suffix not in VALID_EXTENSIONS: + suffix = ".xlsx" + file_name = f"{Path(file_name).stem or 'pchome-sales'}{suffix}" + temp_path = _make_temp_path(suffix) + written = 0 + try: + with open(temp_path, "wb") as handle: + while True: + chunk = response.read(1024 * 1024) + if not chunk: + break + written += len(chunk) + if written > max_bytes: + raise ProviderError("http_file_too_large", "授權報表超過檔案大小上限，未執行匯入。") + handle.write(chunk) + _validate_excel_file(temp_path, file_name) + except Exception: + Path(temp_path).unlink(missing_ok=True) + raise + except ProviderError: + raise + except (OSError, ssl.SSLError, http.client.HTTPException, ValueError) as exc: + raise ProviderError("http_fetch_failed", "授權 HTTPS 報表來源目前無法連線。") from exc + finally: + connection.close() + fingerprint = _sha256_file(temp_path) + return [AcquisitionCandidate( + source_type="authorized_https", + file_path=temp_path, + file_name=file_name, + fingerprint=fingerprint, + source_ref_hash=_hash_text(f"https://{hostname}{parsed.path}"), + finalize_success=lambda: None, + finalize_rejected=lambda: None, + cleanup=lambda candidate_path=temp_path: Path(candidate_path).unlink(missing_ok=True), + )] + + def _mark_imap_seen(self, uid: bytes) -> None: + host = os.getenv("PCHOME_SALES_IMAP_HOST", "").strip() + user = os.getenv("PCHOME_SALES_IMAP_USER", "").strip() + password = os.getenv("PCHOME_SALES_IMAP_PASSWORD", "") + port = env_int("PCHOME_SALES_IMAP_PORT", 993, 1, 65535) + mailbox = os.getenv("PCHOME_SALES_IMAP_MAILBOX", "INBOX").strip() or "INBOX" + client = imaplib.IMAP4_SSL(host, port, ssl_context=ssl.create_default_context(), timeout=20) + try: + client.login(user, password) + client.select(mailbox, readonly=False) + client.uid("store", uid, "+FLAGS", "(\\Seen)") + finally: + try: + client.logout() + except Exception: + pass + + def imap_candidates(self) -> List[AcquisitionCandidate]: + if not env_bool("PCHOME_SALES_IMAP_ENABLED"): + return [] + host = os.getenv("PCHOME_SALES_IMAP_HOST", "").strip() + user = os.getenv("PCHOME_SALES_IMAP_USER", "").strip() + password = os.getenv("PCHOME_SALES_IMAP_PASSWORD", "") + if not host or not user or not password: + raise ProviderError("imap_not_configured", "授權信箱來源已啟用但連線設定不完整。") + port = env_int("PCHOME_SALES_IMAP_PORT", 993, 1, 65535) + mailbox = os.getenv("PCHOME_SALES_IMAP_MAILBOX", "INBOX").strip() or "INBOX" + subject_pattern = os.getenv("PCHOME_SALES_IMAP_SUBJECT", "即時業績").strip() + max_messages = env_int("PCHOME_SALES_IMAP_MAX_MESSAGES", 20, 1, 100) + candidates: List[AcquisitionCandidate] = [] + client = None + try: + client = imaplib.IMAP4_SSL(host, port, ssl_context=ssl.create_default_context(), timeout=20) + client.login(user, password) + status, _ = client.select(mailbox, readonly=True) + if status != "OK": + raise ProviderError("imap_mailbox_failed", "授權信箱資料夾目前無法開啟。") + status, data = client.uid("search", None, "UNSEEN") + if status != "OK": + raise ProviderError("imap_search_failed", "授權信箱目前無法搜尋新報表。") + uids = list(reversed((data[0] or b"").split()))[:max_messages] + for uid in uids: + status, payload = client.uid("fetch", uid, "(RFC822)") + if status != "OK" or not payload or not isinstance(payload[0], tuple): + continue + message = BytesParser(policy=policy.default).parsebytes(payload[0][1]) + if subject_pattern and subject_pattern not in _decode_mail_header(message.get("Subject")): + continue + for attachment in message.iter_attachments(): + file_name = _safe_file_name(_decode_mail_header(attachment.get_filename())) + suffix = Path(file_name).suffix.lower() + if suffix not in VALID_EXTENSIONS: + continue + temp_path = _make_temp_path(suffix) + try: + with open(temp_path, "wb") as handle: + handle.write(attachment.get_payload(decode=True) or b"") + _validate_excel_file(temp_path, file_name) + except Exception: + Path(temp_path).unlink(missing_ok=True) + raise + fingerprint = _sha256_file(temp_path) + candidates.append(AcquisitionCandidate( + source_type="authorized_imap", + file_path=temp_path, + file_name=file_name, + fingerprint=fingerprint, + source_ref_hash=_hash_text(f"{host}:{mailbox}:{uid.decode(errors='ignore')}"), + finalize_success=lambda message_uid=uid: self._mark_imap_seen(message_uid), + finalize_rejected=lambda message_uid=uid: self._mark_imap_seen(message_uid), + cleanup=lambda candidate_path=temp_path: Path(candidate_path).unlink(missing_ok=True), + )) + except ProviderError: + for candidate in candidates: + candidate.cleanup() + raise + except (imaplib.IMAP4.error, OSError, ssl.SSLError) as exc: + for candidate in candidates: + candidate.cleanup() + raise ProviderError("imap_connection_failed", "授權信箱來源目前無法連線或驗證。") from exc + finally: + if client is not None: + try: + client.logout() + except Exception: + pass + return candidates + + def local_candidates(self) -> List[AcquisitionCandidate]: + configured = os.getenv("PCHOME_SALES_LOCAL_DROP_DIR", "").strip() + if not configured: + return [] + root = Path(configured).expanduser().resolve() + if not root.is_dir(): + raise ProviderError("local_drop_missing", "受控落地目錄不存在，未執行檔案存取。") + max_files = env_int("PCHOME_SALES_MAX_FILES_PER_RUN", 5, 1, 25) + sources = sorted( + ( + path for path in root.iterdir() + if path.is_file() and not path.is_symlink() and path.suffix.lower() in VALID_EXTENSIONS + ), + key=lambda path: path.stat().st_mtime, + reverse=True, + )[:max_files] + candidates: List[AcquisitionCandidate] = [] + for source in sources: + temp_path = _make_temp_path(source.suffix.lower()) + try: + shutil.copyfile(str(source), temp_path) + _validate_excel_file(temp_path, source.name) + except Exception: + Path(temp_path).unlink(missing_ok=True) + for candidate in candidates: + candidate.cleanup() + raise + fingerprint = _sha256_file(temp_path) + candidates.append(AcquisitionCandidate( + source_type="controlled_local_drop", + file_path=temp_path, + file_name=_safe_file_name(source.name), + fingerprint=fingerprint, + source_ref_hash=_hash_text(str(source)), + finalize_success=lambda item=source, fp=fingerprint: _archive_file(item, "archive", fp), + finalize_rejected=lambda item=source, fp=fingerprint: _archive_file(item, "quarantine", fp), + cleanup=lambda candidate_path=temp_path: Path(candidate_path).unlink(missing_ok=True), + )) + return candidates diff --git a/services/pchome_sales_acquisition_service.py b/services/pchome_sales_acquisition_service.py new file mode 100644 index 0000000..1d3025a --- /dev/null +++ b/services/pchome_sales_acquisition_service.py @@ -0,0 +1,498 @@ +"""Authorized, replay-safe PChome sales report acquisition and import.""" + +from __future__ import annotations + +import json +import logging +import os +import uuid +from datetime import date, datetime +from pathlib import Path +from typing import Dict, List, Optional + +import pytz +from sqlalchemy import text + +from database.import_models import SalesAcquisitionReceipt +from services.google_drive_service import drive_service +from services.import_service import Session, humanize_import_error, import_service +from services.pchome_sales_acquisition_providers import ( + AcquisitionCandidate, + AuthorizedSalesProviders, + ProviderError, + env_bool, + env_int, +) + + +logger = logging.getLogger(__name__) +TAIPEI_TZ = pytz.timezone("Asia/Taipei") +WORK_ITEM_ID = "P0-PCHOME-SALES-AUTHORIZED-ACQUISITION" + + +class PChomeSalesAcquisitionService: + """Acquire authorized reports and close the import loop without manual review.""" + + def __init__(self, providers: Optional[AuthorizedSalesProviders] = None): + self.providers = providers or AuthorizedSalesProviders() + + def _freshness(self) -> Dict[str, object]: + latest = None + try: + session = Session() + try: + latest = session.execute( + text("SELECT MAX(snapshot_date) FROM daily_sales_snapshot") + ).scalar() + finally: + session.close() + except Exception: + logger.warning("Unable to read PChome sales freshness", exc_info=True) + + normalized = None + if latest: + try: + normalized = latest if isinstance(latest, date) else date.fromisoformat(str(latest)[:10]) + except (TypeError, ValueError): + normalized = None + lag_days = (datetime.now(TAIPEI_TZ).date() - normalized).days if normalized else None + return { + "latest_sales_date": normalized.isoformat() if normalized else None, + "data_lag_days": lag_days, + "decision_ready": lag_days is not None and lag_days <= 1, + "freshness_status": ( + "fresh" if lag_days is not None and lag_days <= 1 + else "degraded" if lag_days is not None + else "missing" + ), + } + + def _latest_receipt(self) -> Optional[Dict[str, object]]: + session = Session() + try: + receipt = session.query(SalesAcquisitionReceipt).order_by( + SalesAcquisitionReceipt.started_at.desc() + ).first() + return receipt.to_public_dict() if receipt else None + except Exception: + logger.warning("Unable to read latest sales acquisition receipt", exc_info=True) + return None + finally: + session.close() + + def readiness(self) -> Dict[str, object]: + try: + drive = drive_service.check_auth_readiness(refresh_expired=False) + drive_ready = bool(drive.get("ready")) + drive_status = drive.get("kind") or "unknown" + except Exception: + drive_ready = False + drive_status = "probe_failed" + + http_enabled = env_bool("PCHOME_SALES_HTTP_ENABLED") + http_ready = bool( + http_enabled + and os.getenv("PCHOME_SALES_HTTP_URL", "").strip() + and os.getenv("PCHOME_SALES_HTTP_ALLOWED_HOSTS", "").strip() + ) + imap_enabled = env_bool("PCHOME_SALES_IMAP_ENABLED") + imap_ready = bool( + imap_enabled + and os.getenv("PCHOME_SALES_IMAP_HOST", "").strip() + and os.getenv("PCHOME_SALES_IMAP_USER", "").strip() + and os.getenv("PCHOME_SALES_IMAP_PASSWORD", "").strip() + ) + local_path = os.getenv("PCHOME_SALES_LOCAL_DROP_DIR", "").strip() + local_ready = bool(local_path and Path(local_path).is_dir()) + sources = [ + {"id": "google_drive", "enabled": True, "ready": drive_ready, "status": drive_status}, + { + "id": "authorized_https", + "enabled": http_enabled, + "ready": http_ready, + "status": "ready" if http_ready else "disabled" if not http_enabled else "incomplete", + }, + { + "id": "authorized_imap", + "enabled": imap_enabled, + "ready": imap_ready, + "status": "ready" if imap_ready else "disabled" if not imap_enabled else "incomplete", + }, + { + "id": "controlled_local_drop", + "enabled": bool(local_path), + "ready": local_ready, + "status": "ready" if local_ready else "disabled" if not local_path else "path_missing", + }, + ] + ready_count = sum(1 for source in sources if source["ready"]) + freshness = self._freshness() + return { + "success": True, + "automation_mode": "controlled_apply", + "manual_review_required": False, + "risk_level": "medium", + "source_priority": [source["id"] for source in sources], + "sources": sources, + "asset_coverage": { + "ready": ready_count, + "total": len(sources), + "percent": round(ready_count / len(sources) * 100, 1), + }, + "freshness": freshness, + "runtime_closure": "closed" if freshness["decision_ready"] else "blocked_by_upstream_freshness", + "last_receipt": self._latest_receipt(), + } + + def _persist_receipt( + self, + *, + receipt_id: str, + trace_id: str, + span_id: str, + run_id: str, + status: str, + source_type: str, + candidate: Optional[AcquisitionCandidate], + decision: str, + before: Dict[str, object], + stages: Dict[str, object], + verifier: Dict[str, object], + import_job_id: Optional[int], + rows_imported: int, + error_kind: Optional[str], + message: str, + started_at: datetime, + ) -> bool: + session = Session() + try: + session.add(SalesAcquisitionReceipt( + receipt_id=receipt_id, + trace_id=trace_id, + span_id=span_id, + run_id=run_id, + work_item_id=WORK_ITEM_ID, + status=status, + source_type=source_type, + source_fingerprint=candidate.fingerprint if candidate else None, + source_ref_hash=candidate.source_ref_hash if candidate else None, + source_file_name=candidate.file_name if candidate else None, + risk_level="medium", + policy_decision="controlled_apply_allowed", + decision=decision, + import_job_id=import_job_id, + rows_imported=rows_imported, + before_freshness_json=json.dumps(before, ensure_ascii=False), + stage_receipts_json=json.dumps(stages, ensure_ascii=False), + verifier_json=json.dumps(verifier, ensure_ascii=False), + error_kind=error_kind, + public_message=message, + started_at=started_at.replace(tzinfo=None), + completed_at=datetime.now(TAIPEI_TZ).replace(tzinfo=None), + )) + session.commit() + return True + except Exception: + session.rollback() + logger.error("Unable to persist PChome sales acquisition receipt", exc_info=True) + return False + finally: + session.close() + + def run(self, *, trigger: str = "scheduler") -> Dict[str, object]: + started_at = datetime.now(TAIPEI_TZ) + receipt_id = uuid.uuid4().hex + trace_id = uuid.uuid4().hex + span_id = uuid.uuid4().hex[:16] + run_id = uuid.uuid4().hex + before = self._freshness() + 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"}, + "source_diff": {"before": before}, + "ai_decision": {"decision": "probe_authorized_sources_in_policy_order"}, + "risk_policy": {"risk": "medium", "decision": "controlled_apply_allowed"}, + "check_mode": {"status": "passed", "max_files": env_int("PCHOME_SALES_MAX_FILES_PER_RUN", 5, 1, 25)}, + "providers": [], + } + candidates: List[AcquisitionCandidate] = [] + successful_results: List[Dict[str, object]] = [] + failed_results: List[Dict[str, object]] = [] + duplicate_results: List[Dict[str, object]] = [] + finalization_failures: List[Dict[str, object]] = [] + selected_candidate = None + error_kind = None + + try: + drive_result = import_service.auto_import_from_drive() + stages["providers"].append({ + "source": "google_drive", + "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), + "error_kind": drive_result.get("error_kind"), + }) + if drive_result.get("imported_count", 0) > 0: + successful_results.append({ + "source_type": "google_drive", + "job_id": None, + "rows": int(drive_result.get("total_rows") or 0), + "date_range": drive_result.get("date_range"), + "result": drive_result, + }) + if not drive_result.get("source_finalize_ok", True): + finalization_failures.append({ + "source_type": "google_drive", + "error_kind": "source_finalize_failed", + }) + elif not drive_result.get("success"): + failed_results.append({ + "source_type": "google_drive", + "error_kind": drive_result.get("error_kind") or "drive_acquisition_failed", + "message": drive_result.get("message") or "Google Drive 授權來源目前無法取得報表。", + }) + + if not successful_results: + for source_name, provider in ( + ("authorized_https", self.providers.http_candidates), + ("authorized_imap", self.providers.imap_candidates), + ("controlled_local_drop", self.providers.local_candidates), + ): + try: + provider_candidates = provider() + candidates.extend(provider_candidates) + stages["providers"].append({ + "source": source_name, + "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}) + + max_files = env_int("PCHOME_SALES_MAX_FILES_PER_RUN", 5, 1, 25) + for candidate in candidates[:max_files]: + selected_candidate = candidate + try: + result = import_service.import_local_daily_sales( + candidate.file_path, + source_type=candidate.source_type, + source_fingerprint=candidate.fingerprint, + source_file_name=candidate.file_name, + ) + job = result.get("job") or {} + summary = job.get("import_summary") or {} + record = { + "source_type": candidate.source_type, + "job_id": result.get("job_id"), + "rows": int(summary.get("imported_count") or 0), + "date_range": ( + {"min": summary.get("date_min"), "max": summary.get("date_max")} + if summary.get("date_min") and summary.get("date_max") + else None + ), + "status": result.get("status"), + } + if result.get("duplicate"): + duplicate_results.append(record) + try: + candidate.finalize_success() + except Exception: + logger.error("Duplicate source finalization failed", exc_info=True) + finalization_failures.append({ + "source_type": candidate.source_type, + "error_kind": "source_finalize_failed", + }) + elif result.get("success"): + successful_results.append(record) + try: + candidate.finalize_success() + except Exception: + logger.error("Imported source finalization failed", exc_info=True) + finalization_failures.append({ + "source_type": candidate.source_type, + "error_kind": "source_finalize_failed", + }) + else: + current_step = str(job.get("current_step") or "") + if "欄位" in current_step or "日期" in current_step: + try: + candidate.finalize_rejected() + except Exception: + logger.error("Rejected source finalization failed", exc_info=True) + finalization_failures.append({ + "source_type": candidate.source_type, + "error_kind": "source_finalize_failed", + }) + failed_results.append({**record, "error_kind": result.get("error_kind") or "import_failed_rolled_back"}) + finally: + candidate.cleanup() + except Exception as exc: + logger.error("PChome sales acquisition run failed", exc_info=True) + error_kind = "acquisition_runtime_error" + failed_results.append({ + "source_type": selected_candidate.source_type if selected_candidate else "orchestrator", + "error_kind": error_kind, + "message": humanize_import_error(exc), + }) + finally: + for candidate in candidates: + try: + candidate.cleanup() + except Exception: + logger.warning("Unable to clean sales acquisition temp file", exc_info=True) + + 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) + source_types = sorted({ + item.get("source_type") + for item in successful_results + duplicate_results + failed_results + if item.get("source_type") + }) + source_type = source_types[0] if len(source_types) == 1 else "multiple" if source_types else "none" + date_mins = [ + item["date_range"]["min"] + for item in successful_results + if item.get("date_range") and item["date_range"].get("min") + ] + date_maxes = [ + item["date_range"]["max"] + for item in successful_results + if item.get("date_range") and item["date_range"].get("max") + ] + date_range = ( + {"min": min(date_mins), "max": max(date_maxes)} + if date_mins and date_maxes + else None + ) + + if finalization_failures and (successful_results or duplicate_results): + status = "partial" + decision = "runtime_verified_source_finalize_failed" + error_kind = "source_finalize_failed" + message = "業績資料已通過正式驗證，但來源封存回讀失敗，系統將自動重試收尾。" + elif successful_results and verifier["decision_ready"]: + status = "completed" + decision = "imported_and_verified" + message = f"已自動匯入 {rows_imported} 筆業績資料，雙表與新鮮度驗證通過。" + elif successful_results: + status = "partial" + decision = "imported_but_freshness_not_closed" + message = "業績檔已匯入並通過雙表驗證，但資料日期仍未達決策新鮮度。" + elif duplicate_results and verifier["decision_ready"]: + status = "completed_no_write" + decision = "duplicate_verified_no_write" + message = "來源內容已匯入過，冪等略過寫入且正式資料仍通過新鮮度驗證。" + elif not failed_results and verifier["decision_ready"]: + status = "completed_no_write" + decision = "no_candidate_fresh_no_write" + message = "目前沒有新業績檔，正式資料仍通過新鮮度驗證，本輪無需寫入。" + elif failed_results: + error_kind = error_kind or failed_results[0].get("error_kind") + import_failed = any( + str(item.get("error_kind") or "") in { + "import_failed_rolled_back", + "acquisition_runtime_error", + } + for item in failed_results + ) + status = "failed_rolled_back" if import_failed else "degraded_no_write" + decision = "import_failed_rolled_back" if import_failed else "provider_failed_no_write" + message = ( + "匯入驗證未通過，正式資料已回滾。" + if import_failed + else "授權來源目前無法取得報表，系統將自動重試。" + ) + else: + status = "blocked_with_safe_next_action" + decision = "no_authorized_candidate" + message = "所有已啟用授權來源目前都沒有新業績檔，正式資料未變更。" + + stages["execution"] = { + "status": status, + "successful_files": len(successful_results), + "duplicate_files": len(duplicate_results), + "failed_files": len(failed_results), + "source_finalize_failed": len(finalization_failures), + "rows_imported": rows_imported, + "idempotent": True, + "bounded": True, + } + stages["post_verifier"] = verifier + stages["terminal"] = { + "status": status, + "rollback": status == "failed_rolled_back", + "no_write": status in { + "completed_no_write", + "blocked_with_safe_next_action", + "failed_rolled_back", + "degraded_no_write", + }, + } + stages["learning_writeback"] = { + "target": "pchome_sales_acquisition_receipts", + "ack": "committed_with_receipt", + "km_rag_payload": "not_applicable_for_raw_sales_payload", + } + persisted = self._persist_receipt( + receipt_id=receipt_id, + trace_id=trace_id, + span_id=span_id, + run_id=run_id, + status=status, + source_type=source_type, + candidate=selected_candidate, + decision=decision, + before=before, + stages=stages, + verifier=verifier, + import_job_id=import_job_id, + rows_imported=rows_imported, + error_kind=error_kind, + message=message, + started_at=started_at, + ) + if not persisted and status in {"completed", "completed_no_write"}: + status = "partial" + decision = "runtime_verified_receipt_write_failed" + message = "業績資料已通過驗證，但治理 receipt 寫入失敗，閉環尚未完成。" + + return { + "success": status in {"completed", "completed_no_write"}, + "status": status, + "message": message, + "trace_id": trace_id, + "span_id": span_id, + "run_id": run_id, + "work_item_id": WORK_ITEM_ID, + "receipt_id": receipt_id, + "receipt_persisted": persisted, + "source_type": source_type, + "file_count": len(successful_results) + len(duplicate_results) + len(failed_results), + "imported_count": len(successful_results), + "failed_count": len(failed_results), + "duplicate_count": len(duplicate_results), + "source_finalize_failed_count": len(finalization_failures), + "total_rows": rows_imported, + "date_range": date_range, + "latest_sales_date": verifier.get("latest_sales_date"), + "data_lag_days": verifier.get("data_lag_days"), + "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" + ), + "providers": stages["providers"], + "verifier": verifier, + } + + +pchome_sales_acquisition_service = PChomeSalesAcquisitionService() diff --git a/templates/auto_import_index.html b/templates/auto_import_index.html index fd36db9..5097bd6 100644 --- a/templates/auto_import_index.html +++ b/templates/auto_import_index.html @@ -1,6 +1,6 @@ {% extends "ewoooc_base.html" %} -{% block title %}當日業績報表匯入{% endblock %} +{% block title %}業績資料自動化{% endblock %} {% block page_attrs %}data-page-group="monitor" data-page-id="auto-import"{% endblock %} {% block extra_head %} @@ -11,429 +11,357 @@ {% block content %}
- - {# ─────────── 頁首 ─────────── #}
- +
-

當日業績報表匯入

-

- - 保持 PChome 業績新鮮，讓評估、分析與建議有可靠資料。 -

+

業績資料自動化

+

正在讀取正式資料狀態

+
+
+ +
- {# ─────────── 配置區 ─────────── #} -
-
-

- Google Drive 自動匯入配置 -

-
-
-
- -
- -

每 30 分鐘檢查雲端業績檔，讓日報、成長分析與作戰清單保持新鮮。

-
- -
-
- - - 設定要監控的 Google Drive 資料夾路徑 -
-
- - - 用於過濾特定名稱的檔案 -
-
- -
- - - - -
+
+
+ 最新業績 + -- + 讀取中 +
+
+ 決策狀態 + -- + 讀取中 +
+
+ 來源覆蓋 + -- + 讀取中 +
+
+ 最近執行 + -- + 尚無 receipt
- {# ─────────── 手動上傳 ─────────── #} -
-
-

- 手動上傳匯入 -

+ + +
+
+
+ 授權來源 +

資料來源

+
+ 自動安全執行
-
-
-
- -
-

每日業績快照

-

- 上傳當日業績檔，更新日報、成長分析與今日作戰清單。 -

-

- 檔名建議：即時業績_當日_YYYYMMDD.xlsx；送出後更新日報、成長分析與今日作戰清單。 -

-
-
- -
-
- - -
-
- -
-
+
+
正在讀取來源狀態
+ +
+ Drive 來源設定 +
+
+
+
+ + +
+
+ + +
+
+
+ + + +
+
+
- {# ─────────── 任務歷史 ─────────── #} -
-
-

- 匯入任務歷史 -

-
-
- - - - - - + + +
-
{% endblock %} {% block extra_scripts %} {% endblock %} diff --git a/tests/test_auto_import_failure_boundaries.py b/tests/test_auto_import_failure_boundaries.py index d065354..c7a5a8f 100644 --- a/tests/test_auto_import_failure_boundaries.py +++ b/tests/test_auto_import_failure_boundaries.py @@ -94,12 +94,14 @@ def test_daily_sales_import_fails_when_monthly_sync_fails(monkeypatch, tmp_path) job = session.query(import_service.ImportJob).filter_by(id=job_id).one() assert job.status == "failed" assert job.progress_percent == 95 - assert job.current_step == "業績分析儀表板同步失敗" + assert job.current_step == "交易驗證失敗，已回滾" assert "monthly sync boom" in job.error_message summary = json.loads(job.import_summary) assert summary["sync_success"] is False assert summary["synced_to"] is None assert "monthly sync boom" in summary["sync_error"] + assert summary["atomic"] is True + assert summary["rolled_back"] is True finally: session.close() @@ -107,11 +109,45 @@ def test_daily_sales_import_fails_when_monthly_sync_fails(monkeypatch, tmp_path) snapshot_rows = conn.execute(text("SELECT COUNT(*) FROM daily_sales_snapshot")).scalar() monthly_rows = conn.execute(text("SELECT COUNT(*) FROM realtime_sales_monthly")).scalar() - assert snapshot_rows == 1 + assert snapshot_rows == 0 assert monthly_rows == 0 assert FakeNotificationManager.sent_messages +def test_daily_sales_import_preserves_today_fallback_when_date_column_is_missing(monkeypatch, tmp_path): + import_service = _load_import_service(monkeypatch, f"sqlite:///{tmp_path / 'momo.db'}") + _prepare_daily_sales_tables(import_service) + source_df = pd.DataFrame([{ + "商品ID": "A001", + "商品名稱": "測試商品", + "銷售金額": 1200, + }]) + monkeypatch.setattr( + import_service, + "_read_daily_sales_excel", + lambda _path: ( + source_df.copy(), + {"date_col": None, "sheet_name": "即時業績明細", "header_row": 1}, + ), + ) + service = import_service.ImportService() + job_id = service.create_import_job("daily_sales", "drive-file-1", "daily.xlsx", 1024) + + assert service.process_daily_sales_import(job_id, str(tmp_path / "daily.xlsx")) is True + + expected_date = import_service.datetime.now(import_service.TAIPEI_TZ).date().isoformat() + with import_service.engine.connect() as conn: + snapshot = conn.execute(text( + 'SELECT "日期", snapshot_date FROM daily_sales_snapshot' + )).one() + monthly = conn.execute(text( + 'SELECT "日期" FROM realtime_sales_monthly' + )).one() + assert snapshot[0] == expected_date + assert str(snapshot[1])[:10] == expected_date + assert monthly[0] == expected_date + + def test_auto_import_does_not_move_drive_file_when_import_fails(monkeypatch, tmp_path): import_service = _load_import_service(monkeypatch, f"sqlite:///{tmp_path / 'momo.db'}") import_service.Base.metadata.create_all(import_service.engine) @@ -145,6 +181,47 @@ def test_auto_import_does_not_move_drive_file_when_import_fails(monkeypatch, tmp assert fake_drive.moved_files == [] +def test_auto_import_reports_partial_when_drive_archive_fails(monkeypatch, tmp_path): + import_service = _load_import_service(monkeypatch, f"sqlite:///{tmp_path / 'momo.db'}") + import_service.Base.metadata.create_all(import_service.engine) + monkeypatch.chdir(tmp_path) + + class FakeDriveService: + last_error_kind = None + last_error = None + + def list_files_in_folder(self, folder_path, file_pattern): + return [{"id": "drive-file-1", "name": "daily.xlsx", "size": 1024}] + + def download_file(self, file_id, local_path): + os.makedirs(os.path.dirname(local_path), exist_ok=True) + with open(local_path, "wb") as handle: + handle.write(b"test") + return True + + def move_file(self, file_id, folder, create_missing=False): + return False + + monkeypatch.setattr(import_service, "drive_service", FakeDriveService()) + service = import_service.ImportService() + monkeypatch.setattr(service, "process_daily_sales_import", lambda job_id, path: True) + + result = service.auto_import_from_drive() + + assert result["success"] is False + assert result["status"] == "partial_source_finalize_failed" + assert result["imported_count"] == 1 + assert result["source_finalize_ok"] is False + assert result["archive_failed_count"] == 1 + session = import_service.Session() + try: + job = session.query(import_service.ImportJob).one() + assert job.status == "completed" + assert job.current_step == "匯入完成，來源封存待自動重試" + finally: + session.close() + + def test_auto_import_fails_closed_when_drive_auth_fails(monkeypatch, tmp_path): import_service = _load_import_service(monkeypatch, f"sqlite:///{tmp_path / 'momo.db'}") import_service.Base.metadata.create_all(import_service.engine) diff --git a/tests/test_frontend_v2_assets.py b/tests/test_frontend_v2_assets.py index b3e1185..13497f5 100644 --- a/tests/test_frontend_v2_assets.py +++ b/tests/test_frontend_v2_assets.py @@ -793,8 +793,15 @@ def test_utility_pages_keep_operator_copy_professional(): assert "簡報線上預覽" in ppt_preview assert "下載簡報檔" in ppt_history - assert "送出後更新日報、成長分析與今日作戰清單" in auto_import - assert "等待系統更新任務狀態；若重複停在異常" in auto_import + assert "業績資料自動化" in auto_import + assert "正式業績可供分析與決策" in auto_import + assert "雙表已驗證" in auto_import + assert "自動安全執行" in auto_import + assert "Controlled apply" not in auto_import + assert "AUTHORIZED SOURCES" not in auto_import + assert "VERIFIED RUNS" not in auto_import + assert "手動上傳匯入" not in auto_import + assert "uploadManualFile" not in auto_import assert "return raw ||" not in auto_import assert "供貨風險匯入" in stockout_import assert "缺少必要內容時，會先停止匯入" in stockout_import diff --git a/tests/test_import_service_sql_params.py b/tests/test_import_service_sql_params.py index 3d0dcff..f91d4e8 100644 --- a/tests/test_import_service_sql_params.py +++ b/tests/test_import_service_sql_params.py @@ -29,11 +29,23 @@ def test_daily_snapshot_delete_casts_text_date_column_on_postgres(monkeypatch): def test_daily_snapshot_delete_query_uses_snapshot_date_cast_helper(): source = Path("services/import_service.py").read_text(encoding="utf-8") - assert 'snapshot_date_expr = _date_filter_expr("snapshot_date")' in source - assert "DELETE FROM {table_name} WHERE {snapshot_date_expr} IN" in source + assert 'snapshot_expr = _date_filter_expr("snapshot_date")' in source + assert "f\"WHERE {snapshot_expr} IN ({snapshot_placeholders})\"" in source assert "DELETE FROM daily_sales_snapshot WHERE snapshot_date IN" not in source +def test_sales_table_replacement_uses_one_transaction_connection(): + source = Path("services/import_service.py").read_text(encoding="utf-8") + start = source.index("def _replace_sales_tables_atomic") + end = source.index("# 資料庫設定", start) + block = source[start:end] + + assert "with engine.begin() as conn:" in block + assert 'df.to_sql(\n "daily_sales_snapshot",\n conn,' in block + assert 'df_monthly.to_sql(\n "realtime_sales_monthly",\n conn,' in block + assert "conn.commit()" not in block + + def test_daily_snapshot_delete_uses_iso_dates_on_sqlite(monkeypatch): monkeypatch.setattr(import_service, "_db_dialect_name", lambda: "sqlite") diff --git a/tests/test_pchome_revenue_growth_service.py b/tests/test_pchome_revenue_growth_service.py index 4cc0695..ad9f65e 100644 --- a/tests/test_pchome_revenue_growth_service.py +++ b/tests/test_pchome_revenue_growth_service.py @@ -728,7 +728,7 @@ def test_primary_pages_use_growth_outcome_copy_instead_of_feature_explaining(): expected = { "templates/daily_sales.html": "找出下滑與價差壓力", "templates/ai_recommend.html": "把價差、商品證據與趨勢轉成主推、調價、補比價動作", - "templates/auto_import_index.html": "保持 PChome 業績新鮮", + "templates/auto_import_index.html": "正式業績可供分析與決策", "templates/price_comparison.html": "確認同款、判斷價差、決定下一步", "templates/vendor_stockout_index_v2.html": "避免主推商品斷貨", "templates/monthly_summary_analysis.html": "判斷成長、毛利與品類結構", @@ -889,14 +889,14 @@ def test_governance_and_low_frequency_pages_avoid_engineering_status_copy(): "templates/403.html": ["權限守門", "未授權操作影響營運資料", "權限控管"], "templates/maintenance.html": ["服務維護", "確認業績、比價與匯入狀態", "台北時間"], "templates/auto_import_index.html": [ - "更新日報、成長分析與今日作戰清單", - "送出後更新日報、成長分析與今日作戰清單", - "作戰清單保持新鮮", - "共更新", - "buildImportActionHint", - "重新確認 Google Drive 授權", - "改用當日業績明細檔", - "重新匯入最新檔案", + "業績資料自動化", + "來源覆蓋", + "自動安全執行", + "執行自動取得", + "雙表已驗證", + "等待最新業績", + "loadReadiness", + "/api/auto_import/run", ], "templates/settings.html": ["比價來源同步", "補齊 MOMO 參考來源"], "templates/system_settings.html": [ diff --git a/tests/test_pchome_sales_acquisition_service.py b/tests/test_pchome_sales_acquisition_service.py new file mode 100644 index 0000000..7a94ef2 --- /dev/null +++ b/tests/test_pchome_sales_acquisition_service.py @@ -0,0 +1,334 @@ +import importlib +import json +import os +from datetime import datetime + +import pandas as pd +import pytz +from flask import Flask +from sqlalchemy import text + + +TAIPEI_TZ = pytz.timezone("Asia/Taipei") + + +def _load_services(monkeypatch, database_url): + os.environ.setdefault("MOMO_ALLOW_INSECURE_CONFIG_FOR_TESTS", "true") + import config + + monkeypatch.setattr(config, "DATABASE_PATH", database_url) + import services.import_service as import_service + + import_service = importlib.reload(import_service) + import services.pchome_sales_acquisition_service as acquisition + + acquisition = importlib.reload(acquisition) + return import_service, acquisition + + +def _prepare_tables(import_service): + import_service.Base.metadata.create_all(import_service.engine) + with import_service.engine.begin() as conn: + conn.execute(text("DROP TABLE IF EXISTS daily_sales_snapshot")) + conn.execute(text("DROP TABLE IF EXISTS realtime_sales_monthly")) + conn.execute(text(""" + CREATE TABLE daily_sales_snapshot ( + "日期" TEXT, + "商品ID" TEXT, + "商品名稱" TEXT, + "銷售金額" INTEGER, + snapshot_date TEXT + ) + """)) + conn.execute(text(""" + CREATE TABLE realtime_sales_monthly ( + "日期" TEXT, + "商品ID" TEXT, + "商品名稱" TEXT, + "銷售金額" INTEGER + ) + """)) + + +class EmptyDrive: + last_error_kind = None + last_error = None + + def list_files_in_folder(self, _folder_path, _file_pattern): + return [] + + +class ReadyDrive(EmptyDrive): + def check_auth_readiness(self, refresh_expired=False): + return {"ready": True, "kind": "ready"} + + +class FailedDrive(EmptyDrive): + last_error_kind = "authentication_failed" + last_error = "credentials missing" + + +def _write_sales_file(path): + today = datetime.now(TAIPEI_TZ).strftime("%Y-%m-%d") + pd.DataFrame([{ + "日期": today, + "商品ID": "P001", + "商品名稱": "測試商品", + "銷售金額": 1680, + }]).to_excel(path, index=False, sheet_name="即時業績明細") + + +def _disable_remote_sources(monkeypatch): + monkeypatch.setenv("PCHOME_SALES_HTTP_ENABLED", "false") + monkeypatch.setenv("PCHOME_SALES_IMAP_ENABLED", "false") + + +def test_local_drop_closes_import_and_archives_only_after_verification(monkeypatch, tmp_path): + import_service, acquisition = _load_services(monkeypatch, f"sqlite:///{tmp_path / 'momo.db'}") + _prepare_tables(import_service) + _disable_remote_sources(monkeypatch) + drop_dir = tmp_path / "drop" + drop_dir.mkdir() + source = drop_dir / "即時業績_當日.xlsx" + _write_sales_file(source) + monkeypatch.setenv("PCHOME_SALES_LOCAL_DROP_DIR", str(drop_dir)) + monkeypatch.setattr(import_service, "drive_service", EmptyDrive()) + + result = acquisition.PChomeSalesAcquisitionService().run(trigger="test") + + assert result["success"] is True + assert result["status"] == "completed" + assert result["source_type"] == "controlled_local_drop" + assert result["total_rows"] == 1 + assert result["decision_ready"] is True + assert result["receipt_persisted"] is True + assert len(result["trace_id"]) == 32 + assert len(result["span_id"]) == 16 + assert result["trace_id"] == result["trace_id"].lower() + assert result["span_id"] == result["span_id"].lower() + assert not source.exists() + assert (drop_dir / "archive" / source.name).exists() + + with import_service.engine.connect() as conn: + assert conn.execute(text("SELECT COUNT(*) FROM daily_sales_snapshot")).scalar() == 1 + assert conn.execute(text("SELECT COUNT(*) FROM realtime_sales_monthly")).scalar() == 1 + assert conn.execute(text("SELECT COUNT(*) FROM pchome_sales_acquisition_receipts")).scalar() == 1 + + +def test_same_content_is_idempotent_and_does_not_duplicate_rows(monkeypatch, tmp_path): + import_service, acquisition = _load_services(monkeypatch, f"sqlite:///{tmp_path / 'momo.db'}") + _prepare_tables(import_service) + _disable_remote_sources(monkeypatch) + drop_dir = tmp_path / "drop" + drop_dir.mkdir() + source = drop_dir / "即時業績_當日.xlsx" + _write_sales_file(source) + original_bytes = source.read_bytes() + monkeypatch.setenv("PCHOME_SALES_LOCAL_DROP_DIR", str(drop_dir)) + monkeypatch.setattr(import_service, "drive_service", EmptyDrive()) + service = acquisition.PChomeSalesAcquisitionService() + + first = service.run(trigger="test") + source.write_bytes(original_bytes) + second = service.run(trigger="test") + + assert first["status"] == "completed" + assert second["status"] == "completed_no_write" + assert second["duplicate_count"] == 1 + assert second["imported_count"] == 0 + with import_service.engine.connect() as conn: + assert conn.execute(text("SELECT COUNT(*) FROM daily_sales_snapshot")).scalar() == 1 + assert conn.execute(text("SELECT COUNT(*) FROM realtime_sales_monthly")).scalar() == 1 + + +def test_verified_import_is_partial_when_source_archive_ack_fails(monkeypatch, tmp_path): + import_service, acquisition = _load_services(monkeypatch, f"sqlite:///{tmp_path / 'momo.db'}") + import services.pchome_sales_acquisition_providers as providers + + _prepare_tables(import_service) + _disable_remote_sources(monkeypatch) + drop_dir = tmp_path / "drop" + drop_dir.mkdir() + source = drop_dir / "即時業績_當日.xlsx" + _write_sales_file(source) + monkeypatch.setenv("PCHOME_SALES_LOCAL_DROP_DIR", str(drop_dir)) + monkeypatch.setattr(import_service, "drive_service", EmptyDrive()) + monkeypatch.setattr( + providers, + "_archive_file", + lambda *_args, **_kwargs: (_ for _ in ()).throw(PermissionError("archive denied")), + ) + + result = acquisition.PChomeSalesAcquisitionService().run(trigger="test") + + assert result["success"] is False + assert result["status"] == "partial" + assert result["source_finalize_failed_count"] == 1 + assert result["safe_next_action"] == "retry_source_finalization" + assert source.exists() + with import_service.engine.connect() as conn: + assert conn.execute(text("SELECT COUNT(*) FROM daily_sales_snapshot")).scalar() == 1 + assert conn.execute(text("SELECT COUNT(*) FROM realtime_sales_monthly")).scalar() == 1 + + +def test_no_candidate_with_missing_freshness_is_blocked_and_receipted(monkeypatch, tmp_path): + import_service, acquisition = _load_services(monkeypatch, f"sqlite:///{tmp_path / 'momo.db'}") + _prepare_tables(import_service) + _disable_remote_sources(monkeypatch) + monkeypatch.delenv("PCHOME_SALES_LOCAL_DROP_DIR", raising=False) + monkeypatch.setattr(import_service, "drive_service", EmptyDrive()) + + 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["receipt_persisted"] is True + with import_service.engine.connect() as conn: + receipt = conn.execute(text( + "SELECT status, decision FROM pchome_sales_acquisition_receipts" + )).one() + assert receipt.status == "blocked_with_safe_next_action" + assert receipt.decision == "no_authorized_candidate" + + +def test_provider_failure_keeps_product_message_compact(monkeypatch, tmp_path): + import_service, acquisition = _load_services(monkeypatch, f"sqlite:///{tmp_path / 'momo.db'}") + _prepare_tables(import_service) + _disable_remote_sources(monkeypatch) + monkeypatch.delenv("PCHOME_SALES_LOCAL_DROP_DIR", raising=False) + monkeypatch.setattr(import_service, "drive_service", FailedDrive()) + + result = acquisition.PChomeSalesAcquisitionService().run(trigger="test") + + assert result["success"] is False + assert result["status"] == "degraded_no_write" + assert result["message"] == "授權來源目前無法取得報表，系統將自動重試。" + assert result["providers"][0]["error_kind"] == "authentication_failed" + + +def test_readiness_never_exposes_provider_credentials_or_urls(monkeypatch, tmp_path): + import_service, acquisition = _load_services(monkeypatch, f"sqlite:///{tmp_path / 'momo.db'}") + _prepare_tables(import_service) + monkeypatch.setattr(acquisition, "drive_service", ReadyDrive()) + monkeypatch.setenv("PCHOME_SALES_HTTP_ENABLED", "true") + monkeypatch.setenv("PCHOME_SALES_HTTP_URL", "https://reports.example.test/private/report.xlsx") + monkeypatch.setenv("PCHOME_SALES_HTTP_ALLOWED_HOSTS", "reports.example.test") + monkeypatch.setenv("PCHOME_SALES_HTTP_BEARER_TOKEN", "top-secret-http-token") + monkeypatch.setenv("PCHOME_SALES_IMAP_ENABLED", "true") + monkeypatch.setenv("PCHOME_SALES_IMAP_HOST", "mail.example.test") + monkeypatch.setenv("PCHOME_SALES_IMAP_USER", "private-user@example.test") + monkeypatch.setenv("PCHOME_SALES_IMAP_PASSWORD", "top-secret-imap-password") + + payload = acquisition.PChomeSalesAcquisitionService().readiness() + public_text = json.dumps(payload, ensure_ascii=False) + + assert payload["manual_review_required"] is False + assert payload["asset_coverage"]["ready"] == 3 + assert "top-secret" not in public_text + assert "reports.example.test" not in public_text + assert "mail.example.test" not in public_text + assert "private-user" not in public_text + + +def test_https_provider_rejects_non_https_and_non_allowlisted_hosts(monkeypatch): + import services.pchome_sales_acquisition_service as acquisition + import services.pchome_sales_acquisition_providers as provider_module + + service = acquisition.PChomeSalesAcquisitionService() + monkeypatch.setenv("PCHOME_SALES_HTTP_ALLOWED_HOSTS", "reports.example.test") + + try: + service.providers.validate_http_url("http://reports.example.test/report.xlsx") + assert False, "HTTP URL should be rejected" + except acquisition.ProviderError as exc: + assert exc.kind == "http_url_rejected" + + try: + service.providers.validate_http_url("https://untrusted.example.test/report.xlsx") + assert False, "Non-allowlisted host should be rejected" + except acquisition.ProviderError as exc: + assert exc.kind == "http_host_not_allowed" + + monkeypatch.setenv("PCHOME_SALES_HTTP_ALLOWED_HOSTS", "reports.example.test") + try: + service.providers.validate_http_url("https://reports.example.test:8443/report.xlsx") + assert False, "Non-allowlisted port should be rejected" + except acquisition.ProviderError as exc: + assert exc.kind == "http_port_rejected" + + monkeypatch.setattr( + provider_module.socket, + "getaddrinfo", + lambda *_args, **_kwargs: [(2, 1, 6, "", ("127.0.0.1", 443))], + ) + try: + service.providers.validate_http_url("https://reports.example.test/report.xlsx") + assert False, "Private resolved address should be rejected" + except acquisition.ProviderError as exc: + assert exc.kind == "http_private_address_rejected" + + +def test_auto_import_api_uses_governed_acquisition_service(monkeypatch): + import routes.auto_import_routes as routes + + class FakeAcquisitionService: + def readiness(self): + return {"success": True, "runtime_closure": "blocked_by_upstream_freshness"} + + def run(self, *, trigger): + return { + "success": False, + "status": "blocked_with_safe_next_action", + "trigger": trigger, + "receipt_persisted": True, + } + + monkeypatch.setattr(routes, "pchome_sales_acquisition_service", FakeAcquisitionService()) + app = Flask(__name__) + app.register_blueprint(routes.auto_import_bp) + client = app.test_client() + + readiness = client.get("/api/auto_import/readiness") + run = client.post("/api/auto_import/run") + legacy = client.post("/api/manual_import") + + assert readiness.status_code == 200 + assert readiness.get_json()["runtime_closure"] == "blocked_by_upstream_freshness" + assert run.get_json()["status"] == "blocked_with_safe_next_action" + assert run.get_json()["trigger"] == "operator_command" + assert legacy.get_json()["trigger"] == "operator_command" + + +def test_scheduler_continues_when_legacy_hitl_pause_is_set(monkeypatch): + import scheduler + import services.agent_actions as agent_actions + import services.pchome_sales_acquisition_service as acquisition + + calls = [] + saved = {} + monkeypatch.setattr(agent_actions, "is_task_paused", lambda _name: True) + monkeypatch.setattr( + acquisition.pchome_sales_acquisition_service, + "run", + lambda *, trigger: calls.append(trigger) or { + "success": False, + "status": "blocked_with_safe_next_action", + "message": "no candidate", + "trace_id": "trace-test", + "receipt_id": "receipt-test", + "receipt_persisted": True, + "latest_sales_date": "2026-07-10", + "data_lag_days": 1, + "decision_ready": True, + "safe_next_action": "scheduler_retry_authorized_sources", + }, + ) + monkeypatch.setattr(scheduler, "_save_stats", lambda name, stats: saved.update({"name": name, "stats": stats})) + + scheduler.run_auto_import_task() + + assert calls == ["scheduler"] + assert saved["name"] == "auto_import_task" + assert saved["stats"]["status"] == "Degraded" + assert saved["stats"]["receipt_persisted"] is True diff --git a/web/static/css/page-auto-import-bem.css b/web/static/css/page-auto-import-bem.css index 65f0b47..126695f 100644 --- a/web/static/css/page-auto-import-bem.css +++ b/web/static/css/page-auto-import-bem.css @@ -271,7 +271,7 @@ font-weight: var(--momo-font-semibold); font-size: var(--momo-text-xs); text-transform: uppercase; - letter-spacing: 0.04em; + letter-spacing: 0; padding: var(--momo-space-3); border-bottom: 1px solid var(--momo-border); white-space: nowrap; @@ -520,3 +520,265 @@ min-width: 0; } } + +/* ---------- V10.776 automation cockpit ---------- */ +.ai-head__actions { + display: flex; + align-items: center; + gap: var(--momo-space-2); + margin-left: auto; +} + +.ai-icon-button, +.ai-row-action { + display: inline-flex; + align-items: center; + justify-content: center; + width: 40px; + height: 40px; + padding: 0; + border: 1px solid var(--momo-border); + border-radius: var(--momo-radius-sm); + background: var(--momo-surface-0); + color: var(--momo-ink-secondary); +} + +.ai-icon-button:hover, +.ai-icon-button:focus-visible, +.ai-row-action:hover, +.ai-row-action:focus-visible { + border-color: #2f6f8f; + color: #205771; + outline: 2px solid rgba(47, 111, 143, 0.2); + outline-offset: 1px; +} + +.ai-cockpit { + display: grid; + grid-template-columns: repeat(4, minmax(0, 1fr)); + margin-bottom: var(--momo-space-5); + border-top: 1px solid var(--momo-border); + border-bottom: 1px solid var(--momo-border); + background: var(--momo-surface-0); +} + +.ai-metric { + display: flex; + min-width: 0; + min-height: 118px; + flex-direction: column; + justify-content: center; + padding: var(--momo-space-4) var(--momo-space-5); + border-right: 1px solid var(--momo-border-light); +} + +.ai-metric:last-child { border-right: 0; } + +.ai-metric__label, +.ai-section__eyebrow { + color: var(--momo-ink-tertiary); + font-family: var(--momo-font-mono); + font-size: var(--momo-text-xs); + font-weight: var(--momo-font-semibold); + letter-spacing: 0; + text-transform: uppercase; +} + +.ai-metric__value { + margin: var(--momo-space-2) 0 var(--momo-space-1); + color: var(--momo-ink-primary); + font-family: var(--momo-font-heading); + font-size: var(--momo-text-xl); + font-weight: var(--momo-font-bold); + line-height: 1.25; + overflow-wrap: anywhere; +} + +.ai-metric__value--compact { font-size: var(--momo-text-base); } +.ai-metric__value[data-state="ready"] { color: #287a55; } +.ai-metric__value[data-state="blocked"] { color: #a33e4a; } + +.ai-metric__meta { + color: var(--momo-ink-tertiary); + font-size: var(--momo-text-xs); + overflow-wrap: anywhere; +} + +.ai-inline-status { + display: flex; + align-items: center; + gap: var(--momo-space-2); + margin: 0 0 var(--momo-space-5); + padding: var(--momo-space-3) var(--momo-space-4); + border-left: 3px solid #2f6f8f; + background: rgba(47, 111, 143, 0.08); + color: #205771; + font-size: var(--momo-text-sm); +} + +.ai-inline-status[hidden] { display: none; } +.ai-inline-status--success { border-color: #287a55; background: rgba(40, 122, 85, 0.09); color: #1f6245; } +.ai-inline-status--warning { border-color: #a06b15; background: rgba(160, 107, 21, 0.09); color: #76500f; } +.ai-inline-status--danger { border-color: #a33e4a; background: rgba(163, 62, 74, 0.09); color: #7b2f38; } + +.ai-section { + padding: var(--momo-space-5) 0; + border-top: 1px solid var(--momo-border-light); +} + +.ai-section__head { + display: flex; + align-items: center; + justify-content: space-between; + gap: var(--momo-space-3); + margin-bottom: var(--momo-space-4); +} + +.ai-section__title { + margin: var(--momo-space-1) 0 0; + color: var(--momo-ink-primary); + font-family: var(--momo-font-heading); + font-size: var(--momo-text-lg); + font-weight: var(--momo-font-semibold); + letter-spacing: 0; +} + +.ai-mode-badge { + display: inline-flex; + align-items: center; + gap: var(--momo-space-1); + padding: var(--momo-space-1) var(--momo-space-2); + border: 1px solid rgba(40, 122, 85, 0.35); + border-radius: var(--momo-radius-sm); + background: rgba(40, 122, 85, 0.08); + color: #1f6245; + font-family: var(--momo-font-mono); + font-size: var(--momo-text-xs); + font-weight: var(--momo-font-semibold); +} + +.ai-source-list { + border-top: 1px solid var(--momo-border-light); +} + +.ai-source-row { + display: grid; + grid-template-columns: 14px minmax(0, 1fr) auto; + gap: var(--momo-space-3); + align-items: center; + min-height: 52px; + padding: var(--momo-space-3) 0; + border-bottom: 1px solid var(--momo-border-light); +} + +.ai-source-row--loading { + display: block; + color: var(--momo-ink-tertiary); + font-size: var(--momo-text-sm); +} + +.ai-source-row__indicator { + width: 9px; + height: 9px; + border-radius: 50%; + background: var(--momo-border-strong); +} + +.ai-source-row__indicator--ready { background: #287a55; } +.ai-source-row__indicator--degraded { background: #a06b15; } +.ai-source-row__indicator--disabled { background: #8a9198; } + +.ai-source-row__name { + min-width: 0; + color: var(--momo-ink-primary); + font-size: var(--momo-text-sm); + font-weight: var(--momo-font-semibold); + overflow-wrap: anywhere; +} + +.ai-source-row__state { + color: var(--momo-ink-tertiary); + font-family: var(--momo-font-mono); + font-size: var(--momo-text-xs); +} + +.ai-disclosure { + margin-top: var(--momo-space-4); + border-bottom: 1px solid var(--momo-border-light); +} + +.ai-disclosure summary { + display: flex; + align-items: center; + gap: var(--momo-space-2); + padding: var(--momo-space-3) 0; + color: var(--momo-ink-secondary); + cursor: pointer; + font-size: var(--momo-text-sm); + font-weight: var(--momo-font-semibold); + list-style: none; +} + +.ai-disclosure summary::-webkit-details-marker { display: none; } +.ai-disclosure summary::after { + margin-left: auto; + color: var(--momo-ink-tertiary); + content: "+"; +} +.ai-disclosure[open] summary::after { content: "−"; } +.ai-disclosure__body { padding: var(--momo-space-3) 0 var(--momo-space-5); } + +.ai-count-separator { + margin: 0 var(--momo-space-1); + color: var(--momo-ink-tertiary); +} + +.ai-row-action { + width: 28px; + height: 28px; + margin-left: var(--momo-space-2); + color: #a33e4a; +} + +@media (max-width: 900px) { + .ai-cockpit { grid-template-columns: repeat(2, minmax(0, 1fr)); } + .ai-metric:nth-child(2) { border-right: 0; } + .ai-metric:nth-child(-n + 2) { border-bottom: 1px solid var(--momo-border-light); } +} + +@media (max-width: 768px) { + .ai-head { flex-wrap: wrap; } + .ai-head__main { flex-basis: calc(100% - 52px); } + .ai-head__actions { + width: 100%; + margin-left: 0; + } + .ai-head__actions .btn { flex: 1; min-height: 44px; } + .ai-cockpit { grid-template-columns: 1fr; } + .ai-metric, + .ai-metric:nth-child(2) { + min-height: 96px; + border-right: 0; + border-bottom: 1px solid var(--momo-border-light); + } + .ai-metric:last-child { border-bottom: 0; } + .ai-section__head { align-items: flex-start; } + .ai-mode-badge { max-width: 48%; text-align: right; } + .ai-source-row { grid-template-columns: 14px minmax(0, 1fr) minmax(80px, auto); } + + .ai-jobtable td:nth-child(1)::before { content: "檔案"; } + .ai-jobtable td:nth-child(2)::before { content: "狀態"; } + .ai-jobtable td:nth-child(3)::before { content: "進度"; } + .ai-jobtable td:nth-child(4)::before { content: "筆數"; } + .ai-jobtable td:nth-child(5)::before { content: "完成"; } + .ai-jobtable td:nth-child(6)::before { content: "驗證"; } +} + +@media (max-width: 380px) { + .ai-source-row { + grid-template-columns: 14px minmax(0, 1fr); + } + .ai-source-row__state { + grid-column: 2; + } +} diff --git a/web/static/css/page-auto-import.css b/web/static/css/page-auto-import.css index 7ba0070..4e6e966 100644 --- a/web/static/css/page-auto-import.css +++ b/web/static/css/page-auto-import.css @@ -234,7 +234,7 @@ font-weight: var(--momo-font-semibold); text-transform: uppercase; font-size: var(--momo-text-xs); - letter-spacing: 0.04em; + letter-spacing: 0; padding: var(--momo-space-3); white-space: nowrap; }