#!/usr/bin/env python3 """扫描奇成,自动生成/增量补全 config/mapping.yaml。 工作流: 1. 读 resources/cards.csv 拿到 iccid 列表 2. 连奇成只读查询每张卡的 account/category / agent_id / 当前生效套餐 / 套餐系列 3. 加载已有的 config/mapping.yaml(若没有则创建空的) 4. 把扫描到的 legacy_* 增量补到 mapping 里(已有项的 target_* 字段保留,新项的 target_* 为 null) 5. 写回 config/mapping.yaml 6. 输出"待人工填写"清单到 console + config/mapping_todo.txt 用法: python3 scan_legacy.py [--config-dir config] [--resources-dir resources] """ from __future__ import annotations import argparse import csv import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from lib import legacy_query, mapping_loader # noqa: E402 from lib.mapping_loader import ( # noqa: E402 AgentMapping, CarrierMapping, Mapping, PackageMapping, SeriesMapping, ) def _read_iccid_pairs(cards_csv: Path) -> list[tuple[str, str, bool]]: """从 cards.csv 读 iccid_19 + iccid_20 列(只关心这两列,其他列忽略)。 返回 (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, 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 [] if "iccid_19" not in headers: raise ValueError(f"{cards_csv} 必须包含 iccid_19 列") if "iccid_20" not in headers: raise ValueError(f"{cards_csv} 必须包含 iccid_20 列(CTCC 卡留空)") for line_no, row in enumerate(reader, start=2): i19 = (row.get("iccid_19") or "").strip() i20 = (row.get("iccid_20") or "").strip() 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: parser = argparse.ArgumentParser(description="奇成元数据预扫描,自动生成 mapping.yaml") parser.add_argument("--config-dir", default="config", help="yaml 配置目录") parser.add_argument("--resources-dir", default="resources", help="CSV 资源目录") args = parser.parse_args() base = Path(__file__).resolve().parent config_dir = (base / args.config_dir).resolve() resources_dir = (base / args.resources_dir).resolve() 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,无需扫描", file=sys.stderr) return 1 print(f"读到 {len(iccid_pairs)} 个 ICCID,连接奇成扫描元数据...") 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) 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): print(f" - {x}") print("这些卡 migrate_assets 阶段会写 errors.csv 阻断,请业务方人工确认。") existing = mapping_loader.load_mapping(config_dir) merged = _merge(existing, card_metas, meal_metas) out_path = mapping_loader.merge_and_save(config_dir, merged) todo = _collect_todo(merged) _write_todo(config_dir / "mapping_todo.txt", todo) _print_summary(out_path, card_metas, meal_metas, todo) return 0 def _merge( existing: Mapping, card_metas: dict[str, "legacy_query.LegacyCardMeta"], meal_metas: dict[str, "legacy_query.LegacyMealMeta"], ) -> Mapping: """把扫描到的 legacy_* 增量合并到现有 mapping(保留已填的 target_*)。""" merged = Mapping( migration_user_id=existing.migration_user_id, migration_batch_no=existing.migration_batch_no, carriers=dict(existing.carriers), packages=dict(existing.packages), series=dict(existing.series), agents=dict(existing.agents), ) for meta in card_metas.values(): # 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.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 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, ) # agent if meta.agent_id and meta.agent_id not in merged.agents: merged.agents[meta.agent_id] = AgentMapping( legacy_agent_id=meta.agent_id, legacy_agent_name=meta.agent_name, target_shop_code=None, ) for mm in meal_metas.values(): if mm.meal_id not in merged.packages: merged.packages[mm.meal_id] = PackageMapping( legacy_meal_id=mm.meal_id, legacy_meal_name=mm.meal_name, target_package_id=None, ) if mm.series_id and mm.series_id not in merged.series: merged.series[mm.series_id] = SeriesMapping( legacy_series_id=mm.series_id, legacy_series_name=mm.series_name, target_series_id=None, ) return merged 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_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"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})") for a in sorted(mapping.agents.values(), key=lambda x: x.legacy_agent_id): if not (a.target_shop_code or "").strip(): todo["agents"].append(f"agent_id={a.legacy_agent_id} ({a.legacy_agent_name})") return todo def _write_todo(path: Path, todo: dict[str, list[str]]) -> None: lines = ["奇成迁移映射待填写清单", "=" * 40, ""] for kind in ("carriers", "packages", "agents"): lines.append(f"## {kind} (待填: {len(todo[kind])})") for item in todo[kind]: lines.append(f" - {item}") lines.append("") path.write_text("\n".join(lines), encoding="utf-8") def _print_summary(out_path: Path, card_metas, meal_metas, todo: dict[str, list[str]]) -> None: print(f"扫描完成,mapping 写入: {out_path}") print(f" 奇成卡命中 : {len(card_metas)}") print(f" 奇成套餐命中 : {len(meal_metas)}") print(f" 待填 carriers : {len(todo['carriers'])}") print(f" 待填 packages : {len(todo['packages'])}") print(f" 待填 agents : {len(todo['agents'])}") if any(todo.values()): print(f"详见: {out_path.parent / 'mapping_todo.txt'}") if __name__ == "__main__": sys.exit(main())