name_parser.py

#!/usr/bin/env python3
from __future__ import annotations

import re
import unicodedata
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Literal, Optional, Tuple

import joblib

RecordType = Literal["PERSON", "JOINT_PERSON", "ORG", "TRUST", "UNKNOWN"]
PlanLabel = Literal[
    "PERSON_COMMA",
    "PERSON_SPACE",
    "JOINT",
    "ORG",
    "TRUST",
    "UNKNOWN",
]

SUFFIXES = {"JR", "SR", "II", "III", "IV", "V", "MD", "PHD", "ESQ", "CPA"}
PARTICLES = {
    "DE",
    "DEL",
    "DELA",
    "DELLA",
    "DI",
    "LA",
    "LE",
    "VAN",
    "VON",
    "ST",
    "SAINT",
}
JOINT_SPLIT_REGEX = re.compile(r"\s(?:&|AND)\s")


@dataclass
class ParsedPerson:
    full_name: str
    first_name: Optional[str]
    middle_name: Optional[str]
    last_name: Optional[str]
    surname: Optional[str]
    suffix: Optional[str]

    def as_dict(self) -> Dict[str, Optional[str]]:
        return {
            "full_name": self.full_name,
            "first_name": self.first_name,
            "middle_name": self.middle_name,
            "last_name": self.last_name,
            "surname": self.surname,
            "suffix": self.suffix,
        }


def normalize_name(input_name: str) -> str:
    return (
        unicodedata.normalize("NFKC", input_name)
        .replace("“", '"')
        .replace("”", '"')
        .replace("‘", "'")
        .replace("’", "'")
        .strip()
        .upper()
    )


def clean_token(token: str) -> str:
    return re.sub(r"^,+|,+$", "", token.strip())


def split_tokens(text: str) -> List[str]:
    raw = (text or "").upper()
    return [
        tok for tok in (clean_token(t) for t in re.split(r"[^A-Z0-9\\.]+", raw)) if tok
    ]


def suffix_of(token: str) -> Optional[str]:
    norm = token.replace(".", "").replace(",", "").upper().strip()
    if norm in SUFFIXES:
        return token.strip()
    return None


def parse_person_comma(name: str) -> Optional[ParsedPerson]:
    if "," not in name:
        return None
    left, right = name.split(",", 1)
    left_tokens = split_tokens(left)
    right_tokens = split_tokens(right)
    if not left_tokens or not right_tokens:
        return None

    suffix = None
    left_last = left_tokens[-1]
    left_suffix = suffix_of(left_last)
    if left_suffix:
        suffix = left_suffix
        left_tokens = left_tokens[:-1]
    if not left_tokens:
        return None

    right_last = right_tokens[-1]
    right_suffix = suffix_of(right_last)
    if right_suffix:
        suffix = right_suffix
        right_tokens = right_tokens[:-1]
    if not right_tokens:
        return None

    first_name = right_tokens[0]
    middle_name = " ".join(right_tokens[1:]) if len(right_tokens) > 1 else None
    last_name = " ".join(left_tokens)
    full_parts = [*right_tokens, *left_tokens]
    if suffix:
        full_parts.append(suffix)

    return ParsedPerson(
        full_name=" ".join(full_parts),
        first_name=first_name,
        middle_name=middle_name,
        last_name=last_name,
        surname=last_name,
        suffix=suffix,
    )


def parse_person_space(name: str) -> Optional[ParsedPerson]:
    tokens = split_tokens(name)
    if not tokens:
        return None
    suffix = None
    maybe_suffix = suffix_of(tokens[-1])
    if maybe_suffix:
        suffix = maybe_suffix
        tokens = tokens[:-1]
    if not tokens:
        return None

    if len(tokens) == 1:
        token = tokens[0]
        full_parts = [token]
        if suffix:
            full_parts.append(suffix)
        return ParsedPerson(
            full_name=" ".join(full_parts),
            first_name=None,
            middle_name=None,
            last_name=token,
            surname=token,
            suffix=suffix,
        )

    span = 1
    if len(tokens) >= 3 and tokens[1].upper() in PARTICLES:
        span = 2
        if len(tokens) >= 4 and tokens[2].upper() in PARTICLES:
            span = 3

    last_tokens = tokens[:span]
    given_tokens = tokens[span:]
    if not given_tokens and len(tokens) >= 2:
        last_tokens = tokens[:1]
        given_tokens = tokens[1:]
    if not given_tokens:
        return None

    first_name = given_tokens[0]
    middle_name = " ".join(given_tokens[1:]) if len(given_tokens) > 1 else None
    last_name = " ".join(last_tokens)
    full_parts = [*given_tokens, *last_tokens]
    if suffix:
        full_parts.append(suffix)

    return ParsedPerson(
        full_name=" ".join(full_parts),
        first_name=first_name,
        middle_name=middle_name,
        last_name=last_name,
        surname=last_name,
        suffix=suffix,
    )


def parse_joint(name: str) -> List[ParsedPerson]:
    parts = [p.strip() for p in JOINT_SPLIT_REGEX.split(name) if p.strip()]
    if len(parts) <= 1:
        single = parse_person_comma(name) if "," in name else parse_person_space(name)
        return [single] if single else []

    parsed = [
        parse_person_comma(part) if "," in part else parse_person_space(part)
        for part in parts
    ]

    if (
        len(parts) == 2
        and parsed[0] is None
        and parsed[1] is not None
        and len(split_tokens(parts[0])) == 1
        and parsed[1].last_name
    ):
        patched_name = f"{parts[0]} {parsed[1].last_name}"
        patched = (
            parse_person_comma(patched_name)
            if "," in patched_name
            else parse_person_space(patched_name)
        )
        if patched:
            parsed[0] = patched

    return [person for person in parsed if person is not None]


def canonical_from_people(people: List[ParsedPerson]) -> Optional[str]:
    if not people:
        return None
    return " | ".join(person.full_name for person in people)


def plan_to_record_type(plan: PlanLabel) -> RecordType:
    if plan == "PERSON_COMMA" or plan == "PERSON_SPACE":
        return "PERSON"
    if plan == "JOINT":
        return "JOINT_PERSON"
    return plan  # type: ignore[return-value]


def predict_parse(
    plan: PlanLabel, normalized_name: str
) -> Tuple[RecordType, Optional[str], List[ParsedPerson]]:
    if plan == "PERSON_COMMA":
        person = parse_person_comma(normalized_name)
        people = [person] if person else []
        return "PERSON", canonical_from_people(people), people
    if plan == "PERSON_SPACE":
        person = parse_person_space(normalized_name)
        people = [person] if person else []
        return "PERSON", canonical_from_people(people), people
    if plan == "JOINT":
        people = parse_joint(normalized_name)
        return "JOINT_PERSON", canonical_from_people(people), people
    return plan_to_record_type(plan), None, []


def load_model(path: str | Path = "model.joblib") -> Any:
    return joblib.load(path)


def parse_name(model: Any, raw_name: str) -> Dict[str, Any]:
    normalized_name = normalize_name(raw_name)
    pred_plan = model.predict([normalized_name])[0]
    pred_rt, pred_canonical, pred_people = predict_parse(pred_plan, normalized_name)

    return {
        "raw_name": raw_name,
        "normalized_name": normalized_name,
        "pred_plan": pred_plan,
        "record_type": pred_rt,
        "canonical_full_name": pred_canonical,
        "parsed_persons": [p.as_dict() for p in pred_people],
    }