break
b38df737e1
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 7m51s
660 lines
26 KiB
Python
660 lines
26 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
|
|
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)
|
|
|
|
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 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
|
|
|
|
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_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([
|
|
"",
|
|
"# ============== 少量资产级覆盖 ==============",
|
|
"# 覆盖优先级: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
|