# -*- coding: utf-8 -*-
"""ODS 层批量校验器

校验逻辑：对比 API 源数据与 ODS 表数据
- 主键 + content_hash 对比
- 批量 UPSERT 补齐缺失/不一致数据
"""

import logging
from datetime import datetime
from typing import Any, Dict, List, Optional, Set, Tuple

from psycopg2.extras import execute_values

from api.local_json_client import LocalJsonClient

from .base_verifier import BaseVerifier, VerificationFetchError


class OdsVerifier(BaseVerifier):
    """ODS 层校验器"""
    
    def __init__(
        self,
        db_connection: Any,
        api_client: Any = None,
        logger: Optional[logging.Logger] = None,
        fetch_from_api: bool = False,
        local_dump_dirs: Optional[Dict[str, str]] = None,
        use_local_json: bool = False,
    ):
        """
        初始化 ODS 校验器
        
        Args:
            db_connection: 数据库连接
            api_client: API 客户端（用于重新获取数据）
            logger: 日志器
            fetch_from_api: 是否从 API 获取源数据进行校验（默认 False，仅校验 ODS 内部一致性）
            local_dump_dirs: 本地 JSON dump 目录映射（task_code -> 目录）
            use_local_json: 是否优先使用本地 JSON 作为源数据
        """
        super().__init__(db_connection, logger)
        self.api_client = api_client
        self.fetch_from_api = fetch_from_api
        self.local_dump_dirs = local_dump_dirs or {}
        self.use_local_json = bool(use_local_json or self.local_dump_dirs)
        
        # 缓存从 API 获取的数据（避免重复调用）
        self._api_data_cache: Dict[str, List[dict]] = {}
        self._api_key_cache: Dict[str, Set[Tuple]] = {}
        self._api_hash_cache: Dict[str, Dict[Tuple, str]] = {}
        self._table_column_cache: Dict[Tuple[str, str], bool] = {}
        self._table_pk_cache: Dict[str, List[str]] = {}
        self._local_json_clients: Dict[str, LocalJsonClient] = {}
        
        # ODS 表配置：{表名: {pk_columns, time_column, api_endpoint}}
        self._table_config = self._load_table_config()
    
    @property
    def layer_name(self) -> str:
        return "ODS"
    
    def _load_table_config(self) -> Dict[str, dict]:
        """加载 ODS 表配置"""
        # 从任务定义中动态获取配置
        try:
            from tasks.ods.ods_tasks import ODS_TASK_SPECS
            config = {}
            for spec in ODS_TASK_SPECS:
                # time_fields 是一个元组 (start_field, end_field)，取第一个作为时间列
                # 或者使用 fetched_at 作为默认
                time_column = "fetched_at"
                
                # 使用 table_name 属性（不是 table）
                table_name = spec.table_name
                # 提取不带 schema 前缀的表名作为 key
                if "." in table_name:
                    table_key = table_name.split(".")[-1]
                else:
                    table_key = table_name
                
                # 从 sources 中提取 ODS 表的实际主键列名
                # sources 格式如 ("settleList.id", "id")，最后一个简单名称是 ODS 列名
                pk_columns = []
                for col in spec.pk_columns:
                    ods_col_name = self._extract_ods_column_name(col)
                    pk_columns.append(ods_col_name)
                
                # 如果 pk_columns 为空，尝试使用 conflict_columns_override 或跳过校验
                # 一些特殊表（如 goods_stock_summary）没有标准主键
                if not pk_columns:
                    # 跳过没有明确主键定义的表
                    self.logger.debug("表 %s 没有定义主键列，跳过校验配置", table_key)
                    continue
                
                config[table_key] = {
                    "full_table_name": table_name,
                    "pk_columns": pk_columns,
                    "time_column": time_column,
                    "api_endpoint": spec.endpoint,
                    "task_code": spec.code,
                }
            return config
        except ImportError:
            self.logger.warning("无法加载 ODS 任务定义，使用默认配置")
            return {}
    
    def _extract_ods_column_name(self, col) -> str:
        """
        从 ColumnSpec 中提取 ODS 表的实际列名
        
        ODS 表使用原始 JSON 字段名（小写），而 col.column 是 DWD 层的命名。
        sources 中的最后一个简单字段名通常就是 ODS 表的列名。
        """
        # 如果 sources 为空，使用 column（假设 column 就是 ODS 列名）
        if not col.sources:
            return col.column
        
        # 遍历 sources，找到最简单的字段名（不含点号的）
        for source in reversed(col.sources):
            if "." not in source:
                return source.lower()  # ODS 列名通常是小写
        
        # 如果都是复杂路径，取最后一个路径的最后一部分
        last_source = col.sources[-1]
        if "." in last_source:
            return last_source.split(".")[-1].lower()
        return last_source.lower()
    
    def get_tables(self) -> List[str]:
        """获取需要校验的 ODS 表列表"""
        if self._table_config:
            return list(self._table_config.keys())
        
        # 从数据库查询 ODS schema 中的表
        sql = """
            SELECT table_name 
            FROM information_schema.tables 
            WHERE table_schema = 'ods'
            AND table_type = 'BASE TABLE'
            ORDER BY table_name
        """
        try:
            with self.db.conn.cursor() as cur:
                cur.execute(sql)
                return [row[0] for row in cur.fetchall()]
        except Exception as e:
            self.logger.warning("获取 ODS 表列表失败: %s", e)
            try:
                self.db.conn.rollback()
            except Exception:
                pass
            return []
    
    def get_primary_keys(self, table: str) -> List[str]:
        """获取表的主键列"""
        if table in self._table_config:
            return self._table_config[table].get("pk_columns", [])
        # 表不在配置中，返回空列表表示无法校验
        return []
    
    def get_time_column(self, table: str) -> Optional[str]:
        """获取表的时间列"""
        if table in self._table_config:
            return self._table_config[table].get("time_column", "fetched_at")
        return "fetched_at"
    
    def _get_full_table_name(self, table: str) -> str:
        """获取完整的表名（包含 schema）"""
        if table in self._table_config:
            return self._table_config[table].get("full_table_name", f"ods.{table}")
        # 如果表名已经包含 schema，直接返回
        if "." in table:
            return table
        return f"ods.{table}"
    
    def fetch_source_keys(
        self,
        table: str,
        window_start: datetime,
        window_end: datetime,
    ) -> Set[Tuple]:
        """
        从源获取主键集合
        
        根据 fetch_from_api 参数决定数据来源：
        - fetch_from_api=True: 从 API 获取数据（真正的源到目标校验）
        - fetch_from_api=False: 从 ODS 表获取（ODS 内部一致性校验）
        """
        if self._has_external_source():
            return self._fetch_keys_from_api(table, window_start, window_end)
        else:
            # ODS 内部校验：直接从 ODS 表获取
            return self._fetch_keys_from_db(table, window_start, window_end)
    
    def _fetch_keys_from_api(
        self,
        table: str,
        window_start: datetime,
        window_end: datetime,
    ) -> Set[Tuple]:
        """从 API 获取源数据主键集合"""
        # 尝试获取缓存的 API 数据
        cache_key = f"{table}_{window_start}_{window_end}"
        if cache_key in self._api_key_cache:
            return self._api_key_cache[cache_key]
        if cache_key not in self._api_data_cache:
            # 调用 API 获取数据
            api_records = self._call_api_for_table(table, window_start, window_end)
            self._api_data_cache[cache_key] = api_records
        
        api_records = self._api_data_cache.get(cache_key, [])
        
        if not api_records:
            self.logger.debug("表 %s 从 API 未获取到数据", table)
            return set()
        
        # 获取主键列
        pk_cols = self.get_primary_keys(table)
        if not pk_cols:
            self.logger.debug("表 %s 没有主键配置，跳过 API 校验", table)
            return set()
        
        # 提取主键
        keys = set()
        for record in api_records:
            pk_values = []
            for col in pk_cols:
                # API 返回的字段名可能是原始格式（如 id, Id, ID）
                # 尝试多种格式
                value = record.get(col)
                if value is None:
                    value = record.get(col.lower())
                if value is None:
                    value = record.get(col.upper())
                pk_values.append(value)
            if all(v is not None for v in pk_values):
                keys.add(tuple(pk_values))
        
        self.logger.info("表 %s 从源数据获取 %d 条记录，%d 个唯一主键", table, len(api_records), len(keys))
        self._api_key_cache[cache_key] = keys
        return keys
    
    def _call_api_for_table(
        self,
        table: str,
        window_start: datetime,
        window_end: datetime,
    ) -> List[dict]:
        """调用源数据获取表对应的数据"""
        config = self._table_config.get(table, {})
        task_code = config.get("task_code")
        endpoint = config.get("api_endpoint")
        
        if not task_code or not endpoint:
            self.logger.warning(
                "表 %s 没有完整的任务配置（task_code=%s, endpoint=%s），无法获取源数据",
                table, task_code, endpoint
            )
            return []

        source_client = self._get_source_client(task_code)
        if not source_client:
            self.logger.warning("表 %s 未找到可用源（API/本地JSON），跳过获取源数据", table)
            return []

        source_label = "本地 JSON" if self._is_using_local_json(task_code) else "API"
        self.logger.info(
            "从 %s 获取数据: 表=%s, 端点=%s, 时间窗口=%s ~ %s",
            source_label, table, endpoint, window_start, window_end
        )
        
        try:
            # 获取 ODS 任务规格以获取正确的参数配置
            from tasks.ods.ods_tasks import ODS_TASK_SPECS
            
            # 查找对应的任务规格
            spec = None
            for s in ODS_TASK_SPECS:
                if s.code == task_code:
                    spec = s
                    break
            
            if not spec:
                self.logger.warning("未找到任务规格: %s", task_code)
                return []
            
            # 构建 API 参数
            params = {}
            if spec.include_site_id:
                # 从 API 客户端获取 store_id（如果可用）
                store_id = getattr(self.api_client, 'store_id', None)
                if store_id:
                    params["siteId"] = store_id
            
            if spec.requires_window and spec.time_fields:
                start_key, end_key = spec.time_fields
                # 格式化时间戳
                params[start_key] = window_start.strftime("%Y-%m-%d %H:%M:%S")
                params[end_key] = window_end.strftime("%Y-%m-%d %H:%M:%S")
            
            # 合并额外参数
            params.update(spec.extra_params)
            
            # 调用源数据
            all_records = []
            for _, page_records, _, _ in source_client.iter_paginated(
                endpoint=spec.endpoint,
                params=params,
                page_size=200,
                data_path=spec.data_path,
                list_key=spec.list_key,
            ):
                all_records.extend(page_records)
            
            self.logger.info("源数据返回 %d 条原始记录", len(all_records))
            return all_records
            
        except Exception as e:
            self.logger.warning("获取源数据失败: 表=%s, error=%s", table, e)
            import traceback
            self.logger.debug("调用栈: %s", traceback.format_exc())
            raise VerificationFetchError(f"获取源数据失败: {table}") from e
    
    def fetch_target_keys(
        self,
        table: str,
        window_start: datetime,
        window_end: datetime,
    ) -> Set[Tuple]:
        """从 ODS 表获取目标数据主键集合"""
        if self._has_external_source():
            cache_key = f"{table}_{window_start}_{window_end}"
            api_keys = self._api_key_cache.get(cache_key)
            if api_keys is None:
                api_keys = self._fetch_keys_from_api(table, window_start, window_end)
            return self._fetch_keys_from_db_by_keys(table, api_keys)
        return self._fetch_keys_from_db(table, window_start, window_end)
    
    def _fetch_keys_from_db(
        self,
        table: str,
        window_start: datetime,
        window_end: datetime,
    ) -> Set[Tuple]:
        """从数据库获取主键集合"""
        pk_cols = self.get_primary_keys(table)
        
        # 如果没有主键列配置，跳过校验
        if not pk_cols:
            self.logger.debug("表 %s 没有主键配置，跳过获取主键", table)
            return set()
        
        time_col = self.get_time_column(table)
        full_table = self._get_full_table_name(table)
        
        pk_select = ", ".join(pk_cols)
        sql = f"""
            SELECT {pk_select}
            FROM {full_table}
            WHERE {time_col} >= %s AND {time_col} < %s
        """
        
        try:
            with self.db.conn.cursor() as cur:
                cur.execute(sql, (window_start, window_end))
                return {tuple(row) for row in cur.fetchall()}
        except Exception as e:
            self.logger.warning("获取 ODS 主键失败: %s, error=%s", table, e)
            try:
                self.db.conn.rollback()
            except Exception:
                pass
            raise VerificationFetchError(f"获取 ODS 主键失败: {table}") from e

    def _fetch_keys_from_db_by_keys(self, table: str, keys: Set[Tuple]) -> Set[Tuple]:
        """按主键集合反查 ODS 表是否存在记录（不依赖时间窗口）"""
        if not keys:
            return set()
        pk_cols = self.get_primary_keys(table)
        if not pk_cols:
            self.logger.debug("表 %s 没有主键配置，跳过按主键反查", table)
            return set()
        full_table = self._get_full_table_name(table)
        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 {full_table} t "
            f"JOIN (VALUES %s) AS v({value_cols}) ON {join_cond}"
        )
        existing: Set[Tuple] = set()
        try:
            with self.db.conn.cursor() as cur:
                for chunk in self._chunked(list(keys), 500):
                    execute_values(cur, sql, chunk, page_size=len(chunk))
                    for row in cur.fetchall():
                        existing.add(tuple(row))
        except Exception as e:
            self.logger.warning("按主键反查 ODS 失败: %s, error=%s", table, e)
            try:
                self.db.conn.rollback()
            except Exception:
                pass
            raise VerificationFetchError(f"按主键反查 ODS 失败: {table}") from e
        return existing
    
    def fetch_source_hashes(
        self,
        table: str,
        window_start: datetime,
        window_end: datetime,
    ) -> Dict[Tuple, str]:
        """获取源数据的主键->content_hash 映射"""
        if self._has_external_source():
            return self._fetch_hashes_from_api(table, window_start, window_end)
        else:
            # ODS 表自带 content_hash 列
            return self._fetch_hashes_from_db(table, window_start, window_end)
    
    def _fetch_hashes_from_api(
        self,
        table: str,
        window_start: datetime,
        window_end: datetime,
    ) -> Dict[Tuple, str]:
        """从 API 数据计算哈希"""
        cache_key = f"{table}_{window_start}_{window_end}"
        if cache_key in self._api_hash_cache:
            return self._api_hash_cache[cache_key]
        api_records = self._api_data_cache.get(cache_key, [])
        
        if not api_records:
            # 尝试从 API 获取
            api_records = self._call_api_for_table(table, window_start, window_end)
            self._api_data_cache[cache_key] = api_records
        
        if not api_records:
            return {}
        
        pk_cols = self.get_primary_keys(table)
        if not pk_cols:
            return {}
        
        result = {}
        for record in api_records:
            # 提取主键
            pk_values = []
            for col in pk_cols:
                value = record.get(col)
                if value is None:
                    value = record.get(col.lower())
                if value is None:
                    value = record.get(col.upper())
                pk_values.append(value)
            
            if all(v is not None for v in pk_values):
                pk = tuple(pk_values)
                # 计算内容哈希
                content_hash = self._compute_hash(record)
                result[pk] = content_hash
        
        self._api_hash_cache[cache_key] = result
        if cache_key not in self._api_key_cache:
            self._api_key_cache[cache_key] = set(result.keys())
        return result
    
    def fetch_target_hashes(
        self,
        table: str,
        window_start: datetime,
        window_end: datetime,
    ) -> Dict[Tuple, str]:
        """获取目标数据的主键->content_hash 映射"""
        if self.fetch_from_api and self.api_client:
            cache_key = f"{table}_{window_start}_{window_end}"
            api_hashes = self._api_hash_cache.get(cache_key)
            if api_hashes is None:
                api_hashes = self._fetch_hashes_from_api(table, window_start, window_end)
            api_keys = set(api_hashes.keys())
            return self._fetch_hashes_from_db_by_keys(table, api_keys)
        return self._fetch_hashes_from_db(table, window_start, window_end)
    
    def _fetch_hashes_from_db(
        self,
        table: str,
        window_start: datetime,
        window_end: datetime,
    ) -> Dict[Tuple, str]:
        """从数据库获取主键->hash 映射"""
        pk_cols = self.get_primary_keys(table)
        
        # 如果没有主键列配置，跳过校验
        if not pk_cols:
            self.logger.debug("表 %s 没有主键配置，跳过获取哈希", table)
            return {}
        
        time_col = self.get_time_column(table)
        full_table = self._get_full_table_name(table)
        
        pk_select = ", ".join(pk_cols)
        sql = f"""
            SELECT {pk_select}, content_hash
            FROM {full_table}
            WHERE {time_col} >= %s AND {time_col} < %s
        """
        
        result = {}
        try:
            with self.db.conn.cursor() as cur:
                cur.execute(sql, (window_start, window_end))
                for row in cur.fetchall():
                    pk = tuple(row[:-1])
                    content_hash = row[-1]
                    result[pk] = content_hash or ""
        except Exception as e:
            # 查询失败时回滚事务，避免影响后续查询
            self.logger.warning("获取 ODS hash 失败: %s, error=%s", table, e)
            try:
                self.db.conn.rollback()
            except Exception:
                pass
            raise VerificationFetchError(f"获取 ODS hash 失败: {table}") from e
        
        return result

    def _fetch_hashes_from_db_by_keys(self, table: str, keys: Set[Tuple]) -> Dict[Tuple, str]:
        """按主键集合反查 ODS 的对比哈希（不依赖时间窗口）"""
        if not keys:
            return {}
        pk_cols = self.get_primary_keys(table)
        if not pk_cols:
            self.logger.debug("表 %s 没有主键配置，跳过按主键反查 hash", table)
            return {}
        full_table = self._get_full_table_name(table)
        has_payload = self._table_has_column(full_table, "payload")
        select_tail = 't."payload"' if has_payload else 't."content_hash"'
        select_cols = ", ".join([*(f't."{c}"' for c in pk_cols), select_tail])
        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 {full_table} t "
            f"JOIN (VALUES %s) AS v({value_cols}) ON {join_cond}"
        )
        result: Dict[Tuple, str] = {}
        try:
            with self.db.conn.cursor() as cur:
                for chunk in self._chunked(list(keys), 500):
                    execute_values(cur, sql, chunk, page_size=len(chunk))
                    for row in cur.fetchall():
                        pk = tuple(row[:-1])
                        tail_value = row[-1]
                        if has_payload:
                            compare_hash = self._compute_compare_hash_from_payload(tail_value)
                            result[pk] = compare_hash or ""
                        else:
                            result[pk] = tail_value or ""
        except Exception as e:
            self.logger.warning("按主键反查 ODS hash 失败: %s, error=%s", table, e)
            try:
                self.db.conn.rollback()
            except Exception:
                pass
            raise VerificationFetchError(f"按主键反查 ODS hash 失败: {table}") from e
        return result

    @staticmethod
    def _chunked(items: List[Tuple], chunk_size: int) -> List[List[Tuple]]:
        """将列表按固定大小分块"""
        if chunk_size <= 0:
            return [items]
        return [items[i:i + chunk_size] for i in range(0, len(items), chunk_size)]
    
    def backfill_missing(
        self,
        table: str,
        missing_keys: Set[Tuple],
        window_start: datetime,
        window_end: datetime,
    ) -> int:
        """
        批量补齐缺失数据
        
        ODS 层补齐需要重新从 API 获取数据
        """
        if not self._has_external_source():
            self.logger.warning("未配置 API/本地JSON 源，无法补齐 ODS 缺失数据")
            return 0
        
        if not missing_keys:
            return 0
        
        # 获取表配置
        config = self._table_config.get(table, {})
        task_code = config.get("task_code")
        
        if not task_code:
            self.logger.warning("未找到表 %s 的任务配置，跳过补齐", table)
            return 0
        
        self.logger.info(
            "ODS 补齐缺失: 表=%s, 数量=%d, 任务=%s",
            table, len(missing_keys), task_code
        )
        
        # ODS 层的补齐实际上是重新执行 ODS 任务从 API 获取数据
        # 但由于 ODS 任务已经在 "校验前先从 API 获取数据" 步骤执行过了，
        # 这里补齐失败是预期的（数据已经在 ODS 表中，只是校验窗口可能不一致）
        # 
        # 实际的 ODS 补齐应该在 verify_only 模式下启用 fetch_before_verify 选项，
        # 这会先执行 ODS 任务获取 API 数据，然后再校验。
        # 
        # 如果仍然有缺失，说明：
        # 1. API 返回的数据时间窗口与校验窗口不完全匹配
        # 2. 或者 ODS 任务的时间参数配置问题
        self.logger.info(
            "ODS 补齐提示: 表=%s 有 %d 条缺失记录，建议使用 '校验前先从 API 获取数据' 选项获取完整数据",
            table, len(missing_keys)
        )
        return 0
    
    def backfill_mismatch(
        self,
        table: str,
        mismatch_keys: Set[Tuple],
        window_start: datetime,
        window_end: datetime,
    ) -> int:
        """
        批量更新不一致数据
        
        ODS 层更新也需要重新从 API 获取
        """
        # 与 backfill_missing 类似，重新获取数据会自动 UPSERT
        return self.backfill_missing(table, mismatch_keys, window_start, window_end)

    def _has_external_source(self) -> bool:
        return bool(self.fetch_from_api and (self.api_client or self.use_local_json))

    def _is_using_local_json(self, task_code: str) -> bool:
        return bool(self.use_local_json and task_code in self.local_dump_dirs)

    def _get_local_json_client(self, task_code: str) -> Optional[LocalJsonClient]:
        if task_code in self._local_json_clients:
            return self._local_json_clients[task_code]
        dump_dir = self.local_dump_dirs.get(task_code)
        if not dump_dir:
            return None
        try:
            client = LocalJsonClient(dump_dir)
        except Exception as exc:  # noqa: BLE001
            self.logger.warning(
                "本地 JSON 目录不可用: task=%s, dir=%s, error=%s",
                task_code, dump_dir, exc,
            )
            return None
        self._local_json_clients[task_code] = client
        return client

    def _get_source_client(self, task_code: str):
        if self.use_local_json:
            return self._get_local_json_client(task_code)
        return self.api_client
    
    def verify_against_api(
        self,
        table: str,
        window_start: datetime,
        window_end: datetime,
        auto_backfill: bool = False,
    ) -> Dict[str, Any]:
        """
        与 API 源数据对比校验
        
        这是更严格的校验，直接调用 API 获取数据进行对比
        """
        if not self.api_client:
            return {"error": "未配置 API 客户端"}
        
        config = self._table_config.get(table, {})
        endpoint = config.get("api_endpoint")
        
        if not endpoint:
            return {"error": f"未找到表 {table} 的 API 端点配置"}
        
        self.logger.info("开始与 API 对比校验: %s", table)
        
        # 1. 从 API 获取数据
        try:
            api_records = self.api_client.fetch_records(
                endpoint=endpoint,
                start_time=window_start,
                end_time=window_end,
            )
        except Exception as e:
            return {"error": f"API 调用失败: {e}"}
        
        # 2. 从 ODS 获取数据
        ods_hashes = self.fetch_target_hashes(table, window_start, window_end)
        
        # 3. 计算 API 数据的 hash
        pk_cols = self.get_primary_keys(table)
        api_hashes = {}
        for record in api_records:
            pk = tuple(record.get(col) for col in pk_cols)
            content_hash = self._compute_hash(record)
            api_hashes[pk] = content_hash
        
        # 4. 对比
        api_keys = set(api_hashes.keys())
        ods_keys = set(ods_hashes.keys())
        
        missing = api_keys - ods_keys
        extra = ods_keys - api_keys
        mismatch = {
            k for k in (api_keys & ods_keys)
            if api_hashes[k] != ods_hashes[k]
        }
        
        result = {
            "table": table,
            "api_count": len(api_hashes),
            "ods_count": len(ods_hashes),
            "missing_count": len(missing),
            "extra_count": len(extra),
            "mismatch_count": len(mismatch),
            "is_consistent": len(missing) == 0 and len(mismatch) == 0,
        }
        
        # 5. 自动补齐
        if auto_backfill and (missing or mismatch):
            # 需要重新获取的主键
            keys_to_refetch = missing | mismatch
            
            # 筛选需要重新插入的记录
            records_to_upsert = [
                r for r in api_records
                if tuple(r.get(col) for col in pk_cols) in keys_to_refetch
            ]
            
            if records_to_upsert:
                backfilled = self._batch_upsert(table, records_to_upsert)
                result["backfilled_count"] = backfilled
        
        return result
    
    def _table_has_column(self, full_table: str, column: str) -> bool:
        """检查表是否包含指定列（带缓存）"""
        cache_key = (full_table, column)
        if cache_key in self._table_column_cache:
            return self._table_column_cache[cache_key]
        schema = "public"
        table = full_table
        if "." in full_table:
            schema, table = full_table.split(".", 1)
        sql = """
            SELECT 1
            FROM information_schema.columns
            WHERE table_schema = %s AND table_name = %s AND column_name = %s
            LIMIT 1
        """
        try:
            with self.db.conn.cursor() as cur:
                cur.execute(sql, (schema, table, column))
                exists = cur.fetchone() is not None
        except Exception:
            exists = False
            try:
                self.db.conn.rollback()
            except Exception:
                pass
        self._table_column_cache[cache_key] = exists
        return exists

    def _get_db_primary_keys(self, full_table: str) -> List[str]:
        """Read primary key columns from database metadata (ordered)."""
        if full_table in self._table_pk_cache:
            return self._table_pk_cache[full_table]

        schema = "public"
        table = full_table
        if "." in full_table:
            schema, table = full_table.split(".", 1)

        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
             AND tc.table_name = kcu.table_name
            WHERE tc.table_schema = %s
              AND tc.table_name = %s
              AND tc.constraint_type = 'PRIMARY KEY'
            ORDER BY kcu.ordinal_position
        """
        try:
            with self.db.conn.cursor() as cur:
                cur.execute(sql, (schema, table))
                rows = cur.fetchall()
                cols = [r[0] if not isinstance(r, dict) else r.get("column_name") for r in rows]
                result = [c for c in cols if c]
        except Exception:
            result = []
            try:
                self.db.conn.rollback()
            except Exception:
                pass

        self._table_pk_cache[full_table] = result
        return result
    
    def _compute_compare_hash_from_payload(self, payload: Any) -> Optional[str]:
        """使用 ODS 任务的算法计算对比哈希"""
        try:
            from tasks.ods.ods_tasks import BaseOdsTask
            return BaseOdsTask._compute_compare_hash_from_payload(payload)
        except Exception:
            return None
    
    def _compute_hash(self, record: dict) -> str:
        """计算记录的对比哈希（与 ODS 入库一致，不包含 fetched_at）"""
        compare_hash = self._compute_compare_hash_from_payload(record)
        return compare_hash or ""
    
    def _batch_upsert(self, table: str, records: List[dict]) -> int:
        """Batch backfill in snapshot-safe mode (insert-only on PK conflict)."""
        if not records:
            return 0
        
        full_table = self._get_full_table_name(table)
        db_pk_cols = self._get_db_primary_keys(full_table)
        if not db_pk_cols:
            self.logger.warning("表 %s 未找到主键，跳过回填", full_table)
            return 0
        has_content_hash_col = self._table_has_column(full_table, "content_hash")
        
        # 获取所有列（从第一条记录），并在存在 content_hash 列时补齐该列。
        all_cols = list(records[0].keys())
        if has_content_hash_col and "content_hash" not in all_cols:
            all_cols.append("content_hash")
        
        # Snapshot-safe strategy: never update historical rows; only insert new snapshots.
        col_list = ", ".join(all_cols)
        placeholders = ", ".join(["%s"] * len(all_cols))
        pk_list = ", ".join(db_pk_cols)
        
        sql = f"""
            INSERT INTO {full_table} ({col_list})
            VALUES ({placeholders})
            ON CONFLICT ({pk_list}) DO NOTHING
        """
        
        count = 0
        with self.db.conn.cursor() as cur:
            for record in records:
                row = dict(record)
                if has_content_hash_col:
                    row["content_hash"] = self._compute_hash(record)
                values = [row.get(col) for col in all_cols]
                try:
                    cur.execute(sql, values)
                    affected = int(cur.rowcount or 0)
                    if affected > 0:
                        count += affected
                except Exception as e:
                    self.logger.warning("UPSERT 失败: %s, error=%s", record, e)
        
        self.db.commit()
        return count
