接受

etl_billiards/build_exe.py

@@ -0,0 +1,181 @@
+# -*- coding: utf-8 -*-
+"""
+ETL GUI 打包脚本
+
+使用 PyInstaller 将 GUI 应用打包为 Windows EXE
+
+用法:
+    python build_exe.py [--onefile] [--console] [--clean]
+
+参数:
+    --onefile   打包为单个 EXE 文件（默认为目录模式）
+    --console   显示控制台窗口（调试用）
+    --clean     打包前清理旧的构建文件
+"""
+
+import os
+import sys
+import shutil
+import subprocess
+from pathlib import Path
+
+
+def get_project_root() -> Path:
+    """获取项目根目录"""
+    return Path(__file__).resolve().parent
+
+
+def clean_build():
+    """清理旧的构建文件"""
+    project_root = get_project_root()
+    
+    dirs_to_clean = [
+        project_root / "build",
+        project_root / "dist",
+    ]
+    
+    files_to_clean = [
+        project_root / "etl_gui.spec",
+    ]
+    
+    for d in dirs_to_clean:
+        if d.exists():
+            print(f"清理目录: {d}")
+            shutil.rmtree(d)
+    
+    for f in files_to_clean:
+        if f.exists():
+            print(f"清理文件: {f}")
+            f.unlink()
+
+
+def build_exe(onefile: bool = False, console: bool = False):
+    """构建 EXE"""
+    project_root = get_project_root()
+    
+    # 主入口
+    main_script = project_root / "gui" / "main.py"
+    
+    # 资源文件
+    resources_dir = project_root / "gui" / "resources"
+    database_dir = project_root / "database"
+    
+    # 构建 PyInstaller 命令
+    # 使用 ASCII 名称避免 Windows 控制台编码问题
+    cmd = [
+        sys.executable, "-m", "PyInstaller",
+        "--name", "ETL_Manager",
+        "--noconfirm",
+    ]
+    
+    # 单文件或目录模式
+    if onefile:
+        cmd.append("--onefile")
+    else:
+        cmd.append("--onedir")
+    
+    # 窗口模式
+    if not console:
+        cmd.append("--windowed")
+    
+    # 添加数据文件
+    # 样式表
+    if resources_dir.exists():
+        cmd.extend(["--add-data", f"{resources_dir};gui/resources"])
+    
+    # 数据库 SQL 文件
+    if database_dir.exists():
+        for sql_file in database_dir.glob("*.sql"):
+            cmd.extend(["--add-data", f"{sql_file};database"])
+    
+    # 隐式导入
+    hidden_imports = [
+        "PySide6.QtCore",
+        "PySide6.QtGui",
+        "PySide6.QtWidgets",
+        "psycopg2",
+        "psycopg2.extras",
+        "psycopg2.extensions",
+        # GUI 模块
+        "gui.models.task_model",
+        "gui.models.schedule_model",
+        "gui.utils.cli_builder",
+        "gui.utils.config_helper",
+        "gui.utils.app_settings",
+        "gui.workers.task_worker",
+        "gui.workers.db_worker",
+        "gui.widgets.settings_dialog",
+    ]
+    for imp in hidden_imports:
+        cmd.extend(["--hidden-import", imp])
+    
+    # 排除不需要的模块（减小体积）
+    excludes = [
+        "matplotlib",
+        "numpy",
+        "pandas",
+        "scipy",
+        "PIL",
+        "cv2",
+        "tkinter",
+    ]
+    for exc in excludes:
+        cmd.extend(["--exclude-module", exc])
+    
+    # 工作目录
+    cmd.extend(["--workpath", str(project_root / "build")])
+    cmd.extend(["--distpath", str(project_root / "dist")])
+    cmd.extend(["--specpath", str(project_root)])
+    
+    # 主脚本
+    cmd.append(str(main_script))
+    
+    print("执行命令:")
+    print(" ".join(cmd))
+    print()
+    
+    # 执行打包
+    result = subprocess.run(cmd, cwd=str(project_root))
+    
+    if result.returncode == 0:
+        print()
+        print("=" * 50)
+        print("打包成功!")
+        print(f"输出目录: {project_root / 'dist'}")
+        print("=" * 50)
+    else:
+        print()
+        print("打包失败，请检查错误信息")
+        sys.exit(1)
+
+
+def main():
+    """主函数"""
+    import argparse
+    
+    parser = argparse.ArgumentParser(description="ETL GUI 打包工具")
+    parser.add_argument("--onefile", action="store_true", help="打包为单个 EXE")
+    parser.add_argument("--console", action="store_true", help="显示控制台窗口")
+    parser.add_argument("--clean", action="store_true", help="打包前清理")
+    
+    args = parser.parse_args()
+    
+    # 检查 PyInstaller
+    try:
+        import PyInstaller
+        print(f"PyInstaller 版本: {PyInstaller.__version__}")
+    except ImportError:
+        print("错误: 未安装 PyInstaller")
+        print("请运行: pip install pyinstaller")
+        sys.exit(1)
+    
+    # 清理
+    if args.clean:
+        clean_build()
+    
+    # 构建
+    build_exe(onefile=args.onefile, console=args.console)
+
+
+if __name__ == "__main__":
+    main()

