junhong_cmp_fiber/scripts/migration/lib/mapping_loader.py

"""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