"""奇成 MySQL 只读查询封装。

接入 kyhl 库,仅 SELECT,强制 READ ONLY 事务防止误写。
所有查询按 ICCID/agent_id 批量分块,避免单条 SQL 过大。
"""
from __future__ import annotations

import contextlib
import time
from dataclasses import dataclass
from decimal import Decimal
from typing import Any, Iterable, Iterator, Optional


# 重构后分阶段索引查询已无大 JOIN,1000 为安全默认值
BATCH_SIZE = 1000


class _StageTimer:
    """阶段耗时计时器,记录迁移各阶段用时。"""

    def __init__(self, func_name: str) -> None:
        self._func = func_name
        self._t = time.perf_counter()

    def log(self, stage: str, **kw: Any) -> None:
        elapsed = time.perf_counter() - self._t
        parts = " ".join(f"{k}={v}" for k, v in kw.items())
        suffix = f" {parts}" if parts else ""
        print(f"[legacy] {self._func}.{stage}:{suffix} elapsed={elapsed:.1f}s")
        self._t = time.perf_counter()


def _normalize_iccid_pairs(iccid_pairs: Iterable[tuple]) -> list[tuple[str, str, bool]]:
    """标准化 ICCID 查询参数。

    tuple 第三项表示是否允许用 iccid_19 查询奇成 tbl_card 主表:
    - 只填 iccid_20 的固定 20 位卡:false,避免 19 位前缀误撞其他卡
    - 显式填了 iccid_19:true,允许作为主表查询键或 20 位缺失时兜底
    """
    out: list[tuple[str, str, bool]] = []
    for item in iccid_pairs:
        i19 = str(item[0] or "").strip() if len(item) > 0 else ""
        i20 = str(item[1] or "").strip() if len(item) > 1 else ""
        allow_i19_lookup = bool(item[2]) if len(item) > 2 else True
        if i19:
            out.append((i19, i20, allow_i19_lookup))
    return out


def _import_pymysql():
    """延迟导入 pymysql,这样脚本1(纯离线)不需要安装 pymysql 也能运行。"""
    try:
        import pymysql
        from pymysql.cursors import DictCursor
        return pymysql, DictCursor
    except ImportError as exc:
        raise ImportError("缺少 pymysql,请先 pip install -r requirements.txt") from exc


@dataclass(frozen=True)
class LegacyPackage:
    """奇成当前生效正式套餐快照(取自 tbl_card_life)。"""

    iccid: str
    meal_id: str
    meal_name: str
    meal_type: str
    start_date: Optional[str]
    expire_date: Optional[str]
    flow_size_mb: int  # 套餐总量,单位 MB
    status: int


@dataclass(frozen=True)
class LegacyPackageLifecycle:
    """奇成正式套餐生命周期记录。"""

    iccid: str
    source_table: str
    life_id: str
    meal_id: str
    meal_name: str
    meal_type: str
    status: int
    start_date: Optional[str]
    expire_date: Optional[str]
    flow_size_mb: int
    stable_sort_key: str
    migration_status: str
    skip_reason: str = ""


@dataclass(frozen=True)
class LegacyCardUsage:
    """奇成卡累计流量快照。

    奇成 ``tbl_card.total_bytes_cnt`` 存的是经过当前套餐 ``flow_add_discount``
    加成后的"虚用量",我们这里同时拉取当前生效正式套餐(``tbl_card_life`` status=1,type!=1
    且 ``expire_date`` 最大)的 ``flow_add_discount``,反算回新系统期望的"真用量"。

    精度说明:跨多个套餐周期累计时,``total_bytes_cnt`` 是按各时段对应套餐
    虚比加权累加的,这里只用"当前生效套餐"的虚比反算,**有误差但可接受**——
    奇成没有按套餐分段的用量明细,无更好做法。
    """

    iccid: str                       # 业务方视角的完整 ICCID
    virtual_used_mb: Decimal         # 奇成 total_bytes_cnt 原始值(虚 MB)
    flow_add_discount_pct: Decimal   # 当前生效套餐虚比百分比(0~100)
    has_active_package: bool         # 是否在 tbl_card_life 中找到当前生效正式套餐

    @property
    def real_used_mb(self) -> Decimal:
        """反算真用量(Decimal MB,保留小数)。

        虚 = 真 × (1 + pct/100) → 真 = 虚 / (1 + pct/100)
        """
        if self.virtual_used_mb <= 0:
            return Decimal("0")
        ratio = Decimal(1) + self.flow_add_discount_pct / Decimal(100)
        return self.virtual_used_mb / ratio

    @property
    def real_used_mb_floor(self) -> int:
        """适配新系统 ``data_usage_mb``(BIGINT)的整数 MB。

        取整规则:0 → 0,(0, 1) → 1,≥1 → 向下取整。
        sub-MB 上取整为 1,避免"明明用过的卡显示 0"。
        """
        real = self.real_used_mb
        if real <= 0:
            return 0
        if real < 1:
            return 1
        return int(real)


