From 31232ea8996f0e5789ce1d3e619a4e03c147815e Mon Sep 17 00:00:00 2001 From: break Date: Fri, 10 Jul 2026 13:10:00 +0900 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E8=BF=81=E7=A7=BB=E8=84=9A?= =?UTF-8?q?=E6=9C=AC=E9=80=9F=E5=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/migration/lib/legacy_query.py | 528 ++++++++++++++++++-------- scripts/migration/lib/sql_builder.py | 1 - scripts/migration/migrate_assets.py | 21 +- scripts/migration/migrate_runtime.py | 35 +- 4 files changed, 405 insertions(+), 180 deletions(-) diff --git a/scripts/migration/lib/legacy_query.py b/scripts/migration/lib/legacy_query.py index 7864a59..e179ccf 100644 --- a/scripts/migration/lib/legacy_query.py +++ b/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 diff --git a/scripts/migration/lib/sql_builder.py b/scripts/migration/lib/sql_builder.py index 4118ed8..3cb60cf 100644 --- a/scripts/migration/lib/sql_builder.py +++ b/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], diff --git a/scripts/migration/migrate_assets.py b/scripts/migration/migrate_assets.py index fb95ed4..a42344a 100644 --- a/scripts/migration/migrate_assets.py +++ b/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 diff --git a/scripts/migration/migrate_runtime.py b/scripts/migration/migrate_runtime.py index d9879cc..b1e6a10 100644 --- a/scripts/migration/migrate_runtime.py +++ b/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