| """Deterministically normalize Korean 13-digit ID labels after NER inference.""" |
| from __future__ import annotations |
|
|
| import re |
| from typing import Any |
|
|
|
|
| KOREAN_ID_PATTERN = re.compile(r"(?<!\d)\d{6}[- ]?\d{7}(?!\d)") |
| KOREAN_ID_LABELS = {"RRN", "ALIEN_NUMBER"} |
|
|
|
|
| def label_from_seventh_digit(value: str) -> str | None: |
| """Return RRN for codes 1-4 and ALIEN_NUMBER for codes 5-8.""" |
| digits = "".join(character for character in value if character.isdigit()) |
| if len(digits) != 13: |
| return None |
| if digits[6] in "1234": |
| return "RRN" |
| if digits[6] in "5678": |
| return "ALIEN_NUMBER" |
| return None |
|
|
|
|
| def _entity_label(entity: dict[str, Any]) -> str | None: |
| return entity.get("entity_group") or entity.get("label") or entity.get("entity") |
|
|
|
|
| def normalize_korean_id_entities( |
| text: str, |
| entities: list[dict[str, Any]], |
| *, |
| add_missing_structured_candidates: bool = False, |
| ) -> list[dict[str, Any]]: |
| """Merge/override overlapping Korean-ID entities using the seventh digit. |
| |
| `entities` can be the output of a Transformers aggregation pipeline. The |
| function does not alter unrelated entities. When fragmented RRN/alien-ID |
| predictions overlap one 13-digit value, they are replaced by one full-span |
| entity and the conservative minimum fragment score is retained. |
| |
| If `add_missing_structured_candidates=True`, a regex-only match with no NER |
| overlap is also returned with score 0.0 and `candidate_only=True`; downstream |
| checksum validation should decide whether to accept it. |
| """ |
| remaining = [dict(entity) for entity in entities] |
| normalized: list[dict[str, Any]] = [] |
| for match in KOREAN_ID_PATTERN.finditer(text): |
| label = label_from_seventh_digit(match.group(0)) |
| if label is None: |
| continue |
| overlaps = [ |
| entity |
| for entity in remaining |
| if _entity_label(entity) in KOREAN_ID_LABELS |
| and int(entity.get("start", -1)) < match.end() |
| and int(entity.get("end", -1)) > match.start() |
| ] |
| if not overlaps and not add_missing_structured_candidates: |
| continue |
| for entity in overlaps: |
| remaining.remove(entity) |
| scores = [float(entity["score"]) for entity in overlaps if entity.get("score") is not None] |
| normalized.append( |
| { |
| "entity_group": label, |
| "score": min(scores) if scores else 0.0, |
| "word": match.group(0), |
| "start": match.start(), |
| "end": match.end(), |
| "normalization_rule": "korean_id_seventh_digit", |
| "candidate_only": not overlaps, |
| } |
| ) |
| return sorted(remaining + normalized, key=lambda entity: (int(entity.get("start", -1)), int(entity.get("end", -1)))) |
|
|
|
|
