break
370fd3e67f
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 10m49s
726 lines
30 KiB
Python
726 lines
30 KiB
Python
"""mapping.yaml 单文件加载与增量合并。
|
||
|
||
设计要点:
|
||
- 一个文件搞定所有映射(migration_user_id / carriers / packages / series / agents)
|
||
- 由 scan_legacy.py 增量补全 legacy_* 字段;人工只需填 target_*
|
||
- lookup 时检查 target_* 是否已填,未填则抛 MissingTargetError 让调用方写入 errors.csv
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Any, Optional
|
||
|
||
import yaml
|
||
|
||
|
||
class MissingTargetError(Exception):
|
||
"""target_* 字段未填的具体错误,带 legacy 标识信息。"""
|
||
|
||
def __init__(self, kind: str, legacy_key: str, legacy_name: str, missing_fields: list[str]):
|
||
self.kind = kind
|
||
self.legacy_key = legacy_key
|
||
self.legacy_name = legacy_name
|
||
self.missing_fields = missing_fields
|
||
super().__init__(
|
||
f"{kind} 映射未填 target_*: legacy_key={legacy_key!r} ({legacy_name!r}), 缺失={missing_fields}"
|
||
)
|
||
|
||
|
||
SHOP_CODE_RE = re.compile(r"^[A-Za-z0-9_-]{2,64}$")
|
||
ALLOWED_OWNERSHIP_MODES = {"default_shop", "legacy_agent", "none"}
|
||
ALLOWED_PACKAGE_STATUSES = {"active", "pending"}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class OwnershipDeviceRule:
|
||
"""设备批量归属规则。"""
|
||
|
||
mode: str = "none"
|
||
current_slot: Optional[int] = None
|
||
package_source_slot: Optional[int] = None
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class OwnershipStandaloneCardRule:
|
||
"""独立卡批量归属规则。"""
|
||
|
||
mode: str = "none"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class OwnershipRules:
|
||
"""资产归属决策规则。"""
|
||
|
||
default_target_shop_code: Optional[str] = None
|
||
device: OwnershipDeviceRule = field(default_factory=OwnershipDeviceRule)
|
||
standalone_card: OwnershipStandaloneCardRule = field(default_factory=OwnershipStandaloneCardRule)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PackageRules:
|
||
"""套餐迁移规则。"""
|
||
|
||
migrate_statuses: tuple[str, ...] = ("active", "pending")
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class DeviceOverride:
|
||
"""设备级覆盖项。"""
|
||
|
||
virtual_no: str
|
||
target_shop_code: Optional[str] = None
|
||
current_slot: Optional[int] = None
|
||
package_source_slot: Optional[int] = None
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CardOverride:
|
||
"""卡级覆盖项。"""
|
||
|
||
iccid: str
|
||
target_shop_code: Optional[str] = None
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Overrides:
|
||
"""少量资产级覆盖项。"""
|
||
|
||
devices: dict[str, DeviceOverride] = field(default_factory=dict)
|
||
cards: dict[str, CardOverride] = field(default_factory=dict)
|
||
|
||
|
||
@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]
|
||
target_carrier_type: Optional[str]
|
||
target_carrier_name: Optional[str]
|
||
|
||
def require_target(self) -> "CarrierMapping":
|
||
missing = [
|
||
n for n, v in (
|
||
("target_carrier_id", self.target_carrier_id),
|
||
("target_carrier_type", self.target_carrier_type),
|
||
("target_carrier_name", self.target_carrier_name),
|
||
) if v in (None, "", 0)
|
||
]
|
||
if missing:
|
||
raise MissingTargetError("carrier", self.legacy_account_id, self.legacy_account_name, missing)
|
||
return self
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PackageMapping:
|
||
legacy_meal_id: str
|
||
legacy_meal_name: str
|
||
target_package_id: Optional[int]
|
||
|
||
def require_target(self) -> "PackageMapping":
|
||
if self.target_package_id in (None, 0):
|
||
raise MissingTargetError("package", self.legacy_meal_id, self.legacy_meal_name, ["target_package_id"])
|
||
return self
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SeriesMapping:
|
||
legacy_series_id: str
|
||
legacy_series_name: str
|
||
target_series_id: Optional[int] # 允许 null;填了可作为资产 series_id 兜底
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class AgentMapping:
|
||
legacy_agent_id: str
|
||
legacy_agent_name: str
|
||
target_shop_code: Optional[str] # legacy_agent 模式下允许 null,表示该代理资产进平台库存
|
||
no_downtime: bool = False # 迁移时免停机:有生效套餐的卡直接以 network_status=1 导入
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PackageLifecycleOverride:
|
||
"""单卡旧套餐生命周期的人工裁决。"""
|
||
|
||
iccid: str
|
||
active_life_id: str
|
||
pending_life_ids: tuple[str, ...] = ()
|
||
skipped_life_ids: tuple[str, ...] = ()
|
||
|
||
|
||
@dataclass
|
||
class Mapping:
|
||
"""整体映射(对应单个 mapping.yaml)。"""
|
||
|
||
migration_user_id: int = 0
|
||
migration_batch_no: str = "MIGRATION-QICHENG"
|
||
ownership_rules: OwnershipRules = field(default_factory=OwnershipRules)
|
||
package_rules: PackageRules = field(default_factory=PackageRules)
|
||
overrides: Overrides = field(default_factory=Overrides)
|
||
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)
|
||
package_lifecycle_overrides: dict[str, PackageLifecycleOverride] = field(default_factory=dict)
|
||
|
||
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())
|
||
|
||
def lookup_series(self, legacy_series_id: str) -> Optional[SeriesMapping]:
|
||
if not legacy_series_id:
|
||
return None
|
||
return self.series.get(str(legacy_series_id).strip())
|
||
|
||
def lookup_package_lifecycle_override(self, iccid: str) -> Optional[PackageLifecycleOverride]:
|
||
return self.package_lifecycle_overrides.get(str(iccid or "").strip())
|
||
|
||
def is_no_downtime_agent(self, legacy_agent_id: str) -> bool:
|
||
"""判断代理是否配置了免停机迁移。"""
|
||
agent = self.agents.get(str(legacy_agent_id or "").strip())
|
||
return agent.no_downtime if agent else False
|
||
|
||
def lookup_agent_shop_code(self, legacy_agent_id: str) -> Optional[str]:
|
||
"""按 agent_id 查 target_shop_code,空字符串视为 None(进平台库存)。"""
|
||
if not legacy_agent_id:
|
||
return None
|
||
agent = self.agents.get(str(legacy_agent_id).strip())
|
||
if agent is None:
|
||
return None
|
||
code = (agent.target_shop_code or "").strip()
|
||
return code or None
|
||
|
||
def lookup_device_override(self, virtual_no: str) -> Optional[DeviceOverride]:
|
||
"""按设备号查覆盖项。"""
|
||
return self.overrides.devices.get(str(virtual_no or "").strip())
|
||
|
||
def lookup_card_override(self, *iccids: str) -> Optional[CardOverride]:
|
||
"""按完整 ICCID 或 19 位 ICCID 查覆盖项。"""
|
||
for iccid in iccids:
|
||
item = self.overrides.cards.get(str(iccid or "").strip())
|
||
if item is not None:
|
||
return item
|
||
return None
|
||
|
||
|
||
# ---------------- 加载 ----------------
|
||
|
||
|
||
MAPPING_FILE = "mapping.yaml"
|
||
DSN_FILE = "legacy_dsn.yaml"
|
||
|
||
|
||
def load_mapping(config_dir: Path) -> Mapping:
|
||
"""加载 config/mapping.yaml。文件不存在时返回空 Mapping(供 scan_legacy 首次创建)。"""
|
||
path = config_dir / MAPPING_FILE
|
||
if not path.exists():
|
||
return Mapping()
|
||
with path.open("r", encoding="utf-8") as f:
|
||
raw = yaml.safe_load(f) or {}
|
||
if not isinstance(raw, dict):
|
||
raise ValueError(f"{path} 顶层必须是 mapping 类型")
|
||
if "ownership_rules" not in raw:
|
||
raise ValueError(
|
||
f"{path} 缺少 ownership_rules。新迁移流程必须显式声明资产归属模式,请参考 "
|
||
"config/mapping.yaml.example 添加 ownership_rules、package_rules 和 overrides 后重跑。"
|
||
)
|
||
|
||
m = Mapping(
|
||
migration_user_id=int(raw.get("migration_user_id", 0)),
|
||
migration_batch_no=_opt_str(raw.get("migration_batch_no")) or "MIGRATION-QICHENG",
|
||
ownership_rules=_load_ownership_rules(raw.get("ownership_rules") or {}),
|
||
package_rules=_load_package_rules(raw.get("package_rules") or {}),
|
||
overrides=_load_overrides(raw.get("overrides") or {}),
|
||
)
|
||
|
||
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_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_account_id] = c
|
||
|
||
package_items = list(raw.get("packages") or [])
|
||
package_rules_raw = raw.get("package_rules") or {}
|
||
if isinstance(package_rules_raw, dict):
|
||
package_items.extend(package_rules_raw.get("packages") or [])
|
||
for item in package_items:
|
||
p = PackageMapping(
|
||
legacy_meal_id=str(item["legacy_meal_id"]).strip(),
|
||
legacy_meal_name=str(item.get("legacy_meal_name") or ""),
|
||
target_package_id=_opt_int(item.get("target_package_id")),
|
||
)
|
||
m.packages[p.legacy_meal_id] = p
|
||
|
||
for item in raw.get("series") or []:
|
||
s = SeriesMapping(
|
||
legacy_series_id=str(item["legacy_series_id"]).strip(),
|
||
legacy_series_name=str(item.get("legacy_series_name") or ""),
|
||
target_series_id=_opt_int(item.get("target_series_id")),
|
||
)
|
||
m.series[s.legacy_series_id] = s
|
||
|
||
m.package_lifecycle_overrides = _load_package_lifecycle_overrides(raw.get("package_lifecycle_overrides") or [])
|
||
|
||
for item in raw.get("agents") or []:
|
||
a = AgentMapping(
|
||
legacy_agent_id=str(item["legacy_agent_id"]).strip(),
|
||
legacy_agent_name=str(item.get("legacy_agent_name") or ""),
|
||
target_shop_code=_opt_shop_code(item.get("target_shop_code"), "agents[].target_shop_code"),
|
||
no_downtime=bool(item.get("no_downtime", False)),
|
||
)
|
||
m.agents[a.legacy_agent_id] = a
|
||
|
||
return m
|
||
|
||
|
||
def _load_ownership_rules(raw: dict) -> OwnershipRules:
|
||
"""加载归属规则配置。"""
|
||
if not isinstance(raw, dict):
|
||
raise ValueError("mapping.yaml.ownership_rules 必须是 mapping 类型")
|
||
|
||
default_shop_code = _opt_shop_code(raw.get("default_target_shop_code"), "ownership_rules.default_target_shop_code")
|
||
device_raw = raw.get("device") or {}
|
||
standalone_raw = raw.get("standalone_card") or {}
|
||
if not isinstance(device_raw, dict):
|
||
raise ValueError("mapping.yaml.ownership_rules.device 必须是 mapping 类型")
|
||
if not isinstance(standalone_raw, dict):
|
||
raise ValueError("mapping.yaml.ownership_rules.standalone_card 必须是 mapping 类型")
|
||
|
||
device_mode = _mode(device_raw.get("mode"), "ownership_rules.device.mode")
|
||
standalone_mode = _mode(standalone_raw.get("mode"), "ownership_rules.standalone_card.mode")
|
||
if (device_mode == "default_shop" or standalone_mode == "default_shop") and not default_shop_code:
|
||
raise ValueError("mapping.yaml.ownership_rules.default_target_shop_code 不能为空:default_shop 模式必须指定目标店铺码")
|
||
|
||
return OwnershipRules(
|
||
default_target_shop_code=default_shop_code,
|
||
device=OwnershipDeviceRule(
|
||
mode=device_mode,
|
||
current_slot=_slot_or_none(device_raw.get("current_slot"), "ownership_rules.device.current_slot"),
|
||
package_source_slot=_slot_or_none(
|
||
device_raw.get("package_source_slot"), "ownership_rules.device.package_source_slot"
|
||
),
|
||
),
|
||
standalone_card=OwnershipStandaloneCardRule(mode=standalone_mode),
|
||
)
|
||
|
||
|
||
def _load_package_rules(raw: dict) -> PackageRules:
|
||
"""加载套餐迁移规则。"""
|
||
if not isinstance(raw, dict):
|
||
raise ValueError("mapping.yaml.package_rules 必须是 mapping 类型")
|
||
statuses_raw = raw.get("migrate_statuses")
|
||
if statuses_raw is None:
|
||
statuses = ("active", "pending")
|
||
elif not isinstance(statuses_raw, list):
|
||
raise ValueError("mapping.yaml.package_rules.migrate_statuses 必须是列表")
|
||
else:
|
||
statuses = tuple(str(x).strip() for x in statuses_raw if str(x).strip())
|
||
invalid = [x for x in statuses if x not in ALLOWED_PACKAGE_STATUSES]
|
||
if invalid:
|
||
raise ValueError(f"mapping.yaml.package_rules.migrate_statuses 仅支持 active/pending,无效值: {invalid}")
|
||
if not statuses:
|
||
raise ValueError("mapping.yaml.package_rules.migrate_statuses 至少要包含 active 或 pending")
|
||
return PackageRules(migrate_statuses=statuses)
|
||
|
||
|
||
def _load_package_lifecycle_overrides(raw: Any) -> dict[str, PackageLifecycleOverride]:
|
||
if not isinstance(raw, list):
|
||
raise ValueError("mapping.yaml.package_lifecycle_overrides 必须是列表")
|
||
overrides: dict[str, PackageLifecycleOverride] = {}
|
||
for idx, item in enumerate(raw, start=1):
|
||
if not isinstance(item, dict):
|
||
raise ValueError(f"mapping.yaml.package_lifecycle_overrides[{idx}] 必须是 mapping 类型")
|
||
iccid = _opt_str(item.get("iccid"))
|
||
active = _opt_str(item.get("active_life_id"))
|
||
if not iccid or len(iccid) != 20 or not iccid.isdigit():
|
||
raise ValueError(f"mapping.yaml.package_lifecycle_overrides[{idx}].iccid 必须是 20 位数字")
|
||
if not active:
|
||
raise ValueError(f"mapping.yaml.package_lifecycle_overrides[{idx}].active_life_id 不能为空")
|
||
pending = _life_ids(item.get("pending_life_ids") or [], f"package_lifecycle_overrides[{idx}].pending_life_ids")
|
||
skipped = _life_ids(item.get("skipped_life_ids") or [], f"package_lifecycle_overrides[{idx}].skipped_life_ids")
|
||
all_ids = (active,) + pending + skipped
|
||
if len(all_ids) != len(set(all_ids)):
|
||
raise ValueError(f"mapping.yaml.package_lifecycle_overrides[{idx}] 生命周期 ID 不可重复或跨状态重复")
|
||
if iccid in overrides:
|
||
raise ValueError(f"mapping.yaml.package_lifecycle_overrides iccid 重复: {iccid}")
|
||
overrides[iccid] = PackageLifecycleOverride(iccid, active, pending, skipped)
|
||
return overrides
|
||
|
||
|
||
def _life_ids(raw: Any, field_name: str) -> tuple[str, ...]:
|
||
if not isinstance(raw, list):
|
||
raise ValueError(f"mapping.yaml.{field_name} 必须是列表")
|
||
values = tuple(_opt_str(value) or "" for value in raw)
|
||
if not all(values):
|
||
raise ValueError(f"mapping.yaml.{field_name} 不可包含空生命周期 ID")
|
||
return values
|
||
|
||
|
||
def _load_overrides(raw: dict) -> Overrides:
|
||
"""加载资产覆盖项。"""
|
||
if not isinstance(raw, dict):
|
||
raise ValueError("mapping.yaml.overrides 必须是 mapping 类型")
|
||
|
||
devices: dict[str, DeviceOverride] = {}
|
||
for idx, item in enumerate(raw.get("devices") or [], start=1):
|
||
if not isinstance(item, dict):
|
||
raise ValueError(f"mapping.yaml.overrides.devices[{idx}] 必须是 mapping 类型")
|
||
virtual_no = _opt_str(item.get("virtual_no"))
|
||
if not virtual_no:
|
||
raise ValueError(f"mapping.yaml.overrides.devices[{idx}].virtual_no 不能为空")
|
||
if virtual_no in devices:
|
||
raise ValueError(f"mapping.yaml.overrides.devices virtual_no 重复: {virtual_no}")
|
||
devices[virtual_no] = DeviceOverride(
|
||
virtual_no=virtual_no,
|
||
target_shop_code=_opt_shop_code(item.get("target_shop_code"), f"overrides.devices[{idx}].target_shop_code"),
|
||
current_slot=_slot_or_none(item.get("current_slot"), f"overrides.devices[{idx}].current_slot"),
|
||
package_source_slot=_slot_or_none(
|
||
item.get("package_source_slot"), f"overrides.devices[{idx}].package_source_slot"
|
||
),
|
||
)
|
||
|
||
cards: dict[str, CardOverride] = {}
|
||
for idx, item in enumerate(raw.get("cards") or [], start=1):
|
||
if not isinstance(item, dict):
|
||
raise ValueError(f"mapping.yaml.overrides.cards[{idx}] 必须是 mapping 类型")
|
||
iccid = _opt_str(item.get("iccid"))
|
||
if not iccid:
|
||
raise ValueError(f"mapping.yaml.overrides.cards[{idx}].iccid 不能为空")
|
||
if len(iccid) not in (19, 20):
|
||
raise ValueError(f"mapping.yaml.overrides.cards[{idx}].iccid 必须是 19 或 20 位")
|
||
if iccid in cards:
|
||
raise ValueError(f"mapping.yaml.overrides.cards iccid 重复: {iccid}")
|
||
cards[iccid] = CardOverride(
|
||
iccid=iccid,
|
||
target_shop_code=_opt_shop_code(item.get("target_shop_code"), f"overrides.cards[{idx}].target_shop_code"),
|
||
)
|
||
|
||
return Overrides(devices=devices, cards=cards)
|
||
|
||
|
||
def merge_and_save(config_dir: Path, mapping: Mapping) -> Path:
|
||
"""把 mapping 写回 config/mapping.yaml,保留人工已填的 target_* 字段。
|
||
|
||
由 scan_legacy 调用:先 load,再用扫描结果 _补充_ legacy_* 项(不存在的追加),最后调用本函数写回。
|
||
"""
|
||
config_dir.mkdir(parents=True, exist_ok=True)
|
||
path = config_dir / MAPPING_FILE
|
||
path.write_text(_render_commented_mapping_yaml(mapping), encoding="utf-8")
|
||
return path
|
||
|
||
|
||
def _render_commented_mapping_yaml(mapping: Mapping) -> str:
|
||
"""渲染带中文注释的 mapping.yaml。
|
||
|
||
PyYAML 写回时会丢弃注释,这里用固定模板承载说明,动态列表仍按扫描结果增量输出。
|
||
"""
|
||
lines: list[str] = [
|
||
"# 奇成迁移映射配置(由 scan_legacy.py 生成/增量补全)",
|
||
"#",
|
||
"# 使用方式:",
|
||
"# 1. cards.csv 必填;devices.csv 可选。没有 devices.csv 时就是纯卡迁移。",
|
||
"# 2. legacy_* 字段来自奇成扫描,一般不要手工修改。",
|
||
"# 3. target_* 字段需要人工填写,否则生成 SQL 时会写入 errors.csv 并阻断对应资产。",
|
||
"# 4. 资产归属先看 overrides,再看 ownership_rules;只有显式 legacy_agent 模式才读取 agents。",
|
||
"",
|
||
"# ============== 全局配置 ==============",
|
||
"# 迁移操作者账号 ID,写入 creator/updater/operator_id。线上建议使用后台超级管理员 ID。",
|
||
f"migration_user_id: {_yaml_scalar(mapping.migration_user_id)}",
|
||
"# 本次迁移批次号。每批线上迁移保持唯一,方便收尾、验收和回滚定位。",
|
||
f"migration_batch_no: {_yaml_scalar(mapping.migration_batch_no)}",
|
||
"",
|
||
"# ============== 资产归属与槽位规则 ==============",
|
||
"# default_shop 模式下的批量目标店铺编码;mode=legacy_agent/none 时可保持 null。",
|
||
"ownership_rules:",
|
||
f" default_target_shop_code: {_yaml_scalar(mapping.ownership_rules.default_target_shop_code)}",
|
||
" # 设备规则只在存在 resources/devices.csv 时生效。",
|
||
" device:",
|
||
" # default_shop=默认店铺;legacy_agent=按默认/行级旧套餐读取槽位绑定卡的奇成 agent_id 映射店铺;none=平台库存。",
|
||
f" mode: {_yaml_scalar(mapping.ownership_rules.device.mode)}",
|
||
" # 默认当前卡槽。devices.csv 未填 current_slot 时使用这里;写入 tb_device_sim_binding.is_current。",
|
||
f" current_slot: {_yaml_scalar(mapping.ownership_rules.device.current_slot)}",
|
||
" # 默认旧套餐读取卡槽。devices.csv 未填 package_source_slot 时使用这里;新系统套餐仍挂设备;不填则回退 current_slot。",
|
||
f" package_source_slot: {_yaml_scalar(mapping.ownership_rules.device.package_source_slot)}",
|
||
" standalone_card:",
|
||
" # default_shop=默认店铺;legacy_agent=按奇成 agent_id 映射店铺;none=平台库存。",
|
||
f" mode: {_yaml_scalar(mapping.ownership_rules.standalone_card.mode)}",
|
||
"",
|
||
"# ============== 套餐迁移规则 ==============",
|
||
"package_rules:",
|
||
" # active=当前生效正式套餐;pending=未生效待生效正式套餐。",
|
||
" migrate_statuses:",
|
||
]
|
||
for status in mapping.package_rules.migrate_statuses:
|
||
lines.append(f" - {_yaml_scalar(status)}")
|
||
|
||
lines.extend([
|
||
"",
|
||
"# ============== 套餐生命周期特殊裁决 ==============",
|
||
"# life_id 来自 tbl_card_life.id；每项必须明确裁决该卡所有可迁移正式套餐。",
|
||
])
|
||
if not mapping.package_lifecycle_overrides:
|
||
lines.append("package_lifecycle_overrides: []")
|
||
else:
|
||
lines.append("package_lifecycle_overrides:")
|
||
for item in sorted(mapping.package_lifecycle_overrides.values(), key=lambda x: x.iccid):
|
||
lines.extend([
|
||
f" - iccid: {_yaml_scalar(item.iccid)}",
|
||
f" active_life_id: {_yaml_scalar(item.active_life_id)}",
|
||
f" pending_life_ids: {json.dumps(list(item.pending_life_ids), ensure_ascii=False)}",
|
||
f" skipped_life_ids: {json.dumps(list(item.skipped_life_ids), ensure_ascii=False)}",
|
||
])
|
||
|
||
lines.extend([
|
||
"",
|
||
"# ============== 少量资产级覆盖 ==============",
|
||
"# 覆盖优先级:devices.csv 行级槽位 > overrides.devices 槽位 > ownership_rules.device 默认槽位。",
|
||
"overrides:",
|
||
" # 设备级特殊处理。virtual_no 对应 devices.csv 的 virtual_no/imei 兜底值。",
|
||
])
|
||
_append_device_overrides(lines, mapping.overrides)
|
||
lines.append(" # 独立卡特殊处理。iccid 可填 19 位或 20 位。")
|
||
_append_card_overrides(lines, mapping.overrides)
|
||
|
||
lines.extend([
|
||
"",
|
||
"# ============== 运营商映射 ==============",
|
||
"# scan_legacy 按奇成 tbl_card.account_id/account_name 分组写入;只需要人工填写 target_*。",
|
||
"# target_carrier_type 只能填 CMCC / CUCC / CTCC / CBN。",
|
||
])
|
||
_append_carriers(lines, mapping)
|
||
|
||
lines.extend([
|
||
"",
|
||
"# ============== 套餐映射 ==============",
|
||
"# scan_legacy 按奇成 tbl_card_life.meal_id 分组写入;target_package_id 必须指向新库正式套餐。",
|
||
])
|
||
_append_packages(lines, mapping)
|
||
|
||
lines.extend([
|
||
"",
|
||
"# ============== 套餐系列快照(可选) ==============",
|
||
"# 资产 series_id 优先从 target_package_id 对应的 tb_package.series_id 反查;target_series_id 可作无套餐时的兜底。",
|
||
])
|
||
_append_series(lines, mapping)
|
||
|
||
lines.extend([
|
||
"",
|
||
"# ============== 代理-店铺映射(legacy_agent 归属 + 代理钱包) ==============",
|
||
"# standalone_card/device mode=legacy_agent 时,资产按这里的 target_shop_code 分配。",
|
||
"# 不使用 legacy_agent 时,这里只影响 migrate_runtime.py 的代理钱包初始化。",
|
||
])
|
||
_append_agents(lines, mapping)
|
||
|
||
return "\n".join(lines) + "\n"
|
||
|
||
|
||
def _append_device_overrides(lines: list[str], overrides: Overrides) -> None:
|
||
if not overrides.devices:
|
||
lines.append(" devices: []")
|
||
return
|
||
lines.append(" devices:")
|
||
for item in sorted(overrides.devices.values(), key=lambda x: x.virtual_no):
|
||
lines.extend([
|
||
f" - virtual_no: {_yaml_scalar(item.virtual_no)}",
|
||
f" target_shop_code: {_yaml_scalar(item.target_shop_code)}",
|
||
f" current_slot: {_yaml_scalar(item.current_slot)}",
|
||
f" package_source_slot: {_yaml_scalar(item.package_source_slot)}",
|
||
])
|
||
|
||
|
||
def _append_card_overrides(lines: list[str], overrides: Overrides) -> None:
|
||
if not overrides.cards:
|
||
lines.append(" cards: []")
|
||
return
|
||
lines.append(" cards:")
|
||
for item in sorted(overrides.cards.values(), key=lambda x: x.iccid):
|
||
lines.extend([
|
||
f" - iccid: {_yaml_scalar(item.iccid)}",
|
||
f" target_shop_code: {_yaml_scalar(item.target_shop_code)}",
|
||
])
|
||
|
||
|
||
def _append_carriers(lines: list[str], mapping: Mapping) -> None:
|
||
if not mapping.carriers:
|
||
lines.append("carriers: []")
|
||
return
|
||
lines.append("carriers:")
|
||
for item in sorted(mapping.carriers.values(), key=lambda x: (x.legacy_account_name, x.legacy_account_id)):
|
||
lines.extend([
|
||
f" - legacy_account_id: {_yaml_scalar(item.legacy_account_id)}",
|
||
f" legacy_account_name: {_yaml_scalar(item.legacy_account_name)}",
|
||
f" legacy_category: {_yaml_scalar(item.legacy_category)}",
|
||
f" legacy_category_name: {_yaml_scalar(item.legacy_category_name)}",
|
||
f" target_carrier_id: {_yaml_scalar(item.target_carrier_id)}",
|
||
f" target_carrier_type: {_yaml_scalar(item.target_carrier_type)}",
|
||
f" target_carrier_name: {_yaml_scalar(item.target_carrier_name)}",
|
||
])
|
||
|
||
|
||
def _append_packages(lines: list[str], mapping: Mapping) -> None:
|
||
if not mapping.packages:
|
||
lines.append("packages: []")
|
||
return
|
||
lines.append("packages:")
|
||
for item in sorted(mapping.packages.values(), key=lambda x: x.legacy_meal_id):
|
||
lines.extend([
|
||
f" - legacy_meal_id: {_yaml_scalar(item.legacy_meal_id)}",
|
||
f" legacy_meal_name: {_yaml_scalar(item.legacy_meal_name)}",
|
||
f" target_package_id: {_yaml_scalar(item.target_package_id)}",
|
||
])
|
||
|
||
|
||
def _append_series(lines: list[str], mapping: Mapping) -> None:
|
||
if not mapping.series:
|
||
lines.append("series: []")
|
||
return
|
||
lines.append("series:")
|
||
for item in sorted(mapping.series.values(), key=lambda x: x.legacy_series_id):
|
||
lines.extend([
|
||
f" - legacy_series_id: {_yaml_scalar(item.legacy_series_id)}",
|
||
f" legacy_series_name: {_yaml_scalar(item.legacy_series_name)}",
|
||
f" target_series_id: {_yaml_scalar(item.target_series_id)}",
|
||
])
|
||
|
||
|
||
def _append_agents(lines: list[str], mapping: Mapping) -> None:
|
||
if not mapping.agents:
|
||
lines.append("agents: []")
|
||
return
|
||
lines.append("agents:")
|
||
for item in sorted(mapping.agents.values(), key=lambda x: x.legacy_agent_id):
|
||
lines.extend([
|
||
f" - legacy_agent_id: {_yaml_scalar(item.legacy_agent_id)}",
|
||
f" legacy_agent_name: {_yaml_scalar(item.legacy_agent_name)}",
|
||
f" target_shop_code: {_yaml_scalar(item.target_shop_code)}",
|
||
f" no_downtime: {_yaml_scalar(item.no_downtime)}",
|
||
])
|
||
|
||
|
||
def _yaml_scalar(value: Any) -> str:
|
||
"""把 Python 标量转成稳定 YAML 标量。"""
|
||
if value is None:
|
||
return "null"
|
||
if isinstance(value, bool):
|
||
return "true" if value else "false"
|
||
if isinstance(value, int):
|
||
return str(value)
|
||
return json.dumps(str(value), ensure_ascii=False)
|
||
|
||
|
||
def _dump_ownership_rules(rules: OwnershipRules) -> dict:
|
||
"""序列化归属规则。"""
|
||
return {
|
||
"default_target_shop_code": rules.default_target_shop_code,
|
||
"device": {
|
||
"mode": rules.device.mode,
|
||
"current_slot": rules.device.current_slot,
|
||
"package_source_slot": rules.device.package_source_slot,
|
||
},
|
||
"standalone_card": {
|
||
"mode": rules.standalone_card.mode,
|
||
},
|
||
}
|
||
|
||
|
||
def _dump_package_rules(rules: PackageRules) -> dict:
|
||
"""序列化套餐规则。"""
|
||
return {
|
||
"migrate_statuses": list(rules.migrate_statuses),
|
||
}
|
||
|
||
|
||
def _dump_overrides(overrides: Overrides) -> dict:
|
||
"""序列化资产覆盖项。"""
|
||
return {
|
||
"devices": [
|
||
{
|
||
"virtual_no": x.virtual_no,
|
||
"target_shop_code": x.target_shop_code,
|
||
"current_slot": x.current_slot,
|
||
"package_source_slot": x.package_source_slot,
|
||
}
|
||
for x in sorted(overrides.devices.values(), key=lambda item: item.virtual_no)
|
||
],
|
||
"cards": [
|
||
{
|
||
"iccid": x.iccid,
|
||
"target_shop_code": x.target_shop_code,
|
||
}
|
||
for x in sorted(overrides.cards.values(), key=lambda item: item.iccid)
|
||
],
|
||
}
|
||
|
||
|
||
def load_legacy_dsn(config_dir: Path) -> dict:
|
||
"""加载奇成 MySQL 只读连接配置。"""
|
||
path = config_dir / DSN_FILE
|
||
if not path.exists():
|
||
raise FileNotFoundError(f"找不到 {path},请基于 legacy_dsn.yaml.example 创建并填写真实 DSN")
|
||
with path.open("r", encoding="utf-8") as f:
|
||
data = yaml.safe_load(f) or {}
|
||
for required in ("host", "port", "user", "password", "database"):
|
||
if required not in data:
|
||
raise ValueError(f"{path} 缺少必填项 {required}")
|
||
return data
|
||
|
||
|
||
# ---------------- 小工具 ----------------
|
||
|
||
|
||
def _opt_int(v: Any) -> Optional[int]:
|
||
if v is None or v == "":
|
||
return None
|
||
return int(v)
|
||
|
||
|
||
def _opt_str(v: Any) -> Optional[str]:
|
||
if v is None:
|
||
return None
|
||
s = str(v).strip()
|
||
return s or None
|
||
|
||
|
||
def _opt_shop_code(v: Any, field_name: str) -> Optional[str]:
|
||
code = _opt_str(v)
|
||
if code is None:
|
||
return None
|
||
if not SHOP_CODE_RE.match(code):
|
||
raise ValueError(f"mapping.yaml.{field_name} 店铺码格式无效: {code!r},仅允许 2-64 位字母/数字/_/-")
|
||
return code
|
||
|
||
|
||
def _mode(v: Any, field_name: str) -> str:
|
||
mode = _opt_str(v) or "none"
|
||
if mode not in ALLOWED_OWNERSHIP_MODES:
|
||
raise ValueError(f"mapping.yaml.{field_name} 仅支持 default_shop/legacy_agent/none,无效值: {mode!r}")
|
||
return mode
|
||
|
||
|
||
def _slot_or_none(v: Any, field_name: str) -> Optional[int]:
|
||
if v is None or v == "":
|
||
return None
|
||
try:
|
||
slot = int(v)
|
||
except (TypeError, ValueError) as exc:
|
||
raise ValueError(f"mapping.yaml.{field_name} 必须是 1-4 的整数") from exc
|
||
if slot < 1 or slot > 4:
|
||
raise ValueError(f"mapping.yaml.{field_name} 必须在 1-4 之间")
|
||
return slot
|