@dataclass(frozen=True)
class LegacyCommissionAccount:
    """奇成代理佣金账户快照。"""

    agent_id: str
    agent_name: str
    can_draw_amount_fen: int      # 当前可提现佣金(分)
    applying_draw_amount_fen: int # 申请提现中佣金(分)
    total_commission_amount_fen: int  # 累计总佣金(分,做参考)
    total_draw_amount_fen: int    # 累计已提现(分,做参考)


@dataclass(frozen=True)
class LegacyAgentBalance:
    """奇成代理主钱包余额快照(取自 tbl_agent.balance_money)。"""

    agent_id: str
    balance_fen: int


@dataclass(frozen=True)
class LegacyCardMeta:
    """奇成卡基础元数据(由 scan_legacy / migrate_assets 共享)。

    业务方提供的 iccid_19 + iccid_20 才是新系统的权威值,奇成的 iccid_mark 长度不可信。
    legacy_raw_iccid 仅作调试参考。
    """

    legacy_raw_iccid: str        # 奇成实际存的 iccid_mark(可能 19 或 20 位)
    account_id: str              # tbl_card.account_id(开卡公司/运营商账号 ID)
    account_name: str            # tbl_card.account_name(开卡公司/运营商账号名称)
    category: int                # tbl_card.category(运营商分类编号)
    category_name: str           # tbl_vendor_category.category_name
    agent_id: str
    agent_name: str
    msisdn: str                  # tbl_card.phone
    virtual_no: str              # tbl_virtual_number.vcode(如有)
    current_meal_id: str         # 当前生效正式套餐 ID(可空,取自 tbl_card_life status=1,type!=1,expire_date 最大)
    current_meal_name: str
    current_meal_type: str       # tbl_card_life.type,1=叠加包/加油包,2=月卡,3=季卡,4=半年卡,5=年卡
    current_expire_date: Optional[str]  # ISO 字符串
    current_series_id: str       # 当前生效正式套餐所属奇成套餐系列 ID
    current_series_name: str


@dataclass(frozen=True)
class LegacyMealMeta:
    """奇成套餐 + 系列元数据(由 scan_legacy 用)。"""

    meal_id: str
    meal_name: str
    series_id: str
    series_name: str


# ---------------- 连接管理 ----------------


@contextlib.contextmanager
def connect_readonly(dsn: dict) -> Iterator[Any]:
    """打开只读连接。

    用法:
        with connect_readonly(dsn) as conn:
            rows = fetch_current_packages(conn, iccids)
    """
    pymysql, DictCursor = _import_pymysql()
    conn = pymysql.connect(
        host=dsn["host"],
        port=int(dsn.get("port", 3306)),
        user=dsn["user"],
        password=dsn["password"],
        database=dsn["database"],
        charset=dsn.get("charset", "utf8mb4"),
        connect_timeout=int(dsn.get("connect_timeout", 10)),
        cursorclass=DictCursor,
        autocommit=False,
    )
    try:
        with conn.cursor() as cur:
            cur.execute("SET SESSION TRANSACTION READ ONLY")
            cur.execute("SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ")
        conn.begin()
        yield conn
        conn.rollback()  # 只读事务,回滚即结束
    finally:
        conn.close()


def _chunks(items: list[str], size: int = BATCH_SIZE) -> Iterator[list[str]]:
    for start in range(0, len(items), size):
        yield items[start : start + size]


def _to_fen(value) -> int:
    """元 → 分。奇成的 decimal 是元单位。"""
    if value is None:
        return 0
    try:
        d = Decimal(str(value))
    except Exception:
        return 0
    return int((d * Decimal("100")).quantize(Decimal("1")))


def _to_mb_decimal(value) -> Decimal:
    """字符串/数字 → Decimal MB。奇成 total_bytes_cnt 是小数 MB 读数。"""
    if value is None or value == "":
        return Decimal("0")
    try:
        d = Decimal(str(value))
        if d < 0:
            return Decimal("0")
        return d
    except Exception:
        return Decimal("0")


def _build_card_lookup(
    pairs: list[tuple[str, str, bool]],
) -> dict[str, tuple[str, int]]:
    """构建 tbl_card 查询键 → (完整 ICCID, rank) 映射。

    rank=0 表示完整精确命中,rank=1 表示 19 位兜底命中。
    """
    by_full: dict[str, tuple[str, int]] = {}
    for i19, i20, allow_i19_lookup in pairs:
        card_key = i20 or i19
        by_full[card_key] = (card_key, 0)
        if i20 and allow_i19_lookup:
            by_full[i19] = (card_key, 1)
    return by_full


