# -*- coding: utf-8 -*-
"""Manual ingestion task that replays archived JSON into ODS tables."""
from __future__ import annotations

import json
import os
from datetime import datetime
from typing import Iterable, Iterator

from .base_task import BaseTask


class ManualIngestTask(BaseTask):
    """
    Load archived API responses (tests/source-data-doc) into billiards_ods.* tables.
    Used when upstream API is unavailable and we need to replay captured payloads.
    """

    # 仅保留已确认的新表映射，其余表待逐一校准后再添加
    FILE_MAPPING: list[tuple[tuple[str, ...], str]] = [
        (("助教账号",), "billiards_ods.assistant_accounts_master"),
        (("助教流水",), "billiards_ods.assistant_service_records"),
        (("助教废除", "助教作废"), "billiards_ods.assistant_cancellation_records"),
        (("库存变化记录1",), "billiards_ods.goods_stock_movements"),
        (("库存汇总", "库存汇总记录"), "billiards_ods.goods_stock_summary"),
        (("团购套餐", "套餐定义"), "billiards_ods.group_buy_packages"),
        (("团购核销", "套餐使用", "团购使用"), "billiards_ods.group_buy_redemption_records"),
        (("余额变动", "余额变更"), "billiards_ods.member_balance_changes"),
        (("会员档案", "会员列表"), "billiards_ods.member_profiles"),
        (("储值卡", "会员储值卡"), "billiards_ods.member_stored_value_cards"),
        (("支付记录", "支付流水"), "billiards_ods.payment_transactions"),
        (("平台验券", "团购验券", "平台券核销"), "billiards_ods.platform_coupon_redemption_records"),
        (("充值结算", "充值记录"), "billiards_ods.recharge_settlements"),
        (("退款流水", "退款记录"), "billiards_ods.refund_transactions"),
        (("结账记录", "结算记录"), "billiards_ods.settlement_records"),
        (("小票详情", "结账小票"), "billiards_ods.settlement_ticket_details"),
        (("台桌列表", "台桌维表"), "billiards_ods.site_tables_master"),
        (("商品分类树", "库存分类"), "billiards_ods.stock_goods_category_tree"),
        (("门店商品档案",), "billiards_ods.store_goods_master"),
        (("门店商品销售", "商品销售流水"), "billiards_ods.store_goods_sales_records"),
        (("台费折扣", "台费调账"), "billiards_ods.table_fee_discount_records"),
        (("台费流水", "台费计费"), "billiards_ods.table_fee_transactions"),
        (("租户商品档案", "商品档案"), "billiards_ods.tenant_goods_master"),
    ]
    WRAPPER_META_KEYS = {"code", "message", "msg", "success", "error", "status"}

    def get_task_code(self) -> str:
        return "MANUAL_INGEST"

    def execute(self) -> dict:
        self.logger.info("Starting Manual Ingest Task")

        data_dir = self.config.get(
            "manual.data_dir",
            r"c:\dev\LLTQ\ETL\feiqiu-ETL\etl_billiards\tests\testdata_json",
        )
        if not os.path.exists(data_dir):
            self.logger.error("Data directory not found: %s", data_dir)
            return {"status": "error", "message": "Directory not found"}

        counts = {"fetched": 0, "inserted": 0, "updated": 0, "skipped": 0, "errors": 0}

        for filename in sorted(os.listdir(data_dir)):
            if not filename.endswith(".json"):
                continue

            filepath = os.path.join(data_dir, filename)
            try:
                with open(filepath, "r", encoding="utf-8") as fh:
                    raw_entries = json.load(fh)
            except Exception:
                counts["errors"] += 1
                self.logger.exception("Failed to read %s", filename)
                continue

            if not isinstance(raw_entries, list):
                raw_entries = [raw_entries]

            records = self._normalize_records(raw_entries)
            if not records:
                counts["skipped"] += 1
                continue

            target_table = self._match_by_filename(filename) or self._match_by_content(
                records, raw_entries
            )
            if not target_table:
                self.logger.warning("No mapping found for file: %s", filename)
                counts["skipped"] += 1
                continue

            self.logger.info("Ingesting %s into %s", filename, target_table)

            try:
                rows = []
                for record in records:
                    site_id = self._extract_store_id(record) or self.config.get(
                        "app.store_id"
                    )
                    pk_value = self._extract_pk(record, target_table)
                    pk_tuple = self._ensure_tuple(pk_value)
                    if not all(value not in (None, "") for value in pk_tuple):
                        continue

                    row = {
                        "site_id": site_id,
                        "payload": json.dumps(record, ensure_ascii=False),
                        "source_file": filename,
                        "fetched_at": datetime.now(),
                    }
                    for column, value in zip(
                        self._get_conflict_columns(target_table), pk_tuple
                    ):
                        row[column] = value
                    self._enrich_row(row, record, target_table)
                    rows.append(row)

                if rows:
                    self._bulk_insert(target_table, rows)
                    counts["inserted"] += len(rows)
                else:
                    counts["skipped"] += 1
                counts["fetched"] += 1

            except Exception:
                counts["errors"] += 1
                self.logger.exception("Error processing %s", filename)
                self.db.rollback()

        try:
            self.db.commit()
        except Exception:
            self.db.rollback()
            raise

        return self._build_result("SUCCESS", counts)

    # ------------------------------------------------------------------ helpers
    def _match_by_filename(self, filename: str) -> str | None:
        for keywords, table in self.FILE_MAPPING:
            if any(keyword and keyword in filename for keyword in keywords):
                return table
        return None

    def _match_by_content(
        self, records: list[dict], raw_entries: list[dict]
    ) -> str | None:
        """
        Map content to PRD ODS tables.
        """
        sample_record = records[0] if records else None
        wrapper = self._extract_sample(raw_entries)
        data_node = wrapper.get("data") if isinstance(wrapper, dict) else None
        data_keys = set(data_node.keys()) if isinstance(data_node, dict) else set()
        record_keys = set(sample_record.keys()) if isinstance(sample_record, dict) else set()

        # Data node based hints
        if "tenantMemberInfos" in data_keys:
            return "billiards_ods.ods_member_profile"
        if "tenantMemberCards" in data_keys:
            return "billiards_ods.ods_member_card"
        if "queryDeliveryRecordsList" in data_keys:
            return "billiards_ods.ods_inventory_change"
        if "goodsStockA" in data_keys or "rangeStartStock" in data_keys:
            return "billiards_ods.ods_inventory_stock"
        if "goodsCategoryList" in data_keys:
            return "billiards_ods.ods_goods_category"
        if "orderAssistantDetails" in data_keys:
            return "billiards_ods.ods_assistant_service_log"
        if "abolitionAssistants" in data_keys:
            return "billiards_ods.ods_assistant_cancel_log"
        if "siteTableUseDetailsList" in data_keys:
            return "billiards_ods.ods_table_use_log"
        if "taiFeeAdjustInfos" in data_keys:
            return "billiards_ods.ods_table_fee_adjust"
        if "orderGoodsLedgers" in data_keys or "orderGoodsList" in data_keys:
            return "billiards_ods.ods_store_sale_item"
        if "tenantGoodsList" in data_keys:
            return "billiards_ods.ods_store_product"
        if "packageCouponList" in data_keys:
            return "billiards_ods.ods_group_package"
        if "settleList" in data_keys and "total" in data_keys:
            return "billiards_ods.ods_order_settle"

        # Record key based hints
        if sample_record:
            if {"pay_amount", "pay_status"} <= record_keys or {"payAmount", "payStatus"} <= record_keys:
                return "billiards_ods.ods_payment_record"
            if "refundAmount" in record_keys or "refund_amount" in record_keys:
                return "billiards_ods.ods_refund_record"
            if "orderSettleId" in record_keys or "order_settle_id" in record_keys:
                return "billiards_ods.ods_order_receipt_detail"
            if "coupon_channel" in record_keys or "groupPackageId" in record_keys:
                return "billiards_ods.ods_platform_coupon_log"
            if "packageId" in record_keys or "package_id" in record_keys:
                return "billiards_ods.ods_group_package_log"
            if "memberCardId" in record_keys or "cardId" in record_keys:
                return "billiards_ods.ods_member_card"
            if "memberId" in record_keys:
                return "billiards_ods.ods_member_profile"
            if "siteGoodsId" in record_keys and "currentStock" in record_keys:
                return "billiards_ods.ods_inventory_stock"
            if "goodsId" in record_keys:
                return "billiards_ods.ods_product"

        return None

    def _extract_sample(self, payloads: Iterable[dict]) -> dict:
        for item in payloads:
            if isinstance(item, dict):
                return item
        return {}

    def _normalize_records(self, payloads: list[dict]) -> list[dict]:
        records: list[dict] = []
        for payload in payloads:
            records.extend(self._unwrap_payload(payload))
        return records

    def _unwrap_payload(self, payload) -> list[dict]:
        if isinstance(payload, dict):
            data_node = payload.get("data")
            extra_keys = set(payload.keys()) - {"data"} - self.WRAPPER_META_KEYS
            if isinstance(data_node, dict) and not extra_keys:
                flattened: list[dict] = []
                found_list = False
                for value in data_node.values():
                    if isinstance(value, list):
                        flattened.extend(value)
                        found_list = True
                if found_list:
                    return flattened
                return [data_node]
            return [payload]

        if isinstance(payload, list):
            flattened: list[dict] = []
            for item in payload:
                flattened.extend(self._unwrap_payload(item))
            return flattened

        return []

    def _extract_store_id(self, item: dict):
        """Extract site_id from record/siteProfile wrappers."""
        site_profile = item.get("siteProfile") or item.get("site_profile")
        if isinstance(site_profile, dict) and site_profile.get("id"):
            return site_profile["id"]

        for key in ("site_id", "siteId", "register_site_id"):
            if item.get(key):
                return item[key]

        data_node = item.get("data")
        if isinstance(data_node, dict):
            return data_node.get("siteId") or data_node.get("site_id")

        return self.config.get("app.store_id")

    def _extract_pk(self, item: dict, table: str):
        if "ods_order_receipt_detail" in table:
            return item.get("orderSettleId") or item.get("order_settle_id") or item.get("id")
        if "ods_order_settle" in table:
            settle = item.get("settleList") or item.get("settle") or item
            if isinstance(settle, dict):
                return settle.get("id") or settle.get("settleId") or item.get("id")
            return item.get("id")

        if "ods_payment_record" in table:
            return item.get("payId") or item.get("id")

        if "ods_refund_record" in table:
            return item.get("refundId") or item.get("id")

        if "ods_platform_coupon_log" in table:
            return item.get("couponId") or item.get("id")

        if "ods_assistant_service_log" in table or "ods_table_use_log" in table:
            return item.get("ledgerId") or item.get("ledger_id") or item.get("id")

        if "ods_assistant_cancel_log" in table:
            return item.get("cancel_id") or item.get("cancelId") or item.get("abolishId") or item.get("id")

        if "ods_store_sale_item" in table:
            return (
                item.get("sale_item_id")
                or item.get("saleItemId")
                or item.get("orderGoodsId")
                or item.get("order_goods_id")
                or item.get("id")
            )

        if "ods_inventory_change" in table:
            return item.get("siteGoodsStockId") or item.get("id")

        if "ods_inventory_stock" in table:
            return (
                item.get("siteGoodsId")
                or item.get("id"),
                item.get("snapshotKey") or item.get("snapshot_key") or "default",
            )

        if "ods_member_card" in table:
            return item.get("cardId") or item.get("memberCardId") or item.get("id")

        if "ods_member_profile" in table:
            return item.get("memberId") or item.get("id")

        if "ods_group_package_log" in table:
            return item.get("usage_id") or item.get("usageId") or item.get("couponId") or item.get("id")

        if "ods_group_package" in table:
            return item.get("package_id") or item.get("packageId") or item.get("groupPackageId") or item.get("id")

        if "ods_goods_category" in table:
            return item.get("category_id") or item.get("categoryId") or item.get("id")

        if "ods_table_fee_adjust" in table:
            return item.get("adjust_id") or item.get("adjustId") or item.get("id")

        if "ods_table_info" in table:
            return item.get("table_id") or item.get("tableId") or item.get("id")

        if "ods_assistant_account" in table:
            return item.get("assistantId") or item.get("assistant_id") or item.get("id")

        if "ods_store_product" in table:
            return item.get("siteGoodsId") or item.get("site_goods_id") or item.get("id")

        if "ods_product" in table:
            return item.get("goodsId") or item.get("goods_id") or item.get("id")

        if "ods_balance_change" in table:
            return item.get("change_id") or item.get("changeId") or item.get("id")

        if "ods_recharge_record" in table:
            return item.get("recharge_id") or item.get("rechargeId") or item.get("id")

        return item.get("id")

    def _get_conflict_columns(self, table: str) -> list[str]:
        if "ods_order_receipt_detail" in table:
            return ["order_settle_id"]
        if "ods_payment_record" in table:
            return ["pay_id"]
        if "ods_refund_record" in table:
            return ["refund_id"]
        if "ods_platform_coupon_log" in table:
            return ["coupon_id"]
        if "ods_assistant_service_log" in table or "ods_table_use_log" in table:
            return ["ledger_id"]
        if "ods_assistant_cancel_log" in table:
            return ["cancel_id"]
        if "ods_store_sale_item" in table:
            return ["sale_item_id"]
        if "ods_order_settle" in table:
            return ["order_settle_id"]
        if "ods_inventory_change" in table:
            return ["change_id"]
        if "ods_inventory_stock" in table:
            return ["site_goods_id", "snapshot_key"]
        if "ods_member_card" in table:
            return ["card_id"]
        if "ods_member_profile" in table:
            return ["member_id"]
        if "ods_group_package_log" in table:
            return ["usage_id"]
        if "ods_group_package" in table:
            return ["package_id"]
        if "ods_goods_category" in table:
            return ["category_id"]
        if "ods_table_info" in table:
            return ["table_id"]
        if "ods_table_fee_adjust" in table:
            return ["adjust_id"]
        if "ods_assistant_account" in table:
            return ["assistant_id"]
        if "ods_store_product" in table:
            return ["site_goods_id"]
        if "ods_product" in table:
            return ["goods_id"]
        if "ods_balance_change" in table:
            return ["change_id"]
        if "ods_recharge_record" in table:
            return ["recharge_id"]
        return ["id"]

    def _enrich_row(self, row: dict, record: dict, table: str):
        """Best-effort populate important columns from payload for PRD ODS schema."""
        def pick(obj, *keys):
            for k in keys:
                if isinstance(obj, dict) and obj.get(k) not in (None, ""):
                    return obj.get(k)
            return None
        def to_bool(value):
            if value in (None, ""):
                return None
            try:
                return bool(int(value))
            except Exception:
                if isinstance(value, bool):
                    return value
                return None
        def to_int(value):
            if value in (None, ""):
                return None
            try:
                return int(value)
            except Exception:
                return None

        if "ods_member_profile" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["system_member_id"] = record.get("system_member_id")
            row["register_site_id"] = record.get("register_site_id")
            row["site_name"] = record.get("site_name")
            row["member_name"] = pick(record, "name", "memberName")
            row["nickname"] = record.get("nickname")
            row["mobile"] = record.get("mobile")
            row["gender"] = record.get("sex")
            row["birthday"] = record.get("birthday")
            row["register_time"] = record.get("register_time") or record.get("registerTime")
            row["member_type_id"] = pick(record, "cardTypeId", "member_type_id")
            row["member_type_name"] = record.get("cardTypeName")
            row["member_card_grade_code"] = record.get("member_card_grade_code")
            row["status"] = pick(record, "status", "state")
            row["user_status"] = record.get("user_status")
            row["balance"] = record.get("balance")
            row["points"] = record.get("points") or record.get("point")
            row["growth_value"] = record.get("growth_value")
            row["last_visit_time"] = record.get("lastVisitTime")
            row["wechat_id"] = record.get("wechatId")
            row["alipay_id"] = record.get("alipayId")
            row["member_card_no"] = record.get("cardNo")
            row["referrer_member_id"] = record.get("referrer_member_id")
            row["remarks"] = record.get("remark")

        if "ods_member_card" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["member_id"] = pick(record, "memberId", "member_id")
            row["tenant_member_id"] = record.get("tenant_member_id")
            row["card_type_id"] = record.get("cardTypeId")
            row["card_type_name"] = record.get("cardTypeName")
            row["card_no"] = record.get("card_no") or record.get("cardNo")
            row["card_physics_type"] = record.get("card_physics_type")
            row["card_balance"] = record.get("balance")
            row["denomination"] = record.get("denomination")
            row["discount_rate"] = record.get("discount") or record.get("discount_rate")
            row["table_discount"] = record.get("table_discount")
            row["goods_discount"] = record.get("goods_discount")
            row["assistant_discount"] = record.get("assistant_discount")
            row["assistant_reward_discount"] = record.get("assistant_reward_discount")
            row["valid_start_date"] = record.get("validStart") or record.get("start_time")
            row["valid_end_date"] = record.get("validEnd") or record.get("end_time")
            row["disable_start_date"] = record.get("disable_start_time")
            row["disable_end_date"] = record.get("disable_end_time")
            row["last_consume_time"] = record.get("lastConsumeTime")
            row["status"] = record.get("status")
            row["is_delete"] = to_bool(record.get("is_delete"))
            row["activate_time"] = record.get("activateTime")
            row["deactivate_time"] = record.get("cancelTime")
            row["register_site_id"] = record.get("register_site_id")
            row["issuer_id"] = record.get("issuerId")
            row["issuer_name"] = record.get("issuerName")

        if "ods_recharge_record" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["member_id"] = pick(record, "memberId", "member_id")
            row["recharge_amount"] = record.get("amount") or record.get("rechargeAmount")
            row["gift_amount"] = record.get("giftAmount")
            row["pay_method"] = record.get("payType") or record.get("pay_method")
            row["pay_trade_no"] = record.get("payTradeNo")
            row["order_trade_no"] = record.get("orderTradeNo")
            row["recharge_time"] = record.get("createTime") or record.get("rechargeTime")
            row["status"] = record.get("status")
            row["operator_id"] = record.get("operatorId")
            row["operator_name"] = record.get("operatorName")

        if "ods_balance_change" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["site_id"] = row.get("site_id") or pick(record, "siteId", "site_id")
            row["member_id"] = pick(record, "memberId", "member_id")
            row["tenant_member_id"] = record.get("tenant_member_id")
            row["tenant_member_card_id"] = record.get("tenant_member_card_id")
            row["member_name"] = record.get("memberName")
            row["member_mobile"] = record.get("memberMobile")
            row["change_amount"] = record.get("change_amount") or record.get("account_data")
            row["balance_before"] = record.get("before_balance") or record.get("before")
            row["balance_after"] = record.get("after_balance") or record.get("after")
            row["change_type"] = record.get("from_type") or record.get("type")
            row["payment_method"] = record.get("payment_method")
            row["refund_amount"] = record.get("refund_amount")
            row["relate_id"] = record.get("relate_id")
            row["register_site_id"] = record.get("register_site_id")
            row["register_site_name"] = record.get("registerSiteName")
            row["pay_site_name"] = record.get("paySiteName")
            row["remark"] = record.get("remark")
            row["operator_id"] = record.get("operatorId")
            row["operator_name"] = record.get("operatorName")
            row["change_time"] = record.get("create_time") or record.get("changeTime")
            row["is_deleted"] = to_bool(record.get("is_delete") or record.get("is_deleted"))
            row["source_file"] = row.get("source_file")
            row["fetched_at"] = row.get("fetched_at")

        if "ods_assistant_account" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["assistant_name"] = record.get("assistantName") or record.get("name")
            row["mobile"] = record.get("mobile")
            row["assistant_no"] = record.get("assistant_no") or record.get("assistantNo")
            row["team_id"] = record.get("teamId")
            row["team_name"] = record.get("teamName")
            row["group_id"] = record.get("group_id")
            row["group_name"] = record.get("group_name")
            row["job_num"] = record.get("job_num")
            row["entry_type"] = record.get("entry_type")
            row["leave_status"] = record.get("leave_status")
            row["assistant_status"] = record.get("assistant_status")
            row["allow_cx"] = to_bool(record.get("allow_cx"))
            row["status"] = record.get("status")
            row["hired_date"] = record.get("hireDate")
            row["left_date"] = record.get("leaveDate")

        if "ods_assistant_service_log" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["assistant_id"] = record.get("assistantId") or record.get("site_assistant_id")
            row["assistant_name"] = record.get("assistantName") or record.get("ledger_name")
            row["service_type"] = record.get("serviceType") or record.get("order_assistant_type")
            row["order_trade_no"] = record.get("orderTradeNo")
            row["order_settle_id"] = record.get("orderSettleId")
            row["start_time"] = record.get("start_use_time") or record.get("startTime") or record.get("ledger_start_time")
            row["end_time"] = record.get("end_time") or record.get("last_use_time") or record.get("ledger_end_time")
            row["duration_seconds"] = record.get("real_use_seconds") or record.get("duration")
            row["original_fee"] = record.get("originFee") or record.get("original_fee") or record.get("ledger_amount")
            row["member_discount_amount"] = record.get("member_discount_amount")
            row["manual_discount_amount"] = record.get("manual_discount_amount")
            row["coupon_discount_amount"] = record.get("coupon_deduct_money")
            row["final_fee"] = record.get("finalFee") or record.get("final_fee") or record.get("service_money")
            row["member_id"] = record.get("memberId") or record.get("tenant_member_id")
            row["operator_id"] = record.get("operator_id")
            row["salesman_id"] = record.get("salesman_user_id")
            row["is_canceled"] = bool(record.get("is_trash"))
            row["cancel_time"] = record.get("trash_time")
            row["skill_grade"] = record.get("skill_grade")
            row["service_grade"] = record.get("service_grade")
            row["composite_grade"] = record.get("composite_grade")
            row["overall_score"] = record.get("sum_grade")
            row["status"] = record.get("status")

        if "ods_assistant_cancel_log" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["ledger_id"] = record.get("ledgerId")
            row["assistant_id"] = record.get("assistantId")
            row["order_trade_no"] = record.get("orderTradeNo")
            row["table_id"] = record.get("tableId")
            row["table_area_id"] = record.get("tableAreaId")
            row["table_area_name"] = record.get("tableArea")
            row["table_name"] = record.get("tableName")
            row["assistant_on"] = record.get("assistantOn")
            row["pd_charge_minutes"] = record.get("pdChargeMinutes")
            row["assistant_abolish_amount"] = record.get("assistantAbolishAmount")
            row["reason"] = record.get("reason")
            row["cancel_time"] = record.get("cancel_time") or record.get("cancelTime")
            row["operator_id"] = record.get("operatorId")
            row["operator_name"] = record.get("operatorName")

        if "ods_table_info" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["table_code"] = record.get("tableCode")
            row["table_name"] = record.get("tableName")
            row["table_type"] = record.get("tableType")
            row["area_name"] = record.get("areaName")
            row["site_table_area_id"] = record.get("site_table_area_id") or record.get("siteTableAreaId")
            row["tenant_table_area_id"] = record.get("tenant_table_area_id") or record.get("tenantTableAreaId")
            row["table_price"] = record.get("table_price") or record.get("tablePrice")
            row["table_status"] = record.get("table_status") or record.get("tableStatusName") or record.get("table_status")
            row["audit_status"] = record.get("audit_status")
            row["show_status"] = record.get("show_status")
            row["light_status"] = record.get("light_status")
            row["virtual_table"] = to_bool(record.get("virtual_table"))
            row["is_rest_area"] = to_bool(record.get("is_rest_area"))
            row["charge_free"] = to_bool(record.get("charge_free"))
            row["table_cloth_use_time"] = record.get("table_cloth_use_time")
            row["table_cloth_use_cycle"] = record.get("table_cloth_use_Cycle")
            row["status"] = record.get("status")
            row["created_time"] = record.get("createTime")
            row["updated_time"] = record.get("updateTime")

        if "ods_table_use_log" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["table_id"] = record.get("tableId") or record.get("site_table_id")
            row["table_name"] = record.get("tableName") or record.get("ledger_name")
            row["order_trade_no"] = record.get("orderTradeNo")
            row["order_settle_id"] = record.get("orderSettleId")
            row["start_time"] = record.get("start_use_time") or record.get("startTime") or record.get("ledger_start_time")
            row["end_time"] = record.get("end_time") or record.get("ledger_end_time")
            row["duration_seconds"] = record.get("real_table_use_seconds") or record.get("duration")
            row["billing_unit_price"] = record.get("ledger_unit_price")
            row["billing_count"] = record.get("ledger_count")
            row["original_table_fee"] = record.get("originFee") or record.get("original_table_fee") or record.get("ledger_amount")
            row["member_discount_amount"] = record.get("member_discount_amount")
            row["coupon_discount_amount"] = record.get("coupon_promotion_amount")
            row["manual_discount_amount"] = record.get("adjust_amount") or record.get("discountAmount")
            row["service_fee"] = record.get("mgmt_fee") or record.get("service_money")
            row["final_table_fee"] = (
                record.get("finalFee")
                or record.get("final_table_fee")
                or record.get("real_table_charge_money")
            )
            row["member_id"] = record.get("memberId")
            row["operator_id"] = record.get("operator_id")
            row["salesman_id"] = record.get("salesman_user_id")
            row["is_canceled"] = bool(record.get("is_delete"))
            row["cancel_time"] = record.get("cancel_time")
            row["site_table_area_id"] = record.get("site_table_area_id")
            row["tenant_table_area_id"] = record.get("tenant_table_area_id")
            row["site_table_area_name"] = record.get("site_table_area_name")
            row["is_single_order"] = to_bool(record.get("is_single_order"))
            row["used_card_amount"] = record.get("used_card_amount")
            row["adjust_amount"] = record.get("adjust_amount")
            row["coupon_promotion_amount"] = record.get("coupon_promotion_amount")
            row["service_money"] = record.get("service_money")
            row["mgmt_fee"] = record.get("mgmt_fee")
            row["fee_total"] = record.get("fee_total")
            row["last_use_time"] = record.get("last_use_time")
            row["ledger_start_time"] = record.get("ledger_start_time")
            row["ledger_end_time"] = record.get("ledger_end_time")
            row["ledger_status"] = record.get("ledger_status")
            row["start_use_time"] = record.get("start_use_time")
            row["add_clock_seconds"] = record.get("add_clock_seconds")
            row["status"] = record.get("status")

        if "ods_table_fee_adjust" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["ledger_id"] = record.get("ledgerId")
            row["order_trade_no"] = record.get("orderTradeNo")
            row["discount_amount"] = record.get("discountAmount")
            row["reason"] = record.get("reason")
            row["operator_id"] = record.get("operatorId")
            row["operator_name"] = record.get("operatorName")
            row["created_at"] = record.get("created_at") or record.get("createTime")

        if "ods_store_product" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["goods_id"] = record.get("goodsId")
            row["goods_name"] = record.get("goodsName")
            row["category_id"] = record.get("categoryId")
            row["category_name"] = record.get("categoryName")
            row["unit"] = record.get("unit")
            row["sale_price"] = record.get("salePrice")
            row["cost_price"] = record.get("costPrice")
            row["sale_num"] = record.get("sale_num")
            row["stock_a"] = record.get("stock_A")
            row["stock"] = record.get("stock")
            row["provisional_total_cost"] = record.get("provisional_total_cost")
            row["total_purchase_cost"] = record.get("total_purchase_cost")
            row["batch_stock_quantity"] = record.get("batch_stock_quantity")
            row["goods_state"] = record.get("goods_state")
            row["status"] = record.get("status")

        if "ods_store_sale_item" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["order_trade_no"] = record.get("orderTradeNo")
            row["order_settle_id"] = record.get("orderSettleId")
            row["order_goods_id"] = record.get("order_goods_id") or record.get("orderGoodsId")
            row["site_goods_id"] = record.get("site_goods_id") or record.get("siteGoodsId")
            row["goods_id"] = record.get("tenant_goods_id") or record.get("goodsId")
            row["goods_name"] = record.get("goodsName") or record.get("ledger_name")
            row["category_id"] = record.get("categoryId") or record.get("tenant_goods_category_id")
            row["quantity"] = record.get("quantity") or record.get("ledger_count")
            row["unit_price"] = record.get("unitPrice") or record.get("ledger_unit_price")
            row["original_amount"] = record.get("originalAmount") or record.get("ledger_amount")
            row["discount_amount"] = record.get("discountAmount") or record.get("discount_money")
            row["final_amount"] = record.get("finalAmount") or record.get("real_goods_money")
            row["discount_price"] = record.get("discount_price")
            row["cost_money"] = record.get("cost_money")
            row["coupon_deduct_amount"] = record.get("coupon_deduct_money")
            row["member_discount_amount"] = record.get("member_discount_amount")
            row["point_discount_money"] = record.get("point_discount_money")
            row["point_discount_cost"] = record.get("point_discount_money_cost")
            row["sales_type"] = record.get("sales_type")
            row["goods_remark"] = record.get("goods_remark")
            row["is_gift"] = to_bool(record.get("isGift"))
            row["sale_time"] = record.get("saleTime") or record.get("create_time")
            row["salesman_id"] = record.get("salesman_user_id")
            row["operator_id"] = record.get("operator_id")
            row["is_refunded"] = to_bool(record.get("is_delete")) or bool(record.get("returns_number"))

        if "ods_group_package_log" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["package_id"] = record.get("packageId")
            row["coupon_id"] = record.get("couponId")
            row["order_trade_no"] = record.get("orderTradeNo")
            row["order_settle_id"] = record.get("orderSettleId")
            row["member_id"] = record.get("memberId")
            row["status"] = record.get("status")
            row["used_time"] = record.get("usedTime")
            row["deduct_amount"] = record.get("deductAmount")
            row["settle_price"] = record.get("settlePrice")
            row["coupon_code"] = record.get("coupon_code") or record.get("couponCode")

        if "ods_group_package" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["package_name"] = record.get("packageName")
            row["table_area_id"] = record.get("table_area_id") or record.get("tableAreaId")
            row["table_area_name"] = record.get("tableAreaName")
            row["platform_code"] = record.get("platformCode")
            row["status"] = record.get("status")
            row["face_price"] = record.get("facePrice")
            row["settle_price"] = record.get("settlePrice")
            row["selling_price"] = record.get("selling_price")
            row["duration"] = record.get("duration")
            row["valid_from"] = record.get("validFrom") or record.get("start_time")
            row["valid_to"] = record.get("validTo") or record.get("end_time")
            row["start_time"] = record.get("start_time")
            row["end_time"] = record.get("end_time")
            row["is_enabled"] = to_bool(record.get("is_enabled"))
            row["is_delete"] = to_bool(record.get("is_delete"))
            row["package_type"] = record.get("type")
            row["usable_count"] = record.get("usable_count")
            row["creator_name"] = record.get("creator_name")
            row["tenant_table_area_id"] = record.get("tenant_table_area_id")

        if "ods_platform_coupon_log" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["platform_code"] = record.get("platformCode")
            row["verify_code"] = record.get("verifyCode")
            row["order_trade_no"] = record.get("orderTradeNo")
            row["order_settle_id"] = record.get("orderSettleId")
            row["member_id"] = record.get("memberId")
            row["status"] = record.get("status")
            row["used_time"] = record.get("usedTime")
            row["deduct_amount"] = record.get("deductAmount")
            row["settle_price"] = record.get("settlePrice")

        if "ods_payment_record" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["order_trade_no"] = record.get("orderTradeNo")
            row["order_settle_id"] = record.get("orderSettleId")
            row["member_id"] = record.get("memberId")
            row["pay_method_code"] = record.get("payMethodCode") or record.get("pay_type") or record.get("payment_method")
            row["pay_method_name"] = record.get("payMethodName")
            row["pay_status"] = record.get("pay_status")
            row["pay_amount"] = record.get("payAmount")
            row["pay_time"] = record.get("payTime")
            row["online_pay_channel"] = record.get("online_pay_channel")
            row["transaction_id"] = record.get("transaction_id")
            row["operator_id"] = record.get("operator_id")
            row["remark"] = record.get("remark")
            row["relate_type"] = record.get("relateType")
            row["relate_id"] = record.get("relateId")

        if "ods_refund_record" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["order_trade_no"] = record.get("orderTradeNo")
            row["order_settle_id"] = record.get("orderSettleId")
            row["member_id"] = record.get("memberId")
            row["pay_sn"] = record.get("pay_sn")
            row["pay_amount"] = record.get("pay_amount")
            row["pay_status"] = record.get("pay_status")
            row["is_revoke"] = to_bool(record.get("is_revoke"))
            row["is_delete"] = to_bool(record.get("is_delete"))
            row["online_pay_channel"] = record.get("online_pay_channel")
            row["pay_method_code"] = record.get("payMethodCode")
            row["refund_amount"] = record.get("refundAmount")
            row["refund_time"] = record.get("refundTime")
            row["action_type"] = record.get("action_type")
            row["pay_terminal"] = record.get("pay_terminal")
            row["pay_config_id"] = record.get("pay_config_id")
            row["cashier_point_id"] = record.get("cashier_point_id")
            row["operator_id"] = record.get("operator_id")
            row["member_card_id"] = record.get("member_card_id")
            row["balance_frozen_amount"] = record.get("balance_frozen_amount")
            row["card_frozen_amount"] = record.get("card_frozen_amount")
            row["round_amount"] = record.get("round_amount")
            row["online_pay_type"] = record.get("online_pay_type")
            row["channel_payer_id"] = record.get("channel_payer_id")
            row["channel_pay_no"] = record.get("channel_pay_no")
            row["check_status"] = record.get("check_status")
            row["channel_fee"] = record.get("channel_fee")
            row["relate_type"] = record.get("relate_type")
            row["relate_id"] = record.get("relate_id")
            row["status"] = record.get("status")
            row["reason"] = record.get("reason")
            row["related_payment_id"] = record.get("related_payment_id") or record.get("payment_id")

        if "ods_inventory_change" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["site_goods_id"] = record.get("siteGoodsId")
            row["goods_id"] = record.get("goodsId")
            row["stock_type"] = record.get("stockType")
            row["change_amount"] = record.get("changeAmount") or record.get("changeNum")
            row["before_stock"] = record.get("beforeStock") or record.get("startNum")
            row["after_stock"] = record.get("afterStock") or record.get("endNum")
            row["change_amount_alt"] = record.get("changeNumA")
            row["before_stock_alt"] = record.get("startNumA")
            row["after_stock_alt"] = record.get("endNumA")
            row["change_type"] = record.get("changeType")
            row["relate_id"] = record.get("relateId")
            row["unit"] = record.get("unit")
            row["price"] = record.get("price")
            row["goods_category_id"] = record.get("goodsCategoryId")
            row["goods_second_category_id"] = record.get("goodsSecondCategoryId")
            row["remark"] = record.get("remark")
            row["operator_id"] = record.get("operatorId")
            row["operator_name"] = record.get("operatorName")
            row["change_time"] = record.get("changeTime") or record.get("createTime")

        if "ods_inventory_stock" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["goods_id"] = record.get("goodsId")
            row["goods_name"] = record.get("goodsName")
            row["goods_unit"] = record.get("goodsUnit")
            row["goods_category_id"] = record.get("goodsCategoryId")
            row["goods_second_category_id"] = record.get("goodsCategorySecondId")
            row["range_start_stock"] = record.get("rangeStartStock")
            row["range_end_stock"] = record.get("rangeEndStock")
            row["range_in"] = record.get("rangeIn")
            row["range_out"] = record.get("rangeOut")
            row["range_inventory"] = record.get("rangeInventory")
            row["range_sale"] = record.get("rangeSale")
            row["range_sale_money"] = record.get("rangeSaleMoney")
            row["current_stock"] = record.get("currentStock")
            row["cost_price"] = record.get("costPrice")
            row["category_name"] = record.get("categoryName")

        if "ods_goods_category" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["category_name"] = record.get("categoryName")
            row["parent_id"] = record.get("parentId")
            row["level_no"] = record.get("levelNo")
            row["status"] = record.get("status")
            row["remark"] = record.get("remark")

        if "ods_order_receipt_detail" in table:
            detail = record
            if isinstance(record.get("data"), dict):
                detail = record["data"].get("data", record["data"])
            row["tenant_id"] = pick(detail, "tenantId", "tenant_id")
            row["order_trade_no"] = detail.get("orderTradeNo")
            row["order_settle_number"] = detail.get("orderSettleNumber")
            row["settle_type"] = to_int(detail.get("settleType"))
            row["receipt_no"] = detail.get("receiptNo")
            row["receipt_time"] = detail.get("receiptTime")
            row["total_amount"] = detail.get("totalAmount") or detail.get("ledgerAmount")
            row["discount_amount"] = detail.get("discountAmount")
            row["final_amount"] = detail.get("finalAmount")
            row["actual_payment"] = detail.get("actualPayment")
            row["ledger_amount"] = detail.get("ledgerAmount")
            row["member_offer_amount"] = detail.get("memberOfferAmount")
            row["delivery_fee"] = detail.get("deliveryFee")
            row["adjust_amount"] = detail.get("adjustAmount")
            row["payment_method"] = detail.get("paymentMethod")
            row["pay_time"] = detail.get("payTime")
            row["member_id"] = detail.get("memberId")
            row["order_remark"] = detail.get("orderRemark")
            row["cashier_name"] = detail.get("cashierName")
            row["ticket_remark"] = detail.get("ticketRemark")
            row["ticket_custom_content"] = detail.get("ticketCustomContent")
            row["voucher_money"] = detail.get("voucherMoney")
            row["reward_name"] = detail.get("rewardName")
            row["consume_money"] = detail.get("consumeMoney")
            row["refund_amount"] = detail.get("refundAmount")
            row["balance_amount"] = detail.get("balanceAmount")
            row["coupon_amount"] = detail.get("couponAmount")
            row["member_deduct_amount"] = detail.get("memberDeductAmount")
            row["prepay_money"] = detail.get("prepayMoney")
            row["delivery_address"] = detail.get("deliveryAddress")
            row["snapshot_raw"] = detail.get("siteProfile") or record.get("site_profile")
            row["member_snapshot"] = detail.get("memberProfile")
            if isinstance(row.get("snapshot_raw"), (dict, list)):
                row["snapshot_raw"] = json.dumps(row["snapshot_raw"], ensure_ascii=False)
            if isinstance(row.get("member_snapshot"), (dict, list)):
                row["member_snapshot"] = json.dumps(row["member_snapshot"], ensure_ascii=False)

        if "ods_order_settle" in table:
            settle_node = record.get("settleList") or record.get("settle")
            if not settle_node and isinstance(record.get("data"), dict):
                settle_node = record["data"].get("settleList") or record["data"].get("settle")
            if isinstance(settle_node, list):
                settle = settle_node[0] if settle_node else {}
            else:
                settle = settle_node or record
            if isinstance(settle, dict):
                row["site_id"] = row.get("site_id") or settle.get("siteId")
                row["tenant_id"] = pick(settle, "tenantId", "tenant_id")
                row["settle_relate_id"] = settle.get("settleRelateId")
                row["settle_name"] = settle.get("settleName")
                row["settle_type"] = settle.get("settleType")
                row["settle_status"] = settle.get("settleStatus")
                row["member_id"] = settle.get("memberId")
                row["member_name"] = settle.get("memberName")
                row["member_phone"] = settle.get("memberPhone")
                row["table_id"] = settle.get("tableId")
                row["consume_money"] = settle.get("consumeMoney")
                row["table_charge_money"] = settle.get("tableChargeMoney")
                row["goods_money"] = settle.get("goodsMoney")
                row["service_money"] = settle.get("serviceMoney")
                row["assistant_pd_money"] = settle.get("assistantPdMoney")
                row["assistant_cx_money"] = settle.get("assistantCxMoney")
                row["pay_amount"] = settle.get("payAmount")
                row["cash_amount"] = settle.get("cashAmount")
                row["online_amount"] = settle.get("onlineAmount")
                row["point_amount"] = settle.get("pointAmount")
                row["coupon_amount"] = settle.get("couponAmount")
                row["card_amount"] = settle.get("cardAmount")
                row["balance_amount"] = settle.get("balanceAmount")
                row["refund_amount"] = settle.get("refundAmount")
                row["prepay_money"] = settle.get("prepayMoney")
                row["adjust_amount"] = settle.get("adjustAmount")
                row["rounding_amount"] = settle.get("roundingAmount")
                row["member_discount_amount"] = settle.get("memberDiscountAmount")
                row["coupon_sale_amount"] = settle.get("couponSaleAmount")
                row["goods_promotion_money"] = settle.get("goodsPromotionMoney")
                row["assistant_promotion_money"] = settle.get("assistantPromotionMoney")
                row["point_discount_price"] = settle.get("pointDiscountPrice")
                row["point_discount_cost"] = settle.get("pointDiscountCost")
                row["real_goods_money"] = settle.get("realGoodsMoney")
                row["assistant_manual_discount"] = settle.get("assistantManualDiscount")
                row["all_coupon_discount"] = settle.get("allCouponDiscount")
                row["is_use_coupon"] = to_bool(settle.get("isUseCoupon"))
                row["is_use_discount"] = to_bool(settle.get("isUseDiscount"))
                row["is_activity"] = to_bool(settle.get("isActivity"))
                row["is_bind_member"] = to_bool(settle.get("isBindMember"))
                row["is_first"] = to_bool(settle.get("isFirst"))
                row["recharge_card_amount"] = settle.get("rechargeCardAmount")
                row["gift_card_amount"] = settle.get("giftCardAmount")
                row["payment_method"] = settle.get("paymentMethod")
                row["create_time"] = settle.get("createTime")
                row["pay_time"] = settle.get("payTime")
                row["revoke_order_id"] = settle.get("revokeOrderId")
                row["revoke_order_name"] = settle.get("revokeOrderName")
                row["revoke_time"] = settle.get("revokeTime")
                row["can_be_revoked"] = settle.get("canBeRevoked")
                row["serial_number"] = settle.get("serialNumber")
                row["sales_man_name"] = settle.get("salesManName")
                row["sales_man_user_id"] = settle.get("salesManUserId")
                row["order_remark"] = settle.get("orderRemark")
                row["operator_id"] = settle.get("operatorId")
                row["operator_name"] = settle.get("operatorName")

        if "ods_product" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["goods_id"] = record.get("goodsId")
            row["goods_name"] = record.get("goodsName")
            row["goods_code"] = record.get("goodsCode")
            row["category_id"] = record.get("categoryId")
            row["category_name"] = record.get("categoryName")
            row["goods_second_category_id"] = record.get("goods_second_category_id") or record.get("goodsSecondCategoryId")
            row["unit"] = record.get("unit")
            row["price"] = record.get("price")
            row["cost_price"] = record.get("cost_price")
            row["market_price"] = record.get("market_price")
            row["goods_state"] = record.get("goods_state")
            row["goods_cover"] = record.get("goods_cover")
            row["goods_bar_code"] = record.get("goods_bar_code")
            row["able_discount"] = record.get("able_discount")
            row["is_delete"] = record.get("is_delete")
            row["status"] = record.get("status")

        if "ods_platform_coupon_log" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")
            row["platform_code"] = record.get("platformCode")
            row["verify_code"] = record.get("verifyCode")
            row["coupon_code"] = record.get("coupon_code")
            row["coupon_channel"] = record.get("coupon_channel")
            row["order_trade_no"] = record.get("orderTradeNo")
            row["order_settle_id"] = record.get("orderSettleId")
            row["member_id"] = record.get("memberId")
            row["status"] = record.get("use_status") or record.get("status")
            row["used_time"] = record.get("consume_time") or record.get("usedTime")
            row["deduct_amount"] = record.get("deductAmount")
            row["settle_price"] = record.get("sale_price") or record.get("settlePrice")
            row["coupon_money"] = record.get("coupon_money")

        if "ods_table_use_log" in table:
            row["tenant_id"] = pick(record, "tenantId", "tenant_id")

    def _ensure_tuple(self, value):
        if isinstance(value, tuple):
            return value
        return (value,)

    def _bulk_insert(self, table: str, rows: list[dict]):
        if not rows:
            return

        columns = list(rows[0].keys())
        col_clause = ", ".join(columns)
        val_clause = ", ".join(f"%({col})s" for col in columns)
        conflict_cols = ["site_id"] + self._get_conflict_columns(table)
        conflict_clause = ", ".join(conflict_cols)

        sql = f"""
            INSERT INTO {table} ({col_clause})
            VALUES ({val_clause})
            ON CONFLICT ({conflict_clause}) DO UPDATE SET
                payload = EXCLUDED.payload,
                fetched_at = EXCLUDED.fetched_at,
                source_file = EXCLUDED.source_file
        """
        self.db.batch_execute(sql, rows)