wooo/ewoooc
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(pchome): automate authorized sales acquisition
Some checks failed
CD Pipeline / deploy (push) Has been cancelled
Details

Browse Source
This commit is contained in:
ogt
2026-07-11 01:30:42 +08:00
parent f68223f0ed
commit 6e65ef00bb
21 changed files with 2428 additions and 617 deletions
Download Patch File Download Diff File Expand all files Collapse all files

523
services/import_service.py
View File

@@ -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

380
services/pchome_sales_acquisition_providers.py Normal file
View File

@@ -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

498
services/pchome_sales_acquisition_service.py Normal file
View File

@@ -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()