# ---------------- 查询函数 ----------------


def fetch_current_packages(
    conn,
    iccids: Iterable[str],
    batch_size: int = BATCH_SIZE,
) -> dict[str, LegacyPackage]:
    """查每张卡的当前生效主套餐:status=1、type!=1 且 expire_date 最大。

    返回: iccid → LegacyPackage(找不到的 iccid 不在结果中)
    """
    timer = _StageTimer("fetch_current_packages")
    iccid_list = sorted({i for i in iccids if i})
    out: dict[str, LegacyPackage] = {}
    if not iccid_list:
        return out

    sql = """
        SELECT
            iccid_mark,
            meal_id,
            meal_name,
            COALESCE(CAST(type AS CHAR), '') AS meal_type,
            start_date,
            expire_date,
            COALESCE(flow_size, 0) AS flow_size,
            status
        FROM tbl_card_life
        WHERE iccid_mark IN %s
          AND status = 1
          AND COALESCE(CAST(type AS CHAR), '') <> '1'
        ORDER BY iccid_mark ASC, expire_date DESC
    """
    seen: set[str] = set()
    rows = 0
    batches = 0
    with conn.cursor() as cur:
        for batch in _chunks(iccid_list, batch_size):
            batches += 1
            cur.execute(sql, (tuple(batch),))
            for row in cur.fetchall():
                iccid = row["iccid_mark"]
                rows += 1
                if not iccid or iccid in seen:
                    continue
                seen.add(iccid)
                out[iccid] = LegacyPackage(
                    iccid=iccid,
                    meal_id=row.get("meal_id") or "",
                    meal_name=row.get("meal_name") or "",
                    meal_type=str(row.get("meal_type") or ""),
                    start_date=row["start_date"].strftime("%Y-%m-%d %H:%M:%S") if row.get("start_date") else None,
                    expire_date=row["expire_date"].strftime("%Y-%m-%d %H:%M:%S") if row.get("expire_date") else None,
                    flow_size_mb=int(row.get("flow_size") or 0),
                    status=int(row.get("status") or 0),
                )
    timer.log("tbl_card_life", keys=len(iccid_list), batches=batches, rows=rows, found=len(out))
    return out


