Commit 5c37925e authored by Data Governance Dev's avatar Data Governance Dev

feat(standards): 国标分析拆分为 24 个细粒度 indicator,并注册为独立 step

国标分析改造两阶段,合并提交:

【阶段 1】拆分 + 新增 5 类分析对象
- STD-001 身份证 → IND-001-a/b/c/d (格式/校验位/出生日期/15位兼容)
- STD-002 USCC  → IND-002-a/b   (格式/校验位)
- STD-003 手机   → IND-003-a/b   (格式/号段)
- STD-004 区划   → IND-004-a/b   (格式/编码存在)
- 新增 IND-101 证件类型枚举(5 值)
- 新增 IND-201~204 4 种证件号码(身份证/护照/港澳台居住证/外国人永居)
- 新增 IND-301 跨字段 (证件类型+号码一致性)
- 新增 IND-302 跨表 (证件唯一性,GROUP BY HAVING)
- 新增 IND-401/402 姓名 (字符集/长度)
- 新增 IND-501/502 出生日期 (格式/年龄范围, 性别推断退休年龄)
- 新增 IND-601/602/603 区划/民族/婚姻

基础设施:
- BaseStandard 加 group/severity/depends_on 类属性 + validate_rows() 默认实现
- registry.list_all_standards / list_standards_by_group
- data_loader.py 惰性加载 data/*.json + extracted_standards.json
- 新增 data/{id_type,marriage,gender,retirement_age}.json
- 旧 STD-001~004 标 [已拆分], applies_to_fields=[] (保留兼容)

【阶段 2】每个 indicator 注册为独立 step
- registry.py 新增 load_standard_class / indicator_step_id /
  step_id_to_indicator_id / list_step_ids 工具函数
- orchestrator.py 动态按 group 顺序注册 24 个 indicator step
  (国标字段规范=100, 证件信息=200, 身份信息=300, 民政信息=400)
- step7_standards.py 新增 run_step7_for_indicator,每个 indicator
  独立 section_key (standards_ind_XXX),独立 tab
- 旧 run_step7() 保留 (向后兼容: 跑全部按 4 group 输出)

校验层:
- IND-301 用 sample_field_pairs.sql
- IND-302 用 sample_field_groups.sql
- 84 单测覆盖所有 indicator (tests/test_indicators.py)

协议层:
- 沿用上轮的 tab_protocol;每个 indicator 一个独立 tab
  (1 KPI 违规字段 + 1 KPI 检查字段 + 2 表: 指标汇总 + 违规明细)

详细工作记录见 docs/WORKLOG.md。
parent 714f714a
......@@ -341,3 +341,107 @@ register_step(
- 当前:`1.0`
- 后续扩展方向:chart(`render.kind="chart"`)、inline-edit、`custom` 特判
---
## 10. 多 tab by group 示例(step7 国标校验 v2)
step7 国标校验从「单 tab 4 个国标」重构为「4 tab × 24 indicator」的版本。每个 tab 对应一个分析主题(group),tab 内仍走「按 indicator 聚合 + 违规明细 + 字段联动」三件套。
### 后端:按 group 拆 sections
```python
# web/core/step_impl/step7_standards.py
from standards.registry import list_standards_by_group
_TABS_BY_GROUP = {
"国标字段规范": _make_group_tab(key="standards_gb_format", title="国标字段规范", ...),
"证件信息": _make_group_tab(key="standards_document", title="证件信息", ...),
"身份信息": _make_group_tab(key="standards_personal", title="身份信息", ...),
"民政信息": _make_group_tab(key="standards_civil", title="民政信息", ...),
}
def _wrap(per_group):
"""per_group = {group_name: data_dict} → 多 tab 注入。"""
sections = {}
for group, data in per_group.items():
tab = _TABS_BY_GROUP.get(group)
sections[group] = inject_protocol(data, step_id="standards", tabs=[tab])
return {"sections": sections}
```
`per_group` 字典的 key 就是 sections 的 key,前端 `displayedTabs` 会自动为每个 section 生成一个 tab。
### 前端:零改动
`web/static/index.html` 的 `displayedTabs` 计算属性已经在迭代 `sections[*]._protocol.tabs`,新增 tab 自动出现。无需改前端代码。
### Indicator 元数据:group / severity / depends_on
`BaseStandard` 加了 3 个类属性([standards/base.py](standards/base.py)):
```python
class BaseStandard(ABC):
standard_id: str = ""
applies_to_fields: list[str] | None = None
description: str = ""
# ── 多 tab 拆分的元数据 ──
group: str = "通用" # 国标字段规范 / 证件信息 / 身份信息 / 民政信息 / 通用
severity: str = "warning" # warning | error
depends_on: list[str] | None = None # 跨字段依赖(如 IND-301 → [IND-201, IND-202])
```
子类按 group 归类即可:
```python
class IdCardFormatIndicator(BaseStandard):
standard_id = "IND-001-a"
group = "国标字段规范"
severity = "warning"
# applies_to_fields = [...] 命中数据库字段
```
### 跨字段 / 跨行校验(IND-301 / IND-302)
单值校验走 `validate(value)`。跨字段校验覆写 `validate_rows(rows)`:
```python
class IdTypeNumberConsistencyIndicator(BaseStandard):
standard_id = "IND-301"
applies_to_fields = [] # ← 不参与单字段自动发现
group = "证件信息"
depends_on = ["IND-101", "IND-201", "IND-202", "IND-203", "IND-204"]
def validate(self, value):
# 单值模式不适用
return ValidationResult(True, "", value, "IND-301 仅支持跨字段校验", self.standard_id)
def validate_rows(self, rows):
# rows 每条含 {id_type, id_no}
out = []
for row in rows:
...
if not indicator.validate(row["id_no"]).valid:
out.append({"value": row["id_no"], "reason": "..."})
return out
```
step7 检测到 indicator 的 `validate` 是「no-op」就走第二轮 SQL 抽样同表多字段,调 `validate_rows`。
跨表聚合(IND-302)走 `GROUP BY HAVING` SQL:
```sql
-- web/sql/standards/sample_field_groups.sql
SELECT ${type_col}, ${no_col}, COUNT(*) AS dup_count
FROM ${table}
WHERE ${type_col} IS NOT NULL AND ${no_col} IS NOT NULL
GROUP BY ${type_col}, ${no_col}
HAVING COUNT(*) >= ${min_count}
ORDER BY dup_count DESC
LIMIT 200
```
### 验证
- **单测**:`tests/test_indicators.py` —— 84 个测试(每 indicator 1-2 合规 + 1-2 违规 + 协议形状)
- **前端编译**:`scripts/verify_template.cjs` —— 验证 `<div id="app">` 在 Vue 3 prod build 下编译产物合法 JS
- **端到端**:勾选「国家标准校验」step,分析结果应出现 4 个 tab(国标字段规范 / 证件信息 / 身份信息 / 民政信息)
\ No newline at end of file
This diff is collapsed.
......@@ -4,7 +4,7 @@ from __future__ import annotations
import re
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from dataclasses import dataclass
@dataclass
......@@ -30,24 +30,74 @@ class BaseStandard(ABC):
"""标准插件基类
子类需实现:
- standard_id: 标准编号,如 "STD-001"
- standard_id: 标准编号,如 "IND-001-a"
- standard_name: 标准中文名
- applies_to_fields: 适用的字段名(list)
- validate(value): 单值校验逻辑
- describe(): 给人看的描述(可选)
扩展元数据(可选覆盖):
- group: "国标字段规范" / "证件信息" / "身份信息" / "民政信息" / "通用"
- severity: "warning" | "error" —— 影响前端 KPI 染色
- depends_on: 列表,声明跨字段依赖(如 IND-301 依赖 IND-201/202)
注意:本类不是 @dataclass(与 ABC 抽象方法不兼容),所以 list 字段
不能用 field(default_factory=list) —— 用 None + __init_subclass__ 兜底,
或干脆让子类各自覆写。基类取值时做 None → [] 的兼容。
"""
standard_id: str = ""
standard_name: str = ""
applies_to_fields: list[str] = field(default_factory=list)
applies_to_fields: list[str] | None = None
description: str = ""
# ── 扩展元数据(向后兼容 —— 缺省走「通用 / warning」) ──
group: str = "通用"
severity: str = "warning"
depends_on: list[str] | None = None
def __init__(self):
# 子类可省略 __init__;基类负责把 None 兜底为 []
if self.applies_to_fields is None:
self.applies_to_fields = []
if self.depends_on is None:
self.depends_on = []
# ── 抽象方法 ──
@abstractmethod
def validate(self, value: str) -> ValidationResult:
"""校验单个值,返回 ValidationResult"""
...
# ── 可选覆写:跨字段 / 跨行校验 ──
# 默认实现:rows 每条独立调 validate(value)
# 跨字段校验器(如 IND-301 类型+号码一致性)可覆写此方法
# 跨表聚合校验器(如 IND-302 唯一性)走 SQL,不走这里
def validate_rows(self, rows: list[dict]) -> list[dict]:
"""批量校验多行,每行至少含 indicator 的若干字段。
返回违规明细 dict 列表(与 validate(value).to_dict() 同结构,
但允许带额外字段如 inconsistency_type)。
默认实现:每行取「主字段」(=applies_to_fields[0])调 validate(value)。
"""
if not self.applies_to_fields:
return []
primary = self.applies_to_fields[0]
out = []
for row in rows:
v = row.get(primary)
if v is None or v == "":
continue
r = self.validate(str(v))
if not r.valid:
out.append({
"value": str(v)[:50],
"reason": r.reason,
"row_keys": list(row.keys()),
})
return out
# ── 辅助:批量校验多个值,统计违规率 ──
def validate_batch(self, values: list[str], field_name: str = "") -> dict:
if not values:
......
{
"_doc": "人的性别代码 —— GB/T 2261.1-2003《个人基本信息分类与代码 第1部分:人的性别代码》",
"std_id": "GB/T 2261.1-2003",
"enum": {
"0": "未知的性别",
"1": "男性",
"2": "女性",
"9": "未说明的性别"
}
}
\ No newline at end of file
{
"_doc": "证件类型枚举 —— 用户业务场景下整理。完整 GA/T 2000.156-2016 三位代码表较繁,本项目按用户给定的 5 值落地。",
"enum": {
"01": {"name": "居民身份证", "std": "GB 11643-1999"},
"02": {"name": "军官证", "std": "(建议核实)"},
"03": {"name": "护照", "std": "ICAO Doc 9303 / 中华人民共和国Passport法"},
"04": {"name": "外国人永久居留身份证", "std": "公安部令第118号 / 2023-12-01新版五星卡"},
"05": {"name": "港澳台居民居住证", "std": "国办发〔2018〕81号"}
}
}
\ No newline at end of file
{
"_doc": "婚姻状况代码 —— GB/T 2261.2-2003《个人基本信息分类与代码 第2部分:婚姻状况代码》",
"std_id": "GB/T 2261.2-2003",
"enum": {
"10": "未婚",
"20": "已婚",
"21": "初婚",
"22": "再婚",
"23": "复婚",
"30": "丧偶",
"40": "离婚",
"90": "未说明的婚姻状况"
}
}
\ No newline at end of file
{
"_doc": "法定退休年龄 —— 国务院《国务院关于工人退休、退职的暂行办法》(国发〔1978〕104号) 及后续干部/职工分类调整",
"rules": {
"default": {"age": 60, "note": "兜底(找不到性别字段时用)"},
"male": {"age": 60, "note": "男职工/干部 60 岁"},
"female_worker": {"age": 50, "note": "女工人 50 岁"},
"female_cadre": {"age": 55, "note": "女干部 55 岁"}
},
"policy_notes": "1978 年国发〔1978〕104 号确立男 60 / 女 50、女干部 55 的基本框架;近年渐进式延迟退休政策(2025-2026 实施)正在调整,本表暂以原框架为准。"
}
\ No newline at end of file
"""数据加载器 —— 惰性加载 + 缓存 standards/data/ 下的枚举 JSON。
设计目标:
- 启动时不读文件,第一次访问时才 load
- 同一份数据多次访问走缓存
- 加载失败抛 RuntimeError 并给出明确路径
"""
from __future__ import annotations
import json
from functools import lru_cache
from pathlib import Path
_DATA_DIR = Path(__file__).parent / "data"
_EXTRACTED_PATH = Path(__file__).resolve().parents[1] / "national_standards" / "extracted_standards.json"
def _load_json(path: Path) -> dict:
if not path.exists():
raise RuntimeError(f"数据文件不存在: {path}")
# extracted_standards.json 是 GBK 编码(内嵌 UTF-8 BOM?实际是 GBK)
# 其他 data/*.json 是 UTF-8
if path.name == "extracted_standards.json":
# 含 4 字节 GBK 序列;用 utf-8 errors=replace 兜底
with open(path, "r", encoding="utf-8", errors="replace") as f:
return json.load(f)
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
# ── 单文件加载 ──
@lru_cache(maxsize=8)
def load_id_type_enum() -> dict[str, dict]:
"""证件类型枚举(id_type_enum.json → {code: {name, std}})"""
return _load_json(_DATA_DIR / "id_type_enum.json")["enum"]
@lru_cache(maxsize=8)
def load_marriage_codes() -> dict[str, str]:
"""GB/T 2261.2-2003 婚姻状况枚举 → {code: name}"""
return _load_json(_DATA_DIR / "marriage_codes.json")["enum"]
@lru_cache(maxsize=8)
def load_gender_codes() -> dict[str, str]:
"""GB/T 2261.1-2003 性别枚举 → {code: name}"""
return _load_json(_DATA_DIR / "gender_codes.json")["enum"]
@lru_cache(maxsize=8)
def load_retirement_age_rules() -> dict[str, dict]:
"""法定退休年龄规则 → {rule_key: {age, note}}"""
return _load_json(_DATA_DIR / "retirement_age.json")["rules"]
# ── 复用 national_standards/extracted_standards.json ──
@lru_cache(maxsize=4)
def load_extracted_standards() -> dict:
"""整份 extracted_standards.json(GB/T 2260 / 3304 / 4658 / 6565 / ISO 3166 / ISO 4217 / GB/T 4754)"""
return _load_json(_EXTRACTED_PATH)
@lru_cache(maxsize=8)
def load_district_codes() -> set[str]:
"""GB/T 2260-2007 区划代码集合(6 位)"""
extracted = load_extracted_standards()
return {r["code"] for r in extracted["GB_T_2260_2007"]["records"]}
@lru_cache(maxsize=8)
def load_district_province_codes() -> set[str]:
"""省级代码集合(前 2 位)"""
return {code[:2] for code in load_district_codes()}
@lru_cache(maxsize=8)
def load_ethnicity_codes() -> set[str]:
"""GB/T 3304-1991 民族代码集合(2 位数字)"""
extracted = load_extracted_standards()
return {r["code"] for r in extracted["GB_T_3304_1991"]["records"]}
@lru_cache(maxsize=8)
def load_iso_country_codes_3() -> set[str]:
"""ISO 3166-1 alpha-3 国家代码集合(3 位字母,外国人永居证国别码用)"""
extracted = load_extracted_standards()
return {r["alpha3"] for r in extracted["ISO_3166_1"]["records"] if r.get("alpha3")}
@lru_cache(maxsize=8)
def load_iso_country_codes_alpha2() -> set[str]:
"""ISO 3166-1 alpha-2 国家代码集合(2 位字母)"""
extracted = load_extracted_standards()
return {r["alpha2"] for r in extracted["ISO_3166_1"]["records"] if r.get("alpha2")}
@lru_cache(maxsize=8)
def load_iso_country_codes_numeric() -> set[str]:
"""ISO 3166-1 numeric-3 国家代码集合(3 位数字,零填充)"""
extracted = load_extracted_standards()
return {r["numeric"] for r in extracted["ISO_3166_1"]["records"] if r.get("numeric")}
# ── 验证快捷接口 ──
def is_valid_id_type(code: str) -> bool:
return code in load_id_type_enum()
def is_valid_marriage(code: str) -> bool:
return code in load_marriage_codes()
def is_valid_gender(code: str) -> bool:
return code in load_gender_codes()
def is_valid_district_6(code: str) -> bool:
return code in load_district_codes()
def is_valid_district_prefix_2(code: str) -> bool:
return code[:2] in load_district_province_codes()
def is_valid_ethnicity(code: str) -> bool:
return code in load_ethnicity_codes()
# ── 调试入口 ──
if __name__ == "__main__":
print("id_type_enum:", len(load_id_type_enum()), "codes")
print("marriage_codes:", list(load_marriage_codes().keys()))
print("district_codes:", len(load_district_codes()))
print("district_province_codes:", sorted(load_district_province_codes()))
print("ethnicity_codes count:", len(load_ethnicity_codes()))
\ No newline at end of file
"""IND-001-a 身份证号格式(18 位 + 正则)
拆分自 STD-001。只检查「长度 + 6+8+3+1 结构」,不做校验位 / 出生日期。
"""
from __future__ import annotations
import re
from .base import BaseStandard, ValidationResult
class IdCardFormatIndicator(BaseStandard):
standard_id = "IND-001-a"
standard_name = "GB 11643-1999 身份证号格式"
applies_to_fields = ["id_card", "id_card_no", "id_number", "identity_card"]
description = "长度 = 18;6 位地址 + 8 位出生日期 + 3 位顺序码 + 1 位校验位"
group = "国标字段规范"
severity = "warning"
_REGEX = re.compile(r"^[1-9]\d{5}(19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[\dXx]$")
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
if len(value) != 18:
return ValidationResult(False, "", value, f"长度 {len(value)} ≠ 18", self.standard_id)
if not self._REGEX.match(value):
return ValidationResult(False, "", value, "不符合 6+8+3+1 字符结构", self.standard_id)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-001-b 身份证号校验位(ISO 7064 MOD 11-2)
拆分自 STD-001。假定格式已通过(由 IND-001-a 保证),仅校验最后一位。
"""
from __future__ import annotations
from .base import BaseStandard, ValidationResult
class IdCardChecksumIndicator(BaseStandard):
standard_id = "IND-001-b"
standard_name = "GB 11643-1999 身份证校验位"
applies_to_fields = ["id_card", "id_card_no", "id_number", "identity_card"]
description = "ISO 7064 MOD 11-2 加权因子:7,9,10,5,8,4,2,1,6,3,7,9,10,5,8,4,2"
group = "国标字段规范"
severity = "warning"
_WEIGHTS = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
_CHECKSUM_MAP = ["1", "0", "X", "9", "8", "7", "6", "5", "4", "3", "2"]
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
if len(value) != 18:
return ValidationResult(
False, "", value,
"校验位要求 18 位(请先确认 IND-001-a 格式)",
self.standard_id,
)
expected = self._calc(value[:17])
if value[17].upper() != expected:
return ValidationResult(
False, "", value,
f"校验位错误,应为 {expected}",
self.standard_id,
)
return ValidationResult(True, "", value, "", self.standard_id)
def _calc(self, seventeen: str) -> str:
total = sum(int(c) * w for c, w in zip(seventeen, self._WEIGHTS))
return self._CHECKSUM_MAP[total % 11]
\ No newline at end of file
"""IND-001-c 身份证号出生日期合法性
拆分自 STD-001。仅校验 value[6:14] 是真实日期(含闰年)。
"""
from __future__ import annotations
from datetime import date
from .base import BaseStandard, ValidationResult
class IdCardBirthdateIndicator(BaseStandard):
standard_id = "IND-001-c"
standard_name = "GB 11643-1999 身份证出生日期"
applies_to_fields = ["id_card", "id_card_no", "id_number", "identity_card"]
description = "value[6:14] 必须能解析为合法日期(含闰年 2-29)"
group = "国标字段规范"
severity = "warning"
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
if len(value) != 18:
return ValidationResult(
False, "", value,
"出生日期要求 18 位完整号码",
self.standard_id,
)
try:
y, m, d = int(value[6:10]), int(value[10:12]), int(value[12:14])
date(y, m, d)
except ValueError:
return ValidationResult(
False, "", value,
f"出生日期 {value[6:14]} 不合法(含闰年 / 月日范围)",
self.standard_id,
)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-001-d 身份证 15 位老证兼容(warning 提示)
GB 11643-1989 老证为 15 位(6 位地址 + 6 位出生年月(YYMMDD)+ 3 位顺序码)。
本 indicator 命中即提示「存在 15 位存量」,不视为 error(可能为历史数据)。
"""
from __future__ import annotations
import re
from .base import BaseStandard, ValidationResult
class IdCard15LegacyIndicator(BaseStandard):
standard_id = "IND-001-d"
standard_name = "GB 11643-1989 身份证 15 位老证提示"
applies_to_fields = ["id_card", "id_card_no", "id_number", "identity_card"]
description = (
"命中即提示存在 15 位老证(已自 1999-07-01 停发)。"
" 仅 warning,不视作违规 —— 业务侧需评估是否升级为 18 位。"
)
group = "国标字段规范"
severity = "warning"
_REGEX_15 = re.compile(r"^[1-9]\d{5}\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}$")
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
if len(value) == 15 and self._REGEX_15.match(value):
return ValidationResult(
False, "", value,
"15 位老证,建议升级为 18 位(GB 11643-1999)",
self.standard_id,
)
# 非 15 位不算违规 —— 交给 IND-001-a/b/c
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-002-a USCC 格式(18 位 + 字符集 + 结构)
拆分自 STD-002。校验「长度 + 字符集 + 1+1+6+9+1 结构」。
"""
from __future__ import annotations
import re
from .base import BaseStandard, ValidationResult
class UsccFormatIndicator(BaseStandard):
standard_id = "IND-002-a"
standard_name = "GB 32100-2015 USCC 格式"
applies_to_fields = ["uscc", "credit_code", "social_credit_code", "unified_social_credit_code"]
description = "长度 18;31 字符集(不含 I/O/Z/S/V);登记管理机关(1)+机构类别(1)+区划(6)+主体标识码(9)+校验码(1)"
group = "国标字段规范"
severity = "warning"
_REGEX = re.compile(r"^[1-9A-HJ-NP-RT-Y]\d[0-9A-HJ-NP-RT-Y]{6}[0-9A-HJ-NP-RT-Y]{9}[0-9A-HJ-NP-RT-Y]$")
_FORBIDDEN = set("IOZSV")
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
if len(value) != 18:
return ValidationResult(False, "", value, f"长度 {len(value)} ≠ 18", self.standard_id)
bad = set(value) & self._FORBIDDEN
if bad:
return ValidationResult(False, "", value, f"包含禁用字符 {''.join(sorted(bad))}", self.standard_id)
if not self._REGEX.match(value):
return ValidationResult(False, "", value, "不符合 1+1+6+9+1 结构", self.standard_id)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-002-b USCC 校验位(MOD 31-3)
拆分自 STD-002。假定格式已通过(IND-002-a),仅校验末位。
"""
from __future__ import annotations
from .base import BaseStandard, ValidationResult
class UsccChecksumIndicator(BaseStandard):
standard_id = "IND-002-b"
standard_name = "GB 32100-2015 USCC 校验位"
applies_to_fields = ["uscc", "credit_code", "social_credit_code", "unified_social_credit_code"]
description = "MOD 31-3:Σ Ci × 3^i mod 31,取字符集倒索引"
group = "国标字段规范"
severity = "warning"
_CHARSET = "0123456789ABCDEFGHJKLMNPQRTUWXY" # 31 个字符
_CHAR_MAP = {c: i for i, c in enumerate(_CHARSET)}
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
if len(value) != 18:
return ValidationResult(
False, "", value,
"校验位要求 18 位(请先确认 IND-002-a 格式)",
self.standard_id,
)
try:
expected = self._calc(value[:17])
except ValueError as e:
return ValidationResult(False, "", value, str(e), self.standard_id)
if value[17] != expected:
return ValidationResult(
False, "", value,
f"校验位错误,应为 {expected}",
self.standard_id,
)
return ValidationResult(True, "", value, "", self.standard_id)
def _calc(self, seventeen: str) -> str:
total = 0
for i, c in enumerate(seventeen):
if c not in self._CHAR_MAP:
raise ValueError(f"字符 {c} 不在 GB 32100 字符集内")
total += self._CHAR_MAP[c] * (3 ** i)
idx = (31 - total % 31) % 31
return self._CHARSET[idx]
\ No newline at end of file
"""IND-003-a 手机号格式(11 位 + 1[3-9] 开头 + 清洗)
拆分自 STD-003。仅做宽松长度 / 头部校验。
"""
from __future__ import annotations
import re
from .base import BaseStandard, ValidationResult
class MobileFormatIndicator(BaseStandard):
standard_id = "IND-003-a"
standard_name = "工信部 手机号格式"
applies_to_fields = [
"mobile", "phone", "mobile_phone", "tel", "telephone",
"contact_phone", "customer_mobile", "legal_person_mobile",
"notify_phone", "maintainer_phone", "manager_phone", "receiver_phone",
]
description = "11 位;以 1 开头,第 2 位 3-9;自动清洗 +86 / 空格 / 横线"
group = "国标字段规范"
severity = "warning"
_REGEX = re.compile(r"^1[3-9]\d{9}$")
@staticmethod
def _clean(raw: str) -> str:
v = re.sub(r"[\s\-]", "", raw)
if v.startswith("+86"):
v = v[3:]
elif v.startswith("86") and len(v) == 13:
v = v[2:]
return v
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
v = self._clean(value)
if len(v) != 11:
return ValidationResult(False, "", value, f"清洗后长度 {len(v)} ≠ 11", self.standard_id)
if not self._REGEX.match(v):
return ValidationResult(False, "", value, "不符合 1[3-9]XXXXXXXXX 格式", self.standard_id)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-003-b 手机号号段(更严格 —— 命中已知号段)
拆分自 STD-003。
"""
from __future__ import annotations
import re
from .base import BaseStandard, ValidationResult
class MobileSegmentIndicator(BaseStandard):
standard_id = "IND-003-b"
standard_name = "工信部 手机号号段"
applies_to_fields = [
"mobile", "phone", "mobile_phone", "tel", "telephone",
"contact_phone", "customer_mobile", "legal_person_mobile",
"notify_phone", "maintainer_phone", "manager_phone", "receiver_phone",
]
description = "在 IND-003-a 通过的前提下,进一步校验号段正则"
group = "国标字段规范"
severity = "warning"
_DETAILED_REGEX = re.compile(
r"^1(3\d|4[5-9]|5[0-35-9]|6[2567]|7[0-8]|8\d|9[0-35-9])\d{8}$"
)
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
v = re.sub(r"[\s\-]", "", value)
if v.startswith("+86"):
v = v[3:]
elif v.startswith("86") and len(v) == 13:
v = v[2:]
if not self._DETAILED_REGEX.match(v):
return ValidationResult(False, "", value, "号段不在已知号段表内", self.standard_id)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-004-a 行政区划代码格式(6 位 / 12 位 + 省级代码合法性)
拆分自 STD-004。
"""
from __future__ import annotations
import re
from .base import BaseStandard, ValidationResult
from .data_loader import load_district_province_codes
class XzqhFormatIndicator(BaseStandard):
standard_id = "IND-004-a"
standard_name = "GB/T 2260 行政区划代码格式"
applies_to_fields = ["xzqhbm", "xzqhjb", "adcode", "district_code", "province_code", "city_code", "region_code"]
description = "6 位标准版或 12 位扩展版;前 2 位必须 ∈ GB/T 2260 省级代码表"
group = "民政信息" # 行政区划本身属于民政
severity = "warning"
_REGEX_6 = re.compile(r"^\d{6}$")
_REGEX_12 = re.compile(r"^\d{12}$")
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
if not (self._REGEX_6.match(value) or self._REGEX_12.match(value)):
return ValidationResult(False, "", value, "长度必须为 6 或 12 位数字", self.standard_id)
province = value[:2]
if province not in load_district_province_codes():
return ValidationResult(
False, "", value,
f"省级代码 {province} 不在 GB/T 2260 表内",
self.standard_id,
)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-004-b 行政区划编码存在性(6 位必须在 GB/T 2260-2007 表内)
拆分自 STD-004。在 IND-004-a 通过的前提下,进一步查 6 位代码是否真实存在。
12 位扩展版本 indicator 仅做长度合法性,不查全表(行政区划代码表只有 6 位版)。
"""
from __future__ import annotations
from .base import BaseStandard, ValidationResult
from .data_loader import load_district_codes
class XzqhExistsIndicator(BaseStandard):
standard_id = "IND-004-b"
standard_name = "GB/T 2260 行政区划编码存在性"
applies_to_fields = ["xzqhbm", "xzqhjb", "adcode", "district_code", "province_code", "city_code", "region_code"]
description = "6 位代码必须在 GB/T 2260-2007 表内(共 3406 条)"
group = "民政信息"
severity = "warning"
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
if len(value) != 6:
return ValidationResult(
False, "", value,
"仅校验 6 位标准版;12 位扩展版交由 IND-004-a 兜底",
self.standard_id,
)
if value not in load_district_codes():
return ValidationResult(
False, "", value,
"6 位代码不在 GB/T 2260-2007 表内",
self.standard_id,
)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-101 证件类型枚举校验
值必须在 {01,02,03,04,05} 内(用户给定 5 值枚举,详见 data/id_type_enum.json)。
- 01=居民身份证
- 02=军官证
- 03=护照
- 04=外国人永久居留身份证
- 05=港澳台居民居住证
注:完整 GA/T 2000.156-2016 三位代码表更细(含军官证细分等),
本项目按用户业务侧给定的 5 值落地。
"""
from __future__ import annotations
from .base import BaseStandard, ValidationResult
from .data_loader import is_valid_id_type
class IdTypeEnumIndicator(BaseStandard):
standard_id = "IND-101"
standard_name = "证件类型枚举"
applies_to_fields = ["id_type", "cert_type", "document_type", "zjlb"]
description = "代码必须在 {01 居民身份证 / 02 军官证 / 03 护照 / 04 永居证 / 05 港澳台居住证} 内"
group = "证件信息"
severity = "warning"
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
v = str(value).strip()
if not is_valid_id_type(v):
return ValidationResult(
False, "", value,
f"证件类型 {v!r} 不在标准枚举内(01/02/03/04/05)",
self.standard_id,
)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-201 公民身份号码(18 位 GB 11643 + 兼容 15 位老证)
复用了 IND-001-a/b/c 的逻辑,但作为「以身份证为核心」的证件号码校验:
- 命中字段同时包含身份证 + 其他证件号时,由 IND-201~204 分流
- 这里与 IND-001-a/b/c 是有意重叠的(用于「证件号码」语义场景)
未做 ID 类型分发(type=01 → IND-201)—— 那由 IND-301 跨字段校验器处理。
"""
from __future__ import annotations
import re
from .base import BaseStandard, ValidationResult
class IdNumberIdCard18Indicator(BaseStandard):
standard_id = "IND-201"
standard_name = "证件号码-居民身份证 18 位(GB 11643-1999)"
applies_to_fields = ["id_no", "cert_no", "zjhm", "document_no"]
description = (
"18 位身份证:长度 + 正则 + 校验位 + 出生日期 全部通过。"
" 适用字段含 id_no / cert_no / zjhm / document_no(与 id_card 等并列,"
" 由前端/业务侧按语义挑选哪个 indicator 命中)"
)
group = "证件信息"
severity = "warning"
_REGEX_18 = re.compile(r"^[1-9]\d{5}(19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[\dXx]$")
_REGEX_15 = re.compile(r"^[1-9]\d{5}\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}$")
_WEIGHTS = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
_CHECKSUM_MAP = ["1", "0", "X", "9", "8", "7", "6", "5", "4", "3", "2"]
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
v = str(value).strip()
if len(v) == 15:
if self._REGEX_15.match(v):
return ValidationResult(
False, "", value,
"15 位老证,请升级为 18 位(GB 11643-1999)",
self.standard_id,
)
return ValidationResult(False, "", value, "15 位格式错误", self.standard_id)
if len(v) != 18:
return ValidationResult(False, "", value, f"长度 {len(v)} ≠ 18", self.standard_id)
if not self._REGEX_18.match(v):
return ValidationResult(False, "", value, "不符合 18 位 6+8+3+1 结构", self.standard_id)
# 出生日期
from datetime import date
try:
date(int(v[6:10]), int(v[10:12]), int(v[12:14]))
except ValueError:
return ValidationResult(False, "", value, f"出生日期 {v[6:14]} 不合法", self.standard_id)
# 校验位
total = sum(int(c) * w for c, w in zip(v[:17], self._WEIGHTS))
expected = self._CHECKSUM_MAP[total % 11]
if v[17].upper() != expected:
return ValidationResult(
False, "", value,
f"校验位错误,应为 {expected}",
self.standard_id,
)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-202 证件号码-中国护照
号码格式(基于 2026-08-07 网络调研结论):
- 普通护照(旧):G + 8 位数字 = 9 字符
- 普通护照(早期电子):E + 8 位数字 = 9 字符(2012-05-15 ~ 2017-04-04)
- 普通护照(电子扩容):E + 1 位字母(排除 I/O)+ 7 位数字 = 9 字符(2017-04-05 起)
- 公务普通:PE + 7 位数字 = 9 字符
- 公务:SE + 7 位数字 = 9 字符
- 外交:DE + 7 位数字 = 9 字符
不在本 indicator 范围内:
- 旧版 P/S/D(未查证到权威起止日期,二手资料冲突 7 vs 8 位)—— 仅 warning 不拒
- ICAO Doc 9303 机读码校验位(属于 MRZ 区,不在可视区号码字符内)
参考资料:
- gov.cn 护照法 https://www.gov.cn/flfg/2006-04/29/content_270216.htm
- ICAO Doc 9303 Part 4
"""
from __future__ import annotations
import re
from .base import BaseStandard, ValidationResult
class IdNumberPassportIndicator(BaseStandard):
standard_id = "IND-202"
standard_name = "证件号码-中国护照"
applies_to_fields = ["passport_no", "hz_no", "huzhao_no"]
description = (
"9 字符:G+8 / E+8 / E+1 字母(排除I/O)+7 / PE+7 / SE+7 / DE+7。"
" 不强校验位算法(公开材料未查到)。"
)
group = "证件信息"
severity = "warning"
# 主流正则(推荐用)
_REGEX = re.compile(
r"^("
r"G[0-9]{8}|" # 普通护照旧版
r"E[0-9]{8}|" # 电子普通护照初期
r"E[A-HJ-NP-Z][0-9]{7}|" # 电子普通护照扩容(字母排除 I/O)
r"PE[0-9]{7}|SE[0-9]{7}|DE[0-9]{7}" # 公务普通 / 公务 / 外交
r")$"
)
# 旧 P/S/D 版本(公开资料冲突 —— 仅作 info 命中,不拒)
_LEGACY_REGEX = re.compile(r"^[PSD][0-9]{7,8}$")
_FORBIDDEN_CHARS = set("'\"`~!@#$%^&*()+=[]{}|\\:;<>?,./")
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
v = str(value).strip().upper()
# 非法字符
if any(c in self._FORBIDDEN_CHARS for c in v):
return ValidationResult(
False, "", value,
"护照号包含非法特殊字符",
self.standard_id,
)
# 长度区间
if not (5 <= len(v) <= 20):
return ValidationResult(
False, "", value,
f"长度 {len(v)} 不在 5-20 区间",
self.standard_id,
)
# 主流正则
if self._REGEX.match(v):
return ValidationResult(True, "", value, "", self.standard_id)
# 旧 P/S/D 兼容 —— 不报错,仅 warning 提示
if self._LEGACY_REGEX.match(v):
return ValidationResult(
False, "", value,
"旧 P/S/D 公务 / 外交护照格式(未查证到权威起止日期,请人工确认)",
self.standard_id,
)
return ValidationResult(
False, "", value,
"不符合已知护照号格式(普通 G+8 / E+8 / E+字母+7 / 公务 PE/SE/DE+7)",
self.standard_id,
)
\ No newline at end of file
"""IND-203 证件号码-港澳台居民居住证(18 位公民身份号码框架)
依据:国办发〔2018〕81 号《港澳台居民居住证申领发放办法》
- 18 位:6 位地址码 + 8 位出生日期 + 3 位顺序码 + 1 位校验码
- 地址码:810000=香港 / 820000=澳门 / 830000=台湾
- 与居民身份证共用 GB 11643-1999 校验算法
注意:GB/T 2260-2007 现行版本只含 810000 / 820000(不含 830000);
830000 由国务院台办配套文件给出。本 indicator 在地址码段位放宽,
接受 81/82/83 三段前缀。
"""
from __future__ import annotations
import re
from datetime import date
from .base import BaseStandard, ValidationResult
class IdNumberHKMacaoResidenceIndicator(BaseStandard):
standard_id = "IND-203"
standard_name = "证件号码-港澳台居民居住证(国办发〔2018〕81 号)"
applies_to_fields = [
"hk_residence_no", "mo_residence_no", "tw_residence_no",
"gat_residence_no", "gatjzz_no",
]
description = (
"18 位 + 地址码 ∈ {810000 香港, 820000 澳门, 830000 台湾} + GB 11643 校验位"
)
group = "证件信息"
severity = "warning"
_REGEX = re.compile(r"^8[123]\d{4}(19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[\dXx]$")
_WEIGHTS = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
_CHECKSUM_MAP = ["1", "0", "X", "9", "8", "7", "6", "5", "4", "3", "2"]
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
v = str(value).strip()
if len(v) != 18:
return ValidationResult(False, "", value, f"长度 {len(v)} ≠ 18", self.standard_id)
if not self._REGEX.match(v):
return ValidationResult(
False, "", value,
"不符合 81/82/83 前缀 + 18 位 GB 11643 结构",
self.standard_id,
)
try:
date(int(v[6:10]), int(v[10:12]), int(v[12:14]))
except ValueError:
return ValidationResult(False, "", value, f"出生日期 {v[6:14]} 不合法", self.standard_id)
total = sum(int(c) * w for c, w in zip(v[:17], self._WEIGHTS))
expected = self._CHECKSUM_MAP[total % 11]
if v[17].upper() != expected:
return ValidationResult(False, "", value, f"校验位错误,应为 {expected}", self.standard_id)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-204 证件号码-外国人永久居留身份证
两种版本共存:
- 旧版(2017-):3 位大写字母(ISO 3166-1 alpha-3 国别码)+ 12 位数字 = 15 位
- 新版「五星卡」(2023-12-01 起):18 位,首位固定 9
仅做格式校验;官方未公开校验位算法(不在二手 MOD 11 推测上做强校验)。
"""
from __future__ import annotations
import re
from .base import BaseStandard, ValidationResult
class IdNumberForeignResidenceIndicator(BaseStandard):
standard_id = "IND-204"
standard_name = "证件号码-外国人永久居留身份证"
applies_to_fields = ["foreign_residence_no", "yd_no", "ywjzj_no", "prc_residence_no"]
description = "旧版 15 位(3 字母 ISO 3166-1 alpha-3 + 12 数字)/ 新版 18 位(首位 9)"
group = "证件信息"
severity = "warning"
_REGEX_OLD = re.compile(r"^[A-Z]{3}\d{12}$")
_REGEX_NEW = re.compile(r"^9\d{16}[\dX]$")
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
v = str(value).strip().upper()
if self._REGEX_NEW.match(v):
return ValidationResult(True, "", value, "(2023 版五星卡,格式通过)", self.standard_id)
if self._REGEX_OLD.match(v):
return ValidationResult(True, "", value, "(2017 旧版 15 位,格式通过)", self.standard_id)
return ValidationResult(
False, "", value,
"不符合永居证格式(旧 15 位 LLL+12 数字 / 新 18 位 9+17 数字+X)",
self.standard_id,
)
\ No newline at end of file
"""IND-301 证件类型 + 证件号 一致性(跨字段校验器)
按证件类型分发到对应 indicator:
- 01 居民身份证 → 复用 IND-201
- 03 护照 → 复用 IND-202
- 05 港澳台居住证 → 复用 IND-203
- 04 永居证 → 复用 IND-204
- 02 军官证 → 当前跳过(无公开权威格式文档)
通过覆写 validate_rows() 实现:rows 元素含 {id_type, id_no} 两个键。
step7 检测到 indicator 实现 validate_rows 时走「跨字段」轮,
SQL 模板走 web/sql/standards/sample_field_pairs.sql。
"""
from __future__ import annotations
from .base import BaseStandard, ValidationResult
from .data_loader import is_valid_id_type
from .ind_201_id_number_id_card import IdNumberIdCard18Indicator
from .ind_202_id_number_passport import IdNumberPassportIndicator
from .ind_203_id_number_hkmacao_residence import IdNumberHKMacaoResidenceIndicator
from .ind_204_id_number_foreign_residence import IdNumberForeignResidenceIndicator
# 类型 → indicator 映射
_DISPATCH = {
"01": IdNumberIdCard18Indicator(),
"03": IdNumberPassportIndicator(),
"04": IdNumberForeignResidenceIndicator(),
"05": IdNumberHKMacaoResidenceIndicator(),
}
class IdTypeNumberConsistencyIndicator(BaseStandard):
standard_id = "IND-301"
standard_name = "证件类型 + 证件号 一致性"
# applies_to_fields 留空 —— 本 indicator 由跨字段 SQL 触发,不走单字段自动注册
applies_to_fields = []
description = (
"type=01 → 必须通过 IND-201(18 位 GB 11643);"
"type=03 → IND-202(护照);type=04 → IND-204(永居证);"
"type=05 → IND-203(港澳台居住证)。"
" SQL 模板:web/sql/standards/sample_field_pairs.sql"
)
group = "证件信息"
severity = "warning"
depends_on = ["IND-101", "IND-201", "IND-202", "IND-203", "IND-204"]
def validate(self, value: str) -> ValidationResult:
# 单值模式不适用 —— 跨字段校验必须走 validate_rows
return ValidationResult(True, "", value, "IND-301 仅支持跨字段校验(validate_rows)", self.standard_id)
def validate_rows(self, rows: list[dict]) -> list[dict]:
out = []
for row in rows:
t = (row.get("id_type") or "").strip()
n = (row.get("id_no") or "").strip()
if not t or not n:
continue
if not is_valid_id_type(t):
out.append({
"value": n,
"id_type": t,
"reason": f"证件类型 {t!r} 不在标准枚举(IND-101 也会命中)",
"rule": "type_invalid",
})
continue
indicator = _DISPATCH.get(t)
if indicator is None:
# 02 军官证等暂跳过
continue
r = indicator.validate(n)
if not r.valid:
out.append({
"value": n,
"id_type": t,
"reason": r.reason,
"rule": f"type_{t}_mismatch",
})
return out
\ No newline at end of file
"""IND-302 证件唯一性(跨表聚合校验器 —— GROUP BY HAVING)
按 (id_type, id_no) 维度跨表去重,疑似重复开户:
- 在多张表里出现相同 (type, no) → 报「疑似重复开户」
- 在同一张表里出现多次 → 报「同表重复」
走 SQL `GROUP BY HAVING COUNT(*) > N`,SQL 模板:
web/sql/standards/sample_field_groups.sql
indicator 类本身不参与单字段自动发现,由 step7 第二轮显式调用。
"""
from __future__ import annotations
from .base import BaseStandard, ValidationResult
class IdUniquenessIndicator(BaseStandard):
standard_id = "IND-302"
standard_name = "证件唯一性(跨表去重)"
applies_to_fields = []
description = (
"GROUP BY (id_type, id_no) HAVING COUNT > N 检出重复。"
" SQL 模板:web/sql/standards/sample_field_groups.sql"
)
group = "证件信息"
severity = "warning"
depends_on = ["IND-101"]
def validate(self, value: str) -> ValidationResult:
# 单值模式不适用
return ValidationResult(True, "", value, "IND-302 仅支持跨表聚合(GROUP BY HAVING)", self.standard_id)
\ No newline at end of file
"""IND-401 姓名字符集
校验规则(基于 GA/T 2000.27 等行业惯例 + 网络调研):
- 允许字符:
- 汉字(基本区 0x4E00-0x9FA5 + 扩展 A 区 0x3400-0x4DBF)
- 维吾尔族等少数民族的「·」间隔符(U+00B7,中点)
- 半角空格(U+0020)
- ASCII 字母(部分业务系统支持拼音名)
- 禁止:
- 纯数字 / 纯特殊符号
- 控制字符 / Emoji 等奇异字符
"""
from __future__ import annotations
import re
from .base import BaseStandard, ValidationResult
_MIDDLE_DOT = "·" # 维吾尔族等少数民族「·」间隔符
_FULLWIDTH_DOT = "." # 全角点(兼容录入)
def _is_han(c: str) -> bool:
return (
"一" <= c <= "鿿" # CJK 基本
or "㐀" <= c <= "䶿" # CJK 扩展 A
)
def _is_allowed(c: str) -> bool:
if _is_han(c):
return True
if c in (_MIDDLE_DOT, _FULLWIDTH_DOT, " "):
return True
if "A" <= c <= "Z" or "a" <= c <= "z":
return True
return False
class NameCharsetIndicator(BaseStandard):
standard_id = "IND-401"
standard_name = "姓名字符集"
applies_to_fields = [
"name", "real_name", "customer_name", "user_name", "xm",
"user_real_name", "person_name",
]
description = (
"允许:汉字 / 维吾尔等少数民族的「·」/ 全角点「.」/ 半角空格 / ASCII 字母。"
" 禁止:纯数字 / 纯特殊符号 / 控制字符 / Emoji"
)
group = "身份信息"
severity = "warning"
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
v = str(value).strip()
if not v:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
has_han = any(_is_han(c) for c in v)
has_letter = bool(re.search(r"[A-Za-z]", v))
if not (has_han or has_letter):
return ValidationResult(
False, "", value,
"姓名不含任何汉字或字母(疑似纯数字 / 纯特殊符号)",
self.standard_id,
)
bad = [c for c in v if not _is_allowed(c)]
if bad:
return ValidationResult(
False, "", value,
f"姓名包含不规范字符 {''.join(bad[:5])!r}",
self.standard_id,
)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-402 姓名长度(2-20 字符)
业务侧惯例;公安部《姓名登记条例(征求意见稿)》及实际办理经验。
少数民族姓名可能较长(如「额尔德木图·巴特巴依尔」13 字),20 字留余。
"""
from __future__ import annotations
from .base import BaseStandard, ValidationResult
class NameLengthIndicator(BaseStandard):
standard_id = "IND-402"
standard_name = "姓名长度"
applies_to_fields = [
"name", "real_name", "customer_name", "user_name", "xm",
"user_real_name", "person_name",
]
description = "字符数 ∈ [2, 20]"
group = "身份信息"
severity = "warning"
_MIN = 2
_MAX = 20
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
v = str(value).strip()
n = len(v)
if n < self._MIN:
return ValidationResult(
False, "", value,
f"姓名长度 {n} < {self._MIN}",
self.standard_id,
)
if n > self._MAX:
return ValidationResult(
False, "", value,
f"姓名长度 {n} > {self._MAX}",
self.standard_id,
)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-501 出生日期格式(8 位 YYYYMMDD + 合法日期)
兼容 ISO 8601(YYYY-MM-DD)、YYYY/MM/DD、YYYY.MM.DD —— 实际业务常见。
"""
from __future__ import annotations
import re
from datetime import date
from .base import BaseStandard, ValidationResult
_RE = re.compile(r"^(\d{4})[-./]?(\d{2})[-./]?(\d{2})$")
class BirthDateFormatIndicator(BaseStandard):
standard_id = "IND-501"
standard_name = "出生日期格式"
applies_to_fields = ["birth_date", "birthday", "csrq", "date_of_birth"]
description = "8 位 YYYYMMDD 或带 - / . 分隔;日期必须真实存在(含闰年)"
group = "身份信息"
severity = "warning"
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
v = str(value).strip()
m = _RE.match(v)
if not m:
return ValidationResult(
False, "", value,
"不符合 8 位 YYYYMMDD 格式(或含 - / . 分隔)",
self.standard_id,
)
try:
y_, m_, d_ = int(m.group(1)), int(m.group(2)), int(m.group(3))
date(y_, m_, d_)
except ValueError:
return ValidationResult(
False, "", value,
f"日期 {v} 不合法(含闰年 / 月日范围)",
self.standard_id,
)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-502 出生日期年龄范围(16 ~ 法定退休年龄)
策略:
- 最低 16 岁(劳动法最低就业年龄)
- 上限按性别字段推断:男 60 / 女 50 / 女干部 55 —— 详见 retirement_age.json
- 找不到性别字段时退化为 60
- 假定出生日期格式已通过(IND-501)
实现为单值校验(需要外部传入性别)—— 不便跨字段,于是放宽为:
- 若同表有 gender 字段,step7 在调用前做 JOIN 拼出 (birth, gender) 后调用本类;
- 默认降级为 60 岁上限(无论性别)。
"""
from __future__ import annotations
from datetime import date
from typing import Iterable
from .base import BaseStandard, ValidationResult
from .data_loader import load_retirement_age_rules
_RE_YMD = __import__("re").compile(r"^(\d{4})[-./]?(\d{2})[-./]?(\d{2})$")
class BirthDateAgeRangeIndicator(BaseStandard):
standard_id = "IND-502"
standard_name = "出生日期年龄范围"
applies_to_fields = ["birth_date", "birthday", "csrq", "date_of_birth"]
description = (
"年龄 ∈ [16, 法定退休年龄]。退休年龄按性别:男 60 / 女(工人)50 / 女(干部)55;"
" 默认 60。"
)
group = "身份信息"
severity = "warning"
_MIN_AGE = 16
def validate(self, value: str) -> ValidationResult:
# 默认按「无性别字段」处理:上限 60
return self._validate_with_gender(value, gender=None)
def validate_with_gender(self, value: str, gender: str | None) -> ValidationResult:
"""step7 入口在能拿到性别字段时显式调用本方法。"""
return self._validate_with_gender(value, gender)
def _validate_with_gender(self, value: str, gender: str | None) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
v = str(value).strip()
m = _RE_YMD.match(v)
if not m:
return ValidationResult(
False, "", value,
"年龄校验要求先通过 IND-501 格式",
self.standard_id,
)
try:
bd = date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
except ValueError:
return ValidationResult(False, "", value, "日期不合法", self.standard_id)
today = date.today()
age = today.year - bd.year - ((today.month, today.day) < (bd.month, bd.day))
if age < self._MIN_AGE:
return ValidationResult(
False, "", value,
f"年龄 {age} < {self._MIN_AGE}(劳动法最低就业年龄)",
self.standard_id,
)
# 退休年龄上限
rules = load_retirement_age_rules()
upper = rules.get("default", {}).get("age", 60)
if gender in ("2", "F", "f", "女"):
# 简化处理:女工 50 / 女干部 55 —— 默认按工人 50
upper = rules.get("female_worker", {}).get("age", 50)
elif gender in ("1", "M", "m", "男"):
upper = rules.get("male", {}).get("age", 60)
if age > upper:
return ValidationResult(
False, "", value,
f"年龄 {age} > {upper}({gender or '未知'}),疑似超龄(贷款业务尤其注意)",
self.standard_id,
)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-601 行政区划代码(GB/T 2260-2007)
6 位代码必须在 GB/T 2260-2007 表内(3406 条)。
与 IND-004-a/b 字段重叠 —— 本 indicator 适用于「以行政区划编码为核心语义」字段(如户籍所在地代码)。
"""
from __future__ import annotations
from .base import BaseStandard, ValidationResult
from .data_loader import is_valid_district_6
class DistrictCodeIndicator(BaseStandard):
standard_id = "IND-601"
standard_name = "GB/T 2260-2007 行政区划代码"
applies_to_fields = [
"hjszd", "hjsxdm", "hukou_xzqh", "registered_district",
"native_district_code", "current_district_code",
]
description = "6 位代码 ∈ GB/T 2260-2007(3406 条)"
group = "民政信息"
severity = "warning"
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
v = str(value).strip()
if len(v) != 6 or not v.isdigit():
return ValidationResult(
False, "", value,
f"行政区划代码须为 6 位数字(实际 {len(v)} 位)",
self.standard_id,
)
if not is_valid_district_6(v):
return ValidationResult(
False, "", value,
"6 位代码不在 GB/T 2260-2007 表内",
self.standard_id,
)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
"""IND-602 民族代码(GB/T 3304-1991)
2 位数字 ∈ {01..56, 97, 98}
- 01-56:正文 56 民族(extracted_standards.json 提供)
- 97:其他(公安部补充码,未在 extracted_standards.json 中)
- 98:外国血统中国籍人士(同上)
- 81(穿青人)是公安部扩展码,未在标准正文 —— 不强校验
"""
from __future__ import annotations
from .base import BaseStandard, ValidationResult
from .data_loader import is_valid_ethnicity
class EthnicityIndicator(BaseStandard):
standard_id = "IND-602"
standard_name = "GB/T 3304-1991 民族代码"
applies_to_fields = ["ethnicity", "nation", "mz", "minzu", "ethnic"]
description = "2 位数字 ∈ {01..56, 97, 98}(GB/T 3304-1991 正文 56 民族 + 补充码 97/98)"
group = "民政信息"
severity = "warning"
# extracted_standards.json 只含正文 01-56;97/98 是补充码,手动接受
_EXTRA = {"97", "98"}
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
v = str(value).strip().zfill(2) # 接受 "1" → "01"
if is_valid_ethnicity(v) or v in self._EXTRA:
return ValidationResult(True, "", value, "", self.standard_id)
return ValidationResult(
False, "", value,
f"民族代码 {v!r} 不在 GB/T 3304-1991 表内(01-56 + 97/98)",
self.standard_id,
)
\ No newline at end of file
"""IND-603 婚姻状况代码(GB/T 2261.2-2003)
2 位 ∈ {10, 20, 21, 22, 23, 30, 40, 90}
- 10 未婚 / 20 已婚 / 21 初婚 / 22 再婚 / 23 复婚
- 30 丧偶 / 40 离婚 / 90 未说明
注意:
- GB/T 4761 是「家庭关系代码」,不是婚姻状况 —— 别用错标准。
- 旧 GB/T 4766-1984 已被本标准代替。
"""
from __future__ import annotations
from .base import BaseStandard, ValidationResult
from .data_loader import is_valid_marriage
class MarriageStatusIndicator(BaseStandard):
standard_id = "IND-603"
standard_name = "GB/T 2261.2-2003 婚姻状况代码"
applies_to_fields = ["marital_status", "marriage", "hyzk"]
description = "2 位 ∈ {10,20,21,22,23,30,40,90}(GB/T 2261.2-2003)"
group = "民政信息"
severity = "warning"
def validate(self, value: str) -> ValidationResult:
if not value:
return ValidationResult(True, "", value, "空值跳过", self.standard_id)
v = str(value).strip()
if not is_valid_marriage(v):
return ValidationResult(
False, "", value,
f"婚姻状况代码 {v!r} 不在 GB/T 2261.2-2003 表内(10/20/21/22/23/30/40/90)",
self.standard_id,
)
return ValidationResult(True, "", value, "", self.standard_id)
\ No newline at end of file
......@@ -44,14 +44,20 @@ def discover_standards() -> dict[str, BaseStandard]:
def list_all_standards() -> list[dict]:
"""返回所有标准的基本信息(用于报告/调试)"""
"""返回所有标准的基本信息(用于报告/调试)
返回字段:
- id, name, applies_to_fields, description: 同原
- group: "国标字段规范" / "证件信息" / "身份信息" / "民政信息" / "通用"
- severity: "warning" | "error"
"""
import standards
out = []
seen: set[type] = set()
package_path = Path(__file__).parent
for module_info in pkgutil.iter_modules([str(package_path)]):
if module_info.name in ("base", "registry"):
if module_info.name in ("base", "registry", "data_loader"):
continue
module = importlib.import_module(f"standards.{module_info.name}")
for attr_name in dir(module):
......@@ -69,5 +75,91 @@ def list_all_standards() -> list[dict]:
"name": inst.standard_name,
"applies_to_fields": inst.applies_to_fields,
"description": inst.description,
"group": getattr(inst, "group", "通用"),
"severity": getattr(inst, "severity", "warning"),
})
return out
def list_standards_by_group() -> dict[str, list[dict]]:
"""按 group 分组返回所有标准 —— step7 拆 tab 用"""
by_group: dict[str, list[dict]] = {}
for s in list_all_standards():
by_group.setdefault(s["group"], []).append(s)
return by_group
def load_standard_class(standard_id: str) -> type[BaseStandard] | None:
"""按 standard_id(如 "IND-001-a")找到对应的标准子类。
返回类本身(不是实例),调用方负责实例化。
用于 orchestrator 按 step_id 找到对应 indicator 类后分发执行。
"""
import standards
seen: set[type] = set()
package_path = Path(__file__).parent
for module_info in pkgutil.iter_modules([str(package_path)]):
if module_info.name in ("base", "registry", "data_loader"):
continue
module = importlib.import_module(f"standards.{module_info.name}")
for attr_name in dir(module):
attr = getattr(module, attr_name)
if (
isinstance(attr, type)
and issubclass(attr, BaseStandard)
and attr is not BaseStandard
and attr not in seen
):
seen.add(attr)
if attr.standard_id == standard_id:
return attr
return None
# 缓存 indicator 的 prefix(IND- 或 STD-),用于 step_id ↔ indicator_id 双向还原
_PREFIX_RE = __import__("re").compile(r"^(IND|STD)-")
def indicator_step_id(standard_id: str) -> str:
"""IND-001-a → std_ind_001_a;STD-001 → std_std_001(保留 prefix 信息)。
标准 ID 形如 "IND-NNN-x" / "IND-NNN" / "STD-NNN";取 prefix + 尾段拼到 step_id 里。
"""
m = _PREFIX_RE.match(standard_id)
prefix = (m.group(1) if m else "IND").lower() # "ind" / "std"
tail = standard_id[m.end():] if m else standard_id # "001-a" / "001"
return f"std_{prefix}_{tail.lower().replace('-', '_')}"
def step_id_to_indicator_id(step_id: str) -> str | None:
"""std_ind_001_a → IND-001-a;std_std_001 → STD-001;非 indicator step 返回 None。"""
m = __import__("re").match(r"^std_(ind|std)_(.+)$", step_id)
if not m:
return None
prefix = m.group(1).upper()
raw = m.group(2)
if "_" in raw:
num, suffix = raw.rsplit("_", 1)
# 标准 ID 后缀保持小写(IND-001-a / IND-001-b ...)
return f"{prefix}-{num}-{suffix.lower()}"
return f"{prefix}-{raw}"
def list_step_ids() -> list[tuple[str, dict]]:
"""列出所有 indicator 的 (step_id, metadata)(给 orchestrator 循环注册用)。
返回顺序按 group(国标字段规范 → 证件信息 → 身份信息 → 民政信息 → 通用),
同 group 内按 standard_id 字典序。
metadata 字段:standard_id / standard_name / group / severity / applies_to_fields
"""
items = list_all_standards()
items.sort(key=lambda s: (
{"国标字段规范": 0, "证件信息": 1, "身份信息": 2, "民政信息": 3, "通用": 4}.get(s["group"], 99),
s["id"],
))
out = []
for s in items:
sid = indicator_step_id(s["id"])
out.append((sid, s))
return out
"""GB 11643-1999 公民身份号码"""
"""GB 11643-1999 公民身份号码 —— 已拆分为 IND-001-a/b/c/d,本类保留为 deprecated 占位。
本类已不再自动注册到字段映射(applies_to_fields=[])。字段命中由
IND-001-a(格式)/ IND-001-b(校验位)/ IND-001-c(出生日期)/
IND-001-d(15 位兼容)四个细粒度 indicator 接管。
保留本文件仅为向后兼容旧引用 + 国标清单页面调试用。
"""
from __future__ import annotations
......@@ -9,10 +16,15 @@ from .base import BaseStandard, ValidationResult
class IdCardStandard(BaseStandard):
standard_id = "STD-001"
standard_name = "GB 11643-1999 公民身份号码"
applies_to_fields = ["id_card", "id_card_no", "id_number", "identity_card"]
description = "18位:6位地址码 + 8位出生日期 + 3位顺序码 + 1位校验码(ISO 7064 MOD 11-2)"
standard_id = "STD-001" # 保留旧 ID 不变(list_all_standards 仍会列出,标 deprecated)
standard_name = "GB 11643-1999 公民身份号码 [已拆分]"
applies_to_fields = [] # 关键:置空 → discover_standards 不再自动注册
description = (
"18位:6位地址码 + 8位出生日期 + 3位顺序码 + 1位校验码(ISO 7064 MOD 11-2)。"
" 本类已拆分 → 字段命中请用 IND-001-a/b/c/d"
)
group = "通用"
severity = "warning"
_REGEX = re.compile(r"^[1-9]\d{5}(19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[\dXx]$")
_WEIGHTS = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
......
"""GB 32100-2015 法人和其他组织统一社会信用代码"""
"""GB 32100-2015 法人和其他组织统一社会信用代码 —— 已拆分为 IND-002-a/b,本类保留为 deprecated 占位。"""
from __future__ import annotations
......@@ -9,9 +9,14 @@ from .base import BaseStandard, ValidationResult
class UsccStandard(BaseStandard):
standard_id = "STD-002"
standard_name = "GB 32100-2015 统一社会信用代码"
applies_to_fields = ["uscc", "credit_code", "social_credit_code", "unified_social_credit_code"]
description = "18位:登记管理机关(1) + 机构类别(1) + 行政区划(6) + 主体标识码(9) + 校验码(1)"
standard_name = "GB 32100-2015 统一社会信用代码 [已拆分]"
applies_to_fields = []
description = (
"18位:登记管理机关(1) + 机构类别(1) + 行政区划(6) + 主体标识码(9) + 校验码(1)。"
" 本类已拆分 → 字段命中请用 IND-002-a/b"
)
group = "通用"
severity = "warning"
# 排除 I、O、Z、S、V
_REGEX = re.compile(r"^[1-9A-HJ-NP-RT-Y]\d[0-9A-HJ-NP-RT-Y]{6}[0-9A-HJ-NP-RT-Y]{9}[0-9A-HJ-NP-RT-Y]$")
......
"""手机号校验(工信部《电信网编号计划》)"""
"""手机号校验(工信部《电信网编号计划》)—— 已拆分为 IND-003-a/b,本类保留为 deprecated 占位。"""
from __future__ import annotations
......@@ -9,13 +9,13 @@ from .base import BaseStandard, ValidationResult
class MobileStandard(BaseStandard):
standard_id = "STD-003"
standard_name = "工信部《电信网编号计划》手机号"
applies_to_fields = [
"mobile", "phone", "mobile_phone", "tel", "telephone",
"contact_phone", "customer_mobile", "legal_person_mobile",
"notify_phone", "maintainer_phone", "manager_phone", "receiver_phone",
]
description = "11位:以 1 开头,第 2 位 3-9"
standard_name = "工信部《电信网编号计划》手机号 [已拆分]"
applies_to_fields = []
description = (
"11位:以 1 开头,第 2 位 3-9。本类已拆分 → 字段命中请用 IND-003-a/b"
)
group = "通用"
severity = "warning"
_REGEX = re.compile(r"^1[3-9]\d{9}$")
# 详细号段正则(更严格)
......
"""GB/T 2260 行政区划代码
仅做格式校验(6位或12位全数字),具体编码合法性需对照官方编码表。
"""
"""GB/T 2260 行政区划代码 —— 已拆分为 IND-004-a/b,本类保留为 deprecated 占位。"""
from __future__ import annotations
......@@ -12,9 +9,13 @@ from .base import BaseStandard, ValidationResult
class XzqhStandard(BaseStandard):
standard_id = "STD-004"
standard_name = "GB/T 2260 行政区划代码"
applies_to_fields = ["xzqhbm", "xzqhjb", "adcode", "district_code", "province_code", "city_code", "region_code"]
description = "6位标准版或12位扩展版,全数字"
standard_name = "GB/T 2260 行政区划代码 [已拆分]"
applies_to_fields = []
description = (
"6位标准版或12位扩展版。本类已拆分 → 字段命中请用 IND-004-a(格式)/ IND-004-b(编码存在)"
)
group = "通用"
severity = "warning"
_REGEX_6 = re.compile(r"^\d{6}$")
_REGEX_12 = re.compile(r"^\d{12}$")
......
This diff is collapsed.
......@@ -106,7 +106,20 @@ def _run_length_check(*, cfg, dict_data, llm, log, cancel_event, table_filter):
return {"section_key": "length_issues", "data": data}
def _run_standards(*, cfg, dict_data, llm, log, cancel_event, table_filter):
def _run_standards_one(standard_id: str):
"""返回单个 indicator step 的 runner(捕获 standard_id 闭包)。"""
def _runner(*, cfg, dict_data, llm, log, cancel_event, table_filter):
from .step_impl.step7_standards import run_step7_for_indicator
return run_step7_for_indicator(
standard_id, cfg, dict_data=dict_data, log=log, table_filter=table_filter,
)
return _runner
def _run_standards(*, cfg, dict_data, llm, log, cancel_event, table_filter): # noqa: ARG001
"""旧版「国家标准校验」聚合入口 —— 已被 per-indicator 模式取代,保留以兼容历史调用。
实际 orchestrator 不再注册此 step;用户勾选的是 std_ind_* 单 indicator。
"""
from .step_impl.step7_standards import run_step7
data = run_step7(cfg, dict_data=dict_data, log=log, table_filter=table_filter)
return {"section_key": "standard_violations", "data": data}
......@@ -141,13 +154,69 @@ register_step(
requires_db=False, llm_mode="none", required=False, order=4,
fn=_run_length_check,
)
register_step(
step_id="standards",
title="国家标准校验",
description="连库抽样:身份证/USCC/手机号/行政区划符合性",
requires_db=True, llm_mode="none", required=False, order=5,
fn=_run_standards,
)
# ── Indicator 步骤:从 standards 包动态注册 24+ 个 ──────────────
# 每个 indicator 是 1 个 step(用户可单独勾选)。
# order:基础步骤 1-4 → 国标字段规范 100-199 → 证件信息 200-299 → 身份信息 300-399 → 民政信息 400-499
_GROUP_ORDER_START = {
"国标字段规范": 100,
"证件信息": 200,
"身份信息": 300,
"民政信息": 400,
"通用": 500,
}
# 在 import time 触发 standards 包扫描,拿到所有 indicator 元数据
try:
from standards.registry import list_step_ids as _list_std_steps
_INDICATOR_STEPS: list[tuple[str, dict]] = _list_std_steps()
except Exception as _e: # pragma: no cover
logger.warning("加载 indicator 步骤列表失败: %s", _e)
_INDICATOR_STEPS = []
def _register_indicator_steps() -> None:
"""动态注册每个 indicator 为一个 orchestrator step。
顺序:先按 group(国标字段规范 → 证件信息 → 身份信息 → 民政信息),
同 group 内按 standard_id 字典序。
过滤:
- 跳过 std_id 以 "STD-" 开头的已拆分旧插件(applies_to_fields=[])
- 保留 IND-301 / IND-302 等无 applies_to_fields 的跨字段 indicator
"""
prev_group: str | None = None
same_group_count = 0
base_order: int = 0
for sid, meta in _INDICATOR_STEPS:
std_id = meta["id"]
# 跳过已 deprecated 的旧插件(STD-*,已拆分为 IND-* 子指标)
if std_id.startswith("STD-"):
logger.info(f"跳过已 deprecated 旧插件: {std_id}(已被 IND-* 拆分)")
continue
group = meta.get("group", "通用")
if group != prev_group:
same_group_count = 0
base_order = _GROUP_ORDER_START.get(group, 900)
prev_group = group
same_group_count += 1
order = base_order + same_group_count
title = f"{std_id} · {meta['name']}"
description = (
f"分组: {group} | 适用字段: {', '.join(meta.get('applies_to_fields', [])) or '(跨字段/跨表)'} | "
f"{meta.get('description', '')}"
)
register_step(
step_id=sid,
title=title,
description=description,
requires_db=True, llm_mode="none", required=False, order=order,
fn=_run_standards_one(std_id),
)
_register_indicator_steps()
# ── 关键 Step(失败直接终止任务) ──────────────────────────────
......
This diff is collapsed.
-- ============================================================================
-- 跨表去重 —— GROUP BY (id_type, id_no) HAVING COUNT(*) > N(IND-302 用)
-- 调用方:web/core/step_impl/step7_standards.py
-- 参数:
-- ${table} 表名(自动加引号)
-- ${type_col} 证件类型字段名(自动加引号)
-- ${no_col} 证件号字段名(自动加引号)
-- ${min_count} 阈值(默认 2),出现次数 >= 该值才纳入疑似重复
-- ============================================================================
SELECT `${type_col}` AS id_type, `${no_col}` AS id_no, COUNT(*) AS dup_count
FROM `${table}`
WHERE `${type_col}` IS NOT NULL AND TRIM(`${type_col}`) <> ''
AND `${no_col}` IS NOT NULL AND TRIM(`${no_col}`) <> ''
GROUP BY `${type_col}`, `${no_col}`
HAVING COUNT(*) >= ${min_count}
ORDER BY dup_count DESC
LIMIT 200
\ No newline at end of file
-- ============================================================================
-- 抽样「证件类型 + 证件号」同行配对(IND-301 跨字段一致性用)
-- 调用方:web/core/step_impl/step7_standards.py
-- 参数:
-- ${table} 表名(自动加引号)
-- ${type_col} 证件类型字段名(自动加引号)
-- ${no_col} 证件号字段名(自动加引号)
-- ${limit} 抽样上限
-- ============================================================================
SELECT `${type_col}` AS id_type, `${no_col}` AS id_no
FROM `${table}`
WHERE `${type_col}` IS NOT NULL AND TRIM(`${type_col}`) <> ''
AND `${no_col}` IS NOT NULL AND TRIM(`${no_col}`) <> ''
LIMIT ${limit}
\ No newline at end of file
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment