#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
自動匯入服務
負責從 Google Drive 自動下載、匯入、刪除檔案
"""

import logging
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
import pytz
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

# 台北時區
TAIPEI_TZ = pytz.timezone('Asia/Taipei')

DAILY_SALES_REQUIRED_COLUMN_GROUPS = {
    "商品名稱類": ["商品名稱", "品名", "Product", "Name"],
    "業績金額類": ["銷售金額", "總業績", "業績", "金額", "Amount", "Sales", "Total"],
}
DAILY_SALES_DATE_ALIASES = ["日期", "訂單日期", "交易日期", "Date"]
DAILY_SALES_DETAIL_SHEET_HINTS = ("即時業績明細", "業績明細", "明細", "detail", "daily")
DAILY_SALES_HEADER_SCAN_ROWS = 15

from services.google_drive_service import drive_service
from services.pchome_sales_freshness_policy import evaluate_pchome_sales_freshness
from database.import_models import ImportJob, ImportConfig, Base
from database.manager import ensure_metadata_initialized

# 設定日誌
logger = logging.getLogger(__name__)

_INTERNAL_ERROR_MARKERS = (
    "psycopg2",
    "sqlalchemy",
    "undefinedfunction",
    "traceback",
    "daily_sales_snapshot",
    "realtime_sales_monthly",
    "snapshot_date",
    "import_jobs",
    "::date",
)
_SNAKE_CASE_RE = re.compile(r"\b[a-z][a-z0-9]+(?:_[a-z0-9]+){1,}\b")


def humanize_import_error(message: Any) -> str:
    """Return operator-facing import error text without SQL/table internals."""
    raw = str(message or "").strip()
    if not raw:
        return ""

    lowered = raw.lower()
    if (
        "could not locate runnable browser" in lowered
        or "reauthorization_required" in lowered
        or "授權檔不存在" in raw
        or "缺少可刷新" in raw
        or "refresh token" in lowered
    ):
        return "Google Drive 授權需要重新確認；請重新完成雲端授權後再執行匯入。"
    if "token_store_failed" in lowered or "不可寫" in raw or ("permission" in lowered and "google" in lowered):
        return "Google Drive 授權無法穩定保存；請通知維護人員確認主機授權檔權限，避免重啟後再次失效。"
    if "google drive" in lowered or "credentials" in lowered or "authenticate" in lowered or "token" in lowered:
        return "Google Drive 連線或授權異常；請確認雲端資料夾權限後再執行匯入。"
    if "欄位" in raw or "excel" in lowered:
        return "匯入檔案格式與業績報表不一致；請確認檔案包含日期、商品與業績金額後重新匯入。"
    if "monthly sync" in lowered or any(marker in lowered for marker in _INTERNAL_ERROR_MARKERS):
        return "業績資料處理未完整完成；請重新匯入最新檔案，若重複失敗請通知維護人員檢查同步流程。"
    if _SNAKE_CASE_RE.search(raw):
        return "匯入處理發生系統異常；請重新匯入最新檔案，若重複失敗請通知維護人員。"

    compact = " ".join(raw.split())
    return compact[:180] + ("…" if len(compact) > 180 else "")


def _public_import_summary(raw_summary: Any) -> Optional[Dict[str, Any]]:
    """Return an operator-facing import summary without table names or SQL sync internals."""
    if not raw_summary:
        return None

    summary = raw_summary
    if isinstance(raw_summary, str):
        try:
            summary = json.loads(raw_summary)
        except Exception:
            return {"message": "匯入摘要暫時無法解析，請以任務狀態為準。"}

    if not isinstance(summary, dict):
        return None

    sync_success = summary.get("sync_success")
    public = {
        "message": summary.get("message") or "匯入任務已完成。",
        "imported_count": summary.get("imported_count"),
        "verified": bool(summary.get("verified")),
        "date_min": summary.get("date_min"),
        "date_max": summary.get("date_max"),
        "source_sheet": summary.get("source_sheet"),
        "source_header_row": summary.get("source_header_row"),
    }
    if sync_success is True:
        public["sync_status"] = "已更新業績分析儀表板"
    elif sync_success is False:
        public["sync_status"] = "業績分析儀表板同步未完成"
        public["sync_error_message"] = humanize_import_error(summary.get("sync_error"))
    else:
        public["sync_status"] = ""

    return {key: value for key, value in public.items() if value not in (None, "")}


def _public_import_job_payload(job: ImportJob) -> Dict[str, Any]:
    payload = job.to_dict()
    display_error = humanize_import_error(payload.get("error_message"))
    payload["error_message"] = display_error
    payload["display_error_message"] = display_error
    payload["import_summary"] = _public_import_summary(payload.get("import_summary"))
    payload.pop("drive_file_id", None)
    payload.pop("local_file_path", None)
    return payload


def _build_in_clause(prefix: str, values) -> tuple:
    """Build a SQLAlchemy-safe IN clause placeholder list and params."""
    params = {}
    placeholders = []
    for idx, value in enumerate(values):
        key = f"{prefix}_{idx}"
        placeholders.append(f":{key}")
        params[key] = value
    return ", ".join(placeholders), params


def _db_dialect_name() -> str:
    try:
        return engine.dialect.name
    except Exception:
        return ""


def _date_filter_expr(column_name: str) -> str:
    """Return a DB-specific date expression for legacy text date columns."""
    if _db_dialect_name() == "postgresql":
        return f"{column_name}::date"
    return f"date({column_name})"


def _table_columns(conn, table_name: str) -> set[str]:
    """Return non-id table columns for PostgreSQL and SQLite-backed tests."""
    if _db_dialect_name() == "postgresql":
        rows = conn.execute(text("""
            SELECT column_name FROM information_schema.columns
            WHERE table_name = :table_name AND column_name != 'id'
            ORDER BY ordinal_position
        """), {"table_name": table_name}).fetchall()
        return {row[0] for row in rows}

    rows = conn.execute(text(f'PRAGMA table_info("{table_name}")')).fetchall()
    return {row[1] for row in rows if row[1] != 'id'}


def _normalise_date_values_for_sql(values):
    """Keep PostgreSQL params as date objects; use ISO strings for SQLite/text comparisons."""
    normalised = []
    use_native_date = _db_dialect_name() == "postgresql"
    for value in values:
        if value is None:
            continue
        if isinstance(value, datetime):
            value = value.date()
        elif hasattr(value, "date") and not isinstance(value, date):
            try:
                value = value.date()
            except Exception:
                pass
        if use_native_date:
            if isinstance(value, date):
                normalised.append(value)
            else:
                parsed = pd.to_datetime(value, errors="coerce")
                if pd.notna(parsed):
                    normalised.append(parsed.date())
        else:
            if isinstance(value, date):
                normalised.append(value.isoformat())
            else:
                parsed = pd.to_datetime(value, errors="coerce")
                normalised.append(parsed.strftime("%Y-%m-%d") if pd.notna(parsed) else str(value))
    return normalised


def _normalise_column_text(value) -> str:
    if pd.isna(value):
        return ""
    return str(value).strip().replace("\n", "").replace("\r", "").replace(" ", "").lower()


def _stringify_column(value) -> str:
    if pd.isna(value):
        return ""
    return str(value).strip()


def _dedupe_columns(columns) -> list[str]:
    seen = {}
    result = []
    for idx, column in enumerate(columns):
        base = _stringify_column(column) or f"unnamed_{idx + 1}"
        count = seen.get(base, 0) + 1
        seen[base] = count
        result.append(base if count == 1 else f"{base}_{count}")
    return result


def _columns_have_any(columns, keywords) -> bool:
    normalised_columns = [_normalise_column_text(column) for column in columns]
    normalised_keywords = [_normalise_column_text(keyword) for keyword in keywords]
    return any(
        keyword and column and keyword in column
        for column in normalised_columns
        for keyword in normalised_keywords
    )


def _missing_daily_sales_column_groups(columns) -> list[str]:
    return [
        label
        for label, keywords in DAILY_SALES_REQUIRED_COLUMN_GROUPS.items()
        if not _columns_have_any(columns, keywords)
    ]


def _find_daily_sales_date_column(columns) -> Optional[str]:
    normalised_aliases = [_normalise_column_text(alias) for alias in DAILY_SALES_DATE_ALIASES]
    for column in columns:
        normalised_column = _normalise_column_text(column)
        if normalised_column in normalised_aliases:
            return column
    for column in columns:
        normalised_column = _normalise_column_text(column)
        if normalised_column and any(alias and alias in normalised_column for alias in normalised_aliases):
            return column
    return None


def _score_daily_sales_candidate(sheet_name: str, header_row: int, columns) -> int:
    missing = _missing_daily_sales_column_groups(columns)
    if missing:
        return 0

    normalised_sheet_name = _normalise_column_text(sheet_name)
    score = 1000
    if _find_daily_sales_date_column(columns):
        score += 250
    if any(_normalise_column_text(hint) in normalised_sheet_name for hint in DAILY_SALES_DETAIL_SHEET_HINTS):
        score += 150
    score += min(len([column for column in columns if _stringify_column(column)]), 80)
    score += max(0, DAILY_SALES_HEADER_SCAN_ROWS - header_row)
    return score


def _clean_daily_sales_dataframe(df: pd.DataFrame) -> pd.DataFrame:
    cleaned = df.dropna(how="all").dropna(axis=1, how="all").copy()
    cleaned.columns = _dedupe_columns(cleaned.columns)
    return cleaned


def _format_daily_sales_diagnostics(candidates: list[dict]) -> str:
    if not candidates:
        return "未找到可辨識的工作表表頭"

    parts = []
    for candidate in sorted(candidates, key=lambda item: item["score"], reverse=True)[:8]:
        columns_sample = [
            column for column in candidate["columns_sample"]
            if column and not str(column).lower().startswith("nan")
        ][:8]
        missing = candidate.get("missing") or []
        parts.append(
            f"{candidate['sheet_name']}!第{candidate['header_row'] + 1}列 "
            f"缺少={missing or '無'} 欄位={columns_sample}"
        )
    return "；".join(parts)


def _read_daily_sales_excel(file_path: str) -> tuple[pd.DataFrame, dict]:
    """
    讀取當日業績 Excel，支援多工作表與非第一列表頭。

    Google Drive 匯出的檔案常把樞紐彙總表放在第一張 sheet，
    真正的匯入明細在「即時業績明細」。此處先掃描所有 sheet 的前幾列，
    再選出同時含商品名稱、業績金額且最好有日期欄位的明細 sheet。
    """
    candidates: list[dict] = []

    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(
                    excel,
                    sheet_name=sheet_name,
                    header=None,
                    nrows=DAILY_SALES_HEADER_SCAN_ROWS,
                    dtype=str,
                )
            except Exception as exc:
                candidates.append({
                    "sheet_name": sheet_name,
                    "header_row": 0,
                    "columns_sample": [f"讀取預覽失敗: {exc}"],
                    "missing": list(DAILY_SALES_REQUIRED_COLUMN_GROUPS.keys()),
                    "score": 0,
                })
                continue

            for header_row, row in preview.iterrows():
                columns = [_stringify_column(value) for value in row.tolist()]
                if not any(columns):
                    continue
                missing = _missing_daily_sales_column_groups(columns)
                score = _score_daily_sales_candidate(sheet_name, int(header_row), columns)
                candidates.append({
                    "sheet_name": sheet_name,
                    "header_row": int(header_row),
                    "columns_sample": columns[:30],
                    "missing": missing,
                    "score": score,
                })

        valid_candidates = [candidate for candidate in candidates if not candidate["missing"]]
        if not valid_candidates:
            diagnostics = _format_daily_sales_diagnostics(candidates)
            raise ValueError(
                "Excel 欄位防禦失敗：找不到可匯入的當日業績明細工作表。"
                f"已檢查：{diagnostics}"
            )

        best = max(valid_candidates, key=lambda candidate: candidate["score"])
        df = pd.read_excel(
            excel,
            sheet_name=best["sheet_name"],
            header=best["header_row"],
            dtype=str,
        )

    df = _clean_daily_sales_dataframe(df)
    if df.empty:
        raise ValueError("Excel 檔案為空")

    missing = _missing_daily_sales_column_groups(df.columns)
    if missing:
        raise ValueError(
            f"Excel 欄位防禦失敗：缺少必要欄位分類 {missing}。"
            f"現有欄位：{list(df.columns)[:30]}"
        )

    metadata = {
        "sheet_name": best["sheet_name"],
        "header_row": best["header_row"] + 1,
        "date_col": _find_daily_sales_date_column(df.columns),
        "candidate_score": best["score"],
    }
    return df, metadata


def _daily_sales_staleness_guard_enabled() -> bool:
    return os.getenv("DAILY_SALES_IMPORT_STALENESS_GUARD_ENABLED", "true").lower() in {
        "1",
        "true",
        "yes",
        "on",
    }


def _daily_sales_max_stale_days() -> int:
    try:
        return max(0, int(os.getenv("DAILY_SALES_IMPORT_MAX_STALE_DAYS", "3")))
    except ValueError:
        return 3


def _should_quarantine_failed_import(error_message: str) -> bool:
    if not error_message:
        return False

    permanent_error_markers = [
        "Excel 欄位防禦失敗",
        "Excel 日期防禦失敗",
        "Excel 檔案為空",
        "找不到可匯入",
        "欄位驗證失敗",
        "日期驗證失敗",
    ]
    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):
    """建立帶有連線池配置的資料庫引擎"""
    if db_path.startswith('postgresql://'):
        return create_engine(
            db_path,
            pool_pre_ping=True,
            pool_size=5,
            max_overflow=10,
            pool_recycle=1800,
            pool_timeout=30,
            connect_args={
                'connect_timeout': 10,
                'options': '-c statement_timeout=120000'  # 匯入需要更長的超時
            }
        )
    elif db_path.startswith('sqlite://'):
        return create_engine(db_path)
    else:
        return create_engine(f'sqlite:///{db_path}')

try:
    from config import DATABASE_PATH as CONFIG_DATABASE_PATH
    engine = _create_engine_with_pool(CONFIG_DATABASE_PATH)
    logger.info(f"使用資料庫: {CONFIG_DATABASE_PATH.split('@')[-1] if '@' in CONFIG_DATABASE_PATH else CONFIG_DATABASE_PATH}")
except ImportError:
    # 備援方案：使用環境變數或預設值
    DATABASE_PATH = os.getenv('DATABASE_PATH', 'data/momo_database.db')
    engine = _create_engine_with_pool(DATABASE_PATH)
    logger.warning(f"無法匯入 config，使用備援資料庫路徑: {DATABASE_PATH}")

Session = sessionmaker(bind=engine)


class ImportService:
    """匯入服務類別"""

    def __init__(self):
        """初始化匯入服務"""
        self._init_database()

    def _init_database(self):
        """初始化資料庫表"""
        try:
            ensure_metadata_initialized(engine, use_postgres_lock=str(engine.url).startswith('postgresql'))
            logger.info("匯入追蹤表已初始化")
        except Exception as e:
            logger.error(f"初始化資料庫表失敗: {str(e)}")

    def get_config(self, key: str, default: str = None) -> Optional[str]:
        """
        取得配置值

        Args:
            key: 配置鍵
            default: 預設值

        Returns:
            Optional[str]: 配置值
        """
        session = Session()
        try:
            config = session.query(ImportConfig).filter_by(config_key=key).first()
            if config:
                return config.config_value
            return default
        finally:
            session.close()

    def set_config(self, key: str, value: str, config_type: str = 'string', description: str = None):
        """
        設定配置值

        Args:
            key: 配置鍵
            value: 配置值
            config_type: 配置類型
            description: 配置說明
        """
        session = Session()
        try:
            config = session.query(ImportConfig).filter_by(config_key=key).first()
            if config:
                config.config_value = value
                config.config_type = config_type
                if description:
                    config.description = description
                config.updated_at = datetime.now(TAIPEI_TZ).replace(tzinfo=None)
            else:
                config = ImportConfig(
                    config_key=key,
                    config_value=value,
                    config_type=config_type,
                    description=description
                )
                session.add(config)

            session.commit()
            logger.info(f"配置已更新: {key} = {value}")

        except Exception as e:
            session.rollback()
            logger.error(f"設定配置失敗: {str(e)}")
        finally:
            session.close()

    def create_import_job(self, job_type: str, drive_file_id: str, drive_file_name: str,
                          drive_file_size: int = None) -> Optional[int]:
        """
        建立匯入任務

        Args:
            job_type: 任務類型（daily_sales 或 vendor_stockout）
            drive_file_id: Google Drive ID or governed source fingerprint
            drive_file_name: Source file name
            drive_file_size: 檔案大小

        Returns:
            Optional[int]: 任務 ID
        """
        session = Session()
        try:
            job = ImportJob(
                job_type=job_type,
                status='pending',
                drive_file_id=drive_file_id,
                drive_file_name=drive_file_name,
                drive_file_size=drive_file_size,
                progress_percent=0.0,
                current_step='等待開始...'
            )

            session.add(job)
            session.commit()

            job_id = job.id
            logger.info(f"已建立匯入任務: ID={job_id}, 檔案={drive_file_name}")
            return job_id

        except Exception as e:
            session.rollback()
            logger.error(f"建立匯入任務失敗: {str(e)}")
            return None
        finally:
            session.close()

    def update_job_status(self, job_id: int, status: str, progress: float = None,
                          current_step: str = None, error_message: str = None):
        """
        更新任務狀態

        Args:
            job_id: 任務 ID
            status: 狀態
            progress: 進度百分比
            current_step: 當前步驟
            error_message: 錯誤訊息
        """
        session = Session()
        try:
            job = session.query(ImportJob).filter_by(id=job_id).first()
            if not job:
                logger.warning(f"找不到任務: ID={job_id}")
                return

            job.status = status

            if progress is not None:
                job.progress_percent = progress

            if current_step:
                job.current_step = current_step

            if error_message:
                job.error_message = error_message

            # 更新時間戳 (2026-01-30 修正：使用台北時區)
            if status == 'downloading' or status == 'importing':
                if not job.started_at:
                    job.started_at = datetime.now(TAIPEI_TZ).replace(tzinfo=None)
            elif status in ['completed', 'failed']:
                job.completed_at = datetime.now(TAIPEI_TZ).replace(tzinfo=None)

            session.commit()
            logger.info(f"任務 {job_id} 狀態已更新: {status} ({progress}%)")

        except Exception as e:
            session.rollback()
            logger.error(f"更新任務狀態失敗: {str(e)}")
        finally:
            session.close()

    def update_job_progress(self, job_id: int, total_rows: int = None, processed_rows: int = None,
                           success_rows: int = None, error_rows: int = None):
        """
        更新任務進度

        Args:
            job_id: 任務 ID
            total_rows: 總行數
            processed_rows: 已處理行數
            success_rows: 成功行數
            error_rows: 錯誤行數
        """
        session = Session()
        try:
            job = session.query(ImportJob).filter_by(id=job_id).first()
            if not job:
                return

            if total_rows is not None:
                job.total_rows = total_rows

            if processed_rows is not None:
                job.processed_rows = processed_rows

            if success_rows is not None:
                job.success_rows = success_rows

            if error_rows is not None:
                job.error_rows = error_rows

            # 計算進度百分比
            if job.total_rows and job.processed_rows:
                job.progress_percent = (job.processed_rows / job.total_rows) * 100

            session.commit()

        except Exception as e:
            session.rollback()
            logger.error(f"更新任務進度失敗: {str(e)}")
        finally:
            session.close()

    def get_job_status(self, job_id: int) -> Optional[Dict[str, Any]]:
        """
        取得任務狀態

        Args:
            job_id: 任務 ID

        Returns:
            Optional[Dict]: 任務資訊
        """
        session = Session()
        try:
            job = session.query(ImportJob).filter_by(id=job_id).first()
            if job:
                return _public_import_job_payload(job)
            return None
        finally:
            session.close()

    def get_recent_jobs(self, limit: int = 20) -> list:
        """
        取得最近的任務清單

        Args:
            limit: 返回數量

        Returns:
            list: 任務清單
        """
        session = Session()
        try:
            jobs = session.query(ImportJob).order_by(
                ImportJob.created_at.desc()
            ).limit(limit).all()

            return [_public_import_job_payload(job) for job in jobs]
        finally:
            session.close()

    def reset_stuck_jobs(self, older_than_minutes: int = 60) -> int:
        """Mark long-running import jobs as failed so operators can retry cleanly."""
        cutoff = datetime.now(TAIPEI_TZ).replace(tzinfo=None) - timedelta(minutes=older_than_minutes)
        session = Session()
        try:
            jobs = session.query(ImportJob).filter(
                ImportJob.status.in_(["pending", "downloading", "importing"]),
                (
                    (ImportJob.started_at.isnot(None) & (ImportJob.started_at < cutoff))
                    | (ImportJob.started_at.is_(None) & (ImportJob.created_at < cutoff))
                )
            ).all()
            for job in jobs:
                job.status = "failed"
                job.progress_percent = 100
                job.current_step = "已重置，請重新匯入"
                job.error_message = "任務逾時未完成，已重置；請重新執行匯入。"
                job.completed_at = datetime.now(TAIPEI_TZ).replace(tzinfo=None)
            session.commit()
            return len(jobs)
        except Exception:
            session.rollback()
            logger.error("重置卡住匯入任務失敗", exc_info=True)
            raise
        finally:
            session.close()

    def fail_job(self, job_id: int) -> bool:
        """Cancel a running import job from the operator UI."""
        session = Session()
        try:
            job = session.query(ImportJob).filter_by(id=job_id).first()
            if not job:
                return False
            job.status = "failed"
            job.progress_percent = max(float(job.progress_percent or 0), 100)
            job.current_step = "已取消"
            job.error_message = "任務已由操作員取消；如需更新資料請重新匯入。"
            job.completed_at = datetime.now(TAIPEI_TZ).replace(tzinfo=None)
            session.commit()
            return True
        except Exception:
            session.rollback()
            logger.error("取消匯入任務失敗: %s", job_id, exc_info=True)
            raise
        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:
        """
        處理當日業績匯入

        Args:
            job_id: 任務 ID
            file_path: Excel 檔案路徑

        Returns:
            bool: 是否成功
        """
        try:
            self.update_job_status(job_id, 'importing', 50, '正在匯入資料...')

            # 讀取 Excel 檔案
            logger.info(f"開始讀取 Excel 檔案: {file_path}")
            try:
                df, excel_metadata = _read_daily_sales_excel(file_path)
            except ValueError as exc:
                error_msg = str(exc)
                step = '欄位驗證失敗' if '欄位' in error_msg else '匯入失敗'
                logger.error(error_msg)
                self.update_job_status(job_id, 'failed', 50, step, error_msg)
                return False

            logger.info(
                "任務 %s 選用 Excel 工作表: %s | 表頭列=%s | 日期欄=%s | 欄位數=%s | 資料列=%s",
                job_id,
                excel_metadata.get("sheet_name"),
                excel_metadata.get("header_row"),
                excel_metadata.get("date_col") or "未提供",
                len(df.columns),
                len(df),
            )

            # 匯入到資料庫
            table_name = 'daily_sales_snapshot'

            # 找到日期欄位
            date_col = excel_metadata.get("date_col") or _find_daily_sales_date_column(df.columns)

            if date_col:
                # 解析日期
                parsed_dates = pd.to_datetime(df[date_col], errors='coerce')
                if parsed_dates.dropna().empty:
                    error_msg = f"Excel 日期防禦失敗：日期欄位「{date_col}」無法解析出有效日期"
                    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}")

                if _daily_sales_staleness_guard_enabled():
                    max_snapshot_date = df['snapshot_date'].dropna().max()
                    today = datetime.now(TAIPEI_TZ).date()
                    lag_days = (today - max_snapshot_date).days
                    max_stale_days = _daily_sales_max_stale_days()
                    if lag_days > max_stale_days:
                        error_msg = (
                            f"Excel 日期防禦失敗：明細最大日期 {max_snapshot_date}，"
                            f"已落後 {lag_days} 天（上限 {max_stale_days} 天）。"
                            "請上傳最新當日業績檔，或將舊檔移至已匯入/匯入失敗資料夾。"
                        )
                        logger.error(error_msg)
                        self.update_job_status(job_id, 'failed', 50, '日期驗證失敗', error_msg)
                        return False
            else:
                # 使用當前日期
                fallback_date = datetime.now(TAIPEI_TZ).date()
                df['snapshot_date'] = fallback_date
                if '日期' not in df.columns:
                    df['日期'] = fallback_date.isoformat()
                logger.info("未找到日期欄位，使用當前日期（台北時區）")

            total_rows = len(df)
            self.update_job_progress(job_id, total_rows=total_rows, processed_rows=0)
            self.update_job_status(job_id, 'importing', 60, '交易式寫入與雙表驗證...')
            max_retries = 2
            write_receipt = None
            sync_error_msg = None
            for attempt in range(max_retries + 1):
                try:
                    if attempt:
                        self.update_job_status(
                            job_id,
                            'importing',
                            60,
                            f'整批交易重試中 ({attempt}/{max_retries})...'
                        )
                    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,
                    )

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

                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
                    NotificationManager()._send_telegram_messages([
                        f"業績匯入失敗（任務 {job_id}）：雙表交易已回滾，正式資料未變更。"
                    ])
                except Exception as notify_error:
                    logger.error("任務 %s 發送回滾告警失敗: %s", job_id, notify_error)
                return False

            self.update_job_status(job_id, 'importing', 90, '雙表交易驗證完成')


            # 計算日期範圍
            date_min = None
            date_max = None
            valid_dates = df['snapshot_date'].dropna().unique()
            if len(valid_dates) > 0:
                sorted_dates = sorted([d for d in valid_dates if d is not None])
                if sorted_dates:
                    date_min = str(sorted_dates[0])
                    date_max = str(sorted_dates[-1])
                    logger.info(f"任務 {job_id} 日期範圍: {date_min} ~ {date_max}")

            # 更新匯入摘要 (2026-01-30 修正：加入同步狀態)
            if sync_success:
                sync_message = f'成功匯入 {total_rows} 筆資料，已同步至業績分析儀表板'
            else:
                sync_message = f'成功匯入 {total_rows} 筆資料，但同步至業績分析儀表板失敗: {sync_error_msg}'

            summary = {
                'imported_count': total_rows,
                'table_name': table_name,
                'synced_to': 'realtime_sales_monthly' if sync_success else None,
                'sync_success': sync_success,
                'sync_error': sync_error_msg,
                'verified': True,  # daily_sales_snapshot 驗證
                'date_min': date_min,
                'date_max': date_max,
                'source_sheet': excel_metadata.get("sheet_name"),
                'source_header_row': excel_metadata.get("header_row"),
                'message': sync_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()

            if not sync_success:
                error_msg = sync_error_msg or '同步至業績分析儀表板失敗'
                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,
                    '業績分析儀表板同步失敗',
                    error_msg,
                )
                logger.error(
                    "任務 %s 匯入未完成：daily_sales_snapshot 已寫入，但 %s 同步失敗；"
                    "保留來源檔案等待重試，不移動 Google Drive 檔案",
                    job_id,
                    monthly_table,
                )
                return False

            # 更新成功資訊
            self.update_job_progress(
                job_id,
                processed_rows=total_rows,
                success_rows=total_rows
            )

            self.update_job_status(
                job_id,
                'completed',
                100,
                '匯入完成（已同步至業績分析儀表板）'
            )

            logger.info(f"任務 {job_id} 匯入成功: {total_rows} 筆")
            try:
                from services.cache_service import clear_growth_cache
                clear_growth_cache()
            except Exception as cache_error:
                logger.warning(f"任務 {job_id} 成長分析快取清除失敗: {cache_error}")
            # daily_sales cache 失效仍靠 _get_data_fingerprint（DB max(snapshot_date)+count(*)）；
            # growth cache 另有 shared file + source fingerprint，匯入後主動清掉避免短 TTL 內看到舊圖。
            return True

        except Exception as e:
            error_msg = f"匯入過程發生異常: {str(e)}"
            self.update_job_status(job_id, 'failed', 50, '匯入失敗', error_msg)
            logger.error(f"任務 {job_id} 匯入異常: {str(e)}")
            import traceback
            logger.error(traceback.format_exc())
            return False

    def auto_import_from_drive(self, *, send_stale_alert: bool = True) -> Dict[str, Any]:
        """
        從 Google Drive 自動匯入檔案

        Returns:
            Dict: 執行結果
        """
        try:
            # 取得配置
            folder_path = self.get_config('gdrive_folder_path', '業績報表/當日業績')
            file_pattern = self.get_config('gdrive_file_pattern', '即時業績_當日')

            logger.info(f"開始檢查 Google Drive: {folder_path}")

            # 列出檔案
            files = drive_service.list_files_in_folder(folder_path, file_pattern)
            drive_error_kind = getattr(drive_service, "last_error_kind", None)
            drive_error = getattr(drive_service, "last_error", None)

            if drive_error_kind:
                public_error = humanize_import_error(drive_error or drive_error_kind)
                message = (
                    "Google Drive 連線或認證失敗，未能確認來源資料夾是否有新檔案。"
                    f"{public_error}"
                )
                logger.error(message)
                return {
                    'success': False,
                    'message': message,
                    'file_count': 0,
                    'imported_count': 0,
                    'failed_count': 1,
                    'connection_error': True,
                    'error_kind': drive_error_kind,
                }

            if not files:
                logger.info("沒有找到待匯入的檔案")
                freshness = evaluate_pchome_sales_freshness(
                    None,
                    now=datetime.now(TAIPEI_TZ),
                )

                # Staleness gate (critic-approved 2026-05-03)
                # 'move-then-success' 反模式：成功 import 後 move_file 把 Excel 搬到
                # 「已匯入」資料夾 → 後續排程 list 回空 → 走此分支 silent return success
                # → 4/27~5/2 daily_sales_snapshot 停更 8 天無告警。補主動偵測：
                # Drive 空時依台北時間到檔 SLA 判斷，避免午夜假性過期，也避免
                # cutoff 後仍把缺檔當成成功。
                try:
                    from sqlalchemy import text
                    from services.openclaw_strategist_service import _send_data_stale_alert

                    _stale_session = Session()
                    try:
                        last_date = _stale_session.execute(
                            text("SELECT MAX(snapshot_date) FROM daily_sales_snapshot")
                        ).scalar()
                    finally:
                        _stale_session.close()

                    if last_date:
                        freshness = evaluate_pchome_sales_freshness(
                            last_date,
                            now=datetime.now(TAIPEI_TZ),
                        )
                        if not freshness["decision_ready"] and send_stale_alert:
                            _send_data_stale_alert(
                                report_type="upstream_drive",
                                last_date=str(freshness["latest_sales_date"]),
                                period=f"超過到檔 SLA {freshness['sla_lag_days']} 天",
                            )
                except Exception:
                    logger.error(
                        "staleness check failed in auto_import_from_drive",
                        exc_info=True,
                    )

                return {
                    'success': True,
                    'status': (
                        'upstream_missing'
                        if freshness.get('latest_sales_date') is None
                        else 'upstream_stale'
                        if not freshness['decision_ready']
                        else 'no_pending_file'
                    ),
                    'message': (
                        f"上游沒有新檔，最新業績超過到檔 SLA {freshness['sla_lag_days']} 天"
                        if freshness.get('latest_sales_date') is not None
                        and not freshness['decision_ready']
                        else '沒有找到待匯入的檔案'
                    ),
                    'file_count': 0,
                    'imported_count': 0,
                    'latest_sales_date': freshness.get('latest_sales_date'),
                    'data_lag_days': freshness.get('data_lag_days'),
                    'sla_lag_days': freshness.get('sla_lag_days'),
                    'freshness_status': freshness.get('freshness_status'),
                    'expected_latest_sales_date': freshness.get('expected_latest_sales_date'),
                    'grace_period_active': freshness.get('grace_period_active'),
                    'report_cutoff_hour': freshness.get('report_cutoff_hour'),
                    'report_due_at': freshness.get('report_due_at'),
                    'freshness_contract': freshness.get('contract_version'),
                    'decision_ready': freshness.get('decision_ready'),
                    'requires_upstream_acquisition': freshness.get('requires_upstream_acquisition'),
                }

            # 處理每個檔案
            imported_count = 0
            total_rows = 0
            all_dates = []  # 收集所有匯入的日期
            failed_files = []
            archive_failed_files = []

            for file in files:
                file_id = file['id']
                file_name = file['name']
                file_size = file.get('size', 0)

                logger.info(f"發現檔案: {file_name}")

                # 建立匯入任務
                job_id = self.create_import_job('daily_sales', file_id, file_name, file_size)
                if not job_id:
                    continue

                # 下載檔案
                self.update_job_status(job_id, 'downloading', 10, '正在下載檔案...')

                temp_dir = 'data/temp'
                os.makedirs(temp_dir, exist_ok=True)
                local_path = os.path.join(temp_dir, file_name)

                if not drive_service.download_file(file_id, local_path):
                    err = '無法從 Google Drive 下載檔案'
                    self.update_job_status(job_id, 'failed', 10, '下載失敗', err)
                    failed_files.append({'file': file_name, 'error': err})
                    continue

                # 更新本地路徑
                session = Session()
                try:
                    job = session.query(ImportJob).filter_by(id=job_id).first()
                    if job:
                        job.local_file_path = local_path
                        session.commit()
                finally:
                    session.close()

                self.update_job_status(job_id, 'downloading', 40, '下載完成')

                # 匯入資料
                if self.process_daily_sales_import(job_id, local_path):
                    # 移動 Google Drive 檔案到「已匯入」資料夾
                    self.update_job_status(job_id, 'completed', 90, '正在移動雲端檔案...')

                    # 取得「已匯入」資料夾路徑配置
                    archive_folder = self.get_config('gdrive_archive_folder', '已匯入')

                    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}")
                        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 取得匯入筆數和日期範圍
                    session = Session()
                    try:
                        job = session.query(ImportJob).filter_by(id=job_id).first()
                        if job and job.import_summary:
                            summary = json.loads(job.import_summary)
                            total_rows += summary.get('imported_count', 0)
                            if summary.get('date_min'):
                                all_dates.append(summary['date_min'])
                            if summary.get('date_max'):
                                all_dates.append(summary['date_max'])
                    finally:
                        session.close()

                    # 清理本地檔案
                    try:
                        os.remove(local_path)
                        logger.info(f"已清理本地檔案: {local_path}")
                    except Exception as e:
                        logger.warning(f"清理本地檔案失敗: {str(e)}")
                else:
                    session = Session()
                    try:
                        job = session.query(ImportJob).filter_by(id=job_id).first()
                        error_message = (job.error_message if job else None) or '匯入失敗，請查看 import_jobs'
                        quarantined = False
                        failed_folder = None
                        if _should_quarantine_failed_import(error_message):
                            default_failed_folder = (
                                f"{folder_path.rstrip('/')}/匯入失敗"
                                if folder_path and folder_path.strip("/")
                                else "匯入失敗"
                            )
                            failed_folder = self.get_config('gdrive_failed_folder', default_failed_folder)
                            if drive_service.move_file(file_id, failed_folder, create_missing=True):
                                quarantined = True
                                error_message = (
                                    f"{error_message}（已移至「{failed_folder}」避免重複告警）"
                                )
                                self.update_job_status(job_id, 'failed', 100, '已移至匯入失敗', error_message)
                                logger.warning(
                                    "已隔離不可自動修復的匯入失敗檔案: %s → %s",
                                    file_name,
                                    failed_folder,
                                )
                            else:
                                logger.error("匯入失敗檔案隔離失敗: %s → %s", file_name, failed_folder)

                        failed_files.append({
                            'file': file_name,
                            'job_id': job_id,
                            'error': error_message,
                            'quarantined': quarantined,
                            'failed_folder': failed_folder,
                        })
                    finally:
                        session.close()

                    try:
                        os.remove(local_path)
                        logger.info(f"已清理失敗匯入本地檔案: {local_path}")
                    except Exception as e:
                        logger.warning(f"清理失敗匯入本地檔案失敗: {str(e)}")

            # 計算日期範圍
            date_range = None
            if all_dates:
                sorted_dates = sorted(set(all_dates))
                if sorted_dates:
                    date_range = {
                        'min': sorted_dates[0],
                        'max': sorted_dates[-1]
                    }

            if failed_files:
                first_error = failed_files[0].get('error') or '未知錯誤'
                public_error = humanize_import_error(first_error) or "請重新匯入最新檔案。"
                message = (
                    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 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
            }

        except Exception as e:
            error_msg = str(e)
            logger.error(f"自動匯入失敗: {error_msg}")

            # 區分連線錯誤和真正的匯入錯誤
            # 連線錯誤（如 Broken pipe、網路問題）不應發送告警
            connection_errors = [
                'Broken pipe',
                'Connection refused',
                'Connection reset',
                'Connection timed out',
                'Name or service not known',
                'No route to host',
                'Network is unreachable',
                'SSL',
                'authenticate',
                'credentials',
                'token'
            ]

            is_connection_error = any(err.lower() in error_msg.lower() for err in connection_errors)

            if is_connection_error:
                # Drive 連線 / 認證錯誤不是「無檔案」，必須 fail-closed 才能觸發告警與人工補件。
                logger.error(f"Google Drive 連線問題，無法確認待匯入檔案: {error_msg}")
                public_error = humanize_import_error(error_msg)
                return {
                    'success': False,
                    'message': f'Google Drive 連線問題，無法確認待匯入檔案。{public_error}',
                    'file_count': 0,
                    'imported_count': 0,
                    'failed_count': 1,
                    'connection_error': True
                }
            else:
                # 真正的匯入錯誤：返回失敗
                return {
                    'success': False,
                    'message': f'自動匯入失敗：{humanize_import_error(error_msg)}',
                    'file_count': 0,
                    'imported_count': 0
                }


# 建立全域服務實例
import_service = ImportService()
