From 283e3e3eb37233d14c0b7ca311360c892548e2b5 Mon Sep 17 00:00:00 2001 From: Break Date: Tue, 9 Jun 2026 11:29:40 +0800 Subject: [PATCH] =?UTF-8?q?=E8=BF=81=E7=A7=BB=E8=84=9A=E6=9C=AC=E7=9A=84?= =?UTF-8?q?=E4=B8=80=E4=BA=9B=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/migration/README.md | 4 +- scripts/migration/lib/csv_loader.py | 69 +++++++++++++----- scripts/migration/lib/legacy_query.py | 101 +++++++++++++++----------- scripts/migration/lib/sql_builder.py | 87 +++++++++++----------- scripts/migration/migrate_assets.py | 4 +- scripts/migration/migrate_runtime.py | 8 +- scripts/migration/resources/README.md | 9 ++- scripts/migration/scan_legacy.py | 22 +++--- 8 files changed, 183 insertions(+), 121 deletions(-) diff --git a/scripts/migration/README.md b/scripts/migration/README.md index 2a3e08e..beddb90 100644 --- a/scripts/migration/README.md +++ b/scripts/migration/README.md @@ -43,7 +43,7 @@ scripts/migration/ 把两个 CSV 放到 `resources/` 下(命名为 `cards.csv` / `devices.csv`): - **cards.csv**:`iccid_19`/`iccid_20` 至少填一个;仅填 `iccid_20` 时脚本自动派生前 19 位,但奇成主表只按 20 位精确查询;`is_industry` 可选 -- **devices.csv**:`imei` / `virtual_no` 至少填一个;`device_name` / `device_model` / `device_type` / `max_sim_slots` 可选;`sim_iccid_*` 引用 `cards.csv` 中的 ICCID(可填 19 或 20 位) +- **devices.csv**:`imei` / `virtual_no` 至少填一个;`device_name` / `device_model` / `device_type` / `max_sim_slots` 可选;`sim_iccid_*` 引用 `cards.csv` 中的 ICCID(可填 19 或 20 位;若 19 位前缀对应多张 20 位卡,必须填完整 20 位) 字段说明详见 [`resources/README.md`](resources/README.md)。 @@ -227,7 +227,7 @@ real_used_mb = total_bytes_cnt / (1 + flow_add_discount / 100) | `errors.csv` 出现 `carrier_not_in_mapping` | `mapping.yaml.carriers` 没覆盖到这个 account_id/account_name,重跑 `scan_legacy.py` 自动补 | | `errors.csv` 出现 `carrier_target_missing` | `mapping.yaml.carriers` 里有这一项但 `target_*` 还是 null,手动填 | | `errors.csv` 出现 `not_in_legacy` | 该 ICCID 在奇成 `tbl_card` 查不到,业务方确认 | -| `errors.csv` 出现 `legacy_duplicate` | 同一 `iccid_19` 在奇成命中多条 `tbl_card`,业务方人工确认是哪张物理卡 | +| `errors.csv` 出现 `legacy_duplicate` | 同一 ICCID 在奇成命中多条 `tbl_card`,业务方人工确认是哪张物理卡 | | `errors.csv` 出现 `carrier_length_conflict` | 业务方填的 `iccid_20` 与奇成查到的运营商不匹配(CTCC 不该有 20 位 / 非 CTCC 必须有 20 位) | | `errors.csv` 出现 `iccid_20_prefix_mismatch` | `iccid_20` 前 19 位 ≠ `iccid_19`,业务方校对 | | `warnings.csv` 出现 `agent_no_legacy` | `mapping.agents` 配了但奇成没有 tbl_agent 记录,业务方确认 | diff --git a/scripts/migration/lib/csv_loader.py b/scripts/migration/lib/csv_loader.py index ff029e9..113572c 100644 --- a/scripts/migration/lib/csv_loader.py +++ b/scripts/migration/lib/csv_loader.py @@ -54,7 +54,7 @@ class DeviceRow: line_no: int imei: str virtual_no: str # 留空时用 imei 兜底,见 _load_devices - sim_iccids: dict[int, str] # slot → iccid(只保留非空) + sim_iccids: dict[int, str] # slot → 完整 ICCID(只保留非空) device_name: str = "" device_model: str = "" device_type: str = "" @@ -159,7 +159,7 @@ def _load_cards(path: Path, result: LoadResult) -> None: if "iccid_20" not in header: raise ValueError(f"{path} 必须包含 iccid_20 列(20 位运营商卡填写,CTCC 留空)") - seen_iccid_19: dict[str, int] = {} + seen_iccid_full: dict[str, int] = {} for idx, raw in enumerate(reader, start=2): iccid_19 = _normalize_cell(raw.get("iccid_19", "")) iccid_20 = _normalize_cell(raw.get("iccid_20", "")) @@ -184,13 +184,14 @@ def _load_cards(path: Path, result: LoadResult) -> None: result.errors.append(ErrorRow("cards.csv", idx, "iccid_20", iccid_20, err.code, err.message)) continue - if iccid_19 in seen_iccid_19: + iccid_full = iccid_20 or iccid_19 + if iccid_full in seen_iccid_full: result.errors.append( - ErrorRow("cards.csv", idx, "iccid_19", iccid_19, "duplicate", - f"与第 {seen_iccid_19[iccid_19]} 行 iccid_19 重复") + ErrorRow("cards.csv", idx, "iccid", iccid_full, "duplicate", + f"与第 {seen_iccid_full[iccid_full]} 行 ICCID 重复") ) continue - seen_iccid_19[iccid_19] = idx + seen_iccid_full[iccid_full] = idx is_industry = _parse_bool(raw.get("is_industry", "")) if is_industry is None: @@ -292,27 +293,59 @@ def _load_devices(path: Path, result: LoadResult) -> None: def _cross_check(result: LoadResult) -> None: """校验设备与卡之间的一致性,并把 bound_device_virtual_no 回填到 CardRow。 - devices.csv 里 sim_iccid_X 业务方可能填 19 位也可能填 20 位, - 统一截取前 19 位回查 cards.iccid_19。 + devices.csv 里 sim_iccid_X 业务方可能填 19 位也可能填 20 位。 + 20 位按完整 ICCID 精确匹配;19 位只有在 cards.csv 中唯一对应一张卡时才允许。 """ from . import iccid_utils - cards_by_iccid_19: dict[str, CardRow] = {c.iccid_19: c for c in result.cards} + cards_by_full: dict[str, CardRow] = {c.iccid_full: c for c in result.cards} + cards_by_i19: dict[str, list[CardRow]] = {} + for card in result.cards: + cards_by_i19.setdefault(card.iccid_19, []).append(card) device_card_refs: dict[str, tuple[str, int]] = {} for device in result.devices: for slot, raw_iccid in device.sim_iccids.items(): - key19 = iccid_utils.prefix_19(raw_iccid) - if key19 not in cards_by_iccid_19: + resolved_card: Optional[CardRow] = None + if len(raw_iccid) == 20: + resolved_card = cards_by_full.get(raw_iccid) + if resolved_card is None: + prefix = iccid_utils.prefix_19(raw_iccid) + candidates = cards_by_i19.get(prefix, []) + if len(candidates) == 1 and not candidates[0].iccid_20: + resolved_card = candidates[0] + elif len(raw_iccid) == 19: + candidates = cards_by_i19.get(raw_iccid, []) + if len(candidates) > 1: + result.errors.append( + ErrorRow( + "devices.csv", device.line_no, f"sim_iccid_{slot}", raw_iccid, + "ambiguous_iccid_19", "该 19 位前缀对应多张 20 位卡,请填写完整 20 位 ICCID", + ) + ) + continue + if candidates: + resolved_card = candidates[0] + else: result.errors.append( ErrorRow( "devices.csv", device.line_no, f"sim_iccid_{slot}", raw_iccid, - "card_not_found", "引用的 ICCID(前 19 位)在 cards.csv 中不存在", + "sim_iccid_length", "设备槽位 ICCID 必须是 19 或 20 位", ) ) continue - if key19 in device_card_refs: - prev_vno, prev_line = device_card_refs[key19] + + if resolved_card is None: + result.errors.append( + ErrorRow( + "devices.csv", device.line_no, f"sim_iccid_{slot}", raw_iccid, + "card_not_found", "引用的 ICCID 在 cards.csv 中不存在", + ) + ) + continue + card_key = resolved_card.iccid_full + if card_key in device_card_refs: + prev_vno, prev_line = device_card_refs[card_key] result.errors.append( ErrorRow( "devices.csv", device.line_no, f"sim_iccid_{slot}", raw_iccid, @@ -320,7 +353,7 @@ def _cross_check(result: LoadResult) -> None: ) ) continue - device_card_refs[key19] = (device.virtual_no, device.line_no) - cards_by_iccid_19[key19].bound_device_virtual_no = device.virtual_no - # 标准化:slot 里的值统一改写为 iccid_19,sql_builder 直接用它查 cards / card_metas - device.sim_iccids[slot] = key19 + device_card_refs[card_key] = (device.virtual_no, device.line_no) + resolved_card.bound_device_virtual_no = device.virtual_no + # 标准化:slot 里的值统一改写为完整 ICCID,sql_builder 直接用它查 cards / card_metas + device.sim_iccids[slot] = card_key diff --git a/scripts/migration/lib/legacy_query.py b/scripts/migration/lib/legacy_query.py index 829a3ef..83c85ac 100644 --- a/scripts/migration/lib/legacy_query.py +++ b/scripts/migration/lib/legacy_query.py @@ -11,7 +11,7 @@ from decimal import Decimal from typing import Any, Iterable, Iterator, Optional -BATCH_SIZE = 500 +BATCH_SIZE = 200 def _normalize_iccid_pairs(iccid_pairs: Iterable[tuple]) -> list[tuple[str, str, bool]]: @@ -67,7 +67,7 @@ class LegacyCardUsage: 奇成没有按套餐分段的用量明细,无更好做法。 """ - iccid: str # 业务方视角的 iccid_19 + 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 中找到当前生效套餐 @@ -273,28 +273,32 @@ def fetch_card_usage(conn, iccid_pairs: Iterable[tuple]) -> dict[str, LegacyCard iccid_pairs: (iccid_19, iccid_20, allow_iccid_19_lookup) 列表。 Returns: - {iccid_19: LegacyCardUsage}。 + {完整 ICCID: LegacyCardUsage}。 """ pairs = _normalize_iccid_pairs(iccid_pairs) out: dict[str, LegacyCardUsage] = {} if not pairs: return out - full_to_19: dict[str, tuple[str, int]] = {} - related_keys_by_19: dict[str, set[str]] = {} + 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 位时兜底。 - full_to_19[i20 or i19] = (i19, 0) + query_to_full[card_key] = (card_key, 0) if i20 and allow_i19_lookup: - full_to_19[i19] = (i19, 1) - related_keys_by_19.setdefault(i19, set()).add(i19) + 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_19[i19].add(i20) - full_set = sorted(full_to_19.keys()) + related_keys_by_full[card_key].add(i20) + full_set = sorted(query_to_full.keys()) - # tbl_card 主表用 iccid_mark IN 全长精确匹配;tbl_card_life 从表用 - # LEFT(iccid_mark, 19) 兜底,兼容奇成两张表 iccid 长度不一致的脏数据。 + # tbl_card 主表用 iccid_mark IN 全长精确匹配;从表只在主表本身是 19 位时 + # 按前缀兜底,避免同前缀 20 位卡互相串套餐。 sql = """ SELECT c.iccid_mark, @@ -314,30 +318,33 @@ def fetch_card_usage(conn, iccid_pairs: Iterable[tuple]) -> dict[str, LegacyCard ON latest.iccid_mark = cl1.iccid_mark AND latest.max_expire = cl1.expire_date WHERE cl1.status = 1 - ) cl ON LEFT(cl.iccid_mark, 19) = LEFT(c.iccid_mark, 19) + ) 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 """ hits: dict[str, list[tuple[int, dict]]] = {} with conn.cursor() as cur: for batch in _chunks(full_set): - batch_i19s = {full_to_19[x][0] for x in batch} - related_batch = sorted({x for i19 in batch_i19s for x in related_keys_by_19[i19]}) + 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 row in cur.fetchall(): legacy_iccid = row["iccid_mark"] if not legacy_iccid: continue - matched = full_to_19.get(legacy_iccid) + matched = query_to_full.get(legacy_iccid) if matched is None: continue - i19, rank = matched - hits.setdefault(i19, []).append((rank, row)) - for i19, rows in hits.items(): + card_key, rank = matched + hits.setdefault(card_key, []).append((rank, 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) - out[i19] = LegacyCardUsage( - iccid=i19, + 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")), @@ -388,31 +395,35 @@ def fetch_card_meta( - tbl_card 主表同时查优先键和显式允许的兜底键:有 iccid_20 时优先取 20 位精确命中;只有业务方也显式填了 iccid_19 时,才允许退回 19 位。 只填 iccid_20 的固定 20 位卡绝不拿 19 位前缀查主表。 - - tbl_card_life / tbl_virtual_number 从表用 LEFT(iccid_mark, 19) 兜底, - 应对奇成两张表 iccid 长度不一致的脏数据 + - tbl_card_life / tbl_virtual_number 从表在主表本身是 19 位时才用 + LEFT(iccid_mark, 19) 兜底,避免同前缀 20 位卡互相串数据 Args: iccid_pairs: (iccid_19, iccid_20, allow_iccid_19_lookup) 列表。 Returns: - (metas, duplicate_iccid_19): - - metas: {iccid_19: LegacyCardMeta} 单匹结果 - - duplicate_iccid_19: 一份业务方 iccid_19 命中奇成 tbl_card 多条记录的 set, + (metas, duplicate_iccids): + - metas: {完整 ICCID: LegacyCardMeta} 单匹结果 + - duplicate_iccids: 一份业务方完整 ICCID 命中奇成 tbl_card 多条记录的 set, 这些卡不进 metas,sql_builder 会写 errors.csv 阻断 """ pairs = _normalize_iccid_pairs(iccid_pairs) if not pairs: return {}, set() - by_full: dict[str, tuple[str, int]] = {} # tbl_card 查询键 → (iccid_19, rank) - related_keys_by_19: dict[str, set[str]] = {} + 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: - by_full[i20 or i19] = (i19, 0) + card_key = i20 or i19 + by_full[card_key] = (card_key, 0) if i20 and allow_i19_lookup: - by_full[i19] = (i19, 1) - related_keys_by_19.setdefault(i19, set()).add(i19) + 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_19[i19].add(i20) + related_keys_by_full[card_key].add(i20) full_set = sorted(by_full.keys()) sql = """ @@ -432,7 +443,10 @@ def fetch_card_meta( cl.expire_date AS expire_date FROM tbl_card c LEFT JOIN tbl_vendor_category vc ON vc.category = c.category - LEFT JOIN tbl_virtual_number vn ON LEFT(vn.iccid_mark, 19) = LEFT(c.iccid_mark, 19) + 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.expire_date FROM tbl_card_life cl1 @@ -445,16 +459,19 @@ def fetch_card_meta( ON latest.iccid_mark = cl1.iccid_mark AND latest.max_expire = cl1.expire_date WHERE cl1.status = 1 - ) cl ON LEFT(cl.iccid_mark, 19) = LEFT(c.iccid_mark, 19) + ) cl ON ( + cl.iccid_mark = c.iccid_mark + OR (CHAR_LENGTH(c.iccid_mark) = 19 AND LEFT(cl.iccid_mark, 19) = c.iccid_mark) + ) WHERE c.iccid_mark IN %s """ - # 收集每个 iccid_19 命中的奇成行(可能多条 = 脏数据) + # 收集每个完整 ICCID 命中的奇成行(可能多条 = 脏数据) hits: dict[str, list[tuple[int, dict]]] = {} with conn.cursor() as cur: for batch in _chunks(full_set): - batch_i19s = {by_full[x][0] for x in batch} - related_batch = sorted({x for i19 in batch_i19s for x in related_keys_by_19[i19]}) + 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 row in cur.fetchall(): legacy_iccid = row["iccid"] @@ -463,12 +480,12 @@ def fetch_card_meta( matched = by_full.get(legacy_iccid) if matched is None: continue - i19, rank = matched - hits.setdefault(i19, []).append((rank, row)) + card_key, rank = matched + hits.setdefault(card_key, []).append((rank, row)) metas: dict[str, LegacyCardMeta] = {} duplicates: set[str] = set() - for i19, rows in hits.items(): + 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 出多行, @@ -477,12 +494,12 @@ def fetch_card_meta( for row in best_rows: rows_by_card_id[str(row.get("card_id") or row.get("iccid") or "")] = row if len(rows_by_card_id) > 1: - duplicates.add(i19) + duplicates.add(card_key) continue row = next(iter(rows_by_card_id.values())) expire = row.get("expire_date") expire_str = expire.strftime("%Y-%m-%d %H:%M:%S") if expire else None - metas[i19] = LegacyCardMeta( + metas[card_key] = LegacyCardMeta( legacy_raw_iccid=row["iccid"], account_id=row.get("account_id") or "", account_name=row.get("account_name") or "", diff --git a/scripts/migration/lib/sql_builder.py b/scripts/migration/lib/sql_builder.py index f769a4b..84f5de4 100644 --- a/scripts/migration/lib/sql_builder.py +++ b/scripts/migration/lib/sql_builder.py @@ -144,8 +144,8 @@ def _device_package_source_iccid(device: DeviceRow) -> str: ) -def _target_series_id_expr_for_iccid_19( - iccid_19: str, +def _target_series_id_expr_for_iccid( + iccid: str, card_metas: dict[str, LegacyCardMeta], mapping: Mapping, ) -> str: @@ -153,9 +153,9 @@ def _target_series_id_expr_for_iccid_19( 系列归属以新系统目标套餐为准，避免奇成套餐系列与新库系列二次映射不一致。 """ - if not iccid_19: + if not iccid: return "NULL" - meta = card_metas.get(iccid_19) + meta = card_metas.get(iccid) if meta is None or not meta.current_meal_id: return "NULL" pkg_map = mapping.lookup_package(meta.current_meal_id) @@ -200,7 +200,7 @@ def _runtime_asset_for_card(card: CardRow, devices_by_virtual_no: dict[str, Devi device = devices_by_virtual_no.get(card.bound_device_virtual_no) if device is None: return None - if card.iccid_19 != _device_package_source_iccid(device): + if card.iccid_full != _device_package_source_iccid(device): return None return "device", device.virtual_no @@ -217,7 +217,7 @@ def write_step1( devices: list[DeviceRow], mapping: Mapping, card_metas: dict[str, LegacyCardMeta], - duplicate_iccid_19s: set[str], + duplicate_iccids: set[str], errors: list[ErrorRow], ) -> None: """生成 step1_* SQL + errors.csv + summary.txt。 @@ -225,21 +225,21 @@ def write_step1( 会就地修改 errors 列表,把映射缺失等问题追加进去。 Args: - card_metas: {iccid_19: LegacyCardMeta} - duplicate_iccid_19s: 同一 iccid_19 在奇成命中多条 tbl_card 记录的集合,直接报错阻断 + card_metas: {完整 ICCID: LegacyCardMeta} + duplicate_iccids: 同一完整 ICCID 在奇成命中多条 tbl_card 记录的集合,直接报错阻断 """ user_id = mapping.migration_user_id # 预检 carrier 映射 + 运营商一致性 + 重复命中,把异常卡剔除 - valid_cards = _filter_cards_by_carrier(cards, card_metas, duplicate_iccid_19s, mapping, errors) + valid_cards = _filter_cards_by_carrier(cards, card_metas, duplicate_iccids, mapping, errors) - cards_by_iccid_19 = {c.iccid_19: c for c in valid_cards} + cards_by_iccid = {c.iccid_full: c for c in valid_cards} _write_iot_cards(output_dir / "step1_01_iot_cards.sql", valid_cards, devices, card_metas, mapping, user_id) _write_devices(output_dir / "step1_02_devices.sql", devices, card_metas, mapping, user_id) _write_asset_identifiers(output_dir / "step1_03_asset_identifiers.sql", valid_cards, devices, card_metas) _write_asset_wallets(output_dir / "step1_04_asset_wallets.sql", valid_cards, devices) - _write_sim_bindings(output_dir / "step1_05_sim_bindings.sql", devices, cards_by_iccid_19) + _write_sim_bindings(output_dir / "step1_05_sim_bindings.sql", devices, cards_by_iccid) _write_shop_allocations( output_dir / "step1_06_shop_allocations.sql", valid_cards, devices, card_metas, mapping, user_id ) @@ -251,29 +251,29 @@ def write_step1( def _filter_cards_by_carrier( cards: list[CardRow], card_metas: dict[str, LegacyCardMeta], - duplicate_iccid_19s: set[str], + duplicate_iccids: set[str], mapping: Mapping, errors: list[ErrorRow], ) -> list[CardRow]: """筛掉:奇成查不到 / 命中多条 / carrier 映射缺失 / target_* 未填 / 运营商与 iccid 长度不一致 的卡。""" valid: list[CardRow] = [] for c in cards: - if c.iccid_19 in duplicate_iccid_19s: + if c.iccid_full in duplicate_iccids: errors.append(ErrorRow( - "cards.csv", c.line_no, "iccid_19", c.iccid_19, - "legacy_duplicate", "奇成 tbl_card 中同一 iccid_19 命中多条记录,请业务方人工确认", + "cards.csv", c.line_no, "iccid", c.iccid_full, + "legacy_duplicate", "奇成 tbl_card 中同一 ICCID 命中多条记录,请业务方人工确认", )) continue - meta = card_metas.get(c.iccid_19) + meta = card_metas.get(c.iccid_full) if meta is None: errors.append(ErrorRow( - "cards.csv", c.line_no, "iccid_19", c.iccid_19, + "cards.csv", c.line_no, "iccid", c.iccid_full, "not_in_legacy", "奇成 tbl_card 中找不到该 ICCID(iccid_19/iccid_20 都没命中)", )) continue if not meta.account_id: errors.append(ErrorRow( - "cards.csv", c.line_no, "iccid_19", c.iccid_19, + "cards.csv", c.line_no, "iccid", c.iccid_full, "no_carrier_account", "奇成 tbl_card.account_id 为空,无法定位具体运营商账号映射", )) continue @@ -320,7 +320,7 @@ def _write_iot_cards( with path.open("w", encoding="utf-8") as f: f.write(_file_header("step1_01 卡资产导入 (tb_iot_card)")) for c in cards: - meta = card_metas[c.iccid_19] + meta = card_metas[c.iccid_full] carrier = mapping.lookup_carrier(meta.account_id) assert carrier is not None # 已被 _filter_cards_by_carrier 保证 @@ -332,7 +332,7 @@ def _write_iot_cards( if c.bound_device_virtual_no: device = devices_by_virtual_no.get(c.bound_device_virtual_no) shop_code = _resolve_device_shop_code(device, card_metas, mapping) if device else "" - series_id_expr = _target_series_id_expr_for_iccid_19(c.iccid_19, card_metas, mapping) + series_id_expr = _target_series_id_expr_for_iccid(c.iccid_full, card_metas, mapping) card_virtual_no = _card_virtual_no_for_import(meta, device_virtual_nos) is_standalone = "FALSE" if c.bound_device_virtual_no else "TRUE" device_virtual_no = c.bound_device_virtual_no or "" @@ -408,7 +408,7 @@ def _write_devices( device_status = 2 if shop_code else 1 shop_id_expr = _shop_id_sub(shop_code) source_iccid = _device_package_source_iccid(d) - series_id_expr = _target_series_id_expr_for_iccid_19(source_iccid, card_metas, mapping) + series_id_expr = _target_series_id_expr_for_iccid(source_iccid, card_metas, mapping) f.write( "INSERT INTO tb_device (\n" " virtual_no, max_sim_slots, imei,\n" @@ -480,7 +480,7 @@ def _write_asset_identifiers( with path.open("w", encoding="utf-8") as f: f.write(_file_header("step1_03 资产标识符注册 (tb_asset_identifier)")) for c in cards: - meta = card_metas[c.iccid_19] + meta = card_metas[c.iccid_full] card_virtual_no = _card_virtual_no_for_import(meta, device_virtual_nos) f.write( "INSERT INTO tb_asset_identifier (identifier, asset_type, asset_id)\n" @@ -563,13 +563,13 @@ def _write_asset_wallets(path: Path, cards: list[CardRow], devices: list[DeviceR def _write_sim_bindings( path: Path, devices: list[DeviceRow], - cards_by_iccid_19: dict[str, CardRow], + cards_by_iccid: dict[str, CardRow], ) -> None: with path.open("w", encoding="utf-8") as f: f.write(_file_header("step1_05 设备-卡绑定 (tb_device_sim_binding)")) for d in devices: - for slot, iccid_19 in sorted(d.sim_iccids.items()): - card = cards_by_iccid_19.get(iccid_19) + for slot, iccid in sorted(d.sim_iccids.items()): + card = cards_by_iccid.get(iccid) if card is None: continue # 该卡可能被 carrier 校验剔除,跳过此 slot 绑定 is_current = "TRUE" if slot == d.current_slot else "FALSE" @@ -632,7 +632,7 @@ def _write_shop_allocations( for c in cards: if c.bound_device_virtual_no: continue - meta = card_metas[c.iccid_19] + meta = card_metas[c.iccid_full] shop_code = _resolve_card_shop_code(c, meta, mapping) if not shop_code: continue @@ -685,6 +685,11 @@ def _write_errors(path: Path, errors: list[ErrorRow]) -> None: writer.writerow(e.to_csv_row()) +def write_runtime_input_errors(output_dir: Path, errors: list[ErrorRow]) -> None: + """写运行态迁移的输入错误。""" + _write_errors(output_dir / "errors.csv", errors) + + def _write_summary_step1(path: Path, cards: list[CardRow], devices: list[DeviceRow], errors: list[ErrorRow]) -> None: bound = sum(1 for c in cards if c.bound_device_virtual_no) lines = [ @@ -723,32 +728,32 @@ def write_step2( warnings: list[tuple[str, str, str]] = [] devices_by_virtual_no = _device_by_virtual_no(devices) # 卡 → package(从 card_meta.current_meal_id 经 mapping.packages 查 target) - resolved_packages: dict[str, tuple[int, str]] = {} # iccid_19 → (target_package_id, expire_date) + resolved_packages: dict[str, tuple[int, str]] = {} # 完整 ICCID → (target_package_id, expire_date) for c in cards: - meta = card_metas.get(c.iccid_19) + meta = card_metas.get(c.iccid_full) if meta is None or not meta.current_meal_id: continue pkg_map = mapping.lookup_package(meta.current_meal_id) if pkg_map is None: - warnings.append(("package_not_in_mapping", c.iccid_19, + warnings.append(("package_not_in_mapping", c.iccid_full, f"奇成 meal_id={meta.current_meal_id} ({meta.current_meal_name}) 未在 mapping.yaml.packages")) continue try: pkg_map.require_target() except MissingTargetError as exc: - warnings.append(("package_target_missing", c.iccid_19, str(exc))) + warnings.append(("package_target_missing", c.iccid_full, str(exc))) continue - resolved_packages[c.iccid_19] = (pkg_map.target_package_id, meta.current_expire_date or "") + resolved_packages[c.iccid_full] = (pkg_map.target_package_id, meta.current_expire_date or "") # 用量预检:虚用量 > 0 但没找到当前生效套餐的卡,无法精确反算真用量 # (按 flow_add_discount=0 处理,即真=虚,可能高估真用量;不阻断) for c in cards: - snap = usage_snapshots.get(c.iccid_19) + snap = usage_snapshots.get(c.iccid_full) if snap is None or snap.virtual_used_mb <= 0: continue if not snap.has_active_package: warnings.append(( - "usage_without_active_package", c.iccid_19, + "usage_without_active_package", c.iccid_full, f"奇成虚用量 {snap.virtual_used_mb}MB,但 tbl_card_life 中没有当前生效套餐" "(status=1),无法精确反算真用量,按 flow_add_discount=0 处理(真=虚)", )) @@ -805,7 +810,7 @@ def _write_migration_orders( with path.open("w", encoding="utf-8") as f: f.write(_file_header("step2_01 迁移伪订单 (tb_order)")) for c in cards: - if c.iccid_19 not in resolved_packages: + if c.iccid_full not in resolved_packages: continue asset = _runtime_asset_for_card(c, devices_by_virtual_no) if asset is None: @@ -873,16 +878,16 @@ def _write_package_usages( with path.open("w", encoding="utf-8") as f: f.write(_file_header("step2_02 套餐使用记录 (tb_package_usage)")) for c in cards: - if c.iccid_19 not in resolved_packages: + if c.iccid_full not in resolved_packages: continue asset = _runtime_asset_for_card(c, devices_by_virtual_no) if asset is None: continue asset_type, _asset_identifier = asset - target_pkg_id, expire = resolved_packages[c.iccid_19] + target_pkg_id, expire = resolved_packages[c.iccid_full] order_no = _make_card_order_no(c.iccid_full) expires_expr = f"{_sql_str(expire)}::timestamp" if expire else "NULL" - snap = usage_snapshots.get(c.iccid_19) + snap = usage_snapshots.get(c.iccid_full) used_mb = snap.real_used_mb_floor if snap is not None else 0 effective_limit_expr = ( "CASE WHEN p.enable_virtual_data AND p.virtual_data_mb > 0 " @@ -935,7 +940,7 @@ def _write_card_data_usage(path: Path, cards: list[CardRow], usage_snapshots: di with path.open("w", encoding="utf-8") as f: f.write(_file_header("step2_03 卡累计流量更新 (tb_iot_card 流量基线)")) for c in cards: - snap = usage_snapshots.get(c.iccid_19) + snap = usage_snapshots.get(c.iccid_full) if snap is None or snap.real_used_mb <= 0: continue real_decimal = _sql_decimal(snap.real_used_mb) @@ -967,9 +972,9 @@ def _write_asset_series_update( with path.open("w", encoding="utf-8") as f: f.write(_file_header("step2_06 资产套餐系列回填 (tb_iot_card / tb_device)")) for c in cards: - if c.iccid_19 not in resolved_packages: + if c.iccid_full not in resolved_packages: continue - target_pkg_id, _expire = resolved_packages[c.iccid_19] + target_pkg_id, _expire = resolved_packages[c.iccid_full] f.write( "UPDATE tb_iot_card c\n" "SET series_id = p.series_id,\n" @@ -1124,7 +1129,7 @@ def _write_summary_step2(path: Path, *, cards, resolved, devices_by_virtual_no, generated_count = sum( 1 for c in cards - if c.iccid_19 in resolved and _runtime_asset_for_card(c, devices_by_virtual_no) is not None + if c.iccid_full in resolved and _runtime_asset_for_card(c, devices_by_virtual_no) is not None ) lines = [ "奇成迁移 - 脚本2 关联数据导入摘要", diff --git a/scripts/migration/migrate_assets.py b/scripts/migration/migrate_assets.py index 253ee6e..73e80f2 100644 --- a/scripts/migration/migrate_assets.py +++ b/scripts/migration/migrate_assets.py @@ -46,7 +46,7 @@ def main() -> int: iccid_pairs = [(c.iccid_19, c.iccid_20, c.allow_iccid_19_lookup) for c in cards] dsn = mapping_loader.load_legacy_dsn(config_dir) with legacy_query.connect_readonly(dsn) as conn: - card_metas, dup_iccid_19 = legacy_query.fetch_card_meta(conn, iccid_pairs) + card_metas, dup_iccids = legacy_query.fetch_card_meta(conn, iccid_pairs) sql_builder.write_step1( output_dir, @@ -54,7 +54,7 @@ def main() -> int: devices=devices, mapping=mapping, card_metas=card_metas, - duplicate_iccid_19s=dup_iccid_19, + duplicate_iccids=dup_iccids, errors=errors, ) diff --git a/scripts/migration/migrate_runtime.py b/scripts/migration/migrate_runtime.py index 3b1f5c1..c140784 100644 --- a/scripts/migration/migrate_runtime.py +++ b/scripts/migration/migrate_runtime.py @@ -37,14 +37,18 @@ def main() -> int: output_dir.mkdir(parents=True, exist_ok=True) mapping = mapping_loader.load_mapping(config_dir) - cards, devices, _errors = csv_loader.load_assets(resources_dir) + cards, devices, errors = csv_loader.load_assets(resources_dir) + 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) + return 1 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()] dsn = mapping_loader.load_legacy_dsn(config_dir) with legacy_query.connect_readonly(dsn) as conn: - card_metas, _dup_iccid_19 = legacy_query.fetch_card_meta(conn, iccid_pairs) + card_metas, _dup_iccids = legacy_query.fetch_card_meta(conn, iccid_pairs) usage_snapshots = legacy_query.fetch_card_usage(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 {} diff --git a/scripts/migration/resources/README.md b/scripts/migration/resources/README.md index d769990..33c3969 100644 --- a/scripts/migration/resources/README.md +++ b/scripts/migration/resources/README.md @@ -8,13 +8,13 @@ ## cards.csv 卡资产清单 每张要迁移的卡一行。`iccid_19` / `iccid_20` 至少填一个。只有 20 位完整 ICCID -时可只填 `iccid_20`,脚本会自动派生前 19 位作为内部稳定 ID。 +时可只填 `iccid_20`,脚本会自动派生前 19 位写入 `iccid_19` 字段。 只填 `iccid_20` 的卡,脚本查询奇成主表时只按 20 位精确匹配;只有同时显式填写 `iccid_19` 和 `iccid_20` 时,才允许 20 位未命中后退回 19 位查询。 | 列名 | 必填 | 说明 | |------|------|------| -| `iccid_19` | 条件必填 | ICCID 前 19 位(内部稳定 ID;必须 19 位;允许奇成历史字母位)。`iccid_20` 已填时可留空 | +| `iccid_19` | 条件必填 | ICCID 前 19 位(必须 19 位;允许奇成历史字母位)。`iccid_20` 已填时可留空 | | `iccid_20` | 视运营商 | 完整 20 位 ICCID;CMCC/CUCC/CBN(20 位运营商) **必填**,CTCC(19 位)**必须留空**;非空时前 19 位必须等于 `iccid_19` | | `is_industry` | 否 | 是否行业卡,留空 = `false`(普通卡) | @@ -22,7 +22,7 @@ `tbl_card.category` 只是奇成大类,仅辅助人工填写 `target_carrier_type`。填完映射后会做一致性校验: - CTCC 卡填了 `iccid_20` → `errors.csv: carrier_length_conflict` 阻断 - 非 CTCC 卡 `iccid_20` 留空 → 同上阻断 -- `iccid_19` 在奇成 `tbl_card` 命中多条记录 → `errors.csv: legacy_duplicate` 阻断,请业务方人工确认 +- 同一完整 ICCID 在奇成 `tbl_card` 命中多条记录 → `errors.csv: legacy_duplicate` 阻断,请业务方人工确认 **店铺归属不在 CSV 里填**,由 `mapping.yaml.agents[].target_shop_code` 通过奇成 `tbl_card.agent_id` 自动推导。如果该卡在奇成没有 agent_id 或代理映射未填 shop_code, @@ -40,7 +40,7 @@ | `device_model` | 否 | 设备型号,写入 `tb_device.device_model` | | `device_type` | 否 | 设备类型,写入 `tb_device.device_type` | | `max_sim_slots` | 否 | 最大卡槽数,范围 1-4。不填时按非空 `sim_iccid_*` 的最大槽位推导 | -| `sim_iccid_1` ~ `sim_iccid_4` | 否 | 各插槽 ICCID,引用 `cards.csv` 的卡。可填 19 或 20 位,内部统一按前 19 位回查 `cards.iccid_19`(至少要有 1 个) | +| `sim_iccid_1` ~ `sim_iccid_4` | 否 | 各插槽 ICCID,引用 `cards.csv` 的卡。可填 19 或 20 位;20 位精确匹配完整 ICCID,19 位仅在该前缀唯一对应一张卡时允许(至少要有 1 个) | **店铺归属规则**:用 `sim_iccid_1`(主卡)的 agent_id 走 `mapping.agents` 推导,与卡归属同口径。 @@ -52,5 +52,6 @@ - `devices.csv` 任何 `sim_iccid_*` 引用的 ICCID **必须**在 `cards.csv` 中存在 - 同一 ICCID 不允许在 `cards.csv` 中重复出现 +- 19 位前缀对应多张 20 位卡时,`devices.csv` 必须填写完整 20 位 ICCID - `imei` 和 `virtual_no` 至少要有一个(用作设备唯一标识) - 同一 ICCID 不能同时被多个设备/插槽引用 diff --git a/scripts/migration/scan_legacy.py b/scripts/migration/scan_legacy.py index 368f0ae..6837c9f 100644 --- a/scripts/migration/scan_legacy.py +++ b/scripts/migration/scan_legacy.py @@ -39,7 +39,7 @@ def _read_iccid_pairs(cards_csv: Path) -> list[tuple[str, str, bool]]: """ if not cards_csv.exists(): raise FileNotFoundError(f"找不到 {cards_csv},请按 resources/README.md 准备") - pairs: dict[str, tuple[str, bool]] = {} + pairs: dict[str, tuple[str, str, bool]] = {} first_seen_line: dict[str, int] = {} errors: list[str] = [] with cards_csv.open("r", encoding="utf-8-sig", newline="") as f: @@ -69,17 +69,19 @@ def _read_iccid_pairs(cards_csv: Path) -> list[tuple[str, str, bool]]: if i20 and i20[:19] != i19: errors.append(f"第 {line_no} 行 iccid_20 前 19 位必须等于 iccid_19: {i20!r}") continue - if i19 in pairs: - errors.append(f"第 {line_no} 行 iccid_19 与第 {first_seen_line[i19]} 行重复: {i19!r}") + # 有 iccid_20 的卡用 iccid_20 做唯一键，避免同批次卡派生出相同 iccid_19 误报重复 + dedup_key = i20 if i20 else i19 + if dedup_key in pairs: + errors.append(f"第 {line_no} 行 {'iccid_20' if i20 else 'iccid_19'} 与第 {first_seen_line[dedup_key]} 行重复: {dedup_key!r}") continue - first_seen_line[i19] = line_no - pairs[i19] = (i20, allow_i19_lookup) + first_seen_line[dedup_key] = line_no + pairs[dedup_key] = (i19, i20, allow_i19_lookup) if errors: preview = "\n".join(f" - {msg}" for msg in errors[:20]) if len(errors) > 20: preview += f"\n - 其余 {len(errors) - 20} 个错误已省略" raise ValueError(f"{cards_csv} 存在无效 ICCID,请先修正后再扫描:\n{preview}") - return [(i19, i20, allow_i19_lookup) for i19, (i20, allow_i19_lookup) in sorted(pairs.items())] + return [v for _, v in sorted(pairs.items())] def main() -> int: @@ -104,13 +106,13 @@ def main() -> int: dsn = mapping_loader.load_legacy_dsn(config_dir) with legacy_query.connect_readonly(dsn) as conn: - card_metas, dup_iccid_19 = legacy_query.fetch_card_meta(conn, iccid_pairs) + card_metas, dup_iccids = legacy_query.fetch_card_meta(conn, iccid_pairs) meal_ids = {m.current_meal_id for m in card_metas.values() if m.current_meal_id} meal_metas = legacy_query.fetch_meal_meta(conn, meal_ids) - if dup_iccid_19: - print(f"警告:{len(dup_iccid_19)} 张卡的 iccid_19 在奇成 tbl_card 命中多条记录:") - for x in sorted(dup_iccid_19): + if dup_iccids: + print(f"警告:{len(dup_iccids)} 张卡的 ICCID 在奇成 tbl_card 命中多条记录:") + for x in sorted(dup_iccids): print(f" - {x}") print("这些卡 migrate_assets 阶段会写 errors.csv 阻断,请业务方人工确认。")