192 lines
7.9 KiB
Python
192 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
|
"""扫描奇成,自动生成/增量补全 config/mapping.yaml。
|
|
|
|
工作流:
|
|
1. 读 resources/cards.csv 拿到 iccid 列表
|
|
2. 连奇成只读查询每张卡的 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]]:
|
|
"""从 cards.csv 读 iccid_19 + iccid_20 列(只关心这两列,其他列忽略)。
|
|
|
|
返回 (iccid_19, iccid_20) 列表;iccid_20 可为空字符串(CTCC 卡留空)。
|
|
"""
|
|
if not cards_csv.exists():
|
|
raise FileNotFoundError(f"找不到 {cards_csv},请按 resources/README.md 准备")
|
|
pairs: dict[str, 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 row in reader:
|
|
i19 = (row.get("iccid_19") or "").strip()
|
|
i20 = (row.get("iccid_20") or "").strip()
|
|
if i19:
|
|
pairs[i19] = i20
|
|
return 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()
|
|
|
|
iccid_pairs = _read_iccid_pairs(resources_dir / "cards.csv")
|
|
if not iccid_pairs:
|
|
print("cards.csv 中没有 iccid_19,无需扫描", 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
|
|
if meta.category and meta.category not in merged.carriers:
|
|
merged.carriers[meta.category] = CarrierMapping(
|
|
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(
|
|
legacy_category=old.legacy_category,
|
|
legacy_category_name=meta.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_category):
|
|
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})")
|
|
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())
|