256 lines
9.2 KiB
Python
256 lines
9.2 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
|
|
|
|
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}"
|
|
)
|
|
|
|
|
|
@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] # 允许 null,表示该代理的卡进平台库存
|
|
|
|
|
|
@dataclass
|
|
class Mapping:
|
|
"""整体映射(对应单个 mapping.yaml)。"""
|
|
|
|
migration_user_id: int = 0
|
|
migration_batch_no: str = "MIGRATION-QICHENG"
|
|
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 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
|
|
|
|
|
|
# ---------------- 加载 ----------------
|
|
|
|
|
|
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 类型")
|
|
|
|
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",
|
|
)
|
|
|
|
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
|
|
|
|
for item in raw.get("packages") or []:
|
|
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_str(item.get("target_shop_code")),
|
|
)
|
|
m.agents[a.legacy_agent_id] = a
|
|
|
|
return m
|
|
|
|
|
|
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
|
|
data = {
|
|
"migration_user_id": mapping.migration_user_id,
|
|
"migration_batch_no": mapping.migration_batch_no,
|
|
"carriers": [
|
|
{
|
|
"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,
|
|
}
|
|
for c in sorted(mapping.carriers.values(), key=lambda x: (x.legacy_account_name, x.legacy_account_id))
|
|
],
|
|
"packages": [
|
|
{
|
|
"legacy_meal_id": p.legacy_meal_id,
|
|
"legacy_meal_name": p.legacy_meal_name,
|
|
"target_package_id": p.target_package_id,
|
|
}
|
|
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)
|
|
],
|
|
}
|
|
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:
|
|
"""加载奇成 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
|