436 lines
18 KiB
Python
436 lines
18 KiB
Python
"""cards.csv / devices.csv 加载与跨文件一致性校验。
|
||
|
||
新设计:CSV 只承载"资产清单和行级槽位决策"——
|
||
- cards.csv 只读 iccid + is_industry
|
||
- devices.csv 读设备元数据 + sim_iccid_*
|
||
- 卡的运营商、套餐、msisdn、agent_id 等元数据全部从奇成查(见 legacy_query.fetch_card_meta)
|
||
- 归属、默认当前槽位、默认旧套餐读取槽位从 mapping.yaml 解析
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
CARD_HEADERS_REQUIRED = ("iccid_19",)
|
||
# 设备列至少包含 imei 或 virtual_no 之一(运行时校验,允许只填一列)
|
||
DEVICE_HEADERS_AT_LEAST_ONE = ("imei", "virtual_no")
|
||
|
||
|
||
@dataclass
|
||
class CardRow:
|
||
"""cards.csv 单行的解析结果(只含业务方决策字段)。
|
||
|
||
业务方至少提供 iccid_19 / iccid_20 之一:
|
||
- iccid_19:前 19 位,可作为奇成主表查询键
|
||
- iccid_20:20 位完整 ICCID,固定 20 位运营商只填这一列即可
|
||
- 运营商类型与 iccid_20 的一致性在查完奇成后由 sql_builder 校验
|
||
新系统 iccid 唯一键 = iccid_20 优先,否则 iccid_19。
|
||
"""
|
||
|
||
line_no: int # 原始 CSV 行号(从 1 开始,含表头)
|
||
iccid_19: str
|
||
iccid_20: str = "" # 留空表示 CTCC(查完奇成会再校验)
|
||
allow_iccid_19_lookup: bool = True # iccid_19 是否为业务方显式提供,可用于奇成主表兜底查询
|
||
is_industry: bool = False # 默认 False(普通卡)
|
||
|
||
# 跨文件校验后填充
|
||
bound_device_virtual_no: str = ""
|
||
|
||
@property
|
||
def iccid_full(self) -> str:
|
||
"""完整 ICCID:有 iccid_20 用 iccid_20,否则用 iccid_19。"""
|
||
return self.iccid_20 or self.iccid_19
|
||
|
||
|
||
@dataclass
|
||
class DeviceRow:
|
||
"""devices.csv 单行的解析结果。
|
||
|
||
简化为业务方核心关心的列:imei / virtual_no / device_* / max_sim_slots / sim_iccid_*。
|
||
店铺归属、批量默认 current_slot 和旧套餐读取槽位 package_source_slot 由 mapping.yaml 决定。
|
||
其他字段(manufacturer 等)直接用默认值,不暴露给业务方。
|
||
"""
|
||
|
||
line_no: int
|
||
imei: str
|
||
virtual_no: str # 留空时用 imei 兜底,见 _load_devices
|
||
sim_iccids: dict[int, str] # slot → 完整 ICCID(只保留非空)
|
||
device_name: str = ""
|
||
device_model: str = ""
|
||
device_type: str = ""
|
||
max_sim_slots: int = 1 # 不填时按非空 sim_iccid_* 的最大槽位推导
|
||
current_slot: int = 1 # 默认 slot 1 为当前生效
|
||
package_source_slot: int = 1 # 设备旧套餐读取槽位
|
||
current_slot_source: str = "default"
|
||
package_source_slot_source: str = "default"
|
||
|
||
|
||
@dataclass
|
||
class ErrorRow:
|
||
"""异常行汇总(写入 output/errors.csv)。"""
|
||
|
||
source_file: str
|
||
line_no: int
|
||
field: str
|
||
value: str
|
||
error_code: str
|
||
message: str
|
||
|
||
def to_csv_row(self) -> list[str]:
|
||
return [self.source_file, str(self.line_no), self.field, self.value, self.error_code, self.message]
|
||
|
||
|
||
@dataclass
|
||
class LoadResult:
|
||
cards: list[CardRow] = field(default_factory=list)
|
||
devices: list[DeviceRow] = field(default_factory=list)
|
||
errors: list[ErrorRow] = field(default_factory=list)
|
||
|
||
|
||
# ---------------- 字段清洗 ----------------
|
||
|
||
|
||
def _normalize_cell(raw: str) -> str:
|
||
if raw is None:
|
||
return ""
|
||
s = raw.strip()
|
||
if s.startswith(""):
|
||
s = s[1:]
|
||
if s.startswith("'"):
|
||
s = s[1:]
|
||
s = s.replace(" ", "")
|
||
return s.strip()
|
||
|
||
|
||
def _parse_bool(raw: str) -> Optional[bool]:
|
||
v = _normalize_cell(raw).lower()
|
||
if v == "":
|
||
return False
|
||
if v in ("true", "1", "yes", "y", "t", "是", "行业卡"):
|
||
return True
|
||
if v in ("false", "0", "no", "n", "f", "否", "普通卡"):
|
||
return False
|
||
return None
|
||
|
||
|
||
def _parse_int(raw: str, default: int = 0) -> int:
|
||
v = _normalize_cell(raw)
|
||
if not v:
|
||
return default
|
||
try:
|
||
return int(v)
|
||
except ValueError:
|
||
return default
|
||
|
||
|
||
# ---------------- 主入口 ----------------
|
||
|
||
|
||
def load_assets(resources_dir: Path, mapping=None) -> tuple[list[CardRow], list[DeviceRow], list[ErrorRow]]:
|
||
"""读 cards.csv + devices.csv,做格式校验与跨文件一致性校验。
|
||
|
||
返回:(合法卡列表, 合法设备列表, 异常行列表)。
|
||
元数据(carrier/agent_id/msisdn 等)由调用方再用 legacy_query 补充。
|
||
"""
|
||
result = LoadResult()
|
||
|
||
cards_path = resources_dir / "cards.csv"
|
||
if not cards_path.exists():
|
||
raise FileNotFoundError(f"缺少 {cards_path},请按 resources/README.md 准备数据")
|
||
_load_cards(cards_path, result)
|
||
|
||
devices_path = resources_dir / "devices.csv"
|
||
if devices_path.exists():
|
||
_load_devices(devices_path, result)
|
||
if mapping is not None:
|
||
_apply_device_slot_rules(result, mapping)
|
||
|
||
_cross_check(result)
|
||
|
||
return result.cards, result.devices, result.errors
|
||
|
||
|
||
def _load_cards(path: Path, result: LoadResult) -> None:
|
||
import csv
|
||
|
||
from . import iccid_utils
|
||
|
||
with path.open("r", encoding="utf-8-sig", newline="") as f:
|
||
reader = csv.DictReader(f)
|
||
header = [_normalize_cell(h) for h in (reader.fieldnames or [])]
|
||
missing = [k for k in CARD_HEADERS_REQUIRED if k not in header]
|
||
if missing:
|
||
raise ValueError(f"{path} 缺少必备列: {missing}")
|
||
if "iccid_20" not in header:
|
||
raise ValueError(f"{path} 必须包含 iccid_20 列(20 位运营商卡填写,CTCC 留空)")
|
||
|
||
seen_iccid_full: dict[str, int] = {}
|
||
for idx, raw in enumerate(reader, start=2):
|
||
iccid_19 = _normalize_cell(raw.get("iccid_19", ""))
|
||
iccid_20 = _normalize_cell(raw.get("iccid_20", ""))
|
||
allow_iccid_19_lookup = bool(iccid_19)
|
||
|
||
if iccid_20 and len(iccid_20) != 20:
|
||
result.errors.append(
|
||
ErrorRow("cards.csv", idx, "iccid_20", iccid_20, "iccid_20_length",
|
||
f"iccid_20 必须是 20 位,当前 {len(iccid_20)} 位")
|
||
)
|
||
continue
|
||
if not iccid_19 and iccid_20:
|
||
iccid_19 = iccid_20[:19]
|
||
|
||
err = iccid_utils.validate_iccid_19(iccid_19)
|
||
if err is not None:
|
||
result.errors.append(ErrorRow("cards.csv", idx, "iccid_19", iccid_19, err.code, err.message))
|
||
continue
|
||
|
||
err = iccid_utils.validate_iccid_20(iccid_20, iccid_19)
|
||
if err is not None:
|
||
result.errors.append(ErrorRow("cards.csv", idx, "iccid_20", iccid_20, err.code, err.message))
|
||
continue
|
||
|
||
iccid_full = iccid_20 or iccid_19
|
||
if iccid_full in seen_iccid_full:
|
||
result.errors.append(
|
||
ErrorRow("cards.csv", idx, "iccid", iccid_full, "duplicate",
|
||
f"与第 {seen_iccid_full[iccid_full]} 行 ICCID 重复")
|
||
)
|
||
continue
|
||
seen_iccid_full[iccid_full] = idx
|
||
|
||
is_industry = _parse_bool(raw.get("is_industry", ""))
|
||
if is_industry is None:
|
||
result.errors.append(
|
||
ErrorRow("cards.csv", idx, "is_industry", raw.get("is_industry", ""), "invalid_bool", "is_industry 必须是 true/false")
|
||
)
|
||
continue
|
||
|
||
result.cards.append(
|
||
CardRow(
|
||
line_no=idx,
|
||
iccid_19=iccid_19,
|
||
iccid_20=iccid_20,
|
||
allow_iccid_19_lookup=allow_iccid_19_lookup,
|
||
is_industry=is_industry,
|
||
)
|
||
)
|
||
|
||
|
||
def _load_devices(path: Path, result: LoadResult) -> None:
|
||
import csv
|
||
|
||
with path.open("r", encoding="utf-8-sig", newline="") as f:
|
||
reader = csv.DictReader(f)
|
||
header = [_normalize_cell(h) for h in (reader.fieldnames or [])]
|
||
if not any(k in header for k in DEVICE_HEADERS_AT_LEAST_ONE):
|
||
raise ValueError(f"{path} 必须至少包含 imei 或 virtual_no 列")
|
||
|
||
seen_virtual_no: dict[str, int] = {}
|
||
for idx, raw in enumerate(reader, start=2):
|
||
imei = _normalize_cell(raw.get("imei", ""))
|
||
virtual_no = _normalize_cell(raw.get("virtual_no", ""))
|
||
device_name = _normalize_cell(raw.get("device_name", ""))
|
||
device_model = _normalize_cell(raw.get("device_model", ""))
|
||
device_type = _normalize_cell(raw.get("device_type", ""))
|
||
# virtual_no 留空时用 imei 兜底(业务方说"虚拟号一般会提供,没有就用 imei")
|
||
effective_vno = virtual_no or imei
|
||
if not effective_vno:
|
||
result.errors.append(
|
||
ErrorRow("devices.csv", idx, "virtual_no", "", "missing_identity",
|
||
"imei 和 virtual_no 不能同时为空")
|
||
)
|
||
continue
|
||
if effective_vno in seen_virtual_no:
|
||
result.errors.append(
|
||
ErrorRow("devices.csv", idx, "virtual_no", effective_vno, "duplicate",
|
||
f"与第 {seen_virtual_no[effective_vno]} 行的设备号重复")
|
||
)
|
||
continue
|
||
seen_virtual_no[effective_vno] = idx
|
||
|
||
sim_iccids: dict[int, str] = {}
|
||
for slot in (1, 2, 3, 4):
|
||
v = _normalize_cell(raw.get(f"sim_iccid_{slot}", ""))
|
||
if v:
|
||
sim_iccids[slot] = v
|
||
|
||
max_sim_slots_raw = _normalize_cell(raw.get("max_sim_slots", ""))
|
||
if max_sim_slots_raw:
|
||
try:
|
||
max_sim_slots = int(max_sim_slots_raw)
|
||
except ValueError:
|
||
result.errors.append(
|
||
ErrorRow("devices.csv", idx, "max_sim_slots", max_sim_slots_raw, "invalid",
|
||
"最大卡槽数必须是 1-4 的整数")
|
||
)
|
||
continue
|
||
if max_sim_slots < 1 or max_sim_slots > 4:
|
||
result.errors.append(
|
||
ErrorRow("devices.csv", idx, "max_sim_slots", max_sim_slots_raw, "invalid",
|
||
"最大卡槽数必须在 1-4 之间")
|
||
)
|
||
continue
|
||
else:
|
||
max_sim_slots = max(sim_iccids.keys(), default=1)
|
||
|
||
overflow_slots = [slot for slot in sim_iccids if slot > max_sim_slots]
|
||
if overflow_slots:
|
||
result.errors.append(
|
||
ErrorRow("devices.csv", idx, "max_sim_slots", str(max_sim_slots), "slot_overflow",
|
||
f"已填写的卡槽 {max(overflow_slots)} 超出最大卡槽数")
|
||
)
|
||
continue
|
||
|
||
current_slot, current_slot_source = _parse_slot_with_source(raw.get("current_slot", ""), 1)
|
||
if current_slot < 1 or current_slot > 4:
|
||
result.errors.append(
|
||
ErrorRow("devices.csv", idx, "current_slot", raw.get("current_slot", ""), "invalid_slot",
|
||
"current_slot 必须是 1-4 的整数")
|
||
)
|
||
continue
|
||
package_source_slot, package_source_slot_source = _parse_slot_with_source(
|
||
raw.get("package_source_slot", ""), current_slot
|
||
)
|
||
if package_source_slot < 1 or package_source_slot > 4:
|
||
result.errors.append(
|
||
ErrorRow("devices.csv", idx, "package_source_slot", raw.get("package_source_slot", ""), "invalid_slot",
|
||
"package_source_slot 必须是 1-4 的整数")
|
||
)
|
||
continue
|
||
|
||
result.devices.append(
|
||
DeviceRow(
|
||
line_no=idx,
|
||
imei=imei,
|
||
virtual_no=effective_vno,
|
||
device_name=device_name,
|
||
device_model=device_model,
|
||
device_type=device_type,
|
||
sim_iccids=sim_iccids,
|
||
max_sim_slots=max_sim_slots,
|
||
current_slot=current_slot,
|
||
package_source_slot=package_source_slot,
|
||
current_slot_source=current_slot_source,
|
||
package_source_slot_source=package_source_slot_source,
|
||
)
|
||
)
|
||
|
||
|
||
def _parse_slot_with_source(raw: str, default: int) -> tuple[int, str]:
|
||
"""解析槽位并返回来源。"""
|
||
v = _normalize_cell(raw)
|
||
if not v:
|
||
return default, "default"
|
||
try:
|
||
return int(v), "row"
|
||
except ValueError:
|
||
return 0, "row"
|
||
|
||
|
||
def _apply_device_slot_rules(result: LoadResult, mapping) -> None:
|
||
"""按 行配置 > 覆盖项 > 批量默认值 解析设备槽位。"""
|
||
for device in result.devices:
|
||
override = mapping.lookup_device_override(device.virtual_no)
|
||
if device.current_slot_source != "row":
|
||
if override and override.current_slot is not None:
|
||
device.current_slot = override.current_slot
|
||
device.current_slot_source = "override"
|
||
elif mapping.ownership_rules.device.current_slot is not None:
|
||
device.current_slot = mapping.ownership_rules.device.current_slot
|
||
device.current_slot_source = "ownership_rules.device"
|
||
if device.package_source_slot_source != "row":
|
||
if override and override.package_source_slot is not None:
|
||
device.package_source_slot = override.package_source_slot
|
||
device.package_source_slot_source = "override"
|
||
elif mapping.ownership_rules.device.package_source_slot is not None:
|
||
device.package_source_slot = mapping.ownership_rules.device.package_source_slot
|
||
device.package_source_slot_source = "ownership_rules.device"
|
||
else:
|
||
device.package_source_slot = device.current_slot
|
||
device.package_source_slot_source = device.current_slot_source
|
||
|
||
for field_name, slot in (
|
||
("current_slot", device.current_slot),
|
||
("package_source_slot", device.package_source_slot),
|
||
):
|
||
if slot < 1 or slot > 4:
|
||
result.errors.append(
|
||
ErrorRow("devices.csv", device.line_no, field_name, str(slot), "invalid_slot",
|
||
f"{field_name} 必须是 1-4 的整数")
|
||
)
|
||
elif slot > device.max_sim_slots:
|
||
result.errors.append(
|
||
ErrorRow("devices.csv", device.line_no, field_name, str(slot), "slot_overflow",
|
||
f"{field_name}={slot} 超出 max_sim_slots={device.max_sim_slots}")
|
||
)
|
||
|
||
|
||
def _cross_check(result: LoadResult) -> None:
|
||
"""校验设备与卡之间的一致性,并把 bound_device_virtual_no 回填到 CardRow。
|
||
|
||
devices.csv 里 sim_iccid_X 业务方可能填 19 位也可能填 20 位。
|
||
20 位按完整 ICCID 精确匹配;19 位只有在 cards.csv 中唯一对应一张卡时才允许。
|
||
"""
|
||
from . import iccid_utils
|
||
|
||
cards_by_full: dict[str, CardRow] = {c.iccid_full: c for c in result.cards}
|
||
cards_by_i19: dict[str, list[CardRow]] = {}
|
||
for card in result.cards:
|
||
cards_by_i19.setdefault(card.iccid_19, []).append(card)
|
||
device_card_refs: dict[str, tuple[str, int]] = {}
|
||
|
||
for device in result.devices:
|
||
for slot, raw_iccid in device.sim_iccids.items():
|
||
resolved_card: Optional[CardRow] = None
|
||
if len(raw_iccid) == 20:
|
||
resolved_card = cards_by_full.get(raw_iccid)
|
||
if resolved_card is None:
|
||
prefix = iccid_utils.prefix_19(raw_iccid)
|
||
candidates = cards_by_i19.get(prefix, [])
|
||
if len(candidates) == 1 and not candidates[0].iccid_20:
|
||
resolved_card = candidates[0]
|
||
elif len(raw_iccid) == 19:
|
||
candidates = cards_by_i19.get(raw_iccid, [])
|
||
if len(candidates) > 1:
|
||
result.errors.append(
|
||
ErrorRow(
|
||
"devices.csv", device.line_no, f"sim_iccid_{slot}", raw_iccid,
|
||
"ambiguous_iccid_19", "该 19 位前缀对应多张 20 位卡,请填写完整 20 位 ICCID",
|
||
)
|
||
)
|
||
continue
|
||
if candidates:
|
||
resolved_card = candidates[0]
|
||
else:
|
||
result.errors.append(
|
||
ErrorRow(
|
||
"devices.csv", device.line_no, f"sim_iccid_{slot}", raw_iccid,
|
||
"sim_iccid_length", "设备槽位 ICCID 必须是 19 或 20 位",
|
||
)
|
||
)
|
||
continue
|
||
|
||
if resolved_card is None:
|
||
result.errors.append(
|
||
ErrorRow(
|
||
"devices.csv", device.line_no, f"sim_iccid_{slot}", raw_iccid,
|
||
"card_not_found", "引用的 ICCID 在 cards.csv 中不存在",
|
||
)
|
||
)
|
||
continue
|
||
card_key = resolved_card.iccid_full
|
||
if card_key in device_card_refs:
|
||
prev_vno, prev_line = device_card_refs[card_key]
|
||
result.errors.append(
|
||
ErrorRow(
|
||
"devices.csv", device.line_no, f"sim_iccid_{slot}", raw_iccid,
|
||
"card_used_twice", f"该 ICCID 已被设备 {prev_vno}(第 {prev_line} 行)使用",
|
||
)
|
||
)
|
||
continue
|
||
device_card_refs[card_key] = (device.virtual_no, device.line_no)
|
||
resolved_card.bound_device_virtual_no = device.virtual_no
|
||
# 标准化:slot 里的值统一改写为完整 ICCID,sql_builder 直接用它查 cards / card_metas
|
||
device.sim_iccids[slot] = card_key
|