# -*- coding: utf-8 -*-
"""DWD 装载任务：从 ODS 增量写入 DWD（维度 SCD2，事实按时间增量）。"""
from __future__ import annotations

import os
import re
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import date, datetime
from decimal import Decimal, InvalidOperation
from typing import Any, Dict, Iterable, List, Sequence

from psycopg2.extras import Json, RealDictCursor, execute_batch, execute_values

from database.connection import DatabaseConnection
from tasks.base_task import BaseTask, TaskContext


class DwdLoadTask(BaseTask):
    """负责 DWD 装载：维度表做 SCD2 合并，事实表按时间增量写入。"""

    # DWD -> ODS 表映射（ODS 表名已与示例 JSON 前缀统一）
    TABLE_MAP: dict[str, str] = {
        # 维度
        # 门店：改用台费流水中的 siteprofile 快照，补齐 org/地址等字段
        "dwd.dim_site": "ods.table_fee_transactions",
        "dwd.dim_site_ex": "ods.table_fee_transactions",
        "dwd.dim_table": "ods.site_tables_master",
        "dwd.dim_table_ex": "ods.site_tables_master",
        "dwd.dim_assistant": "ods.assistant_accounts_master",
        "dwd.dim_assistant_ex": "ods.assistant_accounts_master",
        "dwd.dim_member": "ods.member_profiles",
        "dwd.dim_member_ex": "ods.member_profiles",
        "dwd.dim_member_card_account": "ods.member_stored_value_cards",
        "dwd.dim_member_card_account_ex": "ods.member_stored_value_cards",
        "dwd.dim_tenant_goods": "ods.tenant_goods_master",
        "dwd.dim_tenant_goods_ex": "ods.tenant_goods_master",
        "dwd.dim_store_goods": "ods.store_goods_master",
        "dwd.dim_store_goods_ex": "ods.store_goods_master",
        "dwd.dim_goods_category": "ods.stock_goods_category_tree",
        "dwd.dim_groupbuy_package": "ods.group_buy_packages",
        "dwd.dim_groupbuy_package_ex": "ods.group_buy_packages",
        "dwd.dim_staff": "ods.staff_info_master",
        "dwd.dim_staff_ex": "ods.staff_info_master",
        # 事实
        "dwd.dwd_settlement_head": "ods.settlement_records",
        "dwd.dwd_settlement_head_ex": "ods.settlement_records",
        "dwd.dwd_table_fee_log": "ods.table_fee_transactions",
        "dwd.dwd_table_fee_log_ex": "ods.table_fee_transactions",
        "dwd.dwd_table_fee_adjust": "ods.table_fee_discount_records",
        "dwd.dwd_table_fee_adjust_ex": "ods.table_fee_discount_records",
        "dwd.dwd_store_goods_sale": "ods.store_goods_sales_records",
        "dwd.dwd_store_goods_sale_ex": "ods.store_goods_sales_records",
        "dwd.dwd_assistant_service_log": "ods.assistant_service_records",
        "dwd.dwd_assistant_service_log_ex": "ods.assistant_service_records",
        "dwd.dwd_member_balance_change": "ods.member_balance_changes",
        "dwd.dwd_member_balance_change_ex": "ods.member_balance_changes",
        "dwd.dwd_groupbuy_redemption": "ods.group_buy_redemption_records",
        "dwd.dwd_groupbuy_redemption_ex": "ods.group_buy_redemption_records",
        "dwd.dwd_platform_coupon_redemption": "ods.platform_coupon_redemption_records",
        "dwd.dwd_platform_coupon_redemption_ex": "ods.platform_coupon_redemption_records",
        "dwd.dwd_recharge_order": "ods.recharge_settlements",
        "dwd.dwd_recharge_order_ex": "ods.recharge_settlements",
        "dwd.dwd_payment": "ods.payment_transactions",
        "dwd.dwd_refund": "ods.refund_transactions",
        "dwd.dwd_refund_ex": "ods.refund_transactions",
        "dwd.dwd_goods_stock_summary": "ods.goods_stock_summary",
        "dwd.dwd_goods_stock_movement": "ods.goods_stock_movements",
    }

    SCD_COLS = {"scd2_start_time", "scd2_end_time", "scd2_is_current", "scd2_version"}
    _NUMERIC_RE = re.compile(r"^[+-]?\d+(?:\.\d+)?$")
    _BOOL_STRINGS = {"true", "false", "1", "0", "yes", "no", "y", "n", "t", "f"}

    # 详情表 LEFT JOIN 配置：当 DWD 表需要从额外的 ODS 详情表获取字段时使用
    # detail_columns 中的列在 FACT_MAPPINGS 中以 detail."col" 形式引用
    DETAIL_JOIN_CONFIG: dict[str, dict] = {
        "dwd.dim_groupbuy_package_ex": {
            "detail_table": "ods.group_buy_package_details",
            "join_condition": 'ods_main."id" = detail."coupon_id"',
            "detail_columns": ["table_area_ids", "table_area_names", "assistant_services", "groupon_site_infos"],
        },
    }

    def _strip_scd2_keys(self, pk_cols: Sequence[str]) -> list[str]:
        return [c for c in pk_cols if c.lower() not in self.SCD_COLS]


    @staticmethod
    def _append_where_condition(where_sql: str, condition: str) -> str:
        if not condition:
            return where_sql
        if not where_sql:
            return f"WHERE {condition}"
        return f"{where_sql} AND {condition}"

    def _log_missing_fetched_at(self, cur, ods_table: str) -> None:
        """记录 ODS fetched_at 为空的情况（不抛异常）。"""
        table_sql = self._format_table(ods_table, "ods")
        try:
            cur.execute(f"SELECT 1 FROM {table_sql} WHERE fetched_at IS NULL LIMIT 1")
            if cur.fetchone():
                self.logger.error("ODS 表 %s 存在 fetched_at 为空的记录，已跳过", ods_table)
        except Exception as exc:  # noqa: BLE001
            self.logger.warning("检查 fetched_at 为空记录失败：%s, err=%s", ods_table, exc)

    @staticmethod
    def _pick_snapshot_order_column(cols: Sequence[str]) -> str | None:
        """从 ODS 列中选取用于快照排序的列（fetched_at > update_time > create_time）。
        CHANGE 2026-02-21: BUG 9 fix — 从 integrity_checker 移入本类，解决 AttributeError。
        """
        lower = {c.lower() for c in cols}
        for candidate in ("fetched_at", "update_time", "create_time"):
            if candidate in lower:
                return candidate
        return None

    @staticmethod
    def _latest_snapshot_select_sql(
        select_cols_sql: str,
        ods_table_sql: str,
        key_exprs: Sequence[str],
        order_col: str | None,
        where_sql: str = "",
    ) -> str:
        if key_exprs and order_col:
            distinct_on = ", ".join(key_exprs)
            # order_col 可能是预格式化的表达式（如 ods_main."fetched_at"），此时直接使用；
            # 否则包裹双引号
            order_col_expr = order_col if '"' in order_col else f'"{order_col}"'
            order_by = ", ".join([*key_exprs, f'{order_col_expr} DESC NULLS LAST'])
            return (
                f"SELECT DISTINCT ON ({distinct_on}) {select_cols_sql} "
                f"FROM {ods_table_sql} {where_sql} ORDER BY {order_by}"
            )
        return f"SELECT {select_cols_sql} FROM {ods_table_sql} {where_sql}"

    # 特殊列映射：dwd 列名 -> 源列表达式（可选 CAST）
    FACT_MAPPINGS: dict[str, list[tuple[str, str, str | None]]] = {
        # 维度表（补齐主键/字段差异）
        "dwd.dim_site": [
            ("org_id", "siteprofile->>'org_id'", None),
            ("shop_name", "siteprofile->>'shop_name'", None),
            ("site_label", "siteprofile->>'site_label'", None),
            ("full_address", "siteprofile->>'full_address'", None),
            ("address", "siteprofile->>'address'", None),
            ("longitude", "siteprofile->>'longitude'", "numeric"),
            ("latitude", "siteprofile->>'latitude'", "numeric"),
            ("tenant_site_region_id", "siteprofile->>'tenant_site_region_id'", None),
            ("business_tel", "siteprofile->>'business_tel'", None),
            ("site_type", "siteprofile->>'site_type'", None),
            ("shop_status", "siteprofile->>'shop_status'", None),
            ("tenant_id", "siteprofile->>'tenant_id'", None),
        ],
        "dwd.dim_site_ex": [
            ("auto_light", "siteprofile->>'auto_light'", None),
            ("attendance_enabled", "siteprofile->>'attendance_enabled'", None),
            ("attendance_distance", "siteprofile->>'attendance_distance'", None),
            ("prod_env", "siteprofile->>'prod_env'", None),
            ("light_status", "siteprofile->>'light_status'", None),
            ("light_type", "siteprofile->>'light_type'", None),
            ("light_token", "siteprofile->>'light_token'", None),
            ("address", "siteprofile->>'address'", None),
            ("avatar", "siteprofile->>'avatar'", None),
            ("wifi_name", "siteprofile->>'wifi_name'", None),
            ("wifi_password", "siteprofile->>'wifi_password'", None),
            ("customer_service_qrcode", "siteprofile->>'customer_service_qrcode'", None),
            ("customer_service_wechat", "siteprofile->>'customer_service_wechat'", None),
            ("fixed_pay_qrcode", "siteprofile->>'fixed_pay_qrCode'", None),
            ("longitude", "siteprofile->>'longitude'", "numeric"),
            ("latitude", "siteprofile->>'latitude'", "numeric"),
            ("tenant_site_region_id", "siteprofile->>'tenant_site_region_id'", None),
            ("site_type", "siteprofile->>'site_type'", None),
            ("site_label", "siteprofile->>'site_label'", None),
            ("shop_status", "siteprofile->>'shop_status'", None),
            ("create_time", "siteprofile->>'create_time'", "timestamptz"),
            ("update_time", "siteprofile->>'update_time'", "timestamptz"),
        ],
        "dwd.dim_table": [
            ("table_id", "id", None),
            ("site_table_area_name", "areaname", None),
            ("tenant_table_area_id", "site_table_area_id", None),
            ("order_id", "order_id", None),
        ],
        "dwd.dim_table_ex": [
            ("table_id", "id", None),
            ("table_cloth_use_time", "table_cloth_use_time", None),
            ("create_time", "create_time", None),                     # 台桌创建时间
            ("light_status", "light_status", None),                   # 台灯状态
            ("tablestatusname", "tablestatusname", None),             # 台桌状态中文名（ODS 驼峰 → PG 小写）
            ("sitename", "sitename", None),                           # 门店名称快照（ODS 驼峰 → PG 小写）
            ("applet_qr_code_url", '"appletQrCodeUrl"', None),        # 小程序二维码 URL（ODS 带引号保留大小写）
            ("audit_status", "audit_status", None),                   # 审核状态
            ("charge_free", "charge_free", None),                     # 是否免费台
            ("delay_lights_time", "delay_lights_time", None),         # 台灯熄灭延迟时间
            ("is_rest_area", "is_rest_area", None),                   # 是否休息区
            ("only_allow_groupon", "only_allow_groupon", None),       # 是否仅允许团购
            ("order_delay_time", "order_delay_time", None),           # 订单自动延时时长
            ("self_table", "self_table", None),                       # 是否自有台桌
            ("temporary_light_second", "temporary_light_second", None), # 临时开灯秒数
            ("virtual_table", "virtual_table", None),                 # 是否虚拟台桌
        ],
        "dwd.dim_assistant": [("assistant_id", "id", None), ("user_id", "user_id", None)],
        "dwd.dim_assistant_ex": [
            ("assistant_id", "id", None),
            ("introduce", "introduce", None),
            ("group_name", "group_name", None),
            ("light_equipment_id", "light_equipment_id", None),
            ("system_role_id", "system_role_id", None),       # 系统角色 ID
            ("job_num", "job_num", None),                     # 工号
            ("cx_unit_price", "cx_unit_price", None),         # 促销单价
            ("pd_unit_price", "pd_unit_price", None),         # 陪打单价
        ],
        "dwd.dim_member": [
            ("member_id", "id", None),
            ("pay_money_sum", "pay_money_sum", None),
            ("recharge_money_sum", "recharge_money_sum", None),
            # update_time: 上游 API 不提供此字段，保持 NULL（已在一致性检查白名单中标记）
        ],
        "dwd.dim_member_ex": [
            ("member_id", "id", None),
            ("register_site_name", "site_name", None),
            ("person_tenant_org_id", "person_tenant_org_id", None),
            ("person_tenant_org_name", "person_tenant_org_name", None),
            ("register_source", "register_source", None),
            ("other_pay_money_sum", "payload->>'otherPayMoneySum'", "numeric"),  # CHANGE 2026-03-26: 其他支付金额汇总（从 payload 提取，ODS 列可能为空）
            ("last_consume_time", "payload->>'lastConsumeTime'", "timestamptz"),  # CHANGE 2026-03-26: 最后消费时间
            ("non_consume_day_num", "payload->>'nonConsumeDayNum'", "integer"),  # CHANGE 2026-03-26: 未消费天数
            ("first_consumption", "payload->>'firstConsumption'", "integer"),  # CHANGE 2026-03-26: 是否首次消费（1=是, 2=否）
        ],
        "dwd.dim_member_card_account": [
            ("member_card_id", "id", None),
            ("principal_balance", "principal_balance", None),
            ("member_grade", "member_grade", None),
        ],
        "dwd.dim_member_card_account_ex": [
            ("member_card_id", "id", None),
            ("tenant_name", "tenantname", None),
            ("tenantavatar", "tenantavatar", None),
            ("card_no", "card_no", None),
            ("bind_password", "bind_password", None),
            ("use_scene", "use_scene", None),
            ("tableareaid", "tableareaid", None),
            ("goodscategoryid", "goodscategoryid", None),
            ("able_share_member_discount", "able_share_member_discount", "boolean"),
            ("electricity_deduct_radio", "electricity_deduct_radio", None),
            ("electricity_discount", "electricity_discount", None),
            ("electricity_card_deduct", "electricitycarddeduct", "boolean"),
            ("recharge_freeze_balance", "rechargefreezebalance", None),
            ("pdassisnatlevel", "pdassisnatlevel", None),  # CHANGE 2026-03-26: 陪打助教等级限制
            ("cxassisnatlevel", "cxassisnatlevel", None),  # CHANGE 2026-03-26: 促销助教等级限制
        ],
        "dwd.dim_tenant_goods": [
            ("tenant_goods_id", "id", None),
            ("category_name", "categoryname", None),
            ("not_sale", "not_sale", None),
        ],
        "dwd.dim_tenant_goods_ex": [
            ("tenant_goods_id", "id", None),
            ("remark_name", "remark_name", None),
            ("goods_bar_code", "goods_bar_code", None),
            ("commodity_code_list", "commoditycode", "TEXT[]"),  # CHANGE 2026-02-21: 从 commoditycode（PG 数组格式 {xxx}）映射，CAST 为 TEXT[]
            ("is_in_site", "isinsite", "boolean"),
        ],
        "dwd.dim_store_goods": [
            ("site_goods_id", "id", None),
            ("category_level1_name", "onecategoryname", None),
            ("category_level2_name", "twocategoryname", None),
            ("created_at", "create_time", None),
            ("updated_at", "update_time", None),
            ("avg_monthly_sales", "average_monthly_sales", None),
            ("batch_stock_qty", "batch_stock_quantity", None),  # CHANGE 2026-02-20: 修正映射源 stock→batch_stock_quantity（批次库存，非当前库存）
            ("sale_qty", "sale_num", None),
            ("total_sales_qty", "total_sales", None),
            ("commodity_code", "commodity_code", None),
            ("not_sale", "not_sale", None),
        ],
        "dwd.dim_store_goods_ex": [
            ("site_goods_id", "id", None),
            ("goods_barcode", "goods_bar_code", None),
            ("stock_qty", "stock", None),
            ("stock_secondary_qty", "stock_a", None),
            ("safety_stock_qty", "safe_stock", None),
            ("site_name", "sitename", None),
            ("goods_cover_url", "goods_cover", None),
            ("provisional_total_cost", "provisional_total_cost", None),  # CHANGE 2026-02-20: 修正映射源 total_purchase_cost→provisional_total_cost（暂估成本，非实际采购成本）
            ("is_discountable", "able_discount", None),
            ("freeze_status", "freeze", None),
            ("remark", "remark", None),
            ("days_on_shelf", "days_available", None),
            ("sort_order", "sort", None),
            ("time_slot_sale", "time_slot_sale", None),  # CHANGE 2026-02-21: 新增分时段销售标记
            ("warning_sales_day", "warning_sales_day", None),  # CHANGE 2026-02-24: 库存预警日均销量
            ("warning_day_max", "warning_day_max", None),  # CHANGE 2026-02-24: 预警天数上限
            ("warning_day_min", "warning_day_min", None),  # CHANGE 2026-02-24: 预警天数下限
        ],
        "dwd.dim_goods_category": [
            ("category_id", "id", None),
            ("tenant_id", "tenant_id", None),
            ("category_name", "category_name", None),
            ("alias_name", "alias_name", None),
            ("parent_category_id", "pid", None),
            ("business_name", "business_name", None),
            ("tenant_goods_business_id", "tenant_goods_business_id", None),
            ("sort_order", "sort", None),
            ("open_salesman", "open_salesman", None),
            ("is_warehousing", "is_warehousing", None),
            ("category_level", "CASE WHEN pid = 0 THEN 1 ELSE 2 END", None),
            ("is_leaf", "CASE WHEN categoryboxes IS NULL OR jsonb_array_length(categoryboxes)=0 THEN 1 ELSE 0 END", None),
        ],
        "dwd.dim_groupbuy_package": [
            ("groupbuy_package_id", "id", None),
            ("package_template_id", "package_id", None),
            ("coupon_face_value", "coupon_money", None),
            ("duration_seconds", "duration", None),
            ("sort", "sort", None),
            ("is_first_limit", "is_first_limit", "boolean"),
        ],
        "dwd.dim_groupbuy_package_ex": [
            ("groupbuy_package_id", "id", None),
            ("table_area_id", "table_area_id", None),
            ("tenant_table_area_id", "tenant_table_area_id", None),
            ("usable_range", "usable_range", None),
            ("table_area_id_list", "table_area_id_list", None),
            ("package_type", "type", None),
            ("tenant_coupon_sale_order_item_id", "tenantcouponsaleorderitemid", None),
            # CHANGE 2026-03-05: 团购详情字段（来自 ods.group_buy_package_details，通过 LEFT JOIN 关联）
            ("table_area_ids", 'detail."table_area_ids"', None),
            ("table_area_names", 'detail."table_area_names"', None),
            ("assistant_services", 'detail."assistant_services"', None),
            ("groupon_site_infos", 'detail."groupon_site_infos"', None),
        ],
        "dwd.dim_staff": [
            ("staff_id", "id", None),
            ("entry_time", "entry_time", "timestamptz"),
            ("resign_time", "resign_time", "timestamptz"),
        ],
        "dwd.dim_staff_ex": [
            ("staff_id", "id", None),
            ("rank_name", "rank_name", None),
            ("cashier_point_id", "cashier_point_id", "bigint"),
            ("cashier_point_name", "cashier_point_name", None),
            ("group_id", "group_id", "bigint"),
            ("group_name", "group_name", None),
            ("system_user_id", "system_user_id", "bigint"),
            ("tenant_org_id", "tenant_org_id", "bigint"),
            ("auth_code_create", "auth_code_create", "timestamptz"),
            ("create_time", "create_time", "timestamptz"),
            ("user_roles", "user_roles", "jsonb"),
        ],
        # 事实表主键及关键差异列
        "dwd.dwd_table_fee_log": [
            ("table_fee_log_id", "id", None),
            ("activity_discount_amount", "activity_discount_amount", None),
            ("real_service_money", "real_service_money", None),
        ],
        "dwd.dwd_table_fee_log_ex": [
            ("table_fee_log_id", "id", None),
            ("salesman_name", "salesman_name", None),
            ("order_consumption_type", "order_consumption_type", None),
            ("order_from", "order_from", None),  # CHANGE 2026-03-26: 订单来源
        ],
        "dwd.dwd_table_fee_adjust": [
            ("table_fee_adjust_id", "id", None),
            ("table_id", "site_table_id", None),
            ("table_area_id", "tenant_table_area_id", None),
            ("table_area_name", "tableprofile->>'table_area_name'", None),
            ("adjust_time", "create_time", None),
            ("table_name", "table_name", None),
            ("table_price", "table_price", None),
            ("charge_free", "charge_free", "boolean"),
        ],
        "dwd.dwd_table_fee_adjust_ex": [
            ("table_fee_adjust_id", "id", None),
            ("ledger_name", "ledger_name", None),
            ("area_type_id", "area_type_id", None),
            ("site_table_area_id", "site_table_area_id", None),
            ("site_table_area_name", "site_table_area_name", None),
            ("site_name", "sitename", None),
            ("tenant_name", "tenant_name", None),
        ],
        "dwd.dwd_store_goods_sale": [
            ("store_goods_sale_id", "id", None),
            # CHANGE 2026-02-20: 修正列名误导——原 DWD discount_price 实际映射自 ODS discount_money（折扣金额），
            # 现重命名为 discount_money 以反映真实语义；新增 discount_price 映射 ODS 真正的折后单价。
            ("discount_money", "discount_money", None),
            ("discount_price", "discount_price", None),
            ("coupon_share_money", "coupon_share_money", None),
        ],
        "dwd.dwd_store_goods_sale_ex": [
            ("store_goods_sale_id", "id", None),
            ("option_value_name", "option_value_name", None),
            ("open_salesman_flag", "opensalesman", "integer"),
            ("salesman_name", "salesman_name", None),
            ("salesman_org_id", "sales_man_org_id", None),
            ("legacy_order_goods_id", "ordergoodsid", None),
            ("site_name", "sitename", None),
            ("legacy_site_id", "siteid", None),
            ("activity_amount", "activity_amount", None),  # CHANGE 2026-03-26: 活动金额
            ("activity_id", "activity_id", None),  # CHANGE 2026-03-26: 活动 ID
            ("order_from", "order_from", None),  # CHANGE 2026-03-26: 订单来源
        ],
        "dwd.dwd_assistant_service_log": [
            ("assistant_service_id", "id", None),
            ("assistant_no", "assistantno", None),
            # CHANGE 2026-02-20: 修正映射源——site_assistant_id 应来自 ODS site_assistant_id（助教档案 ID），
            # 而非 order_assistant_id（订单级助教明细 ID）。order_assistant_id 由同名列自动映射。
            ("site_assistant_id", "site_assistant_id", None),
            ("level_name", "levelname", None),
            ("skill_name", "skillname", None),
            ("real_service_money", "real_service_money", None),
        ],
        "dwd.dwd_assistant_service_log_ex": [
            ("assistant_service_id", "id", None),
            ("assistant_name", "assistantname", None),
            ("ledger_group_name", "ledger_group_name", None),
            ("trash_applicant_name", "trash_applicant_name", None),
            ("trash_reason", "trash_reason", None),
            ("salesman_name", "salesman_name", None),
            ("table_name", "tablename", None),
            ("assistant_team_name", "assistantteamname", None),
            ("operator_id", "operator_id", None),             # 操作员 ID
            ("operator_name", "operator_name", None),         # 操作员姓名
            ("deduct_leave_seconds", "deduct_leave_seconds", None),  # CHANGE 2026-03-26: 扣除请假秒数
            ("order_from", "order_from", None),  # CHANGE 2026-03-26: 订单来源（1=线下, 4=小程序）
        ],
        "dwd.dwd_member_balance_change": [
            ("balance_change_id", "id", None),
            ("balance_before", "before", None),
            ("change_amount", "account_data", None),
            ("balance_after", "after", None),
            ("card_type_name", "membercardtypename", None),
            ("change_time", "create_time", None),
            ("member_name", "membername", None),
            ("member_mobile", "membermobile", None),
            ("principal_before", "principal_before", None),
            ("principal_after", "principal_after", None),
            # CHANGE [2026-02-20] intent: 补齐 principal_change_amount，由 principal_after - principal_before 计算
            ("principal_change_amount", "COALESCE(CAST(principal_after AS numeric),0) - COALESCE(CAST(principal_before AS numeric),0)", "numeric"),
        ],
        "dwd.dwd_member_balance_change_ex": [
            ("balance_change_id", "id", None),
            ("pay_site_name", "paysitename", None),
            ("register_site_name", "registersitename", None),
            ("principal_data", "principal_data", None),
            ("relate_id", "relate_id", None),             # 关联业务单据 ID
        ],
        "dwd.dwd_groupbuy_redemption": [
            ("redemption_id", "id", None),
            ("member_discount_money", "member_discount_money", None),
            ("coupon_sale_id", "coupon_sale_id", None),
        ],
        "dwd.dwd_groupbuy_redemption_ex": [
            ("redemption_id", "id", None),
            ("table_area_name", "tableareaname", None),
            ("site_name", "sitename", None),
            ("table_name", "tablename", None),
            ("goods_option_price", "goodsoptionprice", None),
            ("salesman_name", "salesman_name", None),
            ("salesman_org_id", "sales_man_org_id", None),
            ("ledger_group_name", "ledger_group_name", None),
            ("table_share_money", "table_share_money", None),
            ("table_service_share_money", "table_service_share_money", None),
            ("goods_share_money", "goods_share_money", None),
            ("good_service_share_money", "good_service_share_money", None),
            ("assistant_share_money", "assistant_share_money", None),
            ("assistant_service_share_money", "assistant_service_share_money", None),
            ("recharge_share_money", "recharge_share_money", None),
        ],
        "dwd.dwd_platform_coupon_redemption": [("platform_coupon_redemption_id", "id", None)],
        "dwd.dwd_platform_coupon_redemption_ex": [
            ("platform_coupon_redemption_id", "id", None),
            ("coupon_cover", "coupon_cover", None),
        ],
        "dwd.dwd_payment": [
            ("payment_id", "id", None),
            ("pay_date", "pay_time", "date"),
            ("tenant_id", "tenant_id", None),
        ],
        "dwd.dwd_refund": [("refund_id", "id", None)],
        "dwd.dwd_refund_ex": [
            ("refund_id", "id", None),
            ("tenant_name", "tenantname", None),
            ("channel_payer_id", "channel_payer_id", None),
            ("channel_pay_no", "channel_pay_no", None),
        ],
        # 结算头：settlement_records（源列为小写驼峰/无下划线，需要显式映射）
        "dwd.dwd_settlement_head": [
            ("order_settle_id", "id", None),
            ("tenant_id", "tenantid", None),
            ("site_id", "siteid", None),
            ("site_name", "sitename", None),
            ("table_id", "tableid", None),
            ("settle_name", "settlename", None),
            ("order_trade_no", "settlerelateid", None),
            ("create_time", "createtime", None),
            ("pay_time", "paytime", None),
            ("settle_type", "settletype", None),
            ("revoke_order_id", "revokeorderid", None),
            ("member_id", "memberid", None),
            ("member_name", "membername", None),
            ("member_phone", "memberphone", None),
            ("member_card_account_id", "tenantmembercardid", None),
            ("member_card_type_name", "membercardtypename", None),
            ("is_bind_member", "isbindmember", None),
            ("member_discount_amount", "memberdiscountamount", None),
            ("consume_money", "consumemoney", None),
            ("table_charge_money", "tablechargemoney", None),
            ("goods_money", "goodsmoney", None),
            ("real_goods_money", "realgoodsmoney", None),
            ("assistant_pd_money", "assistantpdmoney", None),
            ("assistant_cx_money", "assistantcxmoney", None),
            ("adjust_amount", "adjustamount", None),
            ("pay_amount", "payamount", None),
            ("balance_amount", "balanceamount", None),
            ("recharge_card_amount", "rechargecardamount", None),
            ("gift_card_amount", "giftcardamount", None),
            ("coupon_amount", "couponamount", None),
            ("rounding_amount", "roundingamount", None),
            ("point_amount", "pointamount", None),
            ("electricity_money", "electricitymoney", None),
            ("real_electricity_money", "realelectricitymoney", None),
            ("electricity_adjust_money", "electricityadjustmoney", None),
            ("pl_coupon_sale_amount", "plcouponsaleamount", None),
            ("mervou_sales_amount", "mervousalesamount", None),
        ],
        "dwd.dwd_settlement_head_ex": [
            ("order_settle_id", "id", None),
            ("serial_number", "serialnumber", None),
            ("settle_status", "settlestatus", None),
            ("can_be_revoked", "canberevoked", "boolean"),
            ("revoke_order_name", "revokeordername", None),
            ("revoke_time", "revoketime", None),
            ("is_first_order", "isfirst", "boolean"),
            ("service_money", "servicemoney", None),
            ("cash_amount", "cashamount", None),
            ("card_amount", "cardamount", None),
            ("online_amount", "onlineamount", None),
            ("refund_amount", "refundamount", None),
            ("prepay_money", "prepaymoney", None),
            ("payment_method", "paymentmethod", None),
            ("coupon_sale_amount", "couponsaleamount", None),
            ("all_coupon_discount", "allcoupondiscount", None),
            ("goods_promotion_money", "goodspromotionmoney", None),
            ("assistant_promotion_money", "assistantpromotionmoney", None),
            ("activity_discount", "activitydiscount", None),
            ("assistant_manual_discount", "assistantmanualdiscount", None),
            ("point_discount_price", "pointdiscountprice", None),
            ("point_discount_cost", "pointdiscountcost", None),
            ("is_use_coupon", "isusecoupon", "boolean"),
            ("is_use_discount", "isusediscount", "boolean"),
            ("is_activity", "isactivity", "boolean"),
            ("operator_name", "operatorname", None),
            ("salesman_name", "salesmanname", None),
            ("order_remark", "orderremark", None),
            ("operator_id", "operatorid", None),
            ("salesman_user_id", "salesmanuserid", None),
            ("order_from", "orderfrom", None),  # CHANGE 2026-03-26: 订单来源（来自 settleList.orderFrom）
            # CHANGE: intent=删除 settle_list 映射，该列已从 DWD 表中移除（与 ODS payload 冗余）
            #         assumptions=结算明细可随时从 ODS payload->'settleList' 按需提取
            #         prompt=P20260214-040000
        ],
        # 充值结算：recharge_settlements（字段风格同 settlement_records）
        "dwd.dwd_recharge_order": [
            ("recharge_order_id", "id", None),
            ("tenant_id", "tenantid", None),
            ("site_id", "siteid", None),
            ("member_id", "memberid", None),
            ("member_name_snapshot", "membername", None),
            ("member_phone_snapshot", "memberphone", None),
            ("tenant_member_card_id", "tenantmembercardid", None),
            ("member_card_type_name", "membercardtypename", None),
            ("settle_relate_id", "settlerelateid", None),
            ("settle_type", "settletype", None),
            ("settle_name", "settlename", None),
            ("is_first", "isfirst", None),
            ("pay_amount", "payamount", None),
            ("refund_amount", "refundamount", None),
            ("point_amount", "pointamount", None),
            ("cash_amount", "cashamount", None),
            ("payment_method", "paymentmethod", None),
            ("create_time", "createtime", None),
            ("pay_time", "paytime", None),
            ("pl_coupon_sale_amount", "plcouponsaleamount", None),    # 平台券销售金额
            ("mervou_sales_amount", "mervousalesamount", None),       # 储值券销售金额
            ("electricity_money", "electricitymoney", None),           # 电费金额
            ("real_electricity_money", "realelectricitymoney", None),  # 实际电费金额
            ("electricity_adjust_money", "electricityadjustmoney", None),  # 电费调整金额
        ],
        # CHANGE 2026-02-20: 移除 salesman_name/order_remark/revoke_order_name 的重复条目（原列表首尾各出现一次）
        "dwd.dwd_recharge_order_ex": [
            ("recharge_order_id", "id", None),
            ("site_name_snapshot", "sitename", None),
            ("settle_status", "settlestatus", None),
            ("is_bind_member", "isbindmember", "boolean"),
            ("is_activity", "isactivity", "boolean"),
            ("is_use_coupon", "isusecoupon", "boolean"),
            ("is_use_discount", "isusediscount", "boolean"),
            ("can_be_revoked", "canberevoked", "boolean"),
            ("online_amount", "onlineamount", None),
            ("balance_amount", "balanceamount", None),
            ("card_amount", "cardamount", None),
            ("coupon_amount", "couponamount", None),
            ("recharge_card_amount", "rechargecardamount", None),
            ("gift_card_amount", "giftcardamount", None),
            ("prepay_money", "prepaymoney", None),
            ("consume_money", "consumemoney", None),
            ("goods_money", "goodsmoney", None),
            ("real_goods_money", "realgoodsmoney", None),
            ("table_charge_money", "tablechargemoney", None),
            ("service_money", "servicemoney", None),
            ("activity_discount", "activitydiscount", None),
            ("all_coupon_discount", "allcoupondiscount", None),
            ("goods_promotion_money", "goodspromotionmoney", None),
            ("assistant_promotion_money", "assistantpromotionmoney", None),
            ("assistant_pd_money", "assistantpdmoney", None),
            ("assistant_cx_money", "assistantcxmoney", None),
            ("assistant_manual_discount", "assistantmanualdiscount", None),
            ("coupon_sale_amount", "couponsaleamount", None),
            ("member_discount_amount", "memberdiscountamount", None),
            ("point_discount_price", "pointdiscountprice", None),
            ("point_discount_cost", "pointdiscountcost", None),
            ("adjust_amount", "adjustamount", None),
            ("rounding_amount", "roundingamount", None),
            ("operator_id", "operatorid", None),
            ("operator_name_snapshot", "operatorname", None),
            ("salesman_user_id", "salesmanuserid", None),
            ("salesman_name", "salesmanname", None),
            ("order_remark", "orderremark", None),
            ("table_id", "tableid", None),
            ("serial_number", "serialnumber", None),
            ("revoke_order_id", "revokeorderid", None),
            ("revoke_order_name", "revokeordername", None),
            ("revoke_time", "revoketime", None),
        ],
        # 库存汇总：goods_stock_summary（ODS 列名全小写）
        # CHANGE 2026-02-21: BUG 10 fix — ODS 列名是小写（sitegoodsid），不是驼峰
        # CHANGE 2026-03-01: 补 site_id 映射（ODS 入库时从 app.store_id 注入 siteid）
        "dwd.dwd_goods_stock_summary": [
            ("site_goods_id", '"sitegoodsid"', "bigint"),           # 门店商品 ID（PK）
            ("goods_name", '"goodsname"', None),                     # 商品名称
            ("goods_unit", '"goodsunit"', None),                     # 计量单位
            ("goods_category_id", '"goodscategoryid"', "bigint"),    # 一级分类 ID
            ("goods_category_second_id", '"goodscategorysecondid"', "bigint"),  # 二级分类 ID
            ("category_name", '"categoryname"', None),               # 分类名称
            ("range_start_stock", '"rangestartstock"', "numeric"),   # 期初库存
            ("range_end_stock", '"rangeendstock"', "numeric"),       # 期末库存
            ("range_in", '"rangein"', "numeric"),                    # 入库数量
            ("range_out", '"rangeout"', "numeric"),                  # 出库数量
            ("range_sale", '"rangesale"', "numeric"),                # 销售数量
            ("range_sale_money", '"rangesalemoney"', "numeric"),     # 销售金额
            ("range_inventory", '"rangeinventory"', "numeric"),      # 盘点调整量
            ("current_stock", '"currentstock"', "numeric"),          # 当前库存
            ("site_id", '"siteid"', "bigint"),                       # 门店 ID（ODS 入库时注入）
            ("create_time", '"createtime"', "timestamptz"),  # CHANGE 2026-03-26: 库存记录创建时间
        ],
        # 库存变动流水：goods_stock_movements（ODS 列名全小写）
        # CHANGE 2026-02-21: BUG 10 fix — ODS 列名是小写，不是驼峰
        "dwd.dwd_goods_stock_movement": [
            ("site_goods_stock_id", '"sitegoodsstockid"', "bigint"),       # 库存变动记录 ID（PK）
            ("tenant_id", '"tenantid"', "bigint"),                          # 租户 ID
            ("site_id", '"siteid"', "bigint"),                              # 门店 ID
            ("site_goods_id", '"sitegoodsid"', "bigint"),                   # 门店商品 ID
            ("goods_name", '"goodsname"', None),                            # 商品名称
            ("goods_category_id", '"goodscategoryid"', "bigint"),           # 一级分类 ID
            ("goods_second_category_id", '"goodssecondcategoryid"', "bigint"),  # 二级分类 ID
            ("unit", "unit", None),                                         # 计量单位（ODS 已是小写）
            ("price", "price", "numeric"),                                  # 商品单价（ODS 已是小写）
            ("stock_type", '"stocktype"', "integer"),                       # 库存变动类型
            ("change_num", '"changenum"', "numeric"),                       # 变动数量
            ("start_num", '"startnum"', "numeric"),                         # 变动前库存
            ("end_num", '"endnum"', "numeric"),                             # 变动后库存
            ("change_num_a", '"changenuma"', "numeric"),                    # 辅助单位变动量
            ("start_num_a", '"startnuma"', "numeric"),                      # 辅助单位变动前库存
            ("end_num_a", '"endnuma"', "numeric"),                          # 辅助单位变动后库存
            ("remark", "remark", None),                                     # 备注（ODS 已是小写）
            ("operator_name", '"operatorname"', None),                      # 操作人
            ("create_time", '"createtime"', "timestamptz"),                 # 变动时间
        ],
    }

    def get_task_code(self) -> str:
        """返回任务编码。"""
        return "DWD_LOAD_FROM_ODS"

    def extract(self, context: TaskContext) -> dict[str, Any]:
        """准备运行所需的上下文信息。"""
        return {"now": datetime.now()}

    def load(self, extracted: dict[str, Any], context: TaskContext) -> dict[str, Any]:
        """
        并行遍历映射关系，维度执行 SCD2 合并，事实表按时间增量插入。

        说明：
        - 使用 ThreadPoolExecutor 并行处理多张表，每张表使用独立数据库连接和事务；
        - 单表失败会回滚该表并继续后续表，最终在结果中汇总错误信息；
        - 并行线程数通过 AppConfig 的 dwd.parallel_workers 配置（默认 4）。
        """
        now = extracted["now"]
        summary: List[Dict[str, Any]] = []
        errors: List[Dict[str, Any]] = []
        only_tables_cfg = self.config.get("dwd.only_tables") or []
        # 也支持通过环境变量 DWD_ONLY_TABLES 传递（GUI 使用此方式）
        env_only = os.environ.get("DWD_ONLY_TABLES", "").strip()
        if env_only and not only_tables_cfg:
            only_tables_cfg = [t.strip() for t in env_only.split(",") if t.strip()]
        only_tables = {str(t).strip().lower() for t in only_tables_cfg if str(t).strip()} if only_tables_cfg else set()

        parallel_workers = int(self.config.get("dwd.parallel_workers", 4))

        # 筛选需要处理的表
        tables_to_process: list[tuple[str, str]] = []
        for dwd_table, ods_table in self.TABLE_MAP.items():
            if only_tables and dwd_table.lower() not in only_tables and self._table_base(dwd_table).lower() not in only_tables:
                continue
            tables_to_process.append((dwd_table, ods_table))

        if not tables_to_process:
            return {"tables": summary, "errors": 0, "error_details": errors}

        # 并行调度：每张表在独立线程中执行，使用独立数据库连接
        with ThreadPoolExecutor(max_workers=parallel_workers) as executor:
            futures = {}
            for dwd_table, ods_table in tables_to_process:
                future = executor.submit(
                    self._process_single_table,
                    dwd_table, ods_table, now, context,
                )
                futures[future] = dwd_table

            for future in as_completed(futures):
                dwd_table = futures[future]
                try:
                    table_result = future.result()
                    summary.append(table_result)
                except Exception as exc:  # noqa: BLE001
                    self.logger.error(
                        "DWD 并行装载失败：%s，err=%s", dwd_table, exc,
                    )
                    errors.append({"table": dwd_table, "error": str(exc)})

        return {"tables": summary, "errors": len(errors), "error_details": errors}

    def _process_single_table(
        self,
        dwd_table: str,
        ods_table: str,
        now: datetime,
        context: TaskContext,
    ) -> Dict[str, Any]:
        """在独立线程中处理单张 DWD 表，使用独立数据库连接和事务。

        每张表创建独立的 DatabaseConnection，处理完成后关闭，
        保证线程间事务隔离，单表失败不影响其他表。
        """
        started = time.monotonic()
        self.logger.info("DWD 装载开始：%s <= %s", dwd_table, ods_table)

        # 为当前线程创建独立数据库连接
        thread_db = DatabaseConnection(
            dsn=self.db._dsn,
            session=self.db._session,
            connect_timeout=self.db._connect_timeout,
        )
        try:
            with thread_db.conn.cursor(cursor_factory=RealDictCursor) as cur:
                dwd_cols = self._get_columns(cur, dwd_table)
                ods_cols = self._get_columns(cur, ods_table)
                if not dwd_cols:
                    self.logger.warning("跳过 %s：未能获取 DWD 列信息", dwd_table)
                    return {"table": dwd_table, "mode": "SKIPPED", "inserted": 0, "updated": 0}

                if self._table_base(dwd_table).startswith("dim_"):
                    dim_counts = self._merge_dim(cur, dwd_table, ods_table, dwd_cols, ods_cols, now)
                    thread_db.conn.commit()
                    result = {"table": dwd_table, "mode": "SCD2", **dim_counts}
                else:
                    dwd_types = self._get_column_types(cur, dwd_table, "dwd")
                    ods_types = self._get_column_types(cur, ods_table, "ods")
                    fact_counts = self._merge_fact_increment(
                        cur,
                        dwd_table,
                        ods_table,
                        dwd_cols,
                        ods_cols,
                        dwd_types,
                        ods_types,
                        window_start=context.window_start,
                        window_end=context.window_end,
                    )
                    thread_db.conn.commit()
                    result = {"table": dwd_table, "mode": "INCREMENT", **fact_counts}

            elapsed = time.monotonic() - started
            self.logger.info("DWD 装载完成：%s，用时 %.2fs", dwd_table, elapsed)
            return result
        except Exception as exc:
            try:
                thread_db.conn.rollback()
            except Exception:
                pass
            elapsed = time.monotonic() - started
            self.logger.exception(
                "DWD 装载失败：%s，用时 %.2fs，err=%s", dwd_table, elapsed, exc,
            )
            # 重新抛出，让 future.result() 在主线程捕获
            raise
        finally:
            thread_db.close()

    # ---------------------- 辅助方法 ----------------------
    def _get_columns(self, cur, table: str) -> List[str]:
        """获取指定表的列名（小写）。"""
        schema, name = self._split_table_name(table, default_schema="dwd")
        cur.execute(
            """
            SELECT column_name
            FROM information_schema.columns
            WHERE table_schema = %s AND table_name = %s
            """,
            (schema, name),
        )
        return [r["column_name"].lower() for r in cur.fetchall()]

    def _get_primary_keys(self, cur, table: str) -> List[str]:
        """获取表的主键列名列表。"""
        schema, name = self._split_table_name(table, default_schema="dwd")
        cur.execute(
            """
            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
            """,
            (schema, name),
        )
        return [r["column_name"].lower() for r in cur.fetchall()]

    def _get_column_types(self, cur, table: str, default_schema: str) -> Dict[str, str]:
        """获取列的数据类型（information_schema.data_type）。"""
        schema, name = self._split_table_name(table, default_schema=default_schema)
        cur.execute(
            """
            SELECT column_name, data_type
            FROM information_schema.columns
            WHERE table_schema = %s AND table_name = %s
            """,
            (schema, name),
        )
        return {r["column_name"].lower(): r["data_type"].lower() for r in cur.fetchall()}

    def _build_column_mapping(
        self, cur, dwd_table: str, ods_table: str,
        pk_cols: Sequence[str], ods_cols: Sequence[str]
    ) -> Dict[str, tuple[str, str | None]]:
        """合并显式 FACT_MAPPINGS 与主键兜底映射。"""
        mapping_entries = self.FACT_MAPPINGS.get(dwd_table, [])
        mapping: Dict[str, tuple[str, str | None]] = {
            dst.lower(): (src, cast_type) for dst, src, cast_type in mapping_entries
        }
        ods_set = {c.lower() for c in ods_cols}
        if "fetched_at" not in ods_set:
            self.logger.error("跳过 %s：ODS 表 %s 缺少 fetched_at 列", dwd_table, ods_table)
            return {"processed": 0, "inserted": 0, "updated": 0, "skipped": 0}
        self._log_missing_fetched_at(cur, ods_table)
        for pk in pk_cols:
            pk_lower = pk.lower()
            if pk_lower not in mapping and pk_lower not in ods_set and "id" in ods_set:
                mapping[pk_lower] = ("id", None)
        return mapping


    def _fetch_source_rows(
        self, cur, table: str, columns: Sequence[str], where_sql: str = "", params: Sequence[Any] = None
    ) -> List[Dict[str, Any]]:
        """从源表读取指定列，返回小写键的字典列表。"""
        schema, name = self._split_table_name(table, default_schema="ods")
        cols_sql = ", ".join(f'"{c}"' for c in columns)
        sql = f'SELECT {cols_sql} FROM "{schema}"."{name}" {where_sql}'
        cur.execute(sql, params or [])
        rows = []
        for r in cur.fetchall():
            rows.append({k.lower(): v for k, v in r.items()})
        return rows

    def _expand_goods_category_rows(self, rows: list[Dict[str, Any]]) -> list[Dict[str, Any]]:
        """将分类表中的 categoryboxes 元素展开为子类记录。"""
        expanded: list[Dict[str, Any]] = []
        for r in rows:
            expanded.append(r)
            boxes = r.get("categoryboxes")
            if isinstance(boxes, list):
                for child in boxes:
                    if not isinstance(child, dict):
                        continue
                    child_row: Dict[str, Any] = {}
                    # 继承父级的租户与业务大类信息
                    child_row["tenant_id"] = r.get("tenant_id")
                    child_row["business_name"] = child.get("business_name", r.get("business_name"))
                    child_row["tenant_goods_business_id"] = child.get(
                        "tenant_goods_business_id", r.get("tenant_goods_business_id")
                    )
                    # 合并子类字段
                    child_row.update(child)
                    # 默认父子关系
                    child_row.setdefault("pid", r.get("id"))
                    # 衍生层级/叶子标记
                    child_boxes = child_row.get("categoryboxes")
                    if not isinstance(child_boxes, list):
                        is_leaf = 1
                    else:
                        is_leaf = 1 if len(child_boxes) == 0 else 0
                    child_row.setdefault("category_level", 2)
                    child_row.setdefault("is_leaf", is_leaf)
                    expanded.append(child_row)
        return expanded

    def _merge_dim(
        self,
        cur,
        dwd_table: str,
        ods_table: str,
        dwd_cols: Sequence[str],
        ods_cols: Sequence[str],
        now: datetime,
    ) -> Dict[str, int]:
        """维表合并：统一走 SCD2 路径。"""
        return self._merge_dim_scd2(cur, dwd_table, ods_table, dwd_cols, ods_cols, now)



    def _merge_dim_scd2(
        self,
        cur,
        dwd_table: str,
        ods_table: str,
        dwd_cols: Sequence[str],
        ods_cols: Sequence[str],
        now: datetime,
    ) -> Dict[str, int]:
        """对维表执行 SCD2 合并：对比变更关闭旧版并插入新版。"""
        pk_cols = self._get_primary_keys(cur, dwd_table)
        if not pk_cols:
            raise ValueError(f"{dwd_table} 未配置主键，无法执行 SCD2 合并")

        business_keys = self._strip_scd2_keys(pk_cols)
        if not business_keys:
            raise ValueError(f"{dwd_table} primary key only contains SCD2 columns; cannot merge")

        mapping = self._build_column_mapping(cur, dwd_table, ods_table, business_keys, ods_cols)
        ods_set = {c.lower() for c in ods_cols}
        if "fetched_at" not in ods_set:
            self.logger.error("跳过 %s：ODS 表 %s 缺少 fetched_at 列", dwd_table, ods_table)
            return {"processed": 0, "inserted": 0, "updated": 0, "skipped": 0}
        self._log_missing_fetched_at(cur, ods_table)
        # CHANGE 2026-02-22: BUG 12 — 获取列类型，用于哨兵日期过滤
        dwd_types = self._get_column_types(cur, dwd_table, "dwd")
        ods_types = self._get_column_types(cur, ods_table, "ods")
        ts_types = {"timestamp without time zone", "timestamp with time zone"}
        table_sql = self._format_table(ods_table, "ods")
        # CHANGE 2026-03-05: 详情表 LEFT JOIN 支持 — 当 DWD 表配置了 DETAIL_JOIN_CONFIG 时，
        #   给 ODS 主表加别名 ods_main，LEFT JOIN 详情表为 detail，
        #   非 detail 列引用加 ods_main. 前缀避免歧义
        detail_join = self.DETAIL_JOIN_CONFIG.get(dwd_table)
        ods_alias = "ods_main" if detail_join else ""
        if detail_join:
            detail_table_sql = self._format_table(detail_join["detail_table"], "ods")
            table_sql = (
                f"{table_sql} AS ods_main "
                f'LEFT JOIN {detail_table_sql} AS detail ON {detail_join["join_condition"]}'
            )
        # 构造 SELECT 表达式，支持 JSON/expression 映射
        select_exprs: list[str] = []
        added: set[str] = set()
        for col in dwd_cols:
            lc = col.lower()
            if lc in self.SCD_COLS:
                continue
            if lc in mapping:
                src, cast_type = mapping[lc]
                # detail. 前缀的列直接使用（来自详情表），其他列加 ods_main. 前缀
                if ods_alias and not src.startswith("detail."):
                    src = self._qualify_column_ref(src, ods_alias)
                select_exprs.append(f"{self._cast_expr(src, cast_type)} AS \"{lc}\"")
                added.add(lc)
            elif lc in ods_set:
                col_ref = f'{ods_alias}."{lc}"' if ods_alias else f'"{lc}"'
                # CHANGE 2026-02-22: BUG 12 — 同名列如果是时间类型，加哨兵值过滤
                if dwd_types.get(lc) in ts_types and ods_types.get(lc) in ts_types:
                    select_exprs.append(
                        f"CASE WHEN {col_ref} >= '{self._SENTINEL_DATE_THRESHOLD}'::timestamp "
                        f'THEN {col_ref} ELSE NULL END AS "{lc}"'
                    )
                else:
                    select_exprs.append(f'{col_ref} AS "{lc}"')
                added.add(lc)
        # 分类维度需要额外读取 categoryboxes 以展开子类
        if dwd_table == "dwd.dim_goods_category" and "categoryboxes" not in added and "categoryboxes" in ods_set:
            col_ref = f'{ods_alias}."categoryboxes"' if ods_alias else '"categoryboxes"'
            select_exprs.append(f'{col_ref} AS "categoryboxes"')
            added.add("categoryboxes")
        # 主键兜底确保被选出
        for pk in business_keys:
            lc = pk.lower()
            if lc not in added:
                if lc in mapping:
                    src, cast_type = mapping[lc]
                    if ods_alias and not src.startswith("detail."):
                        src = self._qualify_column_ref(src, ods_alias)
                    select_exprs.append(f"{self._cast_expr(src, cast_type)} AS \"{lc}\"")
                elif lc in ods_set:
                    col_ref = f'{ods_alias}."{lc}"' if ods_alias else f'"{lc}"'
                    select_exprs.append(f'{col_ref} AS "{lc}"')
                added.add(lc)

        if not select_exprs:
            return {"processed": 0, "inserted": 0, "updated": 0, "skipped": 0}

        order_col = self._pick_snapshot_order_column(ods_cols)  # CHANGE 2026-02-21: BUG 9 fix — 方法从 integrity_checker 移入本类
        key_exprs: list[str] = []
        for key in business_keys:
            lc = key.lower()
            if lc in mapping:
                src, cast_type = mapping[lc]
                if ods_alias and not src.startswith("detail."):
                    src = self._qualify_column_ref(src, ods_alias)
                key_exprs.append(self._cast_expr(src, cast_type))
            elif lc in ods_set:
                key_exprs.append(f'{ods_alias}."{lc}"' if ods_alias else f'"{lc}"')

        select_cols_sql = ", ".join(select_exprs)
        fetched_at_ref = f'{ods_alias}."fetched_at"' if ods_alias else '"fetched_at"'
        where_sql = self._append_where_condition("", f'{fetched_at_ref} IS NOT NULL')
        # CHANGE 2026-03-05: order_col 也需要加别名前缀
        qualified_order_col = f'{ods_alias}."{order_col}"' if ods_alias and order_col else (f'"{order_col}"' if order_col else None)
        sql = self._latest_snapshot_select_sql(
            select_cols_sql, table_sql, key_exprs, qualified_order_col, where_sql
        )
        cur.execute(sql)
        rows = [{k.lower(): v for k, v in r.items()} for r in cur.fetchall()]

        # 特殊：分类维度展开子类
        if dwd_table == "dwd.dim_goods_category":
            rows = self._expand_goods_category_rows(rows)

        # 归一化源行并按主键去重
        seen_pk = set()
        src_rows_by_pk: dict[tuple[Any, ...], Dict[str, Any]] = {}
        for row in rows:
            mapped_row: Dict[str, Any] = {}
            for col in dwd_cols:
                lc = col.lower()
                if lc in self.SCD_COLS:
                    continue
                value = row.get(lc)
                if value is None and lc in mapping:
                    src, _ = mapping[lc]
                    value = row.get(src.lower())
                mapped_row[lc] = value

            pk_key = tuple(mapped_row.get(pk) for pk in business_keys)
            if pk_key in seen_pk:
                continue
            if any(v is None for v in pk_key):
                self.logger.warning("跳过 %s：主键缺失 %s", dwd_table, dict(zip(business_keys, pk_key)))
                continue
            seen_pk.add(pk_key)
            src_rows_by_pk[pk_key] = mapped_row

        if not src_rows_by_pk:
            return {"processed": 0, "inserted": 0, "updated": 0, "skipped": 0}

        # 预加载当前版本（scd2_is_current=1），避免逐行 SELECT 造成大量 round-trip
        # CHANGE 2026-02-22: BUG 12 — 用显式列列表替代 SELECT *，对 timestamptz 列做哨兵值过滤
        #   防止 BC 日期导致 psycopg2 fetchall() 抛出 ValueError
        table_sql_dwd = self._format_table(dwd_table, "dwd")
        dwd_select_exprs: list[str] = []
        for col in dwd_cols:
            lc = col.lower()
            if dwd_types.get(lc) in ts_types:
                dwd_select_exprs.append(
                    f'CASE WHEN "{lc}" >= \'{self._SENTINEL_DATE_THRESHOLD}\'::timestamptz '
                    f'THEN "{lc}" ELSE NULL END AS "{lc}"'
                )
            else:
                dwd_select_exprs.append(f'"{lc}"')
        dwd_select_sql = ", ".join(dwd_select_exprs)
        where_current = "COALESCE(scd2_is_current,1)=1"
        cur.execute(f"SELECT {dwd_select_sql} FROM {table_sql_dwd} WHERE {where_current}")
        current_rows = cur.fetchall() or []
        current_by_pk: dict[tuple[Any, ...], Dict[str, Any]] = {}
        for r in current_rows:
            rr = {k.lower(): v for k, v in r.items()}
            pk_key = tuple(rr.get(pk) for pk in business_keys)
            current_by_pk[pk_key] = rr

        # 计算需要关闭/插入的主键集合
        to_close: list[tuple[Any, ...]] = []
        to_insert: list[tuple[Dict[str, Any], int]] = []
        # force_full_update 时跳过逐列对比，视所有已存在行为"已变更"
        force_full = bool(self.config.get("run.force_full_update", False))
        for pk_key, incoming in src_rows_by_pk.items():
            current = current_by_pk.get(pk_key)
            if current and not force_full and not self._is_row_changed(current, incoming, dwd_cols):
                continue
            if current:
                version = (current.get("scd2_version") or 1) + 1
                to_close.append(pk_key)
            else:
                version = 1
            to_insert.append((incoming, version))

        # 先关闭旧版本（同一批次统一 end_time）
        if to_close:
            self._close_current_dim_bulk(cur, dwd_table, business_keys, to_close, now)

        # 批量插入新版本
        if to_insert:
            self._insert_dim_rows_bulk(cur, dwd_table, dwd_cols, to_insert, now, dwd_types=dwd_types)

        processed = len(src_rows_by_pk)
        updated = len(to_close)
        inserted = max(0, len(to_insert) - updated)
        skipped = max(0, processed - inserted - updated)
        return {"processed": processed, "inserted": inserted, "updated": updated, "skipped": skipped}

    def _close_current_dim_bulk(
        self,
        cur,
        table: str,
        pk_cols: Sequence[str],
        pk_keys: Sequence[tuple[Any, ...]],
        now: datetime,
    ) -> None:
        """批量关闭当前版本（scd2_is_current=0 + 填充结束时间）。"""
        table_sql = self._format_table(table, "dwd")
        if len(pk_cols) == 1:
            pk = pk_cols[0]
            ids = [k[0] for k in pk_keys]
            cur.execute(
                f'UPDATE {table_sql} SET scd2_end_time=%s, scd2_is_current=0 '
                f'WHERE COALESCE(scd2_is_current,1)=1 AND "{pk}" = ANY(%s)',
                (now, ids),
            )
            return

        # 复合主键：对"发生变更的键"逐条关闭（数量通常远小于全量行数）
        where_clause = " AND ".join(f'"{pk}" = %s' for pk in pk_cols)
        sql = (
            f"UPDATE {table_sql} SET scd2_end_time=%s, scd2_is_current=0 "
            f"WHERE COALESCE(scd2_is_current,1)=1 AND {where_clause}"
        )
        args_list = [(now, *pk_key) for pk_key in pk_keys]
        execute_batch(cur, sql, args_list, page_size=500)

    def _insert_dim_rows_bulk(
        self,
        cur,
        table: str,
        dwd_cols: Sequence[str],
        rows_with_version: Sequence[tuple[Dict[str, Any], int]],
        now: datetime,
        dwd_types: Dict[str, str] | None = None,
    ) -> None:
        """批量插入新的 SCD2 版本行。"""
        sorted_cols = [c.lower() for c in sorted(dwd_cols)]
        insert_cols_sql = ", ".join(f'"{c}"' for c in sorted_cols)
        table_sql = self._format_table(table, "dwd")
        # 预计算数组类型列集合，避免 list 值被误包装为 Json
        _array_cols: set[str] = set()
        if dwd_types:
            _array_cols = {c for c, t in dwd_types.items() if "ARRAY" in t.upper() or "[]" in t}

        def build_row(src_row: Dict[str, Any], version: int) -> list[Any]:
            values: list[Any] = []
            for c in sorted_cols:
                if c == "scd2_start_time":
                    values.append(now)
                elif c == "scd2_end_time":
                    values.append(datetime(9999, 12, 31, 0, 0, 0))
                elif c == "scd2_is_current":
                    values.append(1)
                elif c == "scd2_version":
                    values.append(version)
                else:
                    val = src_row.get(c)
                    # CHANGE 2026-03-07: 区分数组列和 JSONB 列
                    # 数组列（TEXT[] 等）的 list 值直接传递，psycopg2 自动转为 PG 数组格式
                    # JSONB 列的 dict/list 值需要 Json() 包装
                    if isinstance(val, list) and c not in _array_cols:
                        val = Json(val)
                    elif isinstance(val, dict):
                        val = Json(val)
                    values.append(val)
            return values

        values_rows = [build_row(r, ver) for r, ver in rows_with_version]
        insert_sql = f"INSERT INTO {table_sql} ({insert_cols_sql}) VALUES %s"
        execute_values(cur, insert_sql, values_rows, page_size=500)

    def _is_row_changed(self, current: Dict[str, Any], incoming: Dict[str, Any], dwd_cols: Sequence[str]) -> bool:
        """比较非 SCD2 列，判断是否存在变更。"""
        for col in dwd_cols:
            lc = col.lower()
            if lc in self.SCD_COLS:
                continue
            if not self._values_equal(current.get(lc), incoming.get(lc)):
                return True
        return False

    def _values_equal(self, current_val: Any, incoming_val: Any) -> bool:
        """Normalize common type mismatches (numeric/text, naive/aware datetime) before compare."""
        current_val = self._normalize_empty(current_val)
        incoming_val = self._normalize_empty(incoming_val)
        if current_val is None and incoming_val is None:
            return True

        # 日期时间标准化（朴素时间 vs 时区感知时间）
        if isinstance(current_val, (datetime, date)) or isinstance(incoming_val, (datetime, date)):
            return self._normalize_datetime(current_val) == self._normalize_datetime(incoming_val)

        # 布尔值标准化
        if self._looks_bool(current_val) or self._looks_bool(incoming_val):
            cur_bool = self._coerce_bool(current_val)
            inc_bool = self._coerce_bool(incoming_val)
            if cur_bool is not None and inc_bool is not None:
                return cur_bool == inc_bool

        # 数值标准化（字符串 vs 数值）
        if self._looks_numeric(current_val) or self._looks_numeric(incoming_val):
            cur_num = self._coerce_numeric(current_val)
            inc_num = self._coerce_numeric(incoming_val)
            if cur_num is not None and inc_num is not None:
                return cur_num == inc_num

        return current_val == incoming_val

    def _normalize_empty(self, value: Any) -> Any:
        if isinstance(value, str):
            stripped = value.strip()
            return None if stripped == "" else stripped
        return value

    def _normalize_datetime(self, value: Any) -> Any:
        if value is None:
            return None
        if isinstance(value, date) and not isinstance(value, datetime):
            value = datetime.combine(value, datetime.min.time())
        if not isinstance(value, datetime):
            return value
        try:
            if value.tzinfo is None:
                return value.replace(tzinfo=self.tz)
            return value.astimezone(self.tz)
        except (OverflowError, OSError):
            # 极端日期值（如 9999-12-31）无法转换时区，直接返回原值
            return value

    def _looks_numeric(self, value: Any) -> bool:
        if isinstance(value, (int, float, Decimal)) and not isinstance(value, bool):
            return True
        if isinstance(value, str):
            return bool(self._NUMERIC_RE.match(value.strip()))
        return False

    def _coerce_numeric(self, value: Any) -> Decimal | None:
        value = self._normalize_empty(value)
        if value is None:
            return None
        if isinstance(value, bool):
            return Decimal(int(value))
        if isinstance(value, (int, float, Decimal)):
            try:
                return Decimal(str(value))
            except InvalidOperation:
                return None
        if isinstance(value, str):
            s = value.strip()
            if not self._NUMERIC_RE.match(s):
                return None
            try:
                return Decimal(s)
            except InvalidOperation:
                return None
        return None

    def _looks_bool(self, value: Any) -> bool:
        if isinstance(value, bool):
            return True
        if isinstance(value, str):
            return value.strip().lower() in self._BOOL_STRINGS
        return False

    def _coerce_bool(self, value: Any) -> bool | None:
        value = self._normalize_empty(value)
        if value is None:
            return None
        if isinstance(value, bool):
            return value
        if isinstance(value, (int, Decimal)) and not isinstance(value, bool):
            return bool(int(value))
        if isinstance(value, str):
            s = value.strip().lower()
            if s in {"true", "1", "yes", "y", "t"}:
                return True
            if s in {"false", "0", "no", "n", "f"}:
                return False
        return None

    @staticmethod
    def _count_returning_flags(rows: Iterable[Any]) -> tuple[int, int]:
        """Count inserted vs updated from RETURNING (xmax = 0) rows."""
        inserted = 0
        updated = 0
        for row in rows:
            if isinstance(row, dict):
                flag = row.get("inserted")
            else:
                flag = row[0] if row else None
            if flag:
                inserted += 1
            else:
                updated += 1
        return inserted, updated

    def _merge_fact_increment(
        self,
        cur,
        dwd_table: str,
        ods_table: str,
        dwd_cols: Sequence[str],
        ods_cols: Sequence[str],
        dwd_types: Dict[str, str],
        ods_types: Dict[str, str],
        window_start: datetime,
        window_end: datetime,
    ) -> Dict[str, int]:
        """事实表按时间增量插入，返回真实新增/更新计数。"""
        mapping_entries = self.FACT_MAPPINGS.get(dwd_table) or []
        mapping: Dict[str, tuple[str, str | None]] = {
            dst.lower(): (src, cast_type) for dst, src, cast_type in mapping_entries
        }
        ods_set = {c.lower() for c in ods_cols}
        if "fetched_at" not in ods_set:
            self.logger.error("跳过 %s：ODS 表 %s 缺少 fetched_at 列", dwd_table, ods_table)
            return {"inserted": 0, "updated": 0, "processed": 0}
        self._log_missing_fetched_at(cur, ods_table)
        snapshot_mode = "content_hash" in ods_set
        fact_upsert = bool(self.config.get("dwd.fact_upsert", True))

        mapping_dest = [dst for dst, _, _ in mapping_entries]
        insert_cols: List[str] = list(mapping_dest)
        for col in dwd_cols:
            if col in self.SCD_COLS:
                continue
            if col in insert_cols:
                continue
            if col in ods_cols:
                insert_cols.append(col)

        pk_cols = self._get_primary_keys(cur, dwd_table)
        existing_lower = [c.lower() for c in insert_cols]
        for pk in pk_cols:
            pk_lower = pk.lower()
            if pk_lower in existing_lower:
                continue
            if pk_lower in ods_set:
                insert_cols.append(pk)
                existing_lower.append(pk_lower)
            elif "id" in ods_set:
                insert_cols.append(pk)
                existing_lower.append(pk_lower)
                mapping[pk_lower] = ("id", None)

        # 保持列顺序同时去重
        seen_cols: set[str] = set()
        ordered_cols: list[str] = []
        for col in insert_cols:
            lc = col.lower()
            if lc not in seen_cols:
                seen_cols.add(lc)
                ordered_cols.append(col)
        insert_cols = ordered_cols

        if not insert_cols:
            self.logger.warning("跳过 %s：未找到可插入的列", dwd_table)
            return 0

        # 事实表统一按 fetched_at 做窗口过滤
        order_col = "fetched_at"
        where_sql = ""
        params: List[Any] = []
        dwd_table_sql = self._format_table(dwd_table, "dwd")
        ods_table_sql = self._format_table(ods_table, "ods")
        where_sql = f'WHERE "{order_col}" >= %s AND "{order_col}" < %s'
        params.extend([window_start, window_end])
        where_sql = self._append_where_condition(where_sql, '"fetched_at" IS NOT NULL')

        default_cols = [c for c in insert_cols if c.lower() not in mapping]
        default_expr_map: Dict[str, str] = {}
        if default_cols:
            default_exprs = self._build_fact_select_exprs(default_cols, dwd_types, ods_types)
            default_expr_map = dict(zip(default_cols, default_exprs))

        select_exprs: List[str] = []
        for col in insert_cols:
            key = col.lower()
            if key in mapping:
                src, cast_type = mapping[key]
                select_exprs.append(self._cast_expr(src, cast_type))
            else:
                select_exprs.append(default_expr_map[col])

        select_cols_sql = ", ".join(select_exprs)
        insert_cols_sql = ", ".join(f'"{c}"' for c in insert_cols)
        if snapshot_mode and pk_cols:
            key_exprs: list[str] = []
            for pk in pk_cols:
                pk_lower = pk.lower()
                if pk_lower in mapping:
                    src, cast_type = mapping[pk_lower]
                    key_exprs.append(self._cast_expr(src, cast_type))
                elif pk_lower in ods_set:
                    key_exprs.append(f'"{pk_lower}"')
                elif "id" in ods_set:
                    key_exprs.append('"id"')
            select_sql = self._latest_snapshot_select_sql(
                select_cols_sql,
                ods_table_sql,
                key_exprs,
                order_col,
                where_sql,
            )
        else:
            select_sql = (
                f'SELECT {select_cols_sql} FROM {ods_table_sql} {where_sql}'
            )

        sql = f'INSERT INTO {dwd_table_sql} ({insert_cols_sql}) {select_sql}'

        pk_cols = self._get_primary_keys(cur, dwd_table)
        # force_full_update 时去掉 WHERE IS DISTINCT FROM，无条件 DO UPDATE SET
        force_full = bool(self.config.get("run.force_full_update", False))
        if pk_cols:
            pk_sql = ", ".join(f'"{c}"' for c in pk_cols)
            pk_lower = {c.lower() for c in pk_cols}
            set_exprs = [f'"{c}" = EXCLUDED."{c}"' for c in insert_cols if c.lower() not in pk_lower]
            if snapshot_mode or fact_upsert:
                if set_exprs:
                    if force_full:
                        where_clause = ""
                    else:
                        compare_cols = [c for c in insert_cols if c.lower() not in pk_lower]
                        diff_exprs = [f'{dwd_table_sql}."{c}" IS DISTINCT FROM EXCLUDED."{c}"' for c in compare_cols]
                        where_clause = f" WHERE {' OR '.join(diff_exprs)}" if diff_exprs else ""
                    sql += f" ON CONFLICT ({pk_sql}) DO UPDATE SET {', '.join(set_exprs)}{where_clause}"
                else:
                    sql += f" ON CONFLICT ({pk_sql}) DO NOTHING"
            else:
                sql += f" ON CONFLICT ({pk_sql}) DO NOTHING"

        sql += " RETURNING (xmax = 0) AS inserted"
        cur.execute(sql, params)

        inserted = 0
        updated = 0
        while True:
            rows = cur.fetchmany(10000)
            if not rows:
                break
            ins, upd = self._count_returning_flags(rows)
            inserted += ins
            updated += upd

        return {"inserted": inserted, "updated": updated, "processed": inserted + updated}
    def _build_fact_select_exprs(
        self,
        insert_cols: Sequence[str],
        dwd_types: Dict[str, str],
        ods_types: Dict[str, str],
    ) -> List[str]:
        """构造事实表 SELECT 列表，需要时做类型转换。"""
        numeric_types = {"integer", "bigint", "smallint", "numeric", "double precision", "real", "decimal"}
        text_types = {"text", "character varying", "varchar"}
        ts_types = {"timestamp without time zone", "timestamp with time zone"}
        exprs = []
        for col in insert_cols:
            d_type = dwd_types.get(col)
            o_type = ods_types.get(col)
            if d_type in numeric_types and o_type in text_types:
                exprs.append(f"CAST(NULLIF(CAST(\"{col}\" AS text), '') AS numeric):: {d_type}")
            elif d_type in ts_types and o_type in ts_types:
                # CHANGE 2026-02-22: BUG 12 — 哨兵日期过滤，防止 0001-01-01 转 timestamptz 变 BC
                exprs.append(
                    f'CASE WHEN "{col}" >= \'{self._SENTINEL_DATE_THRESHOLD}\'::timestamp '
                    f'THEN "{col}" ELSE NULL END'
                )
            else:
                exprs.append(f'"{col}"')
        return exprs

    def _split_table_name(self, name: str, default_schema: str) -> tuple[str, str]:
        """拆分 schema.table，若无 schema 则补默认 schema。"""
        parts = name.split(".")
        if len(parts) == 2:
            return parts[0], parts[1].lower()
        return default_schema, name.lower()

    def _table_base(self, name: str) -> str:
        """获取不含 schema 的表名。"""
        return name.split(".")[-1]

    def _format_table(self, name: str, default_schema: str) -> str:
        """返回带引号的 schema.table 名称。"""
        schema, table = self._split_table_name(name, default_schema)
        return f'"{schema}"."{table}"'

    # CHANGE 2026-02-22: BUG 12 fix — 哨兵日期阈值，上游 API 用 0001-01-01 表示"未设置"
    _SENTINEL_DATE_THRESHOLD = "0002-01-01"

    @staticmethod
    def _qualify_column_ref(src: str, alias: str) -> str:
        """为裸列引用添加表别名前缀。

        已包含 detail.、别名前缀、JSON 操作符、表达式（CASE/COALESCE 等）的源不做修改。
        仅对简单列名（如 "col" 或 col）添加 alias."col" 前缀。
        """
        # 已有 detail. 或其他表前缀（含 .）→ 不修改
        if "." in src:
            return src
        # JSON 操作符、SQL 表达式 → 不修改
        if any(tok in src for tok in ("->", "#>>", "::", "CASE ", "COALESCE", "NULLIF", "(")):
            return src
        # 裸列名（可能带引号）→ 加别名前缀
        bare = src.strip('"')
        return f'{alias}."{bare}"'

    def _cast_expr(self, col: str, cast_type: str | None) -> str:
        """构造带可选 CAST 的列表达式。

        对 timestamptz 转换额外包裹哨兵值过滤：ODS 中 0001-01-01 在
        Asia/Shanghai 时区下会变成 BC 日期，psycopg2 无法解析。
        """
        if col.upper() == "NULL":
            base = "NULL"
        else:
            is_expr = not col.isidentifier() or "->" in col or "#>>" in col or "::" in col or "'" in col
            base = col if is_expr else f'"{col}"'
        if cast_type:
            cast_lower = cast_type.lower()
            if cast_lower in {"bigint", "integer", "numeric", "decimal"}:
                return f"CAST(NULLIF(CAST({base} AS text), '') AS numeric):: {cast_type}"
            if cast_lower in {"timestamptz", "timestamp with time zone"}:
                # 哨兵值过滤：< 0002-01-01 的值置为 NULL
                # base 可能是 text（JSONB ->> 提取），需先 CAST 为 timestamp 再比较
                return (f"CASE WHEN ({base})::timestamp >= '{self._SENTINEL_DATE_THRESHOLD}'::timestamp "
                        f"THEN ({base})::timestamptz ELSE NULL END")
            return f"{base}::{cast_type}"
        return base


