更新一版

scripts/migration/lib/mapping_loader.py

@@ -7,6 +7,8 @@
 """
 from __future__ import annotations
 
+import json
+import re
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Optional
@@ -27,6 +29,69 @@ class MissingTargetError(Exception):
         )
 
 
+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
@@ -73,7 +138,7 @@ class SeriesMapping:
 class AgentMapping:
     legacy_agent_id: str
     legacy_agent_name: str
-    target_shop_code: Optional[str]  # 允许 null,表示该代理的卡进平台库存
+    target_shop_code: Optional[str]  # legacy_agent 模式下允许 null,表示该代理资产进平台库存
 
 
 @dataclass
@@ -82,6 +147,9 @@ class Mapping:
 
     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)
@@ -108,6 +176,18 @@ class Mapping:
         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
+
 
 # ---------------- 加载 ----------------
 
@@ -125,10 +205,18 @@ def load_mapping(config_dir: Path) -> Mapping:
         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 []:
@@ -147,7 +235,11 @@ def load_mapping(config_dir: Path) -> Mapping:
         )
         m.carriers[c.legacy_account_id] = c
 
-    for item in raw.get("packages") or []:
+    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 ""),
@@ -167,13 +259,105 @@ def load_mapping(config_dir: Path) -> Mapping:
         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_str(item.get("target_shop_code")),
+            target_shop_code=_opt_shop_code(item.get("target_shop_code"), "agents[].target_shop_code"),
         )
         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_* 字段。
 
@@ -181,49 +365,233 @@ def merge_and_save(config_dir: Path, mapping: Mapping) -> Path:
     """
     config_dir.mkdir(parents=True, exist_ok=True)
     path = config_dir / MAPPING_FILE
-    data = {
-        "migration_user_id": mapping.migration_user_id,
-        "migration_batch_no": mapping.migration_batch_no,
-        "carriers": [
+    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)}",
+        "    # 当前卡槽,写入 tb_device_sim_binding.is_current;纯卡迁移可保持 null。",
+        f"    current_slot: {_yaml_scalar(mapping.ownership_rules.device.current_slot)}",
+        "    # 设备旧套餐读取槽位。新系统套餐挂设备,这里只决定从奇成哪张绑定卡读套餐生命周期;纯卡迁移可保持 null。",
+        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 反查。",
+    ])
+    _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)}",
+        ])
+
+
+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": [
             {
-                "legacy_account_id": c.legacy_account_id,
-                "legacy_account_name": c.legacy_account_name,
-                "legacy_category": c.legacy_category,
-                "legacy_category_name": c.legacy_category_name,
-                "target_carrier_id": c.target_carrier_id,
-                "target_carrier_type": c.target_carrier_type,
-                "target_carrier_name": c.target_carrier_name,
+                "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 c in sorted(mapping.carriers.values(), key=lambda x: (x.legacy_account_name, x.legacy_account_id))
+            for x in sorted(overrides.devices.values(), key=lambda item: item.virtual_no)
         ],
-        "packages": [
+        "cards": [
             {
-                "legacy_meal_id": p.legacy_meal_id,
-                "legacy_meal_name": p.legacy_meal_name,
-                "target_package_id": p.target_package_id,
+                "iccid": x.iccid,
+                "target_shop_code": x.target_shop_code,
             }
-            for p in sorted(mapping.packages.values(), key=lambda x: x.legacy_meal_id)
-        ],
-        "series": [
-            {
-                "legacy_series_id": s.legacy_series_id,
-                "legacy_series_name": s.legacy_series_name,
-                "target_series_id": s.target_series_id,
-            }
-            for s in sorted(mapping.series.values(), key=lambda x: x.legacy_series_id)
-        ],
-        "agents": [
-            {
-                "legacy_agent_id": a.legacy_agent_id,
-                "legacy_agent_name": a.legacy_agent_name,
-                "target_shop_code": a.target_shop_code,
-            }
-            for a in sorted(mapping.agents.values(), key=lambda x: x.legacy_agent_id)
+            for x in sorted(overrides.cards.values(), key=lambda item: item.iccid)
         ],
     }
-    with path.open("w", encoding="utf-8") as f:
-        yaml.safe_dump(data, f, allow_unicode=True, sort_keys=False, default_flow_style=False)
-    return path
 
 
 def load_legacy_dsn(config_dir: Path) -> dict:
@@ -253,3 +621,31 @@ def _opt_str(v: Any) -> Optional[str]:
         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