优化迁移脚本速度

scripts/migration/lib/legacy_query.py

@@ -6,12 +6,29 @@
 from __future__ import annotations
 
 import contextlib
+import time
 from dataclasses import dataclass
 from decimal import Decimal
 from typing import Any, Iterable, Iterator, Optional
 
 
-BATCH_SIZE = 200
+# 重构后分阶段索引查询已无大 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]]:
@@ -236,14 +253,35 @@ def _to_mb_decimal(value) -> Decimal:
         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]) -> dict[str, LegacyPackage]:
+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:
@@ -266,11 +304,15 @@ def fetch_current_packages(conn, iccids: Iterable[str]) -> dict[str, LegacyPacka
         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):
+        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)
@@ -284,15 +326,21 @@ def fetch_current_packages(conn, iccids: Iterable[str]) -> dict[str, LegacyPacka
                     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]) -> dict[str, list[LegacyPackageLifecycle]]:
+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:
@@ -328,8 +376,11 @@ def fetch_package_lifecycles(conn, iccid_pairs: Iterable[tuple]) -> dict[str, li
             COALESCE(expire_date, '9999-12-31') ASC,
             id ASC
     """
+    rows = 0
+    batches = 0
     with conn.cursor() as cur:
-        for batch in _chunks(iccid_list):
+        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 ""
@@ -338,6 +389,7 @@ def fetch_package_lifecycles(conn, iccid_pairs: Iterable[tuple]) -> dict[str, li
                 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
@@ -363,6 +415,7 @@ def fetch_package_lifecycles(conn, iccid_pairs: Iterable[tuple]) -> dict[str, li
                     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
@@ -384,8 +437,11 @@ def fetch_package_lifecycles(conn, iccid_pairs: Iterable[tuple]) -> dict[str, li
             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):
+        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 ""
@@ -394,6 +450,7 @@ def fetch_package_lifecycles(conn, iccid_pairs: Iterable[tuple]) -> dict[str, li
                 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
@@ -417,6 +474,7 @@ def fetch_package_lifecycles(conn, iccid_pairs: Iterable[tuple]) -> dict[str, li
                     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
 
 
@@ -448,13 +506,18 @@ def _classify_next_month_lifecycle(status: int) -> tuple[str, str]:
     return "skipped", f"奇成次月待生效状态 {status} 不在迁移范围"
 
 
-def fetch_card_usage(conn, iccid_pairs: Iterable[tuple]) -> dict[str, LegacyCardUsage]:
-    """查每张卡的累计已用流量,并同步取当前生效正式套餐的 flow_add_discount。
+def fetch_card_usage(
+    conn,
+    iccid_pairs: Iterable[tuple],
+    batch_size: int = BATCH_SIZE,
+) -> dict[str, LegacyCardUsage]:
+    """查每张卡的累计已用流量,分阶段索引查询替代大 JOIN。
 
-    奇成 ``tbl_card.total_bytes_cnt`` 是虚用量(已被套餐虚比加成)。
-    这里 JOIN 当前生效正式套餐(``tbl_card_life`` status=1,type!=1 且 ``expire_date`` 最大)
-    + ``tbl_set_meal`` 取 ``flow_add_discount``,放进 ``LegacyCardUsage``;
-    上层通过 ``real_used_mb_floor`` 反算真用量写入新系统。
+    阶段:
+        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) 列表。
@@ -462,88 +525,119 @@ def fetch_card_usage(conn, iccid_pairs: Iterable[tuple]) -> dict[str, LegacyCard
     Returns:
         {完整 ICCID: LegacyCardUsage}。
     """
+    timer = _StageTimer("fetch_card_usage")
     pairs = _normalize_iccid_pairs(iccid_pairs)
     out: dict[str, LegacyCardUsage] = {}
     if not pairs:
         return out
 
-    query_to_full: dict[str, tuple[str, int]] = {}
-    related_keys_by_full: dict[str, set[str]] = {}
-    for i19, i20, allow_i19_lookup in pairs:
-        card_key = i20 or i19
-        # tbl_card 主表同时查优先键和兜底键,结果选择时按 rank:
-        # 0=业务方完整 ICCID 精确命中,1=20 位卡在奇成主表只存 19 位时兜底。
-        query_to_full[card_key] = (card_key, 0)
-        if i20 and allow_i19_lookup:
-            query_to_full[i19] = (card_key, 1)
-        if not i20 or allow_i19_lookup:
-            related_keys_by_full.setdefault(card_key, set()).add(i19)
-        else:
-            related_keys_by_full.setdefault(card_key, set())
-        if i20:
-            related_keys_by_full[card_key].add(i20)
-    full_set = sorted(query_to_full.keys())
+    by_full = _build_card_lookup(pairs)
+    full_set = sorted(by_full.keys())
 
-    # tbl_card 主表用 iccid_mark IN 全长精确匹配;从表只在主表本身是 19 位时
-    # 按前缀兜底,避免同前缀 20 位卡互相串套餐。
-    sql = """
+    # ── 阶段1: tbl_card ───────────────────────────────────────────────────────
+    card_sql = """
         SELECT
-            c.iccid_mark,
-            c.total_bytes_cnt,
-            COALESCE(sm.flow_add_discount, 0) AS flow_add_discount,
-            CASE WHEN cl.iccid_mark IS NOT NULL THEN 1 ELSE 0 END AS has_active_package
-        FROM tbl_card c
-        LEFT JOIN (
-            SELECT cl1.iccid_mark, cl1.meal_id
-            FROM tbl_card_life cl1
-            INNER JOIN (
-                SELECT iccid_mark, MAX(expire_date) AS max_expire
-                FROM tbl_card_life
-                WHERE status = 1
-                  AND COALESCE(CAST(type AS CHAR), '') <> '1'
-                  AND iccid_mark IN %s
-                GROUP BY iccid_mark
-            ) latest
-              ON latest.iccid_mark = cl1.iccid_mark
-             AND latest.max_expire = cl1.expire_date
-            WHERE cl1.status = 1
-              AND COALESCE(CAST(cl1.type AS CHAR), '') <> '1'
-        ) cl ON (
-            cl.iccid_mark = c.iccid_mark
-            OR (CHAR_LENGTH(c.iccid_mark) = 19 AND LEFT(cl.iccid_mark, 19) = c.iccid_mark)
-        )
-        LEFT JOIN tbl_set_meal sm ON sm.id = cl.meal_id
-        WHERE c.iccid_mark IN %s
+            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_keys = {query_to_full[x][0] for x in batch}
-            related_batch = sorted({x for key in batch_keys for x in related_keys_by_full[key]})
-            cur.execute(sql, (tuple(related_batch), tuple(batch)))
+        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 = query_to_full.get(legacy_iccid)
+                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(rank for rank, _row in rows)
-        row = next(row for rank, row in rows if rank == best_rank)
+        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(row.get("total_bytes_cnt")),
-            flow_add_discount_pct=_to_mb_decimal(row.get("flow_add_discount")),
-            has_active_package=bool(row.get("has_active_package")),
+            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]) -> dict[str, LegacyCommissionAccount]:
+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:
@@ -557,11 +651,15 @@ def fetch_commission_accounts(conn, agent_ids: Iterable[str]) -> dict[str, Legac
         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):