# AI_CHANGELOG:
# - 日期: 2026-02-14
# - Prompt: P20260214-023000 — "settlement_records 的 settlelist 和 payload 数据重复，删掉 ODS 此字段"
# - 直接原因: ODS 层 settlelist 列被删除后，DWD 加载映射需改为从 payload 提取 settleList
# - 变更摘要: FACT_MAPPINGS 中 dwd_settlement_head_ex 的 settle_list 映射从 ("settlelist", None) 改为 ("payload->'settleList'", None)
# - 风险与验证: payload IS NULL 的行 settle_list 将为 NULL；验证：确认 payload->'settleList' 在 settlement_records 中有 54937 行非空

# AI_CHANGELOG:
# - 日期: 2026-02-14
# - Prompt: P20260214-040000 — "dwd_settlement_head_ex.settle_list 也没有必要保留了"
# - 直接原因: settle_list 列与 ODS payload 中的 settleList 完全冗余，DWD 层无需存储该 jsonb 副本
# - 变更摘要: 删除 FACT_MAPPINGS 中 dwd_settlement_head_ex 的 settle_list 映射行及相关 CHANGE 注释
# - 风险与验证: DWD 装载不再写入 settle_list；验证：information_schema 确认列已删除，DWD_LOAD_FROM_ODS 试运行无报错