def fetch_package_lifecycles(
    conn,
    iccid_pairs: Iterable[tuple],
    batch_size: int = BATCH_SIZE,
) -> dict[str, list[LegacyPackageLifecycle]]:
    """查当前生效和未生效待生效正式套餐生命周期。

    只返回可迁移范围(active/pending)和需要在审核文件中说明的跳过记录。
    加油包(type=1)和已过期记录不生成套餐 SQL,但保留 skip_reason 供审核产物输出。
    """
    timer = _StageTimer("fetch_package_lifecycles")
    pairs = _normalize_iccid_pairs(iccid_pairs)
    out: dict[str, list[LegacyPackageLifecycle]] = {}
    if not pairs:
        return out

    query_to_full: dict[str, str] = {}
    for i19, i20, allow_i19_lookup in pairs:
        card_key = i20 or i19
        query_to_full[card_key] = card_key
        if i20 and allow_i19_lookup:
            query_to_full[i19] = card_key
        if not i20:
            query_to_full[i19] = card_key
    iccid_list = sorted(query_to_full.keys())

    sql = """
        SELECT
            'tbl_card_life' AS source_table,
            COALESCE(CAST(id AS CHAR), '') AS life_id,
            iccid_mark,
            COALESCE(meal_id, '') AS meal_id,
            COALESCE(meal_name, '') AS meal_name,
            COALESCE(CAST(type AS CHAR), '') AS meal_type,
            status,
            start_date,
            expire_date,
            COALESCE(flow_size, 0) AS flow_size
        FROM tbl_card_life
        WHERE iccid_mark IN %s
        ORDER BY
            iccid_mark ASC,
            COALESCE(start_date, '1970-01-01') ASC,
            COALESCE(expire_date, '9999-12-31') ASC,
            id ASC
    """
    rows = 0
    batches = 0
    with conn.cursor() as cur:
        for batch in _chunks(iccid_list, batch_size):
            batches += 1
            cur.execute(sql, (tuple(batch),))
            for row in cur.fetchall():
                legacy_iccid = row.get("iccid_mark") or ""
                if not legacy_iccid:
                    continue
                iccid = query_to_full.get(legacy_iccid)
                if not iccid:
                    continue
                rows += 1
                start = row.get("start_date")
                expire = row.get("expire_date")
                start_str = start.strftime("%Y-%m-%d %H:%M:%S") if start else None
                expire_str = expire.strftime("%Y-%m-%d %H:%M:%S") if expire else None
                meal_type = str(row.get("meal_type") or "")
                raw_status = int(row.get("status") or 0)
                migration_status, skip_reason = _classify_package_lifecycle(meal_type, raw_status, start, expire)
                life_id = str(row.get("life_id") or "")
                stable_sort_key = "|".join([start_str or "", expire_str or "", life_id])
                item = LegacyPackageLifecycle(
                    iccid=iccid,
                    source_table=row.get("source_table") or "tbl_card_life",
                    life_id=life_id,
                    meal_id=row.get("meal_id") or "",
                    meal_name=row.get("meal_name") or "",
                    meal_type=meal_type,
                    status=raw_status,
                    start_date=start_str,
                    expire_date=expire_str,
                    flow_size_mb=int(row.get("flow_size") or 0),
                    stable_sort_key=stable_sort_key,
                    migration_status=migration_status,
                    skip_reason=skip_reason,
                )
                out.setdefault(iccid, []).append(item)
    timer.log("tbl_card_life", keys=len(iccid_list), batches=batches, rows=rows)

    next_sql = """
        SELECT
            'tbl_next_month_card_life' AS source_table,
            COALESCE(CAST(id AS CHAR), '') AS life_id,
            iccid_mark,
            COALESCE(meal_id, '') AS meal_id,
            COALESCE(meal_name, '') AS meal_name,
            COALESCE(CAST(order_type AS CHAR), '') AS meal_type,
            status,
            effect_begin_time AS start_date,
            effect_complete_time AS expire_date,
            COALESCE(flow_size, 0) AS flow_size
        FROM tbl_next_month_card_life
        WHERE iccid_mark IN %s
        ORDER BY
            iccid_mark ASC,
            COALESCE(effect_begin_time, '1970-01-01') ASC,
            COALESCE(effect_complete_time, '9999-12-31') ASC,
            id ASC
    """
    next_rows = 0
    next_batches = 0
    with conn.cursor() as cur:
        for batch in _chunks(iccid_list, batch_size):
            next_batches += 1
            cur.execute(next_sql, (tuple(batch),))
            for row in cur.fetchall():
                legacy_iccid = row.get("iccid_mark") or ""
                if not legacy_iccid:
                    continue
                iccid = query_to_full.get(legacy_iccid)
                if not iccid:
                    continue
                next_rows += 1
                start = row.get("start_date")
                expire = row.get("expire_date")
                start_str = start.strftime("%Y-%m-%d %H:%M:%S") if start else None
                expire_str = expire.strftime("%Y-%m-%d %H:%M:%S") if expire else None
                raw_status = int(row.get("status") or 0)
                migration_status, skip_reason = _classify_next_month_lifecycle(raw_status)
                life_id = str(row.get("life_id") or "")
                stable_sort_key = "|".join([start_str or "", expire_str or "", f"next:{life_id}"])
                out.setdefault(iccid, []).append(LegacyPackageLifecycle(
                    iccid=iccid,
                    source_table=row.get("source_table") or "tbl_next_month_card_life",
                    life_id=life_id,
                    meal_id=row.get("meal_id") or "",
                    meal_name=row.get("meal_name") or "",
                    meal_type=str(row.get("meal_type") or ""),
                    status=raw_status,
                    start_date=start_str,
                    expire_date=expire_str,
                    flow_size_mb=int(row.get("flow_size") or 0),
                    stable_sort_key=stable_sort_key,
                    migration_status=migration_status,
                    skip_reason=skip_reason,
                ))
    timer.log("tbl_next_month_card_life", keys=len(iccid_list), batches=next_batches, rows=next_rows)
    return out


def _classify_package_lifecycle(meal_type: str, status: int, start_date, expire_date) -> tuple[str, str]:
    """把奇成生命周期记录分类为 active/pending/skipped。"""
    if meal_type == "1":
        return "skipped", "加油包不迁移"

    from datetime import datetime

    now = datetime.now()
    if expire_date and expire_date <= now:
        return "skipped", "已过期套餐不迁移"
    if status == 1:
        if start_date and start_date > now:
            return "pending", ""
        return "active", ""
    if start_date and start_date > now:
        return "pending", ""
    return "skipped", f"奇成状态 {status} 不在迁移范围"


def _classify_next_month_lifecycle(status: int) -> tuple[str, str]:
    """把奇成次月待生效套餐记录分类为 pending/skipped。"""
    if status == 1:
        return "pending", ""
    if status == 2:
        return "skipped", "次月待生效记录已插入 tbl_card_life,避免重复迁移"
    return "skipped", f"奇成次月待生效状态 {status} 不在迁移范围"


def fetch_card_usage(
    conn,
    iccid_pairs: Iterable[tuple],
    batch_size: int = BATCH_SIZE,
) -> dict[str, LegacyCardUsage]:
    """查每张卡的累计已用流量,分阶段索引查询替代大 JOIN。

    阶段:
        1. tbl_card — iccid_mark + total_bytes_cnt
        2. tbl_card_life — 当前生效正式套餐 meal_id(Python 选 expire_date 最大)
        3. tbl_set_meal — flow_add_discount(按 meal_id 精确查)
        4. Python 组装 LegacyCardUsage

    Args:
        iccid_pairs: (iccid_19, iccid_20, allow_iccid_19_lookup) 列表。

    Returns:
        {完整 ICCID: LegacyCardUsage}。
    """
    timer = _StageTimer("fetch_card_usage")
    pairs = _normalize_iccid_pairs(iccid_pairs)
    out: dict[str, LegacyCardUsage] = {}
    if not pairs:
        return out

    by_full = _build_card_lookup(pairs)
    full_set = sorted(by_full.keys())

    # ── 阶段1: tbl_card ───────────────────────────────────────────────────────
    card_sql = """
        SELECT
            iccid_mark,
            COALESCE(total_bytes_cnt, 0) AS total_bytes_cnt
        FROM tbl_card
        WHERE iccid_mark IN %s
    """
    hits: dict[str, list[tuple[int, dict]]] = {}
    card_rows = 0
    card_batches = 0
    with conn.cursor() as cur:
        for batch in _chunks(full_set, batch_size):
            card_batches += 1
            cur.execute(card_sql, (tuple(batch),))
            for row in cur.fetchall():
                legacy_iccid = row["iccid_mark"]
                if not legacy_iccid:
                    continue
                matched = by_full.get(legacy_iccid)
                if matched is None:
                    continue
                card_key, rank = matched
                hits.setdefault(card_key, []).append((rank, row))
                card_rows += 1
    timer.log("tbl_card", keys=len(full_set), batches=card_batches, rows=card_rows)

    # 取最优命中行,构建 legacy_iccid 列表
    card_data: dict[str, dict] = {}   # card_key → best tbl_card row
    for card_key, rows in hits.items():
        best_rank = min(r for r, _ in rows)
        card_data[card_key] = next(row for r, row in rows if r == best_rank)

    if not card_data:
        return out

    legacy_iccids = sorted({row["iccid_mark"] for row in card_data.values()})

    # ── 阶段2: tbl_card_life 当前生效正式套餐(精确匹配,Python 选最新) ─────────
    life_sql = """
        SELECT
            iccid_mark,
            COALESCE(meal_id, '') AS meal_id
        FROM tbl_card_life
        WHERE iccid_mark IN %s
          AND status = 1
          AND COALESCE(CAST(type AS CHAR), '') <> '1'
        ORDER BY iccid_mark ASC, expire_date DESC, id DESC
    """
    current_meal_by_iccid: dict[str, str] = {}  # legacy_iccid → meal_id
    life_rows = 0
    life_batches = 0
    with conn.cursor() as cur:
        for batch in _chunks(legacy_iccids, batch_size):
            life_batches += 1
            cur.execute(life_sql, (tuple(batch),))
            for row in cur.fetchall():
                iccid = row["iccid_mark"]
                life_rows += 1
                if iccid not in current_meal_by_iccid:
                    current_meal_by_iccid[iccid] = row.get("meal_id") or ""
    timer.log("tbl_card_life", keys=len(legacy_iccids), batches=life_batches, rows=life_rows)

    # ── 阶段3: tbl_set_meal flow_add_discount(按 distinct meal_id 查) ─────────
    meal_ids = sorted({mid for mid in current_meal_by_iccid.values() if mid})
    discount_by_meal: dict[str, Decimal] = {}
    if meal_ids:
        meal_sql = "SELECT id, COALESCE(flow_add_discount, 0) AS flow_add_discount FROM tbl_set_meal WHERE id IN %s"
        meal_batches = 0
        with conn.cursor() as cur:
            for batch in _chunks(meal_ids, batch_size):
                meal_batches += 1
                cur.execute(meal_sql, (tuple(batch),))
                for row in cur.fetchall():
                    mid = str(row["id"] or "")
                    if mid:
                        discount_by_meal[mid] = _to_mb_decimal(row.get("flow_add_discount"))
        timer.log("tbl_set_meal", meal_ids=len(meal_ids), batches=meal_batches)
    else:
        timer.log("tbl_set_meal", meal_ids=0, skipped=True)

    # ── 阶段4: Python 组装 ────────────────────────────────────────────────────
    for card_key, base_row in card_data.items():
        legacy_iccid = base_row["iccid_mark"]
        meal_id = current_meal_by_iccid.get(legacy_iccid, "")
        has_active = bool(meal_id or legacy_iccid in current_meal_by_iccid)
        discount = discount_by_meal.get(str(meal_id), Decimal("0")) if meal_id else Decimal("0")
        out[card_key] = LegacyCardUsage(
            iccid=card_key,
            virtual_used_mb=_to_mb_decimal(base_row.get("total_bytes_cnt")),
            flow_add_discount_pct=discount,
            has_active_package=has_active,
        )
    timer.log("assemble", cards=len(out))
    return out