+        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(
@@ -572,107 +670,69 @@ def fetch_commission_accounts(conn, agent_ids: Iterable[str]) -> dict[str, Legac
                     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 主表同时查优先键和显式允许的兜底键:有 iccid_20 时优先取
-          20 位精确命中;只有业务方也显式填了 iccid_19 时,才允许退回 19 位。
-          只填 iccid_20 的固定 20 位卡绝不拿 19 位前缀查主表。
-        - tbl_card_life / tbl_virtual_number 从表在主表本身是 19 位时才用
-          LEFT(iccid_mark, 19) 兜底,避免同前缀 20 位卡互相串数据
+        - 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,
-              这些卡不进 metas,sql_builder 会写 errors.csv 阻断
+            - 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: dict[str, tuple[str, int]] = {}     # tbl_card 查询键 → (完整 ICCID, rank)
-    related_keys_by_full: dict[str, set[str]] = {}
-    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)
-        if not i20 or allow_i19_lookup:
-            related_keys_by_full.setdefault(card_key, set()).add(i19)
-        else:
-            related_keys_by_full.setdefault(card_key, set())
-        if i20:
-            related_keys_by_full[card_key].add(i20)
+    by_full = _build_card_lookup(pairs)
     full_set = sorted(by_full.keys())
 
-    sql = """
+    # ── 阶段1: tbl_card ───────────────────────────────────────────────────────
+    card_sql = """
         SELECT
-            c.id                       AS card_id,
-            c.iccid_mark               AS iccid,
-            COALESCE(c.account_id, '')  AS account_id,
-            COALESCE(c.account_name, '') AS account_name,
-            COALESCE(c.category, 0)    AS category,
-            COALESCE(vc.category_name, '') AS category_name,
-            COALESCE(c.agent_id, '')   AS agent_id,
-            COALESCE(c.agent_name, '') AS agent_name,
-            COALESCE(c.phone, '')      AS phone,
-            COALESCE(vn.vcode, '')     AS vcode,
-            COALESCE(cl.meal_id, '')   AS meal_id,
-            COALESCE(cl.meal_name, '') AS meal_name,
-            COALESCE(CAST(cl.meal_type AS CHAR), '') AS meal_type,
-            cl.expire_date             AS expire_date,
-            COALESCE(sm.meal_series_id, '') AS series_id,
-            COALESCE(sms.name, '')      AS series_name
-        FROM tbl_card c
-        LEFT JOIN tbl_vendor_category vc ON vc.category = c.category
-        LEFT JOIN tbl_virtual_number vn ON (
-            vn.iccid_mark = c.iccid_mark
-            OR (CHAR_LENGTH(c.iccid_mark) = 19 AND LEFT(vn.iccid_mark, 19) = c.iccid_mark)
-        )
-        LEFT JOIN (
-            SELECT cl1.iccid_mark, cl1.meal_id, cl1.meal_name, cl1.type AS meal_type, cl1.expire_date
-            FROM tbl_card_life cl1
-            INNER JOIN (
-                SELECT iccid_mark, MAX(expire_date) AS max_expire
-                FROM tbl_card_life
-                WHERE status = 1
-                  AND COALESCE(CAST(type AS CHAR), '') <> '1'
-                  AND iccid_mark IN %s
-                GROUP BY iccid_mark
-            ) latest
-              ON latest.iccid_mark = cl1.iccid_mark
-             AND latest.max_expire = cl1.expire_date
-            WHERE cl1.status = 1
-              AND COALESCE(CAST(cl1.type AS CHAR), '') <> '1'
-        ) cl ON (
-            cl.iccid_mark = c.iccid_mark
-            OR (CHAR_LENGTH(c.iccid_mark) = 19 AND LEFT(cl.iccid_mark, 19) = c.iccid_mark)
-        )
-        LEFT JOIN tbl_set_meal sm ON sm.id = cl.meal_id
-        LEFT JOIN tbl_set_meal_series sms ON sms.id = sm.meal_series_id
-        WHERE c.iccid_mark IN %s
+            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
     """
-
-    # 收集每个完整 ICCID 命中的奇成行(可能多条 = 脏数据)
     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_keys = {by_full[x][0] for x in batch}
-            related_batch = sorted({x for key in batch_keys for x in related_keys_by_full[key]})
-            cur.execute(sql, (tuple(related_batch), tuple(batch)))
+        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"]
+                legacy_iccid = row["iccid_mark"]
                 if not legacy_iccid:
                     continue
                 matched = by_full.get(legacy_iccid)
@@ -680,45 +740,162 @@ def fetch_card_meta(
                     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)
 
-    metas: dict[str, LegacyCardMeta] = {}
+    # 去重:同一完整 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(rank for rank, _row in rows)
-        best_rows = [row for rank, row in rows if rank == best_rank]
-        # 同一张 tbl_card 可能因 card_life / virtual_number 前缀兜底 JOIN 出多行,
-        # 这不是主表重复;只有不同 card_id 才算同一业务 ICCID 命中多张卡。
+        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") or "")] = row
+            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()))
-        expire = row.get("expire_date")
+        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=row["iccid"],
-            account_id=row.get("account_id") or "",
-            account_name=row.get("account_name") or "",
-            category=int(row.get("category") or 0),
-            category_name=row.get("category_name") or "",
-            agent_id=row.get("agent_id") or "",
-            agent_name=row.get("agent_name") or "",
-            msisdn=row.get("phone") or "",
-            virtual_no=row.get("vcode") or "",
-            current_meal_id=row.get("meal_id") or "",
-            current_meal_name=row.get("meal_name") or "",
-            current_meal_type=str(row.get("meal_type") or ""),
+            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=row.get("series_id") or "",
-            current_series_name=row.get("series_name") or "",
+            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]) -> dict[str, LegacyMealMeta]:
+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:
@@ -733,11 +910,15 @@ def fetch_meal_meta(conn, meal_ids: Iterable[str]) -> dict[str, LegacyMealMeta]:
         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):
+        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(
@@ -746,11 +927,17 @@ def fetch_meal_meta(conn, meal_ids: Iterable[str]) -> dict[str, LegacyMealMeta]:
                     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]) -> dict[str, LegacyAgentBalance]:
+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:
@@ -761,12 +948,17 @@ def fetch_agent_balances(conn, agent_ids: Iterable[str]) -> dict[str, LegacyAgen
         FROM tbl_agent
         WHERE id IN %s
     """
+    rows = 0
+    batches = 0
     with conn.cursor() as cur:
-        for batch in _chunks(agent_list):
+        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

scripts/migration/lib/sql_builder.py

@@ -1091,7 +1091,6 @@ def write_step2(
     cards: list[CardRow],
     devices: list[DeviceRow],
     mapping: Mapping,
-    card_metas: dict[str, LegacyCardMeta],
     usage_snapshots: dict[str, LegacyCardUsage],
     package_lifecycles: dict[str, list[LegacyPackageLifecycle]],
     commission_snapshots: dict[str, LegacyCommissionAccount],

scripts/migration/migrate_assets.py

@@ -4,7 +4,7 @@
 工作流:
     1. 读 mapping.yaml(必须先跑过 scan_legacy.py)
     2. 读 cards.csv / devices.csv
-    3. 连奇成查每张卡的元数据(category/agent_id/msisdn/virtual_no)
+    3. 连奇成查每张卡的元数据(category/agent_id/msisdn/virtual_no/当前生效套餐)
     4. 应用 mapping 把 legacy_* 翻译成 target_*,自动决定归属
     5. 生成 output/step1_*.sql + errors.csv + summary.txt
 
@@ -14,11 +14,13 @@
 
 用法:
     python3 migrate_assets.py [--config-dir config] [--resources-dir resources] [--output-dir output]
+                              [--batch-size 1000]
 """
 from __future__ import annotations
 
 import argparse
 import sys
+import time
 from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
@@ -31,6 +33,7 @@ def main() -> int:
     parser.add_argument("--config-dir", default="config", help="yaml 配置目录")
     parser.add_argument("--resources-dir", default="resources", help="CSV 资源目录")
     parser.add_argument("--output-dir", default="output", help="SQL/CSV 输出目录")
+    parser.add_argument("--batch-size", type=int, default=legacy_query.BATCH_SIZE, help="老库批量查询大小")
     args = parser.parse_args()
 
     base = Path(__file__).resolve().parent
@@ -38,16 +41,26 @@ def main() -> int:
     resources_dir = (base / args.resources_dir).resolve()
     output_dir = (base / args.output_dir).resolve()
     output_dir.mkdir(parents=True, exist_ok=True)
+    batch_size = args.batch_size
 
+    t0 = time.perf_counter()
     mapping = mapping_loader.load_mapping(config_dir)
     cards, devices, errors = csv_loader.load_assets(resources_dir, mapping)
+    print(f"[migrate] 加载 CSV: cards={len(cards)} devices={len(devices)} elapsed={time.perf_counter()-t0:.1f}s")
 
     # 连奇成查卡元数据(运营商分类、agent_id、msisdn、virtual_no、当前生效套餐)
     iccid_pairs = [(c.iccid_19, c.iccid_20, c.allow_iccid_19_lookup) for c in cards]
+
+    t1 = time.perf_counter()
     dsn = mapping_loader.load_legacy_dsn(config_dir)
     with legacy_query.connect_readonly(dsn) as conn:
-        card_metas, dup_iccids = legacy_query.fetch_card_meta(conn, iccid_pairs)
+        print(f"[migrate] 连接老库: elapsed={time.perf_counter()-t1:.1f}s")
 
+        t2 = time.perf_counter()
+        card_metas, dup_iccids = legacy_query.fetch_card_meta(conn, iccid_pairs, batch_size=batch_size)
+        print(f"[migrate] fetch_card_meta: cards={len(card_metas)} duplicates={len(dup_iccids)} elapsed={time.perf_counter()-t2:.1f}s")
+
+    t3 = time.perf_counter()
     sql_builder.write_step1(
         output_dir,
         cards=cards,
@@ -57,8 +70,10 @@ def main() -> int:
         duplicate_iccids=dup_iccids,
         errors=errors,
     )
+    print(f"[migrate] SQL 生成: elapsed={time.perf_counter()-t3:.1f}s")
 
-    print(f"完成:卡 {len(cards)} 张,设备 {len(devices)} 台,异常 {len(errors)} 行")
+    total = time.perf_counter() - t0
+    print(f"完成:卡 {len(cards)} 张,设备 {len(devices)} 台,异常 {len(errors)} 行,总耗时 {total:.1f}s")
     print(f"SQL 输出: {output_dir}")
     return 0
 

scripts/migration/migrate_runtime.py

@@ -5,17 +5,19 @@
 
 工作流:
     1. 读 mapping.yaml + cards.csv
-    2. 连奇成查:卡元数据(card_meta) + 累计流量(card_usage) + 代理佣金/主钱包余额
+    2. 连奇成查:累计流量(card_usage) + 套餐生命周期 + 代理佣金/主钱包余额
     3. 卡 → 当前生效套餐 → mapping.packages → target_package_id
     4. 生成 step2_*.sql
 
 用法:
     python3 migrate_runtime.py [--config-dir config] [--resources-dir resources] [--output-dir output]
+                               [--batch-size 1000]
 """
 from __future__ import annotations
 
 import argparse
 import sys
+import time
 from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
@@ -28,6 +30,7 @@ def main() -> int:
     parser.add_argument("--config-dir", default="config", help="yaml 配置目录")
     parser.add_argument("--resources-dir", default="resources", help="CSV 资源目录")
     parser.add_argument("--output-dir", default="output", help="SQL/CSV 输出目录")
+    parser.add_argument("--batch-size", type=int, default=legacy_query.BATCH_SIZE, help="老库批量查询大小")
     args = parser.parse_args()
 
     base = Path(__file__).resolve().parent
@@ -35,9 +38,13 @@ def main() -> int:
     resources_dir = (base / args.resources_dir).resolve()
     output_dir = (base / args.output_dir).resolve()
     output_dir.mkdir(parents=True, exist_ok=True)
+    batch_size = args.batch_size
 
+    t0 = time.perf_counter()
     mapping = mapping_loader.load_mapping(config_dir)
     cards, devices, errors = csv_loader.load_assets(resources_dir, mapping)
+    print(f"[migrate] 加载 CSV: cards={len(cards)} devices={len(devices)} elapsed={time.perf_counter()-t0:.1f}s")
+
     if errors:
         sql_builder.write_runtime_input_errors(output_dir, errors)
         print(f"cards.csv/devices.csv 存在 {len(errors)} 个基础错误,已写入 {output_dir / 'errors.csv'}", file=sys.stderr)
@@ -46,27 +53,39 @@ def main() -> int:
     iccid_pairs = [(c.iccid_19, c.iccid_20, c.allow_iccid_19_lookup) for c in cards]
     agent_ids = [a.legacy_agent_id for a in mapping.agents.values() if (a.target_shop_code or "").strip()]
 
+    t1 = time.perf_counter()
     dsn = mapping_loader.load_legacy_dsn(config_dir)
     with legacy_query.connect_readonly(dsn) as conn:
-        card_metas, _dup_iccids = legacy_query.fetch_card_meta(conn, iccid_pairs)
-        usage_snapshots = legacy_query.fetch_card_usage(conn, iccid_pairs)
-        package_lifecycles = legacy_query.fetch_package_lifecycles(conn, iccid_pairs)
-        commission_snapshots = legacy_query.fetch_commission_accounts(conn, agent_ids) if agent_ids else {}
-        agent_balances = legacy_query.fetch_agent_balances(conn, agent_ids) if agent_ids else {}
+        print(f"[migrate] 连接老库: elapsed={time.perf_counter()-t1:.1f}s")
 
+        t2 = time.perf_counter()
+        usage_snapshots = legacy_query.fetch_card_usage(conn, iccid_pairs, batch_size=batch_size)
+        print(f"[migrate] fetch_card_usage: cards={len(usage_snapshots)} elapsed={time.perf_counter()-t2:.1f}s")
+
+        t3 = time.perf_counter()
+        package_lifecycles = legacy_query.fetch_package_lifecycles(conn, iccid_pairs, batch_size=batch_size)
+        print(f"[migrate] fetch_package_lifecycles: cards={len(package_lifecycles)} elapsed={time.perf_counter()-t3:.1f}s")
+
+        t4 = time.perf_counter()
+        commission_snapshots = legacy_query.fetch_commission_accounts(conn, agent_ids, batch_size=batch_size) if agent_ids else {}
+        agent_balances = legacy_query.fetch_agent_balances(conn, agent_ids, batch_size=batch_size) if agent_ids else {}
+        print(f"[migrate] fetch_agent_data: agents={len(agent_ids)} elapsed={time.perf_counter()-t4:.1f}s")
+
+    t5 = time.perf_counter()
     warnings = sql_builder.write_step2(
         output_dir,
         cards=cards,
         devices=devices,
         mapping=mapping,
-        card_metas=card_metas,
         usage_snapshots=usage_snapshots,
         package_lifecycles=package_lifecycles,
         commission_snapshots=commission_snapshots,
         agent_balances=agent_balances,
     )
+    print(f"[migrate] SQL 生成: elapsed={time.perf_counter()-t5:.1f}s")
 
-    print(f"完成:卡 {len(cards)} 张,代理 {len(agent_ids)} 个,警告 {len(warnings)} 条")
+    total = time.perf_counter() - t0
+    print(f"完成:卡 {len(cards)} 张,代理 {len(agent_ids)} 个,警告 {len(warnings)} 条,总耗时 {total:.1f}s")
     print(f"SQL 输出: {output_dir}")
     return 0