From 872769e69e54529184efc51976e7da76582d3ddd Mon Sep 17 00:00:00 2001 From: Break Date: Thu, 28 May 2026 17:22:52 +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=E9=97=AE=E9=A2=98=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 | 12 +- scripts/migration/config/mapping.yaml.example | 7 +- scripts/migration/lib/csv_loader.py | 25 +++- scripts/migration/lib/iccid_utils.py | 11 +- scripts/migration/lib/legacy_query.py | 128 ++++++++++++------ scripts/migration/lib/mapping_loader.py | 24 +++- scripts/migration/lib/sql_builder.py | 127 +++++++++++++++-- scripts/migration/migrate_assets.py | 2 +- scripts/migration/migrate_runtime.py | 5 +- scripts/migration/resources/README.md | 11 +- scripts/migration/resources/cards.csv.example | 4 - .../migration/resources/devices.csv.example | 4 - scripts/migration/scan_legacy.py | 82 ++++++++--- 13 files changed, 328 insertions(+), 114 deletions(-) delete mode 100644 scripts/migration/resources/cards.csv.example delete mode 100644 scripts/migration/resources/devices.csv.example diff --git a/scripts/migration/README.md b/scripts/migration/README.md index 0da132d..8a2262c 100644 --- a/scripts/migration/README.md +++ b/scripts/migration/README.md @@ -42,7 +42,7 @@ scripts/migration/ 把两个 CSV 放到 `resources/` 下(命名为 `cards.csv` / `devices.csv`): -- **cards.csv**:`iccid_19` 必填(所有卡前 19 位),`iccid_20` 视运营商必填或留空,`is_industry` 可选 +- **cards.csv**:`iccid_19`/`iccid_20` 至少填一个;仅填 `iccid_20` 时脚本自动派生前 19 位,但奇成主表只按 20 位精确查询;`is_industry` 可选 - **devices.csv**:`virtual_no` 必填,`sim_iccid_*` 引用 `cards.csv` 中的 ICCID(可填 19 或 20 位) 字段说明详见 [`resources/README.md`](resources/README.md)。 @@ -112,7 +112,9 @@ migration_user_id: 0 # 后台超级管理员 ID,作为 migration_batch_no: "MIGRATION-QICHENG-YYYYMMDD" # 本次迁移批次号,线上每批必须唯一 carriers: # 由 scan_legacy 自动发现 legacy_*,你只填 target_* - - legacy_category: 9 + - legacy_account_id: "72493DB64A2A4F01B5E59E49D58A2A34" + legacy_account_name: "联通10-1" + legacy_category: 9 # 奇成运营商大类,仅作辅助判断 legacy_category_name: "中国联通" target_carrier_id: 1585 ← 必填:新库 tb_carrier.id target_carrier_type: CUCC ← 必填:CMCC/CUCC/CTCC/CBN @@ -203,6 +205,10 @@ real_used_mb = total_bytes_cnt / (1 + flow_add_discount / 100) | `data_limit_mb` | 始终 `real_data_mb`(真总量) | | `data_usage_mb` | `LEAST(real_used_mb_floor, virtual_total_mb_snapshot)`(防卫上限) | +### 6.5 设备绑定卡的套餐归属 + +设备上的卡在新系统运行态会按设备维度判断有效套餐和复机条件，因此迁移时只用设备当前槽位卡（默认 `sim_iccid_1`）作为奇成套餐来源，生成一条 `usage_type=device`、`device_id=设备ID` 的套餐使用记录。其他槽位绑定卡不重复生成主套餐，避免同一设备出现多条并列生效主套餐。 + ## 7. 重要约束 - ❗ 所有奇成查询走 `SET TRANSACTION READ ONLY`,严禁误写。 @@ -215,7 +221,7 @@ real_used_mb = total_bytes_cnt / (1 + flow_add_discount / 100) | 现象 | 排查 | |------|------| | `scan_legacy.py` 报 `pymysql 缺失` | `pip install -r requirements.txt` | -| `errors.csv` 出现 `carrier_not_in_mapping` | `mapping.yaml.carriers` 没覆盖到这个 category,重跑 `scan_legacy.py` 自动补 | +| `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`,业务方人工确认是哪张物理卡 | diff --git a/scripts/migration/config/mapping.yaml.example b/scripts/migration/config/mapping.yaml.example index 65961a0..b673dd6 100644 --- a/scripts/migration/config/mapping.yaml.example +++ b/scripts/migration/config/mapping.yaml.example @@ -16,9 +16,12 @@ migration_user_id: 0 migration_batch_no: "MIGRATION-QICHENG-YYYYMMDD" # ============== 运营商映射 ============== -# scan_legacy 会按"奇成 tbl_card.category"分组写入,你只需填 target_* +# scan_legacy 会按"奇成 tbl_card.account_id/account_name"分组写入,你只需填 target_* +# legacy_category 只是奇成运营商大类,用于人工判断 target_carrier_type,不要把它当具体运营商。 carriers: - - legacy_category: 9 + - legacy_account_id: "72493DB64A2A4F01B5E59E49D58A2A34" + legacy_account_name: "联通10-1" + legacy_category: 9 legacy_category_name: "中国联通" target_carrier_id: null # 必填:新库 tb_carrier.id target_carrier_type: null # 必填:CMCC / CUCC / CTCC / CBN diff --git a/scripts/migration/lib/csv_loader.py b/scripts/migration/lib/csv_loader.py index 2e70e55..e32e1d3 100644 --- a/scripts/migration/lib/csv_loader.py +++ b/scripts/migration/lib/csv_loader.py @@ -20,9 +20,9 @@ DEVICE_HEADERS_AT_LEAST_ONE = ("imei", "virtual_no") class CardRow: """cards.csv 单行的解析结果(只含业务方决策字段)。 - 业务方必须同时提供 iccid_19 和 iccid_20(20 位运营商): - - iccid_19:前 19 位,所有卡必填,作为跨文件稳定 ID - - iccid_20:20 位完整 ICCID,CMCC/CUCC/CBN 必填、CTCC 必须留空 + 业务方至少提供 iccid_19 / iccid_20 之一: + - iccid_19:前 19 位,可作为奇成主表查询键 + - iccid_20:20 位完整 ICCID,固定 20 位运营商只填这一列即可 - 运营商类型与 iccid_20 的一致性在查完奇成后由 sql_builder 校验 新系统 iccid 唯一键 = iccid_20 优先,否则 iccid_19。 """ @@ -30,6 +30,7 @@ class CardRow: line_no: int # 原始 CSV 行号(从 1 开始,含表头) iccid_19: str iccid_20: str = "" # 留空表示 CTCC(查完奇成会再校验) + allow_iccid_19_lookup: bool = True # iccid_19 是否为业务方显式提供,可用于奇成主表兜底查询 is_industry: bool = False # 默认 False(普通卡) # 跨文件校验后填充 @@ -159,6 +160,16 @@ def _load_cards(path: Path, result: LoadResult) -> None: for idx, raw in enumerate(reader, start=2): iccid_19 = _normalize_cell(raw.get("iccid_19", "")) iccid_20 = _normalize_cell(raw.get("iccid_20", "")) + allow_iccid_19_lookup = bool(iccid_19) + + if iccid_20 and len(iccid_20) != 20: + result.errors.append( + ErrorRow("cards.csv", idx, "iccid_20", iccid_20, "iccid_20_length", + f"iccid_20 必须是 20 位,当前 {len(iccid_20)} 位") + ) + continue + if not iccid_19 and iccid_20: + iccid_19 = iccid_20[:19] err = iccid_utils.validate_iccid_19(iccid_19) if err is not None: @@ -186,7 +197,13 @@ def _load_cards(path: Path, result: LoadResult) -> None: continue result.cards.append( - CardRow(line_no=idx, iccid_19=iccid_19, iccid_20=iccid_20, is_industry=is_industry) + CardRow( + line_no=idx, + iccid_19=iccid_19, + iccid_20=iccid_20, + allow_iccid_19_lookup=allow_iccid_19_lookup, + is_industry=is_industry, + ) ) diff --git a/scripts/migration/lib/iccid_utils.py b/scripts/migration/lib/iccid_utils.py index 89b346b..89c5082 100644 --- a/scripts/migration/lib/iccid_utils.py +++ b/scripts/migration/lib/iccid_utils.py @@ -99,7 +99,10 @@ def prefix_19(iccid: str) -> str: def validate_iccid_19(iccid_19: str) -> Optional[ValidationError]: - """iccid_19 必须 19 位纯数字。""" + """iccid_19 必须是 19 位。 + + 奇成历史数据里存在带字母的 ICCID,迁移脚本只校验长度,不强制纯数字。 + """ if not iccid_19: return ValidationError(code="iccid_19_required", message="iccid_19 不能为空") if len(iccid_19) != 19: @@ -107,13 +110,11 @@ def validate_iccid_19(iccid_19: str) -> Optional[ValidationError]: code="iccid_19_length", message=f"iccid_19 必须是 19 位,当前 {len(iccid_19)} 位", ) - if not iccid_19.isdigit(): - return ValidationError(code="iccid_19_format", message="iccid_19 必须是纯数字") return None def validate_iccid_20(iccid_20: str, iccid_19: str) -> Optional[ValidationError]: - """iccid_20 可空(CTCC 卡);非空时必须 20 位纯数字且前 19 位 == iccid_19。""" + """iccid_20 可空(CTCC 卡);非空时必须 20 位且前 19 位 == iccid_19。""" if not iccid_20: return None if len(iccid_20) != 20: @@ -121,8 +122,6 @@ def validate_iccid_20(iccid_20: str, iccid_19: str) -> Optional[ValidationError] code="iccid_20_length", message=f"iccid_20 必须是 20 位,当前 {len(iccid_20)} 位", ) - if not iccid_20.isdigit(): - return ValidationError(code="iccid_20_format", message="iccid_20 必须是纯数字") if iccid_20[:19] != iccid_19: return ValidationError( code="iccid_20_prefix_mismatch", diff --git a/scripts/migration/lib/legacy_query.py b/scripts/migration/lib/legacy_query.py index 345f05f..829a3ef 100644 --- a/scripts/migration/lib/legacy_query.py +++ b/scripts/migration/lib/legacy_query.py @@ -14,6 +14,23 @@ from typing import Any, Iterable, Iterator, Optional BATCH_SIZE = 500 +def _normalize_iccid_pairs(iccid_pairs: Iterable[tuple]) -> list[tuple[str, str, bool]]: + """标准化 ICCID 查询参数。 + + tuple 第三项表示是否允许用 iccid_19 查询奇成 tbl_card 主表: + - 只填 iccid_20 的固定 20 位卡:false,避免 19 位前缀误撞其他卡 + - 显式填了 iccid_19:true,允许作为主表查询键或 20 位缺失时兜底 + """ + out: list[tuple[str, str, bool]] = [] + for item in iccid_pairs: + i19 = str(item[0] or "").strip() if len(item) > 0 else "" + i20 = str(item[1] or "").strip() if len(item) > 1 else "" + allow_i19_lookup = bool(item[2]) if len(item) > 2 else True + if i19: + out.append((i19, i20, allow_i19_lookup)) + return out + + def _import_pymysql(): """延迟导入 pymysql,这样脚本1(纯离线)不需要安装 pymysql 也能运行。""" try: @@ -110,6 +127,8 @@ class LegacyCardMeta: """ legacy_raw_iccid: str # 奇成实际存的 iccid_mark(可能 19 或 20 位) + account_id: str # tbl_card.account_id(开卡公司/运营商账号 ID) + account_name: str # tbl_card.account_name(开卡公司/运营商账号名称) category: int # tbl_card.category(运营商分类编号) category_name: str # tbl_vendor_category.category_name agent_id: str @@ -242,7 +261,7 @@ def fetch_current_packages(conn, iccids: Iterable[str]) -> dict[str, LegacyPacka return out -def fetch_card_usage(conn, iccid_pairs: Iterable[tuple[str, str]]) -> dict[str, LegacyCardUsage]: +def fetch_card_usage(conn, iccid_pairs: Iterable[tuple]) -> dict[str, LegacyCardUsage]: """查每张卡的累计已用流量,并同步取当前生效套餐的 flow_add_discount。 奇成 ``tbl_card.total_bytes_cnt`` 是虚用量(已被套餐虚比加成)。 @@ -251,21 +270,27 @@ def fetch_card_usage(conn, iccid_pairs: Iterable[tuple[str, str]]) -> dict[str, 上层通过 ``real_used_mb_floor`` 反算真用量写入新系统。 Args: - iccid_pairs: (iccid_19, iccid_20) 列表;iccid_20 可为空字符串。 + iccid_pairs: (iccid_19, iccid_20, allow_iccid_19_lookup) 列表。 Returns: {iccid_19: LegacyCardUsage}。 """ - pairs = [(p[0], p[1]) for p in iccid_pairs if p[0]] + pairs = _normalize_iccid_pairs(iccid_pairs) out: dict[str, LegacyCardUsage] = {} if not pairs: return out - full_to_19: dict[str, str] = {} - for i19, i20 in pairs: - full_to_19[i19] = i19 + full_to_19: dict[str, tuple[str, int]] = {} + related_keys_by_19: dict[str, set[str]] = {} + for i19, i20, allow_i19_lookup in pairs: + # tbl_card 主表同时查优先键和兜底键,结果选择时按 rank: + # 0=业务方完整 ICCID 精确命中,1=20 位卡在奇成主表只存 19 位时兜底。 + full_to_19[i20 or i19] = (i19, 0) + if i20 and allow_i19_lookup: + full_to_19[i19] = (i19, 1) + related_keys_by_19.setdefault(i19, set()).add(i19) if i20: - full_to_19[i20] = i19 + related_keys_by_19[i19].add(i20) full_set = sorted(full_to_19.keys()) # tbl_card 主表用 iccid_mark IN 全长精确匹配;tbl_card_life 从表用 @@ -293,25 +318,30 @@ def fetch_card_usage(conn, iccid_pairs: Iterable[tuple[str, str]]) -> dict[str, 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): - cur.execute(sql, (tuple(batch), tuple(batch))) + 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]}) + cur.execute(sql, (tuple(related_batch), tuple(batch))) for row in cur.fetchall(): legacy_iccid = row["iccid_mark"] if not legacy_iccid: continue - i19 = full_to_19.get(legacy_iccid) - if i19 is None: + matched = full_to_19.get(legacy_iccid) + if matched is None: continue - # 取第一个匹配,后续 fetch_card_meta 已经会把重复命中的卡标记为 errors - if i19 in out: - continue - out[i19] = LegacyCardUsage( - iccid=i19, - 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")), - ) + i19, rank = matched + hits.setdefault(i19, []).append((rank, row)) + for i19, 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, + 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")), + ) return out @@ -350,18 +380,19 @@ def fetch_commission_accounts(conn, agent_ids: Iterable[str]) -> dict[str, Legac def fetch_card_meta( conn, - iccid_pairs: Iterable[tuple[str, str]], + iccid_pairs: Iterable[tuple], ) -> tuple[dict[str, LegacyCardMeta], set[str]]: """查每张卡的元数据(卡本体 + 当前生效套餐 + 虚拟号)。 匹配策略: - - tbl_card 主表用 `iccid_mark IN (iccid_19 ∪ iccid_20)` 全长精确; - 这样不会因前 19 位相同(校验位不同的两张 20 位卡)而误匹配多张 + - tbl_card 主表同时查优先键和显式允许的兜底键:有 iccid_20 时优先取 + 20 位精确命中;只有业务方也显式填了 iccid_19 时,才允许退回 19 位。 + 只填 iccid_20 的固定 20 位卡绝不拿 19 位前缀查主表。 - tbl_card_life / tbl_virtual_number 从表用 LEFT(iccid_mark, 19) 兜底, 应对奇成两张表 iccid 长度不一致的脏数据 Args: - iccid_pairs: (iccid_19, iccid_20) 列表;iccid_20 可为空字符串(CTCC 卡)。 + iccid_pairs: (iccid_19, iccid_20, allow_iccid_19_lookup) 列表。 Returns: (metas, duplicate_iccid_19): @@ -369,23 +400,27 @@ def fetch_card_meta( - duplicate_iccid_19: 一份业务方 iccid_19 命中奇成 tbl_card 多条记录的 set, 这些卡不进 metas,sql_builder 会写 errors.csv 阻断 """ - pairs = [(p[0], p[1]) for p in iccid_pairs if p[0]] + pairs = _normalize_iccid_pairs(iccid_pairs) if not pairs: return {}, set() - # 全长 ICCID → 业务方 iccid_19。同时维护 iccid_20 优先级: - # 奇成 iccid_mark 命中时,优先取等于 iccid_20 的映射(20 位精确), - # 找不到再退到 iccid_19(奇成存了 19 位的兜底)。 - by_full: dict[str, str] = {} # 任意长度 ICCID → iccid_19 - for i19, i20 in pairs: - by_full[i19] = i19 + by_full: dict[str, tuple[str, int]] = {} # tbl_card 查询键 → (iccid_19, rank) + related_keys_by_19: dict[str, set[str]] = {} + for i19, i20, allow_i19_lookup in pairs: + by_full[i20 or i19] = (i19, 0) + if i20 and allow_i19_lookup: + by_full[i19] = (i19, 1) + related_keys_by_19.setdefault(i19, set()).add(i19) if i20: - by_full[i20] = i19 + related_keys_by_19[i19].add(i20) full_set = sorted(by_full.keys()) 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, @@ -415,33 +450,42 @@ def fetch_card_meta( """ # 收集每个 iccid_19 命中的奇成行(可能多条 = 脏数据) - hits: dict[str, list[dict]] = {} + hits: dict[str, list[tuple[int, dict]]] = {} with conn.cursor() as cur: for batch in _chunks(full_set): - cur.execute(sql, (tuple(batch), tuple(batch))) + 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]}) + cur.execute(sql, (tuple(related_batch), tuple(batch))) for row in cur.fetchall(): legacy_iccid = row["iccid"] if not legacy_iccid: continue - i19 = by_full.get(legacy_iccid) - if i19 is None: - # WHERE 已经限定 IN,理论不会出现;保险起见再按前 19 位反查 - i19 = by_full.get(legacy_iccid[:19]) - if i19 is None: - continue - hits.setdefault(i19, []).append(row) + matched = by_full.get(legacy_iccid) + if matched is None: + continue + i19, rank = matched + hits.setdefault(i19, []).append((rank, row)) metas: dict[str, LegacyCardMeta] = {} duplicates: set[str] = set() for i19, rows in hits.items(): - if len(rows) > 1: + 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 命中多张卡。 + 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 + if len(rows_by_card_id) > 1: duplicates.add(i19) continue - row = rows[0] + 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( 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 "", diff --git a/scripts/migration/lib/mapping_loader.py b/scripts/migration/lib/mapping_loader.py index 34a89ef..b1fff0e 100644 --- a/scripts/migration/lib/mapping_loader.py +++ b/scripts/migration/lib/mapping_loader.py @@ -29,6 +29,8 @@ class MissingTargetError(Exception): @dataclass(frozen=True) class CarrierMapping: + legacy_account_id: str + legacy_account_name: str legacy_category: int legacy_category_name: str target_carrier_id: Optional[int] @@ -44,7 +46,7 @@ class CarrierMapping: ) if v in (None, "", 0) ] if missing: - raise MissingTargetError("carrier", str(self.legacy_category), self.legacy_category_name, missing) + raise MissingTargetError("carrier", self.legacy_account_id, self.legacy_account_name, missing) return self @@ -80,13 +82,13 @@ class Mapping: migration_user_id: int = 0 migration_batch_no: str = "MIGRATION-QICHENG" - carriers: dict[int, CarrierMapping] = field(default_factory=dict) + carriers: dict[str, CarrierMapping] = field(default_factory=dict) packages: dict[str, PackageMapping] = field(default_factory=dict) series: dict[str, SeriesMapping] = field(default_factory=dict) agents: dict[str, AgentMapping] = field(default_factory=dict) - def lookup_carrier(self, legacy_category: int) -> Optional[CarrierMapping]: - return self.carriers.get(int(legacy_category)) + def lookup_carrier(self, legacy_account_id: str) -> Optional[CarrierMapping]: + return self.carriers.get(str(legacy_account_id or "").strip()) def lookup_package(self, legacy_meal_id: str) -> Optional[PackageMapping]: return self.packages.get(str(legacy_meal_id).strip()) @@ -130,14 +132,20 @@ def load_mapping(config_dir: Path) -> Mapping: ) for item in raw.get("carriers") or []: + legacy_account_id = _opt_str(item.get("legacy_account_id")) + # 旧版按 category 映射的配置会把多个实际运营商账号合成一项,不能安全复用。 + if not legacy_account_id: + continue c = CarrierMapping( - legacy_category=int(item["legacy_category"]), + legacy_account_id=legacy_account_id, + legacy_account_name=str(item.get("legacy_account_name") or ""), + legacy_category=int(item.get("legacy_category") or 0), legacy_category_name=str(item.get("legacy_category_name") or ""), target_carrier_id=_opt_int(item.get("target_carrier_id")), target_carrier_type=_opt_str(item.get("target_carrier_type")), target_carrier_name=_opt_str(item.get("target_carrier_name")), ) - m.carriers[c.legacy_category] = c + m.carriers[c.legacy_account_id] = c for item in raw.get("packages") or []: p = PackageMapping( @@ -178,13 +186,15 @@ def merge_and_save(config_dir: Path, mapping: Mapping) -> Path: "migration_batch_no": mapping.migration_batch_no, "carriers": [ { + "legacy_account_id": c.legacy_account_id, + "legacy_account_name": c.legacy_account_name, "legacy_category": c.legacy_category, "legacy_category_name": c.legacy_category_name, "target_carrier_id": c.target_carrier_id, "target_carrier_type": c.target_carrier_type, "target_carrier_name": c.target_carrier_name, } - for c in sorted(mapping.carriers.values(), key=lambda x: x.legacy_category) + for c in sorted(mapping.carriers.values(), key=lambda x: (x.legacy_account_name, x.legacy_account_id)) ], "packages": [ { diff --git a/scripts/migration/lib/sql_builder.py b/scripts/migration/lib/sql_builder.py index f6e8f2b..3c195d8 100644 --- a/scripts/migration/lib/sql_builder.py +++ b/scripts/migration/lib/sql_builder.py @@ -43,6 +43,8 @@ from .mapping_loader import Mapping, MissingTargetError ORDER_NO_PREFIX_CARD = "MIG-CARD-" ALLOCATION_NO_PREFIX = "MIG-ALLOC-" METADATA_SOURCE = "qicheng" +STOP_REASON_NOT_REALNAME = "not_realname" +STOP_REASON_NO_PACKAGE = "no_package" # ---------------- 通用工具 ---------------- @@ -115,6 +117,53 @@ def _resolve_device_shop_code( return mapping.lookup_agent_shop_code(meta.agent_id) or "" +def _initial_polling_stop_reason(card: CardRow) -> str: + """生成迁移卡初始停机原因。 + + 导入时卡默认 network_status=0。停复机服务只会自动复机轮询系统写入的停机原因， + 因此不能留空；普通卡先等待实名轮询确认，行业卡无需实名则交给套餐轮询重评估。 + """ + if card.is_industry: + return STOP_REASON_NO_PACKAGE + return STOP_REASON_NOT_REALNAME + + +def _device_package_source_iccid(device: DeviceRow) -> str: + """返回设备级套餐迁移使用的来源卡。 + + 设备套餐在新系统挂到 device_id；奇成套餐仍来自卡生命周期表。 + 优先取当前槽位，缺失时回退 slot1，再回退任意已绑定卡。 + """ + if not device.sim_iccids: + return "" + return ( + device.sim_iccids.get(device.current_slot) + or device.sim_iccids.get(1) + or next(iter(device.sim_iccids.values()), "") + ) + + +def _device_by_virtual_no(devices: list[DeviceRow]) -> dict[str, DeviceRow]: + return {d.virtual_no: d for d in devices} + + +def _runtime_asset_for_card(card: CardRow, devices_by_virtual_no: dict[str, DeviceRow]) -> tuple[str, str] | None: + """返回运行态套餐/订单归属：(asset_type, asset_identifier)。 + + 独立卡归属 iot_card；设备绑定卡只用设备当前槽位卡生成一条设备级套餐， + 其他槽位不重复生成，避免同一设备出现多条并列主套餐。 + """ + if not card.bound_device_virtual_no: + return "iot_card", card.iccid_full + + 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): + return None + return "device", device.virtual_no + + # ============================================================ # 脚本1:资产导入 # ============================================================ @@ -181,24 +230,26 @@ def _filter_cards_by_carrier( "not_in_legacy", "奇成 tbl_card 中找不到该 ICCID(iccid_19/iccid_20 都没命中)", )) continue - if not meta.category: + if not meta.account_id: errors.append(ErrorRow( "cards.csv", c.line_no, "iccid_19", c.iccid_19, - "no_category", "奇成 tbl_card.category 为空,无法定位运营商映射", + "no_carrier_account", "奇成 tbl_card.account_id 为空,无法定位具体运营商账号映射", )) continue - carrier = mapping.lookup_carrier(meta.category) + carrier = mapping.lookup_carrier(meta.account_id) if carrier is None: errors.append(ErrorRow( - "cards.csv", c.line_no, "category", str(meta.category), - "carrier_not_in_mapping", f"mapping.yaml.carriers 未配置 category={meta.category} ({meta.category_name})", + "cards.csv", c.line_no, "account_id", meta.account_id, + "carrier_not_in_mapping", + f"mapping.yaml.carriers 未配置 account_id={meta.account_id} " + f"({meta.account_name}, category={meta.category}/{meta.category_name})", )) continue try: carrier.require_target() except MissingTargetError as exc: errors.append(ErrorRow( - "cards.csv", c.line_no, "category", str(meta.category), + "cards.csv", c.line_no, "account_id", meta.account_id, "carrier_target_missing", str(exc), )) continue @@ -226,7 +277,7 @@ def _write_iot_cards( f.write(_file_header("step1_01 卡资产导入 (tb_iot_card)")) for c in cards: meta = card_metas[c.iccid_19] - carrier = mapping.lookup_carrier(meta.category) + carrier = mapping.lookup_carrier(meta.account_id) assert carrier is not None # 已被 _filter_cards_by_carrier 保证 # 业务方直接给的两列;iccid 唯一键 = 有 iccid_20 用 iccid_20,否则 iccid_19 @@ -254,7 +305,7 @@ def _write_iot_cards( f" {_sql_str(card_category)}, {_sql_str(mapping.migration_batch_no)},\n" " 1, 0, 0, 0,\n" " 1, 1, TRUE,\n" - " 'after_order', '',\n" + f" 'after_order', {_sql_str(_initial_polling_stop_reason(c))},\n" f" {_sql_str(meta.msisdn) if meta.msisdn else 'NULL'}, " f"{_sql_str(meta.virtual_no) if meta.virtual_no else 'NULL'}, " f"{_shop_id_sub(shop_code)}, {series_id_expr},\n" @@ -509,6 +560,7 @@ def write_step2( output_dir: Path, *, cards: list[CardRow], + devices: list[DeviceRow], mapping: Mapping, card_metas: dict[str, LegacyCardMeta], usage_snapshots: dict[str, LegacyCardUsage], @@ -518,6 +570,7 @@ def write_step2( """生成 step2_* SQL + warnings.csv + summary.txt。""" user_id = mapping.migration_user_id 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) for c in cards: @@ -549,9 +602,9 @@ def write_step2( "(status=1),无法精确反算真用量,按 flow_add_discount=0 处理(真=虚)", )) - _write_migration_orders(output_dir / "step2_01_migration_orders.sql", cards, resolved_packages, user_id) + _write_migration_orders(output_dir / "step2_01_migration_orders.sql", cards, resolved_packages, devices_by_virtual_no, user_id) _write_package_usages( - output_dir / "step2_02_package_usages.sql", cards, resolved_packages, usage_snapshots, user_id, + output_dir / "step2_02_package_usages.sql", cards, resolved_packages, devices_by_virtual_no, usage_snapshots, user_id, ) _write_card_data_usage(output_dir / "step2_03_card_data_usage_update.sql", cards, usage_snapshots) @@ -571,6 +624,7 @@ def write_step2( output_dir / "summary.txt", cards=cards, resolved=resolved_packages, + devices_by_virtual_no=devices_by_virtual_no, agents=eligible_agents, warnings=warnings, ) @@ -590,6 +644,7 @@ def _write_migration_orders( path: Path, cards: list[CardRow], resolved_packages: dict[str, tuple[int, str]], + devices_by_virtual_no: dict[str, DeviceRow], user_id: int, ) -> None: """伪订单关键字段(详见方案文档 10.2):source=admin / buyer_type=personal / @@ -600,7 +655,36 @@ def _write_migration_orders( for c in cards: if c.iccid_19 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 order_no = _make_card_order_no(c.iccid_full) + if asset_type == "device": + f.write( + "INSERT INTO tb_order (\n" + " order_no, order_type, buyer_type, buyer_id,\n" + " device_id, asset_identifier,\n" + " total_amount, actual_paid_amount,\n" + " payment_method, payment_status, paid_at,\n" + " commission_status, commission_result, commission_config_version,\n" + " source, generation,\n" + " operator_account_type, operator_account_name,\n" + " creator, updater, created_at, updated_at\n" + ")\n" + f"SELECT {_sql_str(order_no)}, 'device', 'personal', 0,\n" + f" d.id, {_sql_str(asset_identifier)},\n" + " 0, 0,\n" + " 'offline', 2, NOW(),\n" + " 2, 2, 0,\n" + " 'admin', 1,\n" + " 'platform', '奇成数据迁移',\n" + f" {user_id}, {user_id}, NOW(), NOW()\n" + "FROM tb_device d\n" + f"WHERE d.virtual_no = {_sql_str(asset_identifier)} AND d.deleted_at IS NULL\n" + "ON CONFLICT (order_no) WHERE deleted_at IS NULL DO NOTHING;\n\n" + ) + continue f.write( "INSERT INTO tb_order (\n" " order_no, order_type, buyer_type, buyer_id,\n" @@ -630,6 +714,7 @@ def _write_package_usages( path: Path, cards: list[CardRow], resolved_packages: dict[str, tuple[int, str]], + devices_by_virtual_no: dict[str, DeviceRow], usage_snapshots: dict[str, LegacyCardUsage], user_id: int, ) -> None: @@ -638,6 +723,10 @@ def _write_package_usages( for c in cards: if c.iccid_19 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] order_no = _make_card_order_no(c.iccid_full) expires_expr = f"{_sql_str(expire)}::timestamp" if expire else "NULL" @@ -648,10 +737,13 @@ def _write_package_usages( "THEN p.virtual_data_mb ELSE p.real_data_mb END" ) usage_expr = f"LEAST({used_mb}, {effective_limit_expr})" + carrier_column = "device_id" if asset_type == "device" else "iot_card_id" + carrier_select = "o.device_id" if asset_type == "device" else "o.iot_card_id" + usage_type = "device" if asset_type == "device" else "single_card" f.write( "INSERT INTO tb_package_usage (\n" " order_id, order_no, package_id, usage_type,\n" - " iot_card_id,\n" + f" {carrier_column},\n" " data_limit_mb, data_usage_mb,\n" " virtual_total_mb_snapshot, display_gain_ratio_snapshot, enable_virtual_data_snapshot,\n" " status, priority, generation,\n" @@ -661,8 +753,8 @@ def _write_package_usages( " data_reset_cycle,\n" " creator, updater, created_at, updated_at\n" ")\n" - "SELECT o.id, o.order_no, p.id, 'single_card',\n" - " o.iot_card_id,\n" + f"SELECT o.id, o.order_no, p.id, {_sql_str(usage_type)},\n" + f" {carrier_select},\n" f" p.real_data_mb, {usage_expr},\n" f" {effective_limit_expr},\n" " CASE WHEN p.enable_virtual_data AND p.virtual_data_mb > 0 THEN p.virtual_ratio ELSE 1.0 END,\n" @@ -832,13 +924,18 @@ def _write_warnings(path: Path, warnings: list[tuple[str, str, str]]) -> None: writer.writerow([kind, key, message]) -def _write_summary_step2(path: Path, *, cards, resolved, agents, warnings) -> None: +def _write_summary_step2(path: Path, *, cards, resolved, devices_by_virtual_no, agents, warnings) -> None: + 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 + ) lines = [ "奇成迁移 - 脚本2 关联数据导入摘要", "=" * 48, f"生成时间 : {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", f"卡总数 : {len(cards)}", - f" 生成伪订单+套餐 : {len(resolved)}", + f" 生成伪订单+套餐 : {generated_count}", f"代理钱包初始化 : {len(agents)}", f"警告条目 : {len(warnings)} (详见 warnings.csv)", ] diff --git a/scripts/migration/migrate_assets.py b/scripts/migration/migrate_assets.py index 7b23766..253ee6e 100644 --- a/scripts/migration/migrate_assets.py +++ b/scripts/migration/migrate_assets.py @@ -43,7 +43,7 @@ def main() -> int: cards, devices, errors = csv_loader.load_assets(resources_dir) # 连奇成查卡元数据(运营商分类、agent_id、msisdn、virtual_no、当前生效套餐) - iccid_pairs = [(c.iccid_19, c.iccid_20) for c in cards] + 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) diff --git a/scripts/migration/migrate_runtime.py b/scripts/migration/migrate_runtime.py index 5830b6c..3b1f5c1 100644 --- a/scripts/migration/migrate_runtime.py +++ b/scripts/migration/migrate_runtime.py @@ -37,9 +37,9 @@ 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) - iccid_pairs = [(c.iccid_19, c.iccid_20) for c in cards] + 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) @@ -52,6 +52,7 @@ def main() -> int: warnings = sql_builder.write_step2( output_dir, cards=cards, + devices=devices, mapping=mapping, card_metas=card_metas, usage_snapshots=usage_snapshots, diff --git a/scripts/migration/resources/README.md b/scripts/migration/resources/README.md index 0cad7e1..113888f 100644 --- a/scripts/migration/resources/README.md +++ b/scripts/migration/resources/README.md @@ -7,16 +7,19 @@ ## cards.csv 卡资产清单 -每张要迁移的卡一行。**必须同时提供 iccid_19 和 iccid_20**(因为奇成历史数据 -长度不可信,业务方提供两列作为权威值)。 +每张要迁移的卡一行。`iccid_19` / `iccid_20` 至少填一个。只有 20 位完整 ICCID +时可只填 `iccid_20`,脚本会自动派生前 19 位作为内部稳定 ID。 +只填 `iccid_20` 的卡,脚本查询奇成主表时只按 20 位精确匹配;只有同时显式填写 +`iccid_19` 和 `iccid_20` 时,才允许 20 位未命中后退回 19 位查询。 | 列名 | 必填 | 说明 | |------|------|------| -| `iccid_19` | 是 | ICCID 前 19 位(所有卡都有,作为跨文件稳定 ID;必须 19 位纯数字) | +| `iccid_19` | 条件必填 | ICCID 前 19 位(内部稳定 ID;必须 19 位;允许奇成历史字母位)。`iccid_20` 已填时可留空 | | `iccid_20` | 视运营商 | 完整 20 位 ICCID;CMCC/CUCC/CBN(20 位运营商) **必填**,CTCC(19 位)**必须留空**;非空时前 19 位必须等于 `iccid_19` | | `is_industry` | 否 | 是否行业卡,留空 = `false`(普通卡) | -运营商类型从奇成 `tbl_card.category` 自动推断,在查完奇成后会做一致性校验: +具体运营商从奇成 `tbl_card.account_id/account_name` 扫描后写入 `mapping.yaml.carriers`; +`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` 阻断,请业务方人工确认 diff --git a/scripts/migration/resources/cards.csv.example b/scripts/migration/resources/cards.csv.example deleted file mode 100644 index c4b2147..0000000 --- a/scripts/migration/resources/cards.csv.example +++ /dev/null @@ -1,4 +0,0 @@ -iccid_19,iccid_20,is_industry -8985200026333877243,89852000263338772439, -8986062300000000001,,true -8986006654000000000,89860066540000000002, diff --git a/scripts/migration/resources/devices.csv.example b/scripts/migration/resources/devices.csv.example deleted file mode 100644 index d7bcd4d..0000000 --- a/scripts/migration/resources/devices.csv.example +++ /dev/null @@ -1,4 +0,0 @@ -imei,virtual_no,sim_iccid_1,sim_iccid_2,sim_iccid_3,sim_iccid_4 -860000000000001,DEV-202604-0001,89852000263338772439,8986062300000000001,, -,DEV-202604-0002,89860066540000000002,,, -860000000000003,,89860000000000004,,, diff --git a/scripts/migration/scan_legacy.py b/scripts/migration/scan_legacy.py index e0ee2e7..368f0ae 100644 --- a/scripts/migration/scan_legacy.py +++ b/scripts/migration/scan_legacy.py @@ -3,7 +3,7 @@ 工作流: 1. 读 resources/cards.csv 拿到 iccid 列表 - 2. 连奇成只读查询每张卡的 category / agent_id / 当前生效套餐 / 套餐系列 + 2. 连奇成只读查询每张卡的 account/category / agent_id / 当前生效套餐 / 套餐系列 3. 加载已有的 config/mapping.yaml(若没有则创建空的) 4. 把扫描到的 legacy_* 增量补到 mapping 里(已有项的 target_* 字段保留,新项的 target_* 为 null) 5. 写回 config/mapping.yaml @@ -31,14 +31,17 @@ from lib.mapping_loader import ( # noqa: E402 ) -def _read_iccid_pairs(cards_csv: Path) -> list[tuple[str, str]]: +def _read_iccid_pairs(cards_csv: Path) -> list[tuple[str, str, bool]]: """从 cards.csv 读 iccid_19 + iccid_20 列(只关心这两列,其他列忽略)。 - 返回 (iccid_19, iccid_20) 列表;iccid_20 可为空字符串(CTCC 卡留空)。 + 返回 (iccid_19, iccid_20, allow_iccid_19_lookup) 列表。 + 只填 iccid_20 时,iccid_19 从前 19 位派生,但不用于奇成主表兜底查询。 """ if not cards_csv.exists(): raise FileNotFoundError(f"找不到 {cards_csv},请按 resources/README.md 准备") - pairs: dict[str, str] = {} + pairs: dict[str, tuple[str, bool]] = {} + first_seen_line: dict[str, int] = {} + errors: list[str] = [] with cards_csv.open("r", encoding="utf-8-sig", newline="") as f: reader = csv.DictReader(f) headers = reader.fieldnames or [] @@ -46,12 +49,37 @@ def _read_iccid_pairs(cards_csv: Path) -> list[tuple[str, str]]: raise ValueError(f"{cards_csv} 必须包含 iccid_19 列") if "iccid_20" not in headers: raise ValueError(f"{cards_csv} 必须包含 iccid_20 列(CTCC 卡留空)") - for row in reader: + for line_no, row in enumerate(reader, start=2): i19 = (row.get("iccid_19") or "").strip() i20 = (row.get("iccid_20") or "").strip() - if i19: - pairs[i19] = i20 - return sorted(pairs.items()) + allow_i19_lookup = bool(i19) + if not i19 and not i20 and not any((v or "").strip() for v in row.values()): + continue + if i20 and len(i20) != 20: + errors.append(f"第 {line_no} 行 iccid_20 必须是 20 位: {i20!r}") + continue + if not i19: + if not i20: + errors.append(f"第 {line_no} 行 iccid_19 和 iccid_20 不能同时为空") + continue + i19 = i20[:19] + if len(i19) != 19: + errors.append(f"第 {line_no} 行 iccid_19 必须是 19 位: {i19!r}") + continue + 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}") + continue + first_seen_line[i19] = line_no + pairs[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())] def main() -> int: @@ -64,9 +92,13 @@ def main() -> int: config_dir = (base / args.config_dir).resolve() resources_dir = (base / args.resources_dir).resolve() - iccid_pairs = _read_iccid_pairs(resources_dir / "cards.csv") + try: + iccid_pairs = _read_iccid_pairs(resources_dir / "cards.csv") + except (FileNotFoundError, ValueError) as exc: + print(str(exc), file=sys.stderr) + return 1 if not iccid_pairs: - print("cards.csv 中没有 iccid_19,无需扫描", file=sys.stderr) + print("cards.csv 中没有有效 ICCID,无需扫描", file=sys.stderr) return 1 print(f"读到 {len(iccid_pairs)} 个 ICCID,连接奇成扫描元数据...") @@ -107,21 +139,28 @@ def _merge( ) for meta in card_metas.values(): - # carrier - if meta.category and meta.category not in merged.carriers: - merged.carriers[meta.category] = CarrierMapping( + # carrier:按开卡公司/运营商账号映射,category 仅作为辅助信息。 + if meta.account_id and meta.account_id not in merged.carriers: + merged.carriers[meta.account_id] = CarrierMapping( + legacy_account_id=meta.account_id, + legacy_account_name=meta.account_name, legacy_category=meta.category, legacy_category_name=meta.category_name, target_carrier_id=None, target_carrier_type=None, target_carrier_name=None, ) - elif meta.category in merged.carriers and not merged.carriers[meta.category].legacy_category_name: - # 补全 legacy_category_name(如果之前是空的) - old = merged.carriers[meta.category] - merged.carriers[meta.category] = CarrierMapping( + elif meta.account_id in merged.carriers and ( + not merged.carriers[meta.account_id].legacy_account_name + or not merged.carriers[meta.account_id].legacy_category_name + ): + # 补全 legacy_* 快照(如果之前是空的) + old = merged.carriers[meta.account_id] + merged.carriers[meta.account_id] = CarrierMapping( + legacy_account_id=old.legacy_account_id, + legacy_account_name=meta.account_name or old.legacy_account_name, legacy_category=old.legacy_category, - legacy_category_name=meta.category_name, + legacy_category_name=meta.category_name or old.legacy_category_name, target_carrier_id=old.target_carrier_id, target_carrier_type=old.target_carrier_type, target_carrier_name=old.target_carrier_name, @@ -154,9 +193,12 @@ def _merge( def _collect_todo(mapping: Mapping) -> dict[str, list[str]]: """收集所有待填写的项,按类别分组。""" todo: dict[str, list[str]] = {"carriers": [], "packages": [], "agents": []} - for c in sorted(mapping.carriers.values(), key=lambda x: x.legacy_category): + for c in sorted(mapping.carriers.values(), key=lambda x: (x.legacy_account_name, x.legacy_account_id)): if not c.target_carrier_id or not c.target_carrier_type or not c.target_carrier_name: - todo["carriers"].append(f"category={c.legacy_category} ({c.legacy_category_name})") + todo["carriers"].append( + f"account_id={c.legacy_account_id} ({c.legacy_account_name}, " + f"category={c.legacy_category}/{c.legacy_category_name})" + ) for p in sorted(mapping.packages.values(), key=lambda x: x.legacy_meal_id): if not p.target_package_id: todo["packages"].append(f"meal_id={p.legacy_meal_id} ({p.legacy_meal_name})")