gendesign/tradein-mvp/backend/app/services/matching/normalize.py

"""Address normalization and fingerprint utilities.

Used by 3-tier matching to produce stable, source-independent keys.
Algorithm reference: decisions/Cross_Source_Matching_Strategy.md sec 3.2
"""

import hashlib
import re
import unicodedata

from app.services.regions import REGIONS as _REGIONS

# Strip punctuation EXCEPT hyphens — keep hyphens so abbreviation rules like
# 'пр-кт', 'б-р', 'пр-д' can match before they are collapsed to spaces.
_PUNCT = re.compile(r"[^\w\s\-]", flags=re.UNICODE)
_WS = re.compile(r"\s+")

# House/building markers ('дом'/'д'/'корпус'/'корп'/'к') are dropped to a single
# canonical form: only the number is kept. This makes the output source-independent —
# SERP writes a bare number ('Ленина 5') while detail/БТИ write a marker ('д. 5' -> 'д 5')
# and both must yield the same normalized_address / fingerprint (bug #1534).
# The marker is removed ONLY when immediately followed by a number, so street-name
# initials like 'К. Либкнехта' / 'Д. Зверева' (marker followed by a word) are left
# untouched and not mis-expanded into 'корпус'/'дом' (bug #1535).
_HOUSE_MARKER = re.compile(r"\b(?:дом|корпус|корп|д|к)\s+(?=\d)", flags=re.UNICODE)

# Abbreviation expansion — applied after lowercasing, bounded by spaces.
# Order matters: longer/more-specific patterns first to avoid partial overlap.
# Hyphenated forms ('пр-кт', 'б-р', 'пр-д') come first so they match before
# the shorter variants ('пр', 'б') would erroneously consume the prefix.
_ABBREV = [
    (" пр-кт ", " проспект "),
    (" пр-кт. ", " проспект "),
    (" б-р ", " бульвар "),
    (" пр-д ", " проезд "),
    (" ш ", " шоссе "),
    (" ул ", " улица "),
    (" ул. ", " улица "),
    (" пр ", " проспект "),
    (" пр. ", " проспект "),
    (" пер ", " переулок "),
    (" пл ", " площадь "),
    (" наб ", " набережная "),
    (" бул ", " бульвар "),
    (" туп ", " тупик "),
    (" стр ", " строение "),
    (" корп ", " корпус "),
    # NOTE: house/building markers ('д'/'дом'/'к'/'корп'/'корпус') are NOT expanded here.
    # They are handled separately by _HOUSE_MARKER below — stripped to a single canonical
    # form (number only) and only when followed by a number — to avoid asymmetric keys
    # (#1534) and mis-expanding street-name initials like 'К. Либкнехта' (#1535).
]


def normalize_address(text: str | None) -> str:
    """Normalize address string for cross-source comparison.

    Steps:
      1. Unicode NFC normalization
      2. Lowercase
      3. Strip punctuation except hyphens (preserve 'пр-кт', 'б-р', 'пр-д' for expansion)
      4. Collapse whitespace
      5. Expand common Russian address abbreviations (hyphenated forms first)
      6. Collapse remaining hyphens to spaces
      7. Drop house/building markers ('дом'/'д'/'корпус'/'корп'/'к') that precede a
         number, keeping only the number — a single canonical form across sources
    """
    if not text:
        return ""
    s = unicodedata.normalize("NFC", text).lower()
    s = _PUNCT.sub(" ", s)  # strip punctuation EXCEPT hyphens
    s = _WS.sub(" ", s).strip()
    # Wrap with spaces for clean boundary matching
    s = f" {s} "
    for short, full in _ABBREV:
        s = s.replace(short, full)
    # Collapse remaining hyphens (e.g. standalone '-' separators or numeric ranges)
    s = s.replace("-", " ")
    s = _WS.sub(" ", s).strip()
    # Drop house/building markers that precede a number ('д 5'/'дом 5'/'корпус 5' -> '5'),
    # leaving street-name initials ('к либкнехта') untouched. Re-collapse whitespace after.
    s = _HOUSE_MARKER.sub("", s)
    return _WS.sub(" ", s).strip()


def has_house_number(normalized: str | None) -> bool:
    """True if the normalized address ends with a house-number token.

    The trailing token must start with a digit:
      'улица мамина сибиряка 126а' -> True
      'улица 8 марта'              -> False  (street-name digit, no house no.)
      'улица 8 марта 5'            -> True   ('5' is the house number)
      'улица мамина сибиряка'      -> False  (bare street, no number)
      ''/None                      -> False

    Detection uses the TRAILING token only — a digit inside the street name
    ('8 марта', '50 лет октября') is NOT a house number.
    """
    if not normalized:
        return False
    return re.search(r"(?:^|\s)\d\S*$", normalized) is not None