def fetch_commission_accounts(
    conn,
    agent_ids: Iterable[str],
    batch_size: int = BATCH_SIZE,
) -> dict[str, LegacyCommissionAccount]:
    """查代理佣金账户。"""
    timer = _StageTimer("fetch_commission_accounts")
    agent_list = sorted({a for a in agent_ids if a})
    out: dict[str, LegacyCommissionAccount] = {}
    if not agent_list:
        return out

    sql = """
        SELECT
            agent_id, agent_name,
            can_draw_amount, applying_draw_amount,
            total_commision_amount, total_draw_amount
        FROM tbl_agent_commission_account
        WHERE agent_id IN %s
    """
    rows = 0
    batches = 0
    with conn.cursor() as cur:
        for batch in _chunks(agent_list, batch_size):
            batches += 1
            cur.execute(sql, (tuple(batch),))
            for row in cur.fetchall():
                aid = row["agent_id"]
                rows += 1
                if not aid:
                    continue
                out[aid] = LegacyCommissionAccount(
                    agent_id=aid,
                    agent_name=row.get("agent_name") or "",
                    can_draw_amount_fen=_to_fen(row.get("can_draw_amount")),
                    applying_draw_amount_fen=_to_fen(row.get("applying_draw_amount")),
                    total_commission_amount_fen=_to_fen(row.get("total_commision_amount")),
                    total_draw_amount_fen=_to_fen(row.get("total_draw_amount")),
                )
    timer.log("tbl_agent_commission_account", agents=len(agent_list), batches=batches, rows=rows)
    return out


