`services/ingestion/naupa2.py`

109 lines
"""NAUPA II fixed-width parser adapter.

NAUPA II is the legacy fixed-width reporting format (NAUPA III moves to schema-validated
XML — a future adapter, wired the same way). A real state's layout carries dozens of
fields and code tables; this is a documented, representative PROPERTY-DETAIL subset
sufficient to prove the adapter-registry pattern end-to-end on synthetic data.

Registered lazily by services.ingestion.ingest._get_parser under format_name 'naupa2_v1'.
Emits the same CanonicalProperty as the CSV adapter, so ingestion/dedup/masking/search are
identical regardless of source format.
"""
from __future__ import annotations

import hashlib
from decimal import Decimal
from typing import Iterable

from services.common.normalize import (
    amount_band, mask_name, normalize_business, normalize_postal, normalize_text,
)

# (field_name, start, length) — 0-indexed. Fixed record width = 213.
LAYOUT: list[tuple[str, int, int]] = [
    ("record_type", 0, 2),        # 'PR' = property detail
    ("source_property_id", 2, 18),
    ("holder_name", 20, 40),
    ("owner_last", 60, 30),
    ("owner_first", 90, 20),
    ("address", 110, 40),
    ("city", 150, 28),
    ("state", 178, 2),
    ("zip", 180, 9),
    ("property_type", 189, 12),
    ("amount_cents", 201, 12),     # right-justified, zero-padded CENTS (implied 2 decimals)
]
RECORD_WIDTH = 213
_PROPERTY_RECORD_TYPE = "PR"


def _field(line: str, start: int, length: int) -> str:
    return line[start:start + length].strip()


def _looks_like_business(name: str) -> bool:
    from services.common.normalize import CORPORATE_SUFFIXES
    return any(tok in CORPORATE_SUFFIXES for tok in normalize_text(name).split())


def build_naupa2_line(fields: dict) -> str:
    """Assemble one fixed-width line from a field dict (used by tests/tools).

    Guarantees column alignment: numeric amount_cents is right-justified zero-padded,
    everything else is left-justified space-padded, each truncated to its width.
    """
    line = [" "] * RECORD_WIDTH
    for name, start, length in LAYOUT:
        val = str(fields.get(name, ""))
        if name == "amount_cents":
            val = val.rjust(length, "0")[:length]
        else:
            val = val.ljust(length)[:length]
        line[start:start + length] = list(val)
    return "".join(line)


def _amount_from_cents(cents_field: str) -> Decimal | None:
    digits = cents_field.strip()
    if not digits or not digits.isdigit():
        return None
    return (Decimal(digits) / Decimal(100)).quantize(Decimal("0.01"))


def parse_naupa2_feed(data: bytes, jurisdiction: str) -> Iterable["object"]:
    from services.ingestion.ingest import CanonicalProperty  # lazy: avoid import cycle

    text = data.decode("utf-8", errors="replace")
    for raw_line in text.splitlines():
        if not raw_line.strip():
            continue
        if _field(raw_line, 0, 2) != _PROPERTY_RECORD_TYPE:
            # header/trailer/holder records are skipped by this property-detail adapter
            continue

        rec = {name: _field(raw_line, start, length) for name, start, length in LAYOUT}
        owner_name = f"{rec['owner_first']} {rec['owner_last']}".strip()
        is_business = _looks_like_business(owner_name)
        amount = _amount_from_cents(rec["amount_cents"])
        raw_hash = hashlib.sha256(raw_line.encode("utf-8")).hexdigest()
        owner_norm = (
            normalize_business(owner_name) if is_business else normalize_text(owner_name)
        )
        yield CanonicalProperty(
            jurisdiction=jurisdiction,
            source_property_id=rec["source_property_id"],
            holder_name_raw=rec["holder_name"],
            owner_type="business" if is_business else "person",
            owner_name_raw=owner_name,
            owner_name_normalized=owner_norm,
            owner_name_masked=mask_name(owner_name),
            amount=amount,
            amount_band=amount_band(amount),
            raw_payload=raw_line,
            raw_record_hash=raw_hash,
            city_normalized=normalize_text(rec["city"]) or None,
            region=normalize_text(rec["state"]) or None,
            postal_code=normalize_postal(rec["zip"]),
            property_type=normalize_text(rec["property_type"]) or None,
        )