Commit c4ed1b5c authored by Data Governance Dev's avatar Data Governance Dev

feat(web2): 身份证合规扫描 — 多线程 + 实时进度接口

需求链路(按用户描述):
  1. 按当前数据库连接抽数据字典(复用 web.core.data_dict)
  2. 按「字段名 / 字段注释」关键字过滤命中列
  3. 每个命中列 SELECT DISTINCT → Python 校验(GB 11643-1999)
  4. 按违规值反查完整行(SELECT * WHERE col IN (...))
  5. 前端按表折叠渲染

架构亮点 —— 三端点异步:
  POST   /api/id-card/scan                     → 立即 202 {scan_id}(不阻塞)
  GET    /api/id-card/scan/{id}/progress       → {done, total, current_column, status}
  GET    /api/id-card/scan/{id}/result         → 完整 IdCardScanResponse / 202 / 500
  进度共享一个 in-memory ProgressTracker(thread-safe,10min TTL)

性能:
  ThreadPoolExecutor(max_workers=5) 并发跑每列
  每个 worker 自己开 DBConnection(oracledb/pymysql/dmPython 不能跨线程共享)
  实测 Oracle 84 列从单线程 221s → 多线程 115s(1.9x 加速)

方言适配(按 web2 库三态):
  MySQL  → LIMIT %s,%s 占位符
  达梦   → LIMIT ?,? 占位符
  Oracle → 子查询 ROWNUM <= N,:1 :2 :3 numbered bind(11g 无 LIMIT;? 占位符会 ORA-01036)

文件:
- web2/backend/idcard_validator.py  新增
    GB 11643-1999 §6 加权和 + mod 11 映射 + 15 位老号升级 18 位
    违规规则:EMPTY / LENGTH / CHARSET / CHECK_DIGIT / ADDRESS_PREFIX /
             BIRTH_DATE / ORDER_CODE
- web2/backend/routes/idcard_scan.py 新增
    异步三端点 + ProgressTracker + ThreadPoolExecutor 5 worker
- web2/src/config/idCardKeywords.js 新增
    默认关键字常量(前端离线兜底,与 web/configs/standards_match.yaml
    的 IND-001-a 保持一致)
    DEFAULT_NAME_KEYWORDS = [id_card, id_card_no, id_number, identity_card]
    DEFAULT_COMMENT_KEYWORDS = [身份证号]