def fetch_card_meta(
    conn,
    iccid_pairs: Iterable[tuple],
    batch_size: int = BATCH_SIZE,
) -> tuple[dict[str, LegacyCardMeta], set[str]]:
    """查每张卡的元数据,分阶段索引查询替代大 JOIN。

    阶段:
        1. tbl_card — 卡基础字段 + 去重检测
        2. tbl_vendor_category — 运营商分类名称(按 distinct category 查)
        3. tbl_card_life — 当前生效正式套餐(Python 选 expire_date 最大,精确匹配)
        4. tbl_set_meal + tbl_set_meal_series — 套餐系列(按 distinct meal_id 查)
        5. tbl_virtual_number — 虚拟号(精确匹配,不做 LEFT() 前缀扫描)
        6. Python 组装 LegacyCardMeta

    匹配策略:
        - tbl_card 主表同时查优先键和显式允许的兜底键;20 位精确命中优先
        - 从表(card_life / virtual_number)用 tbl_card 实际返回的 iccid_mark 精确查询
        - 不再使用 LEFT(vn.iccid_mark, 19) 前缀匹配,避免全表扫描

    Args:
        iccid_pairs: (iccid_19, iccid_20, allow_iccid_19_lookup) 列表。

    Returns:
        (metas, duplicate_iccids):
            - metas: {完整 ICCID: LegacyCardMeta}
            - duplicate_iccids: 一份业务方 ICCID 命中多条 tbl_card 记录的 set
    """
    timer = _StageTimer("fetch_card_meta")
    pairs = _normalize_iccid_pairs(iccid_pairs)
    if not pairs:
        return {}, set()

    by_full = _build_card_lookup(pairs)
    full_set = sorted(by_full.keys())

    # ── 阶段1: tbl_card ───────────────────────────────────────────────────────
    card_sql = """
        SELECT
            id                             AS card_id,
            iccid_mark,
            COALESCE(account_id, '')        AS account_id,
            COALESCE(account_name, '')      AS account_name,
            COALESCE(category, 0)           AS category,
            COALESCE(agent_id, '')          AS agent_id,
            COALESCE(agent_name, '')        AS agent_name,
            COALESCE(phone, '')             AS phone
        FROM tbl_card
        WHERE iccid_mark IN %s
    """
    hits: dict[str, list[tuple[int, dict]]] = {}
    card_rows = 0
    card_batches = 0
    with conn.cursor() as cur:
        for batch in _chunks(full_set, batch_size):
            card_batches += 1
            cur.execute(card_sql, (tuple(batch),))
            for row in cur.fetchall():
                legacy_iccid = row["iccid_mark"]
                if not legacy_iccid:
                    continue
                matched = by_full.get(legacy_iccid)
                if matched is None:
                    continue
                card_key, rank = matched
                hits.setdefault(card_key, []).append((rank, row))
                card_rows += 1
    timer.log("tbl_card", keys=len(full_set), batches=card_batches, rows=card_rows)

    # 去重:同一完整 ICCID 命中多条不同 card_id → duplicates
    metas_base: dict[str, dict] = {}   # card_key → best tbl_card row
    duplicates: set[str] = set()
    categories: set[int] = set()

    for card_key, rows in hits.items():
        best_rank = min(r for r, _ in rows)
        best_rows = [row for r, row in rows if r == best_rank]
        rows_by_card_id: dict[str, dict] = {}
        for row in best_rows:
            rows_by_card_id[str(row.get("card_id") or row.get("iccid_mark") or "")] = row
        if len(rows_by_card_id) > 1:
            duplicates.add(card_key)
            continue
        row = next(iter(rows_by_card_id.values()))
        metas_base[card_key] = row
        categories.add(int(row.get("category") or 0))

    if not metas_base:
        return {}, duplicates

    # legacy_iccids: tbl_card 实际存储的 iccid_mark 值(用于从表精确查询)
    legacy_iccids = sorted({row["iccid_mark"] for row in metas_base.values()})

    # ── 阶段2: tbl_vendor_category(distinct category,通常极少几十个) ─────────
    category_names: dict[int, str] = {}
    cat_batches = 0
    if categories:
        cat_sql = """
            SELECT category, COALESCE(category_name, '') AS category_name
            FROM tbl_vendor_category
            WHERE category IN %s
        """
        cat_list = sorted(categories)
        with conn.cursor() as cur:
            for batch in _chunks(cat_list, batch_size):
                cat_batches += 1
                cur.execute(cat_sql, (tuple(batch),))
                for row in cur.fetchall():
                    category_names[int(row["category"])] = row.get("category_name") or ""
    timer.log("tbl_vendor_category", categories=len(categories), batches=cat_batches)

    # ── 阶段3: tbl_card_life 当前生效正式套餐(精确匹配,Python 选最新) ─────────
    # 精确用 legacy_iccid 查询;不再做 LEFT(iccid_mark, 19) 前缀扫描
    life_sql = """
        SELECT
            iccid_mark,
            COALESCE(meal_id, '')            AS meal_id,
            COALESCE(meal_name, '')          AS meal_name,
            COALESCE(CAST(type AS CHAR), '') AS meal_type,
            expire_date
        FROM tbl_card_life
        WHERE iccid_mark IN %s
          AND status = 1
          AND COALESCE(CAST(type AS CHAR), '') <> '1'
        ORDER BY iccid_mark ASC, expire_date DESC, id DESC
    """
    current_meals: dict[str, dict] = {}  # legacy_iccid → 当前生效套餐行
    meal_ids: set[str] = set()
    life_rows = 0
    life_batches = 0
    with conn.cursor() as cur:
        for batch in _chunks(legacy_iccids, batch_size):
            life_batches += 1
            cur.execute(life_sql, (tuple(batch),))
            for row in cur.fetchall():
                iccid = row["iccid_mark"]
                life_rows += 1
                if iccid not in current_meals:
                    current_meals[iccid] = row
                    if row.get("meal_id"):
                        meal_ids.add(row["meal_id"])
    timer.log("tbl_card_life", keys=len(legacy_iccids), batches=life_batches, rows=life_rows, found=len(current_meals))

    # ── 阶段4: tbl_set_meal + tbl_set_meal_series(按 distinct meal_id) ───────
    meal_series: dict[str, tuple[str, str]] = {}  # meal_id → (series_id, series_name)
    meal_batches = 0
    if meal_ids:
        meal_sql = """
            SELECT
                sm.id                            AS meal_id,
                COALESCE(sm.meal_series_id, '')  AS series_id,
                COALESCE(sms.name, '')           AS series_name
            FROM tbl_set_meal sm
            LEFT JOIN tbl_set_meal_series sms ON sms.id = sm.meal_series_id
            WHERE sm.id IN %s
        """
        meal_id_list = sorted(meal_ids)
        with conn.cursor() as cur:
            for batch in _chunks(meal_id_list, batch_size):
                meal_batches += 1
                cur.execute(meal_sql, (tuple(batch),))
                for row in cur.fetchall():
                    mid = str(row["meal_id"] or "")
                    if mid:
                        meal_series[mid] = (row.get("series_id") or "", row.get("series_name") or "")
    timer.log("tbl_set_meal", meal_ids=len(meal_ids), batches=meal_batches)

    # ── 阶段5: tbl_virtual_number(精确匹配,不做前缀扫描) ─────────────────────
    virtual_nos: dict[str, str] = {}  # legacy_iccid → vcode
    vn_rows = 0
    vn_batches = 0
    with conn.cursor() as cur:
        for batch in _chunks(legacy_iccids, batch_size):
            vn_batches += 1
            cur.execute(
                "SELECT iccid_mark, COALESCE(vcode, '') AS vcode FROM tbl_virtual_number WHERE iccid_mark IN %s",
                (tuple(batch),),
            )
            for row in cur.fetchall():
                iccid = row["iccid_mark"]
                vn_rows += 1
                if iccid:
                    virtual_nos[iccid] = row.get("vcode") or ""
    timer.log("tbl_virtual_number", keys=len(legacy_iccids), batches=vn_batches, rows=vn_rows)

    # ── 阶段6: Python 组装 ────────────────────────────────────────────────────
    metas: dict[str, LegacyCardMeta] = {}
    for card_key, base_row in metas_base.items():
        legacy_iccid = base_row["iccid_mark"]
        meal_row = current_meals.get(legacy_iccid, {})
        expire = meal_row.get("expire_date")
        expire_str = expire.strftime("%Y-%m-%d %H:%M:%S") if expire else None
        meal_id = meal_row.get("meal_id") or ""
        series_id, series_name = meal_series.get(str(meal_id), ("", ""))
        metas[card_key] = LegacyCardMeta(
            legacy_raw_iccid=legacy_iccid,
            account_id=base_row.get("account_id") or "",
            account_name=base_row.get("account_name") or "",
            category=int(base_row.get("category") or 0),
            category_name=category_names.get(int(base_row.get("category") or 0), ""),
            agent_id=base_row.get("agent_id") or "",
            agent_name=base_row.get("agent_name") or "",
            msisdn=base_row.get("phone") or "",
            virtual_no=virtual_nos.get(legacy_iccid, ""),
            current_meal_id=meal_id,
            current_meal_name=meal_row.get("meal_name") or "",
            current_meal_type=str(meal_row.get("meal_type") or ""),
            current_expire_date=expire_str,
            current_series_id=series_id,
            current_series_name=series_name,
        )
    timer.log("assemble", cards=len(metas), duplicates=len(duplicates))
    return metas, duplicates


