ow
This commit is contained in:
owenchen committed
1 parent
419c22cb74
commit
59ec58a1eb
8 files changed
+1163
No files matched your search
@@ -0,0 +1,185 @@
|
||||
"""身份证字段解析器。
|
||||
|
||||
输入 RapidOCR 的识别结果（文本框列表），输出结构化身份证信息。
|
||||
设计原则：不依赖固定像素坐标，只用「关键词锚点 + 身份证号校验位」，
|
||||
对扫描件、手机拍照、轻微倾斜都能工作。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any, Dict, List, Optional, Sequence
|
||||
|
||||
# ISO 7064 MOD 11-2 校验位权重
|
||||
_WEIGHTS = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
|
||||
_CHECK = "10X98765432"
|
||||
|
||||
_ID_RE = re.compile(r"(?<![0-9])([0-9]{17}[0-9Xx])(?![0-9])")
|
||||
_BIRTH_RE = re.compile(r"(\d{4})\s*年\s*(\d{1,2})\s*月\s*(\d{1,2})\s*日")
|
||||
_VALID_RE = re.compile(
|
||||
r"(\d{4})[.\-/](\d{1,2})[.\-/](\d{1,2})\s*[-—~至]\s*(\d{4})[.\-/](\d{1,2})[.\-/](\d{1,2})"
|
||||
)
|
||||
|
||||
# 关键词锚点（允许 OCR 把字读漏或在中间插入空格）
|
||||
_ANCHORS = {
|
||||
"name": r"姓\s*名",
|
||||
"gender": r"性\s*别",
|
||||
"nation": r"民\s*族",
|
||||
"birth": r"出\s*生",
|
||||
"address": r"住\s*址",
|
||||
"id_number": r"公\s*民\s*身\s*份\s*号\s*码",
|
||||
"issuer": r"签\s*发\s*机\s*关",
|
||||
"validity": r"有\s*效\s*期\s*限",
|
||||
}
|
||||
|
||||
# 正面/反面判定词
|
||||
_FRONT_MARKERS = ("公民身份号码", "住址", "民族")
|
||||
_BACK_MARKERS = ("签发机关", "有效期限")
|
||||
|
||||
|
||||
def verify_id_number(number: str) -> bool:
|
||||
"""校验 18 位身份证号（含最后一位校验位）。"""
|
||||
number = (number or "").strip().upper()
|
||||
if not re.fullmatch(r"\d{17}[\dX]", number):
|
||||
return False
|
||||
total = sum(int(number[i]) * _WEIGHTS[i] for i in range(17))
|
||||
return _CHECK[total % 11] == number[17]
|
||||
|
||||
|
||||
def _lines_from_result(result: Optional[Sequence]) -> List[Dict[str, Any]]:
|
||||
"""把 RapidOCR 结果整理成按阅读顺序排列的行。
|
||||
|
||||
result: [[box, text, score], ...] 或 None
|
||||
返回: [{"text": str, "score": float, "box": [...]}, ...]，按先上下后左右排序。
|
||||
"""
|
||||
if not result:
|
||||
return []
|
||||
items = []
|
||||
for entry in result:
|
||||
try:
|
||||
box, text, score = entry
|
||||
except (ValueError, TypeError):
|
||||
continue
|
||||
text = str(text).strip()
|
||||
if not text:
|
||||
continue
|
||||
try:
|
||||
cy = sum(float(p[1]) for p in box) / len(box)
|
||||
cx = sum(float(p[0]) for p in box) / len(box)
|
||||
except (TypeError, ZeroDivisionError):
|
||||
cy, cx = 0.0, 0.0
|
||||
items.append({"text": text, "score": float(score), "cy": cy, "cx": cx})
|
||||
|
||||
items.sort(key=lambda it: (round(it["cy"] / 12.0), it["cx"]))
|
||||
return items
|
||||
|
||||
|
||||
def _cut(text: str, start_pat: str, end_pats: Sequence[str]) -> Optional[str]:
|
||||
"""从 start 关键词之后切到最近的 end 关键词之前。"""
|
||||
m = re.search(start_pat, text)
|
||||
if not m:
|
||||
return None
|
||||
rest = text[m.end():]
|
||||
end = len(rest)
|
||||
for pat in end_pats:
|
||||
m2 = re.search(pat, rest)
|
||||
if m2:
|
||||
end = min(end, m2.start())
|
||||
value = rest[:end].strip()
|
||||
return value.strip(" ：:·,，.。;；\t ") or None
|
||||
|
||||
|
||||
def _clean_name(value: Optional[str]) -> Optional[str]:
|
||||
if not value:
|
||||
return None
|
||||
value = re.sub(r"[^\u4e00-\u9fa5·]", "", value)
|
||||
return value[:30] or None
|
||||
|
||||
|
||||
def _clean_short(value: Optional[str], keep: int) -> Optional[str]:
|
||||
if not value:
|
||||
return None
|
||||
m = re.search(r"[\u4e00-\u9fa5]{1,%d}" % keep, value)
|
||||
return m.group(0) if m else None
|
||||
|
||||
|
||||
def _clean_birth(value: Optional[str]) -> Optional[str]:
|
||||
if not value:
|
||||
return None
|
||||
m = _BIRTH_RE.search(value)
|
||||
if not m:
|
||||
return None
|
||||
y, mo, d = m.groups()
|
||||
return "%s%02d%02d" % (y, int(mo), int(d))
|
||||
|
||||
|
||||
def _clean_address(value: Optional[str]) -> Optional[str]:
|
||||
if not value:
|
||||
return None
|
||||
value = re.sub(r"\s+", "", value)
|
||||
return value[:120] or None
|
||||
|
||||
|
||||
def _find_id_number(text: str) -> Optional[str]:
|
||||
"""全文找身份证号；若多个候选，取通过校验位的那个。"""
|
||||
candidates = _ID_RE.findall(text)
|
||||
for c in candidates:
|
||||
if verify_id_number(c):
|
||||
return c.upper()
|
||||
return candidates[0].upper() if candidates else None
|
||||
|
||||
|
||||
def parse(lines: Sequence[Dict[str, Any]]) -> Dict[str, Any]:
|
||||
"""主入口：OCR 行 -> 结构化身份证信息。"""
|
||||
full = "".join(item["text"] for item in lines)
|
||||
|
||||
number = _find_id_number(full)
|
||||
gender_raw = _cut(full, _ANCHORS["gender"], (_ANCHORS["nation"], _ANCHORS["birth"], _ANCHORS["address"]))
|
||||
nation = _clean_short(_cut(full, _ANCHORS["nation"], (_ANCHORS["birth"], _ANCHORS["address"])), 8)
|
||||
issuer = _cut(full, _ANCHORS["issuer"], (_ANCHORS["validity"],))
|
||||
validity_raw = _cut(full, _ANCHORS["validity"], ()) or ""
|
||||
vm = _VALID_RE.search(validity_raw)
|
||||
validity = "%s.%s.%s-%s.%s.%s" % (vm.groups()) if vm else (validity_raw.strip()[:40] or None)
|
||||
|
||||
is_front = any(m in full for m in _FRONT_MARKERS)
|
||||
is_back = any(m in full for m in _BACK_MARKERS)
|
||||
side = "front" if is_front and not is_back else ("back" if is_back else "unknown")
|
||||
|
||||
data = {
|
||||
"side": side,
|
||||
"name": _clean_name(_cut(full, _ANCHORS["name"], (_ANCHORS["gender"], _ANCHORS["nation"], _ANCHORS["birth"], _ANCHORS["address"]))),
|
||||
"gender": _clean_short(gender_raw, 2),
|
||||
"nation": nation,
|
||||
"birth": _clean_birth(_cut(full, _ANCHORS["birth"], (_ANCHORS["address"], _ANCHORS["id_number"]))),
|
||||
"address": _clean_address(_cut(full, _ANCHORS["address"], (_ANCHORS["id_number"],))),
|
||||
"id_number": number,
|
||||
"id_number_valid": verify_id_number(number) if number else None,
|
||||
"issuer": issuer,
|
||||
"validity": validity,
|
||||
}
|
||||
|
||||
# 只统计 8 个业务字段，side/valid/confidence/raw_text 不计入
|
||||
business_fields = (
|
||||
"name", "gender", "nation", "birth", "address",
|
||||
"id_number", "issuer", "validity",
|
||||
)
|
||||
filled = sum(1 for k in business_fields if data.get(k) not in (None, ""))
|
||||
data["field_count"] = filled
|
||||
data["raw_text"] = " | ".join(item["text"] for item in lines)
|
||||
scores = [item["score"] for item in lines if item.get("score") is not None]
|
||||
data["confidence"] = round(sum(scores) / len(scores), 4) if scores else None
|
||||
return data
|
||||
|
||||
|
||||
def mask(data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""敏感字段脱敏，用于日志或对外返回。"""
|
||||
out = dict(data)
|
||||
if out.get("id_number"):
|
||||
n = out["id_number"]
|
||||
out["id_number"] = n[:6] + "*" * 8 + n[-4:]
|
||||
if out.get("address"):
|
||||
out["address"] = out["address"][:6] + "****"
|
||||
if out.get("name"):
|
||||
n = out["name"]
|
||||
out["name"] = n[0] + "*" * (len(n) - 1)
|
||||
out.pop("raw_text", None)
|
||||
return out
|
||||
Reference in new issue
Block a user