def house_number_token(normalized: str | None) -> str | None:
    """The trailing house-number token (e.g. '126а', '32', '108и'), or None.

    Returns None when the normalized address has no trailing house number.
    Used to compare whether two addresses refer to the same building number
    (geo-proximity consistency guard in match_or_create_house).
    """
    if not normalized:
        return None
    m = re.search(r"(?:^|\s)(\d\S*)$", normalized)
    return m.group(1) if m else None


# Екатеринбург — the home city; every other _CITY_TOKENS entry is a non-ЕКБ обл.66 city.
# The Tier-2a oblast guard treats a resolved non-ЕКБ city specially (see resolve_city_token).
# #3051: значение — city_token региона 66 из реестра (app.services.regions);
# _CITY_TOKENS ниже НЕ из реестра намеренно — это зеркало sweep-раскатки
# (CITY_ANCHORS в scraper_kit.orchestration.pipeline, «какие города скрапим»),
# а реестр отвечает на «какие города узнаём» — списки эволюционируют раздельно.
EKB_CITY_TOKEN = _REGIONS[66].city_token

# Cities covered by the обл.66 sweep rollout (ЕКБ + oblast per-city schedules).
# Normalized form: lowercase, hyphens collapsed to spaces (mirrors normalize_address,
# e.g. 'каменск-уральский' -> 'каменск уральский'). Mirrors CITY_ANCHORS in
# scraper_kit.orchestration.pipeline — extend together as the sweep adds cities.
# EKB_CITY_TOKEN must be first (leftmost-alternation determinism / documentation).
_CITY_TOKENS: tuple[str, ...] = (
    EKB_CITY_TOKEN,
    "нижний тагил",
    "каменск уральский",
    "первоуральск",
    "верхняя пышма",
    "серов",
)
# Token-bounded (not substring) so a street named after a city does NOT false-positive:
# 'улица серова 5' has token 'серова' (trailing 'а'), never bare 'серов'.
_CITY_TOKEN_RE = re.compile(
    r"(?:^|\s)(?:" + "|".join(re.escape(c) for c in _CITY_TOKENS) + r")(?:\s|$)"
)


def resolve_city_token(normalized: str | None) -> str | None:
    """Return the обл.66 city token present in the normalized address, else None.

    Token-bounded (see _CITY_TOKEN_RE) so a street named after a city never matches
    ('улица серова 5' -> None, not 'серов'). When the address names ЕКБ, returns
    EKB_CITY_TOKEN; a non-ЕКБ oblast city returns its own token. Used by the Tier-2a
    oblast guard in match_or_create_house: a coord-less card that resolves a NON-ЕКБ city
    must not fingerprint/normalized_address-match a globally-unique alias that (almost
    certainly) belongs to ЕКБ/another city — that would corrupt ЕКБ house data.
    """
    if not normalized:
        return None
    m = _CITY_TOKEN_RE.search(normalized)
    if m is None:
        return None
    # group(0) carries the (?:^|\s)...(?:\s|$) boundary whitespace — strip it to the token.
    return m.group(0).strip()


def has_city_token(normalized: str | None) -> bool:
    """True if the normalized address carries a known обл.66 city token.

    A city token makes the normalized_address city-scoped — two addresses in different
    cities can then never collapse to the same key, so the alias is a safe cross-source
    building key. Used by the Tier-2b oblast guard in match_or_create_house: a bare
    common-street address with NO city token AND no coords is too ambiguous to match a
    globally-unique normalized_address alias (it would mis-bucket an oblast card into a
    same-named ЕКБ house — bug #2 oblast rollout).
    """
    return resolve_city_token(normalized) is not None


def address_fingerprint(address: str | None, lat: float | None, lon: float | None) -> str:
    """SHA-256 fingerprint of normalized address + rounded coordinates (4 dp ≈ 11 m).

    Returns first 32 hex chars of the digest (128 bits — collision-safe for millions of rows).
    """
    norm_addr = normalize_address(address or "")
    lat_r = f"{lat:.4f}" if lat is not None else ""
    lon_r = f"{lon:.4f}" if lon is not None else ""
    key = f"{norm_addr}|{lat_r}|{lon_r}"
    return hashlib.sha256(key.encode("utf-8")).hexdigest()[:32]