def fetch_meal_meta(
    conn,
    meal_ids: Iterable[str],
    batch_size: int = BATCH_SIZE,
) -> dict[str, LegacyMealMeta]:
    """查套餐 + 系列元数据(供 scan_legacy 收集映射 legacy_* 字段)。"""
    timer = _StageTimer("fetch_meal_meta")
    id_list = sorted({m for m in meal_ids if m})
    out: dict[str, LegacyMealMeta] = {}
    if not id_list:
        return out
    sql = """
        SELECT
            sm.id                            AS meal_id,
            COALESCE(sm.name, '')            AS meal_name,
            COALESCE(sm.meal_series_id, '')  AS series_id,
            COALESCE(sms.name, '')           AS series_name
        FROM tbl_set_meal sm
        LEFT JOIN tbl_set_meal_series sms ON sms.id = sm.meal_series_id
        WHERE sm.id IN %s
    """
    rows = 0
    batches = 0
    with conn.cursor() as cur:
        for batch in _chunks(id_list, batch_size):
            batches += 1
            cur.execute(sql, (tuple(batch),))
            for row in cur.fetchall():
                mid = row["meal_id"]
                rows += 1
                if not mid:
                    continue
                out[mid] = LegacyMealMeta(
                    meal_id=mid,
                    meal_name=row.get("meal_name") or "",
                    series_id=row.get("series_id") or "",
                    series_name=row.get("series_name") or "",
                )
    timer.log("tbl_set_meal", meal_ids=len(id_list), batches=batches, rows=rows)
    return out


def fetch_agent_balances(
    conn,
    agent_ids: Iterable[str],
    batch_size: int = BATCH_SIZE,
) -> dict[str, LegacyAgentBalance]:
    """查代理主钱包余额(tbl_agent.balance_money,元单位)。"""
    timer = _StageTimer("fetch_agent_balances")
    agent_list = sorted({a for a in agent_ids if a})
    out: dict[str, LegacyAgentBalance] = {}
    if not agent_list:
        return out

    sql = """
        SELECT id, balance_money
        FROM tbl_agent
        WHERE id IN %s
    """
    rows = 0
    batches = 0
    with conn.cursor() as cur:
        for batch in _chunks(agent_list, batch_size):
            batches += 1
            cur.execute(sql, (tuple(batch),))
            for row in cur.fetchall():
                aid = row["id"]
                rows += 1
                if not aid:
                    continue
                out[aid] = LegacyAgentBalance(agent_id=aid, balance_fen=_to_fen(row.get("balance_money")))
    timer.log("tbl_agent", agents=len(agent_list), batches=batches, rows=rows)
    return out