踩坑见 work-logs/2026-08-17.md 第三、五、六章。
parent 97a82545
"""GB 11643-1999 身份证号码校验
依据:
GB 11643-1999《公民身份号码》§6 校验位计算
GB/T 7408 日期格式
GB/T 2260 地址码(最简校验:非 00 前缀即可,完整区划表留待外部数据接入)
校验项:
1. LENGTH 长度不是 18 位 / 15 位
2. CHARSET 字符集不合规(前 17 位非数字 / 第 18 位非 0-9 或 X)
3. CHECK_DIGIT 校验位计算错误(GB 11643 §6)
4. ADDRESS_PREFIX 地址码全 0 或前 2 位为 00
5. BIRTH_DATE 出生日期不是真实日期 / 晚于今天 / 早于 1900
6. ORDER_CODE 顺序码不在 001-999 范围
返回结构:
{"ok": bool, "violations": [rule_id, ...], "normalized": str | None}
- ok=True 表示合规;violations 为空
- normalized:如果是 15 位老号 → 升 18 位后的值;否则 None
"""
from __future__ import annotations
import datetime as _dt
# GB 11643-1999 §6 加权因子
_WEIGHTS = (7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2)
# mod 11 → 校验码映射
_CHECK_MAP = ("1", "0", "X", "9", "8", "7", "6", "5", "4", "3", "2")
# 合理出生日期区间(身份证制度从 1985 年实施,但出生日期可追溯)
_MIN_BIRTH_YEAR = 1900
_MAX_BIRTH_DATE = _dt.date.today()
def _is_all_same(s: str) -> bool:
return len(s) > 1 and len(set(s)) == 1
def _validate_18(code: str) -> tuple[bool, list[str], str | None]:
"""校验 18 位身份证号。返回 (ok, violations, normalized)"""
violations: list[str] = []
normalized = code
# 字符集
body, last = code[:17], code[17]
if not body.isdigit():
violations.append("CHARSET")
if not (last.isdigit() or last == "X"):
violations.append("CHARSET")
# 不再继续算校验位(last 不合法)
return bool(violations), violations, normalized
# 全相同
if _is_all_same(code):
violations.append("CHARSET") # 复用 CHARSET 槽位:占位符类
return False, violations, normalized
# 校验位
s = sum(int(body[i]) * _WEIGHTS[i] for i in range(17))
expected = _CHECK_MAP[s % 11]
if expected != last:
violations.append("CHECK_DIGIT")
# 地址码(前 6 位)
region = code[:6]
if region == "000000" or region[:2] == "00":
violations.append("ADDRESS_PREFIX")
# 出生日期(第 7-14 位)
try:
bd = _dt.date(int(code[6:10]), int(code[10:12]), int(code[12:14]))
if bd > _MAX_BIRTH_DATE:
violations.append("BIRTH_DATE")
elif bd.year < _MIN_BIRTH_YEAR:
violations.append("BIRTH_DATE")
except ValueError:
violations.append("BIRTH_DATE")
# 顺序码(15-17 位)
order = code[14:17]
if not order.isdigit() or not (1 <= int(order) <= 999):
violations.append("ORDER_CODE")
return (not violations), violations, normalized
def _lift_15_to_18(code15: str) -> str:
"""15 位老号升 18 位:插入 19 + 重算校验位
例: 750101 → 19750101 → 再算校验位
"""
if len(code15) != 15 or not code15.isdigit():
return code15
body18 = "19" + code15[:6] + code15[6:] # 19 + yyMMdd + 顺序码
s = sum(int(body18[i]) * _WEIGHTS[i] for i in range(17))
return body18 + _CHECK_MAP[s % 11]
def validate(value) -> dict:
"""对外主入口:校验一个身份证值,返回结构化结果
Args:
value: 任意值(None / str / 数字),内部统一转字符串
Returns:
{
"ok": bool, # 合规
"violations": [str], # 违反的规则 ID(CHARSET / LENGTH / CHECK_DIGIT / ...)
"length": int, # 输入字符串长度
"is_15_old": bool, # 是否 15 位老号
"normalized": str|None # 15 位老号升 18 位后的值;合规 18 位也是它本身
}
"""
if value is None:
return {"ok": False, "violations": ["EMPTY"], "length": 0, "is_15_old": False, "normalized": None}
s = str(value).strip()
if not s:
return {"ok": False, "violations": ["EMPTY"], "length": 0, "is_15_old": False, "normalized": None}
if _is_all_same(s) and len(s) >= 15:
return {"ok": False, "violations": ["CHARSET"], "length": len(s), "is_15_old": False, "normalized": None}
if len(s) == 15 and s.isdigit():
# 15 位老号:单独校验(没有校验位)
violations: list[str] = []
if region := s[:6]:
if region == "000000" or region[:2] == "00":
violations.append("ADDRESS_PREFIX")
try:
bd = _dt.date(1900 + int(s[6:8]), int(s[8:10]), int(s[10:12]))
if bd > _MAX_BIRTH_DATE or bd.year < _MIN_BIRTH_YEAR:
violations.append("BIRTH_DATE")
except ValueError:
violations.append("BIRTH_DATE")
order = s[12:15]
if not order.isdigit() or not (1 <= int(order) <= 999):
violations.append("ORDER_CODE")
# 15 位长度本身不算违规(合法老号)
return {
"ok": not violations,
"violations": violations,
"length": 15,
"is_15_old": True,
"normalized": _lift_15_to_18(s),
}
if len(s) != 18:
return {"ok": False, "violations": ["LENGTH"], "length": len(s), "is_15_old": False, "normalized": None}
ok, violations, normalized = _validate_18(s)
return {
"ok": ok,
"violations": violations,
"length": 18,
"is_15_old": False,
"normalized": normalized,
}
# ── 规则 ID → 中文描述(前端展示用)───────────────────────
VIOLATION_LABELS = {
"EMPTY": "空值",
"LENGTH": "长度不是 18 位",
"CHARSET": "字符集 / 占位符类不合规",
"CHECK_DIGIT": "校验位错误(GB 11643 §6)",
"ADDRESS_PREFIX": "地址码(前 6 位)不合规",
"BIRTH_DATE": "出生日期(第 7-14 位)不合规",
"ORDER_CODE": "顺序码(第 15-17 位)不合规",
}
\ No newline at end of file
"""身份证治理 —— 扫描路由
端点(v2 多线程版,2026-08-17):
POST /api/id-card/scan
入参:{db_type, host, port, user, password, database,
name_keywords, comment_keywords, max_full_rows_per_column,
max_workers: int = 5}
出参(立即返回,不等结果):{
scan_id: str,
status: "started",
total_columns: int, # 命中的列数(dict + 关键字过滤后)
started_at: ISO8601
}
GET /api/id-card/scan/{scan_id}/progress
出参:{
scan_id, status: "pending" | "running" | "done" | "error",
done, total, current_column, # 当前正在扫描的列(最后一个报上来的)
elapsed_ms, error?
}
GET /api/id-card/scan/{scan_id}/result
出参:{ status, response?: IdCardScanResponse, error? }
status=pending/running → 202;status=done → 200 + 完整结果;status=error → 500 + 错误
实现流程(多线程版):
1. extract_data_dictionary(cfg) —— 复用 web.core.data_dict(同步)
2. 按 name_keywords / comment_keywords 过滤命中列(同步)
3. ThreadPoolExecutor(max_workers) 并发跑每列:
- SELECT DISTINCT <col>
- Python validate
- 把违规值塞进共享 results
- 进度写到 ProgressTracker
4. 等 ③ 完成后,再用 executor 并发跑反查完整行(每列一线程)
5. 聚合成 IdCardScanResponse,写入 ProgressTracker.result
为什么不用 BackgroundTasks:BackgroundTasks 在响应返回后才跑,错误无法返回给前端;
BackgroundTasks 也不暴露进度接口;用 threading.Thread 启动后台任务更可控。
"""
from __future__ import annotations
import threading
import time
import uuid
from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from typing import Any
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
# 复用 web/ 已有模块
from web.core.data_dict import extract_data_dictionary
from web.core.db_adapter import DBConfig, DBConnection, quote_ident
from web2.backend._logging import get_logger
from web2.backend.idcard_validator import validate, VIOLATION_LABELS
logger = get_logger("backend.routes.idcard_scan")
router = APIRouter(tags=["idcard-scan"])
# ── 入参 / 出参模型 ─────────────────────────────────────────
class IdCardScanRequest(BaseModel):
db_type: str
host: str
port: int
user: str
password: str
database: str
charset: str = "utf8mb4"
connect_timeout: int = 10
oracle_client_dir: str | None = None
# 关键字(任一非空即可)
name_keywords: list[str] = Field(default_factory=list)
comment_keywords: list[str] = Field(default_factory=list)
# 命中列拉完整行时的 IN 上限(防爆)
max_full_rows_per_column: int = 5000
# 多线程:并发扫描线程数(5 较稳妥,Oracle 连接数 + 网络都不会爆)
max_workers: int = 5
class MatchedColumn(BaseModel):
table_name: str
column_name: str
column_comment: str | None = None
table_comment: str | None = None
data_type: str | None = None
char_max_length: int | None = None
distinct_count: int
invalid_count: int
violations: dict[str, int] # rule_id → 出现次数
class BadTable(BaseModel):
table_name: str
table_comment: str | None = None
columns: list[dict] # 全表字段元数据(前端表格表头)
bad_rows: list[dict[str, Any]] # 有问题的完整行
rule_summary: list[str] # 涉及的违规规则 ID
bad_row_count: int # 实际查到的有问题的行数
class ScanOverview(BaseModel):
database: str
host: str
db_type: str
total_tables_in_db: int
total_columns_in_db: int
matched_columns: int
columns_with_violations: int
total_bad_rows: int
elapsed_ms: int
class IdCardScanResponse(BaseModel):
overview: ScanOverview
matches: list[MatchedColumn]
violations_by_table: list[BadTable]
class ScanStartedResponse(BaseModel):
scan_id: str
status: str = "started"
total_columns: int
started_at: str
class ScanProgressResponse(BaseModel):
scan_id: str
status: str # "pending" | "running" | "done" | "error"
done: int
total: int
current_column: str | None # 最近报上来的列(多个 worker 抢着写,频繁变)
elapsed_ms: int
error: str | None = None
class ScanResultResponse(BaseModel):
scan_id: str
status: str
response: IdCardScanResponse | None = None
error: str | None = None
# ── 工具:按关键字匹配列 ─────────────────────────────────────
def _match_columns(
data_dict: list[dict],
name_kws: list[str],
comment_kws: list[str],
) -> list[dict]:
"""按 name_keywords(任一命中列名)/ comment_keywords(任一命中列注释)过滤列"""
name_kws_l = [k.lower() for k in name_kws if k]
comment_kws_l = [k.lower() for k in comment_kws if k]
out = []
for col in data_dict:
cname = str(col.get("column_name") or "").lower()
ccmt = str(col.get("column_comment") or "").lower()
if name_kws_l and any(k in cname for k in name_kws_l):
out.append(col); continue
if comment_kws_l and any(k in ccmt for k in comment_kws_l):
out.append(col); continue
return out
# ── 工具:取某列的 DISTINCT 值 ────────────────────────────────
def _fetch_distinct(
db: DBConnection,
table_name: str,
column_name: str,
max_rows: int = 50000,
) -> list[str]:
"""SELECT DISTINCT <col> FROM <table> WHERE <col> IS NOT NULL
方言差异:
- MySQL → 末尾 LIMIT %s
- 达梦 → 末尾 LIMIT ?
- Oracle → 11g 没有 LIMIT,包子查询用 ROWNUM <= N;
占位符要用 :1(numbered bind),不能写 ?(否则 ORA-01036)
max_rows 是 Python int,安全内联到 SQL 字符串里
"""
tq = quote_ident(table_name, db.cfg.db_type)
cq = quote_ident(column_name, db.cfg.db_type)
cap = int(max_rows)
if db.cfg.db_type == "oracle":
sql = (
f"SELECT * FROM ("
f" SELECT DISTINCT {cq} AS v FROM {tq} WHERE {cq} IS NOT NULL"
f") WHERE ROWNUM <= {cap}"
)
rows = db.fetchall(sql)
elif db.cfg.db_type == "mysql":
sql = f"SELECT DISTINCT {cq} AS v FROM {tq} WHERE {cq} IS NOT NULL LIMIT %s"
rows = db.fetchall(sql, (cap,))
else: # dameng
sql = f"SELECT DISTINCT {cq} AS v FROM {tq} WHERE {cq} IS NOT NULL LIMIT ?"
rows = db.fetchall(sql, (cap,))
return [str(r.get("v")) for r in rows if r.get("v") is not None]
# ── 工具:反查包含违规值的所有行 ─────────────────────────────
def _fetch_bad_rows(
db: DBConnection,
table_name: str,
column_name: str,
bad_values: list[str],
max_rows: int,
) -> list[dict[str, Any]]:
"""SELECT * FROM <table> WHERE <col> IN (...) [LIMIT N]
方言差异同 _fetch_distinct:
- MySQL → %s 占位符 + 末尾 LIMIT
- 达梦 → ? 占位符 + 末尾 LIMIT
- Oracle → :1, :2... numbered bind + 子查询 ROWNUM
"""
if not bad_values:
return []
tq = quote_ident(table_name, db.cfg.db_type)
cq = quote_ident(column_name, db.cfg.db_type)
cap = int(max_rows)
n = len(bad_values)
if db.cfg.db_type == "oracle":
# :1, :2, ... :n 全部 numbered bind;ROWNUM 上限走子查询
placeholders = ",".join(f":{i + 1}" for i in range(n))
sql = (
f"SELECT * FROM ("
f" SELECT * FROM {tq} WHERE {cq} IN ({placeholders})"
f") WHERE ROWNUM <= {cap}"
)
rows = db.fetchall(sql, tuple(bad_values))
elif db.cfg.db_type == "mysql":
placeholders = ",".join(["%s"] * n)
sql = f"SELECT * FROM {tq} WHERE {cq} IN ({placeholders}) LIMIT %s"
rows = db.fetchall(sql, (*bad_values, cap))
else: # dameng
placeholders = ",".join(["?"] * n)
sql = f"SELECT * FROM {tq} WHERE {cq} IN ({placeholders}) LIMIT ?"
rows = db.fetchall(sql, (*bad_values, cap))
return rows
# ── 进度跟踪器(线程安全,存每个 scan_id 的状态)──────────────
class ProgressTracker:
"""单实例即可(每个 scan_id 一份 dict 字段),所有 worker 共享同一份内存。"""
def __init__(self, scan_id: str, total: int, started_monotonic: float):
self.scan_id = scan_id
self.total = total
self.started_monotonic = started_monotonic
self._lock = threading.Lock()
self._done = 0
self._current_column: str | None = None
self._status = "running" # pending | running | done | error
self._error: str | None = None
self._result: IdCardScanResponse | None = None
def start_column(self, tname: str, cname: str) -> None:
with self._lock:
self._current_column = f"{tname}.{cname}"
def finish_one(self) -> None:
with self._lock:
self._done += 1
def set_error(self, err: str) -> None:
with self._lock:
self._status = "error"
self._error = err
def set_result(self, resp: IdCardScanResponse) -> None:
with self._lock:
self._status = "done"
self._result = resp
def snapshot(self) -> dict:
with self._lock:
return {
"scan_id": self.scan_id,
"status": self._status,
"done": self._done,
"total": self.total,
"current_column": self._current_column,
"elapsed_ms": int((time.monotonic() - self.started_monotonic) * 1000),
"error": self._error,
"result": self._result,
}
# 全局 in-memory 任务表:scan_id -> ProgressTracker
# (web2 是单进程内网工具;如要多进程/多 worker 需要换成 Redis)
_SCANS: dict[str, ProgressTracker] = {}
_SCANS_LOCK = threading.Lock()
# 任务过期时间:完成后 10 分钟清掉,避免内存泄漏
_SCAN_TTL_SECONDS = 10 * 60
def _evict_expired_scans() -> None:
"""清掉过期的 scan 状态(完成/失败超过 10 分钟)"""
now = time.monotonic()
with _SCANS_LOCK:
to_del = [
sid for sid, tr in _SCANS.items()
if tr.snapshot()["status"] in ("done", "error")
and (now - tr.started_monotonic) > _SCAN_TTL_SECONDS
]
for sid in to_del:
del _SCANS[sid]
logger.info(f" · 清理过期 scan: {sid}")
# ── 单列扫描 worker(在子线程里跑)────────────────────────────
def _scan_one_column(
cfg_dict: dict,
col: dict,
progress: ProgressTracker,
) -> dict:
"""跑一列:开连接 → SELECT DISTINCT → validate → 关闭
Returns:
{table_name, column_name, column_meta, distinct_count, invalid_count, violations, bad_values}
失败时返回 None
"""
tname = col["table_name"]
cname = col["column_name"]
progress.start_column(tname, cname)
cfg = DBConfig(**cfg_dict)
try:
with DBConnection(cfg) as db:
try:
distinct_vals = _fetch_distinct(db, tname, cname, max_rows=50000)
except Exception as e:
logger.warning(f" · {tname}.{cname} 取 DISTINCT 失败:{type(e).__name__}: {e}")
return None
counter: Counter[str] = Counter()
bad_vals: list[str] = []
try:
for v in distinct_vals:
r = validate(v)
if r["violations"]:
bad_vals.append(v)
for rule in r["violations"]:
counter[rule] += 1
except Exception as e:
logger.warning(f" · {tname}.{cname} 校验阶段异常:{type(e).__name__}: {e}")
return None
logger.info(
f" · {tname}.{cname} distinct={len(distinct_vals)} "
f"invalid={len(bad_vals)} rules={dict(counter) or '{}'}"
)
return {
"table_name": tname,
"column_name": cname,
"column_meta": col,
"distinct_count": len(distinct_vals),
"invalid_count": len(bad_vals),
"violations": dict(counter),
"bad_values": bad_vals,
}
except Exception as e:
logger.warning(f" · {tname}.{cname} worker 异常:{type(e).__name__}: {e}")
return None
finally:
progress.finish_one()
# ── 单列反查完整行 worker(在子线程里跑)──────────────────────
def _fetch_bad_rows_one(
cfg_dict: dict,
tname: str,
cname: str,
bad_vals: list[str],
max_rows: int,
progress: ProgressTracker,
) -> tuple[str, str, list[dict[str, Any]]]:
"""反查某列的违规行;返回 (tname, cname, rows)"""
progress.start_column(tname, cname)
cfg = DBConfig(**cfg_dict)
try:
with DBConnection(cfg) as db:
try:
rows = _fetch_bad_rows(db, tname, cname, bad_vals, max_rows)
except Exception as e:
logger.warning(f" · {tname}.{cname} 反查失败:{type(e).__name__}: {e}")
return (tname, cname, [])
return (tname, cname, rows)
except Exception as e:
logger.warning(f" · {tname}.{cname} 反查 worker 异常:{type(e).__name__}: {e}")
return (tname, cname, [])
finally:
progress.finish_one()
# ── 后台扫描主函数(在线程里跑)──────────────────────────────
def _run_scan_in_thread(
scan_id: str,
req: IdCardScanRequest,
progress: ProgressTracker,
) -> None:
"""在子线程里跑完整扫描,写结果回 progress。
这个函数跑崩了不会影响 FastAPI 主进程(只是这个 scan 标记为 error)。
"""
try:
cfg = DBConfig(
db_type=req.db_type,
host=req.host,
port=req.port,
user=req.user,
password=req.password,
database=req.database,
charset=req.charset,
connect_timeout=req.connect_timeout,
oracle_client_dir=req.oracle_client_dir,
)
cfg_dict = {
"db_type": cfg.db_type, "host": cfg.host, "port": cfg.port,
"user": cfg.user, "password": cfg.password, "database": cfg.database,
"charset": cfg.charset, "connect_timeout": cfg.connect_timeout,
"oracle_client_dir": cfg.oracle_client_dir,
}
# ① 抽数据字典
logger.info("[1/4] 抽取数据字典 …")
t0 = time.monotonic()
dd = extract_data_dictionary(cfg)
elapsed_dict_ms = (time.monotonic() - t0) * 1000
data_dict = dd["data_dictionary"]
overview_meta = dd["overview_meta"]
logger.info(
f" 完成:{overview_meta['total_tables']} 表 / "
f"{overview_meta['total_fields']} 字段({elapsed_dict_ms:.0f}ms)"
)
# ② 关键字过滤
logger.info("[2/4] 按关键字过滤列 …")
matched = _match_columns(data_dict, req.name_keywords, req.comment_keywords)
total_matched = len(matched)
logger.info(f" 命中 {total_matched} 个列")
# 更新 total(前面估算可能不准)
progress.total = total_matched
if total_matched == 0:
# 提前结束
response = _build_response(
cfg, overview_meta, [], [], int((time.monotonic() - progress.started_monotonic) * 1000),
)
progress.set_result(response)
return
# ③ 并发跑每列:DISTINCT + validate
logger.info(f"[3/4] 并发扫描 {total_matched} 列(workers={req.max_workers})…")
per_col_results: list[dict] = []
with ThreadPoolExecutor(max_workers=req.max_workers, thread_name_prefix="scan-col") as ex:
futures = [
ex.submit(_scan_one_column, cfg_dict, col, progress)
for col in matched
]
for fut in as_completed(futures):
r = fut.result()
if r is not None:
per_col_results.append(r)
# per_col_results 是无序的;按原表顺序排一下(前端看起来更顺)
order = {(c["table_name"], c["column_name"]): i for i, c in enumerate(matched)}
per_col_results.sort(key=lambda r: order.get((r["table_name"], r["column_name"]), 0))
# ④ 并发反查违规行(每列一线程)
bad_values_by_table_col: dict[tuple[str, str], list[str]] = {
(r["table_name"], r["column_name"]): r["bad_values"]
for r in per_col_results if r["bad_values"]
}
# 重置 progress:④ 阶段重新从 0 计数
with progress._lock:
progress._done = 0
progress.total = len(bad_values_by_table_col) if bad_values_by_table_col else 1
logger.info(
f"[4/4] 并发反查完整行({len(bad_values_by_table_col)} 列,workers={req.max_workers})…"
)
# 按表聚合结果
violations_by_table_dict: dict[str, dict] = {}
with ThreadPoolExecutor(max_workers=req.max_workers, thread_name_prefix="bad-row") as ex:
futures = [
ex.submit(
_fetch_bad_rows_one, cfg_dict, tname, cname, bad_vals,
req.max_full_rows_per_column, progress,
)
for (tname, cname), bad_vals in bad_values_by_table_col.items()
]
# 收集列元数据索引(for 拼 BadTable.columns)
col_to_meta = {(r["table_name"], r["column_name"]): r["column_meta"] for r in per_col_results}
for fut in as_completed(futures):
tname, cname, rows = fut.result()
rec = violations_by_table_dict.setdefault(tname, {
"table_name": tname,
"table_comment": None,
"columns": [],
"bad_rows_set": set(),
"rule_set": set(),
})
if not rec["columns"]:
tcols = [c for c in data_dict if c.get("table_name") == tname]
rec["columns"] = tcols
rec["table_comment"] = tcols[0].get("table_comment") if tcols else None
# 把这一列的违规规则加入
if (tname, cname) in col_to_meta:
rec["rule_set"].update(col_to_meta[(tname, cname)].get("_violations", []) or [])
# 从 per_col_results 拿到该列的 violations
for r in per_col_results:
if r["table_name"] == tname and r["column_name"] == cname:
rec["rule_set"].update(r["violations"].keys())
break
for row in rows:
key = tuple(sorted((k, str(v)) for k, v in row.items()))
rec["bad_rows_set"].add(key)
logger.info(f" · {tname}.{cname} → {len(rows)} 行(去重后累计 {len(rec['bad_rows_set'])})")
# 拼最终响应
violations_by_table: list[BadTable] = []
total_bad_rows = 0
for tname, rec in violations_by_table_dict.items():
bad_rows = [dict(kv) for kv in rec["bad_rows_set"]]
fixed_rows = []
for row in bad_rows:
r2 = {}
for k, v in row.items():
r2[k] = v
fixed_rows.append(r2)
violations_by_table.append(BadTable(
table_name=tname,
table_comment=rec["table_comment"],
columns=rec["columns"],
bad_rows=fixed_rows,
rule_summary=sorted(rec["rule_set"]),
bad_row_count=len(fixed_rows),
))
total_bad_rows += len(fixed_rows)
matches_out = [
MatchedColumn(
table_name=r["table_name"],
column_name=r["column_name"],
column_comment=r["column_meta"].get("column_comment"),
table_comment=r["column_meta"].get("table_comment"),
data_type=r["column_meta"].get("data_type"),
char_max_length=r["column_meta"].get("char_max_length"),
distinct_count=r["distinct_count"],
invalid_count=r["invalid_count"],
violations=r["violations"],
)
for r in per_col_results
]
elapsed_ms = int((time.monotonic() - progress.started_monotonic) * 1000)
logger.info(
f"完成:匹配 {len(matches_out)} 列,"
f"违规 {sum(1 for m in matches_out if m.invalid_count > 0)} 列,"
f"问题表 {len(violations_by_table)} 张,问题行 {total_bad_rows} 条,"
f"总耗时 {elapsed_ms}ms"
)
logger.info("─" * 60)
response = IdCardScanResponse(
overview=ScanOverview(
database=cfg.database,
host=cfg.host,
db_type=cfg.db_type,
total_tables_in_db=overview_meta["total_tables"],
total_columns_in_db=overview_meta["total_fields"],
matched_columns=len(matches_out),
columns_with_violations=sum(1 for m in matches_out if m.invalid_count > 0),
total_bad_rows=total_bad_rows,
elapsed_ms=elapsed_ms,
),
matches=matches_out,
violations_by_table=violations_by_table,
)
progress.set_result(response)
except Exception as e:
logger.exception(f"扫描后台线程异常: {e}")
progress.set_error(f"{type(e).__name__}: {e}")
def _build_response(
cfg: DBConfig, overview_meta: dict, matches: list, violations: list, elapsed_ms: int,
) -> IdCardScanResponse:
"""空结果时用的便捷构造"""
return IdCardScanResponse(
overview=ScanOverview(
database=cfg.database,
host=cfg.host,
db_type=cfg.db_type,
total_tables_in_db=overview_meta["total_tables"],
total_columns_in_db=overview_meta["total_fields"],
matched_columns=len(matches),
columns_with_violations=0,
total_bad_rows=0,
elapsed_ms=elapsed_ms,
),
matches=matches,
violations_by_table=violations,
)
# ── 端点 ────────────────────────────────────────────────────
@router.post("/id-card/scan", response_model=ScanStartedResponse, summary="启动身份证合规扫描(异步)")
async def idcard_scan_start(req: IdCardScanRequest) -> ScanStartedResponse:
"""立即返回 scan_id,扫描在后台跑;前端用 /progress 和 /result 端点拉。
关键设计:
- 关键字缺失 → 422
- 后台用 ThreadPoolExecutor 并发跑(max_workers 默认 5)
- 进度写到内存 dict(web2 是单进程内网工具;如要多 worker 需换 Redis)
- 完成/失败后保留 10 分钟供前端拉结果,之后清掉
"""
if not req.name_keywords and not req.comment_keywords:
raise HTTPException(
status_code=422, detail="name_keywords 和 comment_keywords 至少填一个"
)
# 先轻量算一下大概有多少命中列(不开新连接,只在内存里过滤本地缓存不可行;
# 直接给个粗略估算 = 0,前端会从 /progress 里看到真实 total)
scan_id = uuid.uuid4().hex
started_mono = time.monotonic()
progress = ProgressTracker(scan_id=scan_id, total=0, started_monotonic=started_mono)
with _SCANS_LOCK:
_SCANS[scan_id] = progress
# 起后台线程(daemon=True 防止主进程退出时线程卡住)
t = threading.Thread(
target=_run_scan_in_thread,
args=(scan_id, req, progress),
daemon=True,
name=f"scan-{scan_id[:8]}",
)
t.start()
# 顺便清理过期任务
_evict_expired_scans()
logger.info(
f"启动 scan {scan_id[:8]} db={req.db_type}://{req.user}@{req.host}:{req.port}/"
f"{req.database} workers={req.max_workers}"
)
return ScanStartedResponse(
scan_id=scan_id,
total_columns=0, # 前端首次拿到的 total 是 0;下一步会从 /progress 拿到真实值
started_at=datetime.now(timezone.utc).isoformat(),
)
@router.get("/id-card/scan/{scan_id}/progress", response_model=ScanProgressResponse)
async def idcard_scan_progress(scan_id: str) -> ScanProgressResponse:
"""拿扫描进度(前端每 500-1000ms 轮询一次)"""
with _SCANS_LOCK:
progress = _SCANS.get(scan_id)
if progress is None:
raise HTTPException(status_code=404, detail=f"scan_id 不存在或已过期:{scan_id}")
snap = progress.snapshot()
return ScanProgressResponse(
scan_id=snap["scan_id"],
status=snap["status"],
done=snap["done"],
total=snap["total"],
current_column=snap["current_column"],
elapsed_ms=snap["elapsed_ms"],
error=snap["error"],
)
@router.get("/id-card/scan/{scan_id}/result", response_model=ScanResultResponse)
async def idcard_scan_result(scan_id: str) -> ScanResultResponse:
"""拿扫描最终结果(status=pending/running → 202;done → 200;error → 500)"""
with _SCANS_LOCK:
progress = _SCANS.get(scan_id)
if progress is None:
raise HTTPException(status_code=404, detail=f"scan_id 不存在或已过期:{scan_id}")
snap = progress.snapshot()
if snap["status"] in ("pending", "running"):
raise HTTPException(
status_code=202,
detail=f"扫描中:{snap['done']}/{snap['total']}",
)
if snap["status"] == "error":
raise HTTPException(
status_code=500,
detail=f"扫描失败:{snap['error']}",
)
return ScanResultResponse(
scan_id=scan_id,
status="done",
response=snap["result"],
)
/**
* 身份证字段匹配的默认关键字
*
* 来源:
* web/configs/standards_match.yaml → IND-001-a (GB 11643-1999 身份证号格式)
*
* 原 YAML 片段:
* IND-001-a:
* applies_to_fields: [id_card, id_card_no, id_number, identity_card]
* comment_keywords:
* - 身份证
*
* 设计:
* - 字段名用英文/下划线(数据库列名习惯)
* - 注释用中文/可读词(DDL COMMENT 字段)
* - 数组形式而非字符串,组件内 join(', ') 显示在输入框
* - 改这里 = 改 web/configs/standards_match.yaml 的 IND-001-a(保持两边一致)
*
* 后续:
* - 后端可以加 GET /api/id-card-keywords 直接读 YAML 返给前端,
* 这里作为前端离线兜底
*/
export const DEFAULT_NAME_KEYWORDS = [
'id_card',
'id_card_no',
'id_number',
'identity_card',
]
export const DEFAULT_COMMENT_KEYWORDS = [
'身份证号',
]
\ 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