etl_billiards/scripts/backfill_missing_data.py

@@ -0,0 +1,685 @@
+# -*- coding: utf-8 -*-
+"""
+补全丢失的 ODS 数据
+
+通过运行数据完整性检查，找出 API 与 ODS 之间的差异，
+然后重新从 API 获取丢失的数据并插入 ODS。
+
+用法:
+    python -m scripts.backfill_missing_data --start 2025-07-01 --end 2026-01-19
+    python -m scripts.backfill_missing_data --from-report reports/ods_gap_check_xxx.json
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import sys
+import time as time_mod
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Set, Tuple
+from zoneinfo import ZoneInfo
+
+from dateutil import parser as dtparser
+from psycopg2.extras import Json, execute_values
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from api.client import APIClient
+from config.settings import AppConfig
+from database.connection import DatabaseConnection
+from models.parsers import TypeParser
+from tasks.ods_tasks import ENABLED_ODS_CODES, ODS_TASK_SPECS, OdsTaskSpec
+from scripts.check_ods_gaps import run_gap_check
+from utils.logging_utils import build_log_path, configure_logging
+
+
+def _reconfigure_stdout_utf8() -> None:
+    if hasattr(sys.stdout, "reconfigure"):
+        try:
+            sys.stdout.reconfigure(encoding="utf-8")
+        except Exception:
+            pass
+
+
+def _parse_dt(value: str, tz: ZoneInfo, *, is_end: bool = False) -> datetime:
+    raw = (value or "").strip()
+    if not raw:
+        raise ValueError("empty datetime")
+    has_time = any(ch in raw for ch in (":", "T"))
+    dt = dtparser.parse(raw)
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=tz)
+    else:
+        dt = dt.astimezone(tz)
+    if not has_time:
+        dt = dt.replace(
+            hour=23 if is_end else 0,
+            minute=59 if is_end else 0,
+            second=59 if is_end else 0,
+            microsecond=0
+        )
+    return dt
+
+
+def _get_spec(code: str) -> Optional[OdsTaskSpec]:
+    """根据任务代码获取 ODS 任务规格"""
+    for spec in ODS_TASK_SPECS:
+        if spec.code == code:
+            return spec
+    return None
+
+
+def _merge_record_layers(record: dict) -> dict:
+    """展开嵌套的 data 层"""
+    merged = record
+    data_part = merged.get("data")
+    while isinstance(data_part, dict):
+        merged = {**data_part, **merged}
+        data_part = data_part.get("data")
+    settle_inner = merged.get("settleList")
+    if isinstance(settle_inner, dict):
+        merged = {**settle_inner, **merged}
+    return merged
+
+
+def _get_value_case_insensitive(record: dict | None, col: str | None):
+    """不区分大小写地获取值"""
+    if record is None or col is None:
+        return None
+    if col in record:
+        return record.get(col)
+    col_lower = col.lower()
+    for k, v in record.items():
+        if isinstance(k, str) and k.lower() == col_lower:
+            return v
+    return None
+
+
+def _normalize_pk_value(value):
+    """规范化 PK 值"""
+    if value is None:
+        return None
+    if isinstance(value, str) and value.isdigit():
+        try:
+            return int(value)
+        except Exception:
+            return value
+    return value
+
+
+def _pk_tuple_from_record(record: dict, pk_cols: List[str]) -> Optional[Tuple]:
+    """从记录中提取 PK 元组"""
+    merged = _merge_record_layers(record)
+    values = []
+    for col in pk_cols:
+        val = _normalize_pk_value(_get_value_case_insensitive(merged, col))
+        if val is None or val == "":
+            return None
+        values.append(val)
+    return tuple(values)
+
+
+def _get_table_pk_columns(conn, table: str) -> List[str]:
+    """获取表的主键列"""
+    if "." in table:
+        schema, name = table.split(".", 1)
+    else:
+        schema, name = "public", table
+    sql = """
+        SELECT kcu.column_name
+        FROM information_schema.table_constraints tc
+        JOIN information_schema.key_column_usage kcu
+          ON tc.constraint_name = kcu.constraint_name
+         AND tc.table_schema = kcu.table_schema
+        WHERE tc.constraint_type = 'PRIMARY KEY'
+          AND tc.table_schema = %s
+          AND tc.table_name = %s
+        ORDER BY kcu.ordinal_position
+    """
+    with conn.cursor() as cur:
+        cur.execute(sql, (schema, name))
+        return [r[0] for r in cur.fetchall()]
+
+
+def _get_table_columns(conn, table: str) -> List[Tuple[str, str, str]]:
+    """获取表的所有列信息"""
+    if "." in table:
+        schema, name = table.split(".", 1)
+    else:
+        schema, name = "public", table
+    sql = """
+        SELECT column_name, data_type, udt_name
+        FROM information_schema.columns
+        WHERE table_schema = %s AND table_name = %s
+        ORDER BY ordinal_position
+    """
+    with conn.cursor() as cur:
+        cur.execute(sql, (schema, name))
+        return [(r[0], (r[1] or "").lower(), (r[2] or "").lower()) for r in cur.fetchall()]
+
+
+def _fetch_existing_pk_set(
+    conn, table: str, pk_cols: List[str], pk_values: List[Tuple], chunk_size: int
+) -> Set[Tuple]:
+    """获取已存在的 PK 集合"""
+    if not pk_values:
+        return set()
+    select_cols = ", ".join(f't."{c}"' for c in pk_cols)
+    value_cols = ", ".join(f'"{c}"' for c in pk_cols)
+    join_cond = " AND ".join(f't."{c}" = v."{c}"' for c in pk_cols)
+    sql = (
+        f"SELECT {select_cols} FROM {table} t "
+        f"JOIN (VALUES %s) AS v({value_cols}) ON {join_cond}"
+    )
+    existing: Set[Tuple] = set()
+    with conn.cursor() as cur:
+        for i in range(0, len(pk_values), chunk_size):
+            chunk = pk_values[i:i + chunk_size]
+            execute_values(cur, sql, chunk, page_size=len(chunk))
+            for row in cur.fetchall():
+                existing.add(tuple(row))
+    return existing
+
+
+def _cast_value(value, data_type: str):
+    """类型转换"""
+    if value is None:
+        return None
+    dt = (data_type or "").lower()
+    if dt in ("integer", "bigint", "smallint"):
+        if isinstance(value, bool):
+            return int(value)
+        try:
+            return int(value)
+        except Exception:
+            return None
+    if dt in ("numeric", "double precision", "real", "decimal"):
+        if isinstance(value, bool):
+            return int(value)
+        try:
+            return float(value)
+        except Exception:
+            return None
+    if dt.startswith("timestamp") or dt in ("date", "time", "interval"):
+        return value if isinstance(value, (str, datetime)) else None
+    return value
+
+
+def _normalize_scalar(value):
+    """规范化标量值"""
+    if value == "" or value == "{}" or value == "[]":
+        return None
+    return value
+
+
+class MissingDataBackfiller:
+    """丢失数据补全器"""
+    
+    def __init__(
+        self,
+        cfg: AppConfig,
+        logger: logging.Logger,
+        dry_run: bool = False,
+    ):
+        self.cfg = cfg
+        self.logger = logger
+        self.dry_run = dry_run
+        self.tz = ZoneInfo(cfg.get("app.timezone", "Asia/Taipei"))
+        self.store_id = int(cfg.get("app.store_id") or 0)
+        
+        # API 客户端
+        self.api = APIClient(
+            base_url=cfg["api"]["base_url"],
+            token=cfg["api"]["token"],
+            timeout=int(cfg["api"].get("timeout_sec") or 20),
+            retry_max=int(cfg["api"].get("retries", {}).get("max_attempts") or 3),
+            headers_extra=cfg["api"].get("headers_extra") or {},
+        )
+        
+        # 数据库连接
+        self.db = DatabaseConnection(dsn=cfg["db"]["dsn"], session=cfg["db"].get("session"))
+        self.db.conn.autocommit = False
+    
+    def close(self):
+        """关闭连接"""
+        if self.db:
+            self.db.close()
+    
+    def backfill_from_gap_check(
+        self,
+        *,
+        start: datetime,
+        end: datetime,
+        task_codes: Optional[str] = None,
+        page_size: int = 200,
+        chunk_size: int = 500,
+    ) -> Dict[str, Any]:
+        """
+        运行 gap check 并补全丢失数据
+        
+        Returns:
+            补全结果统计
+        """
+        self.logger.info("BACKFILL_START start=%s end=%s", start.isoformat(), end.isoformat())
+        
+        # 计算窗口大小
+        total_seconds = max(0, int((end - start).total_seconds()))
+        if total_seconds >= 86400:
+            window_days = max(1, total_seconds // 86400)
+            window_hours = 0
+        else:
+            window_days = 0
+            window_hours = max(1, total_seconds // 3600 or 1)
+        
+        # 运行 gap check
+        self.logger.info("RUNNING_GAP_CHECK...")
+        gap_result = run_gap_check(
+            cfg=self.cfg,
+            start=start,
+            end=end,
+            window_days=window_days,
+            window_hours=window_hours,
+            page_size=page_size,
+            chunk_size=chunk_size,
+            sample_limit=10000,  # 获取所有丢失样本
+            sleep_per_window=0,
+            sleep_per_page=0,
+            task_codes=task_codes or "",
+            from_cutoff=False,
+            cutoff_overlap_hours=24,
+            allow_small_window=True,
+            logger=self.logger,
+        )
+        
+        total_missing = gap_result.get("total_missing", 0)
+        if total_missing == 0:
+            self.logger.info("NO_MISSING_DATA")
+            return {"backfilled": 0, "errors": 0, "details": []}
+        
+        self.logger.info("GAP_CHECK_DONE total_missing=%s", total_missing)
+        
+        # 补全每个任务的丢失数据
+        results = []
+        total_backfilled = 0
+        total_errors = 0
+        
+        for task_result in gap_result.get("results", []):
+            task_code = task_result.get("task_code")
+            missing = task_result.get("missing", 0)
+            missing_samples = task_result.get("missing_samples", [])
+            
+            if missing == 0:
+                continue
+            
+            self.logger.info(
+                "BACKFILL_TASK task=%s missing=%s samples=%s",
+                task_code, missing, len(missing_samples)
+            )
+            
+            try:
+                backfilled = self._backfill_task(
+                    task_code=task_code,
+                    table=task_result.get("table"),
+                    pk_columns=task_result.get("pk_columns", []),
+                    missing_samples=missing_samples,
+                    start=start,
+                    end=end,
+                    page_size=page_size,
+                    chunk_size=chunk_size,
+                )
+                results.append({
+                    "task_code": task_code,
+                    "missing": missing,
+                    "backfilled": backfilled,
+                    "error": None,
+                })
+                total_backfilled += backfilled
+            except Exception as exc:
+                self.logger.exception("BACKFILL_ERROR task=%s", task_code)
+                results.append({
+                    "task_code": task_code,
+                    "missing": missing,
+                    "backfilled": 0,
+                    "error": str(exc),
+                })
+                total_errors += 1
+        
+        self.logger.info(
+            "BACKFILL_DONE total_missing=%s backfilled=%s errors=%s",
+            total_missing, total_backfilled, total_errors
+        )
+        
+        return {
+            "total_missing": total_missing,
+            "backfilled": total_backfilled,
+            "errors": total_errors,
+            "details": results,
+        }
+    
+    def _backfill_task(
+        self,
+        *,
+        task_code: str,
+        table: str,
+        pk_columns: List[str],
+        missing_samples: List[Dict],
+        start: datetime,
+        end: datetime,
+        page_size: int,
+        chunk_size: int,
+    ) -> int:
+        """补全单个任务的丢失数据"""
+        spec = _get_spec(task_code)
+        if not spec:
+            self.logger.warning("SPEC_NOT_FOUND task=%s", task_code)
+            return 0
+        
+        if not pk_columns:
+            pk_columns = _get_table_pk_columns(self.db.conn, table)
+        
+        if not pk_columns:
+            self.logger.warning("NO_PK_COLUMNS task=%s table=%s", task_code, table)
+            return 0
+        
+        # 提取丢失的 PK 值
+        missing_pks: Set[Tuple] = set()
+        for sample in missing_samples:
+            pk_tuple = tuple(sample.get(col) for col in pk_columns)
+            if all(v is not None for v in pk_tuple):
+                missing_pks.add(pk_tuple)
+        
+        if not missing_pks:
+            self.logger.info("NO_MISSING_PKS task=%s", task_code)
+            return 0
+        
+        self.logger.info(
+            "BACKFILL_FETCHING task=%s missing_pks=%s",
+            task_code, len(missing_pks)
+        )
+        
+        # 从 API 获取数据并过滤出丢失的记录
+        params = self._build_params(spec, start, end)
+        
+        backfilled = 0
+        cols_info = _get_table_columns(self.db.conn, table)
+        db_json_cols_lower = {
+            c[0].lower() for c in cols_info
+            if c[1] in ("json", "jsonb") or c[2] in ("json", "jsonb")
+        }
+        col_names = [c[0] for c in cols_info]
+        
+        try:
+            for page_no, records, _, response_payload in self.api.iter_paginated(
+                endpoint=spec.endpoint,
+                params=params,
+                page_size=page_size,
+                data_path=spec.data_path,
+                list_key=spec.list_key,
+            ):
+                # 过滤出丢失的记录
+                records_to_insert = []
+                for rec in records:
+                    if not isinstance(rec, dict):
+                        continue
+                    pk_tuple = _pk_tuple_from_record(rec, pk_columns)
+                    if pk_tuple and pk_tuple in missing_pks:
+                        records_to_insert.append(rec)
+                
+                if not records_to_insert:
+                    continue
+                
+                # 插入丢失的记录
+                if self.dry_run:
+                    backfilled += len(records_to_insert)
+                    self.logger.info(
+                        "DRY_RUN task=%s page=%s would_insert=%s",
+                        task_code, page_no, len(records_to_insert)
+                    )
+                else:
+                    inserted = self._insert_records(
+                        table=table,
+                        records=records_to_insert,
+                        cols_info=cols_info,
+                        pk_columns=pk_columns,
+                        db_json_cols_lower=db_json_cols_lower,
+                    )
+                    backfilled += inserted
+                    self.logger.info(
+                        "INSERTED task=%s page=%s count=%s",
+                        task_code, page_no, inserted
+                    )
+            
+            if not self.dry_run:
+                self.db.conn.commit()
+            
+            self.logger.info("BACKFILL_TASK_DONE task=%s backfilled=%s", task_code, backfilled)
+            return backfilled
+            
+        except Exception:
+            self.db.conn.rollback()
+            raise
+    
+    def _build_params(
+        self,
+        spec: OdsTaskSpec,
+        start: datetime,
+        end: datetime,
+    ) -> Dict:
+        """构建 API 请求参数"""
+        base: Dict[str, Any] = {}
+        if spec.include_site_id:
+            if spec.endpoint == "/TenantGoods/GetGoodsInventoryList":
+                base["siteId"] = [self.store_id]
+            else:
+                base["siteId"] = self.store_id
+        
+        if spec.requires_window and spec.time_fields:
+            start_key, end_key = spec.time_fields
+            base[start_key] = TypeParser.format_timestamp(start, self.tz)
+            base[end_key] = TypeParser.format_timestamp(end, self.tz)
+        
+        # 合并公共参数
+        common = self.cfg.get("api.params", {}) or {}
+        if isinstance(common, dict):
+            merged = {**common, **base}
+        else:
+            merged = base
+        
+        merged.update(spec.extra_params or {})
+        return merged
+    
+    def _insert_records(
+        self,
+        *,
+        table: str,
+        records: List[Dict],
+        cols_info: List[Tuple[str, str, str]],
+        pk_columns: List[str],
+        db_json_cols_lower: Set[str],
+    ) -> int:
+        """插入记录到数据库"""
+        if not records:
+            return 0
+        
+        col_names = [c[0] for c in cols_info]
+        quoted_cols = ", ".join(f'"{c}"' for c in col_names)
+        sql = f"INSERT INTO {table} ({quoted_cols}) VALUES %s"
+        if pk_columns:
+            pk_clause = ", ".join(f'"{c}"' for c in pk_columns)
+            sql += f" ON CONFLICT ({pk_clause}) DO NOTHING"
+        
+        now = datetime.now(self.tz)
+        json_dump = lambda v: json.dumps(v, ensure_ascii=False)
+        
+        params: List[Tuple] = []
+        for rec in records:
+            merged_rec = _merge_record_layers(rec)
+            
+            # 检查 PK
+            if pk_columns:
+                missing_pk = False
+                for pk in pk_columns:
+                    pk_val = _get_value_case_insensitive(merged_rec, pk)
+                    if pk_val is None or pk_val == "":
+                        missing_pk = True
+                        break
+                if missing_pk:
+                    continue
+            
+            row_vals: List[Any] = []
+            for (col_name, data_type, _udt) in cols_info:
+                col_lower = col_name.lower()
+                if col_lower == "payload":
+                    row_vals.append(Json(rec, dumps=json_dump))
+                    continue
+                if col_lower == "source_file":
+                    row_vals.append("backfill")
+                    continue
+                if col_lower == "source_endpoint":
+                    row_vals.append("backfill")
+                    continue
+                if col_lower == "fetched_at":
+                    row_vals.append(now)
+                    continue
+                
+                value = _normalize_scalar(_get_value_case_insensitive(merged_rec, col_name))
+                if col_lower in db_json_cols_lower:
+                    row_vals.append(Json(value, dumps=json_dump) if value is not None else None)
+                    continue
+                
+                row_vals.append(_cast_value(value, data_type))
+            
+            params.append(tuple(row_vals))
+        
+        if not params:
+            return 0
+        
+        inserted = 0
+        with self.db.conn.cursor() as cur:
+            for i in range(0, len(params), 200):
+                chunk = params[i:i + 200]
+                execute_values(cur, sql, chunk, page_size=len(chunk))
+                if cur.rowcount is not None and cur.rowcount > 0:
+                    inserted += int(cur.rowcount)
+        
+        return inserted
+
+
+def run_backfill(
+    *,
+    cfg: AppConfig,
+    start: datetime,
+    end: datetime,
+    task_codes: Optional[str] = None,
+    dry_run: bool = False,
+    page_size: int = 200,
+    chunk_size: int = 500,
+    logger: logging.Logger,
+) -> Dict[str, Any]:
+    """
+    运行数据补全
+    
+    Args:
+        cfg: 应用配置
+        start: 开始时间
+        end: 结束时间
+        task_codes: 指定任务代码（逗号分隔）
+        dry_run: 是否仅预览
+        page_size: API 分页大小
+        chunk_size: 数据库批量大小
+        logger: 日志记录器
+    
+    Returns:
+        补全结果
+    """
+    backfiller = MissingDataBackfiller(cfg, logger, dry_run)
+    try:
+        return backfiller.backfill_from_gap_check(
+            start=start,
+            end=end,
+            task_codes=task_codes,
+            page_size=page_size,
+            chunk_size=chunk_size,
+        )
+    finally:
+        backfiller.close()
+
+
+def main() -> int:
+    _reconfigure_stdout_utf8()
+    
+    ap = argparse.ArgumentParser(description="补全丢失的 ODS 数据")
+    ap.add_argument("--start", default="2025-07-01", help="开始日期 (默认: 2025-07-01)")
+    ap.add_argument("--end", default="", help="结束日期 (默认: 当前时间)")
+    ap.add_argument("--task-codes", default="", help="指定任务代码（逗号分隔，留空=全部）")
+    ap.add_argument("--dry-run", action="store_true", help="仅预览，不实际写入")
+    ap.add_argument("--page-size", type=int, default=200, help="API 分页大小 (默认: 200)")
+    ap.add_argument("--chunk-size", type=int, default=500, help="数据库批量大小 (默认: 500)")
+    ap.add_argument("--log-file", default="", help="日志文件路径")
+    ap.add_argument("--log-dir", default="", help="日志目录")
+    ap.add_argument("--log-level", default="INFO", help="日志级别 (默认: INFO)")
+    ap.add_argument("--no-log-console", action="store_true", help="禁用控制台日志")
+    args = ap.parse_args()
+    
+    log_dir = Path(args.log_dir) if args.log_dir else (PROJECT_ROOT / "logs")
+    log_file = Path(args.log_file) if args.log_file else build_log_path(log_dir, "backfill_missing")
+    log_console = not args.no_log_console
+    
+    with configure_logging(
+        "backfill_missing",
+        log_file,
+        level=args.log_level,
+        console=log_console,
+        tee_std=True,
+    ) as logger:
+        cfg = AppConfig.load({})
+        tz = ZoneInfo(cfg.get("app.timezone", "Asia/Taipei"))
+        
+        start = _parse_dt(args.start, tz)
+        end = _parse_dt(args.end, tz, is_end=True) if args.end else datetime.now(tz)
+        
+        result = run_backfill(
+            cfg=cfg,
+            start=start,
+            end=end,
+            task_codes=args.task_codes or None,
+            dry_run=args.dry_run,
+            page_size=args.page_size,
+            chunk_size=args.chunk_size,
+            logger=logger,
+        )
+        
+        logger.info("=" * 60)
+        logger.info("补全完成!")
+        logger.info("  总丢失: %s", result.get("total_missing", 0))
+        logger.info("  已补全: %s", result.get("backfilled", 0))
+        logger.info("  错误数: %s", result.get("errors", 0))
+        logger.info("=" * 60)
+        
+        # 输出详细结果
+        for detail in result.get("details", []):
+            if detail.get("error"):
+                logger.error(
+                    "  %s: 丢失=%s 补全=%s 错误=%s",
+                    detail.get("task_code"),
+                    detail.get("missing"),
+                    detail.get("backfilled"),
+                    detail.get("error"),
+                )
+            elif detail.get("backfilled", 0) > 0:
+                logger.info(
+                    "  %s: 丢失=%s 补全=%s",
+                    detail.get("task_code"),
+                    detail.get("missing"),
+                    detail.get("backfilled"),
+                )
+    
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())