数据迁移脚本

scripts/migration/lib/iccid_utils.py

@@ -0,0 +1,158 @@
+"""ICCID 19/20 位拆分与校验工具。
+
+移植自 pkg/utils/iccid.go,并按运营商类型增加长度合规校验。
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Optional
+
+# 运营商标准 ICCID 长度
+CARRIER_ICCID_LENGTH = {
+    "CTCC": 19,  # 电信
+    "CMCC": 20,  # 移动
+    "CUCC": 20,  # 联通
+    "CBN": 20,   # 广电
+}
+
+VALID_CARRIER_TYPES = set(CARRIER_ICCID_LENGTH.keys())
+
+
+@dataclass(frozen=True)
+class SplitResult:
+    """ICCID 拆分结果。"""
+
+    iccid: str           # 原始 ICCID(去空格)
+    iccid_19: str        # 前 19 位(所有卡必填)
+    iccid_20: Optional[str]  # 完整 20 位(仅 20 位运营商卡有值,19 位卡为 None → SQL NULL)
+
+
+@dataclass(frozen=True)
+class ValidationError:
+    """ICCID 校验失败信息。"""
+
+    code: str    # 错误码:length_invalid / length_mismatch
+    message: str # 中文描述
+
+
+def split_iccid(iccid: str) -> Optional[SplitResult]:
+    """按 ICCID 长度拆分双列存储值。
+
+    与 Go 版本一致:
+        19 位 → (iccid, iccid, None)
+        20 位 → (iccid, iccid[:19], iccid)
+        其他   → None(调用方按异常处理)
+    """
+    iccid = iccid.strip()
+    if len(iccid) == 19:
+        return SplitResult(iccid=iccid, iccid_19=iccid, iccid_20=None)
+    if len(iccid) == 20:
+        return SplitResult(iccid=iccid, iccid_19=iccid[:19], iccid_20=iccid)
+    return None
+
+
+def validate_and_split(iccid: str, carrier_type: str) -> tuple[Optional[SplitResult], Optional[ValidationError]]:
+    """按运营商类型校验 ICCID 长度并拆分。
+
+    返回 (拆分结果, 错误信息) 二元组:
+        - 长度不是 19/20:返回 (None, length_invalid 错误)
+        - 长度与 carrier_type 不匹配:返回 (拆分结果, length_mismatch 警告) → 调用方决定要不要放行
+        - 合法:返回 (拆分结果, None)
+
+    设计说明:不直接拒绝长度不匹配的情况(电信卡居然是 20 位 / 非电信卡是 19 位),
+    保留拆分结果让脚本写入 errors.csv 让业务方确认。
+    """
+    iccid = iccid.strip()
+    if carrier_type not in VALID_CARRIER_TYPES:
+        return None, ValidationError(
+            code="carrier_type_invalid",
+            message=f"无效的运营商类型 {carrier_type!r},应为 {sorted(VALID_CARRIER_TYPES)} 之一",
+        )
+
+    result = split_iccid(iccid)
+    if result is None:
+        return None, ValidationError(
+            code="length_invalid",
+            message=f"ICCID 长度 {len(iccid)} 不合法,只支持 19/20 位",
+        )
+
+    expected_len = CARRIER_ICCID_LENGTH[carrier_type]
+    if len(iccid) != expected_len:
+        return result, ValidationError(
+            code="length_mismatch",
+            message=f"{carrier_type} 应为 {expected_len} 位但 ICCID 是 {len(iccid)} 位,请人工确认",
+        )
+
+    return result, None
+
+
+def prefix_19(iccid: str) -> str:
+    """ICCID 前 19 位(稳定 ID,无校验位)。
+
+    用于 devices.csv 引用 cards 时的兼容查找:业务方在 devices 里填 19 还是 20 位都接受,
+    内部统一截取前 19 位回查 cards.iccid_19。
+    """
+    s = (iccid or "").strip()
+    if len(s) >= 19:
+        return s[:19]
+    return s
+
+
+def validate_iccid_19(iccid_19: str) -> Optional[ValidationError]:
+    """iccid_19 必须 19 位纯数字。"""
+    if not iccid_19:
+        return ValidationError(code="iccid_19_required", message="iccid_19 不能为空")
+    if len(iccid_19) != 19:
+        return ValidationError(
+            code="iccid_19_length",
+            message=f"iccid_19 必须是 19 位,当前 {len(iccid_19)} 位",
+        )
+    if not iccid_19.isdigit():
+        return ValidationError(code="iccid_19_format", message="iccid_19 必须是纯数字")
+    return None
+
+
+def validate_iccid_20(iccid_20: str, iccid_19: str) -> Optional[ValidationError]:
+    """iccid_20 可空(CTCC 卡);非空时必须 20 位纯数字且前 19 位 == iccid_19。"""
+    if not iccid_20:
+        return None
+    if len(iccid_20) != 20:
+        return ValidationError(
+            code="iccid_20_length",
+            message=f"iccid_20 必须是 20 位,当前 {len(iccid_20)} 位",
+        )
+    if not iccid_20.isdigit():
+        return ValidationError(code="iccid_20_format", message="iccid_20 必须是纯数字")
+    if iccid_20[:19] != iccid_19:
+        return ValidationError(
+            code="iccid_20_prefix_mismatch",
+            message=f"iccid_20 前 19 位 ({iccid_20[:19]!r}) 与 iccid_19 ({iccid_19!r}) 不一致",
+        )
+    return None
+
+
+def validate_carrier_iccid_match(
+    iccid_20: str, carrier_type: str
+) -> Optional[ValidationError]:
+    """业务方填写的 iccid_20 与奇成查到的 carrier_type 是否匹配。
+
+    - CTCC(19 位运营商):iccid_20 必须为空
+    - CMCC/CUCC/CBN(20 位运营商):iccid_20 必填
+    """
+    if carrier_type not in VALID_CARRIER_TYPES:
+        return ValidationError(
+            code="carrier_type_invalid",
+            message=f"未知运营商类型 {carrier_type!r}",
+        )
+    expected_len = CARRIER_ICCID_LENGTH[carrier_type]
+    if expected_len == 19 and iccid_20:
+        return ValidationError(
+            code="carrier_length_conflict",
+            message=f"{carrier_type} 是 19 位运营商,iccid_20 应留空,实际填了 {iccid_20!r}",
+        )
+    if expected_len == 20 and not iccid_20:
+        return ValidationError(
+            code="carrier_length_conflict",
+            message=f"{carrier_type} 是 {expected_len} 位运营商,iccid_20 必填",
+        )
+    return None