Docker/idcard-api/app/parser.py

"""身份证字段解析器。

输入 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