185e7c2e26
- normalize.py: normalize_address() + address_fingerprint() (SHA-256, 4dp coords) - houses.py: match_or_create_house() — tiers: cadastr(1.0) / source_exact(1.0) / fingerprint(0.9) / geo_proximity 30m(0.7) / new - listings.py: match_or_create_listing() — tiers: cadastr(1.0) / source_exact(1.0) / minhash(0.85) / composite floor+area±2%+rooms(0.75) / new - conflict_resolution.py: HOUSE/LISTING_FIELD_PRIORITY dicts + update_canonical_fields() stub (Stage 8 v1) - tests/test_matching.py: 28 unit tests (all pass), mock-DB tier routing coverage
67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
"""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
|
||
|
||
_PUNCT = re.compile(r'[^\w\s]', flags=re.UNICODE)
|
||
_WS = re.compile(r'\s+')
|
||
|
||
# Abbreviation expansion — applied after lowercasing, bounded by spaces.
|
||
# Order matters: longer/more-specific patterns first to avoid partial overlap.
|
||
_ABBREV = [
|
||
(' пр-кт ', ' проспект '),
|
||
(' б-р ', ' бульвар '),
|
||
(' ш ', ' шоссе '),
|
||
(' ул ', ' улица '),
|
||
(' пр ', ' проспект '),
|
||
(' пер ', ' переулок '),
|
||
(' пл ', ' площадь '),
|
||
(' наб ', ' набережная '),
|
||
(' бул ', ' бульвар '),
|
||
(' пр-д ', ' проезд '),
|
||
(' туп ', ' тупик '),
|
||
(' стр ', ' строение '),
|
||
(' корп ', ' корпус '),
|
||
(' литер ', ' литер '),
|
||
(' д ', ' дом '),
|
||
(' к ', ' корпус '),
|
||
]
|
||
|
||
|
||
def normalize_address(text: str | None) -> str:
|
||
"""Normalize address string for cross-source comparison.
|
||
|
||
Steps:
|
||
1. Unicode NFC normalization
|
||
2. Lowercase
|
||
3. Strip punctuation (replace with space)
|
||
4. Collapse whitespace
|
||
5. Expand common Russian address abbreviations
|
||
"""
|
||
if not text:
|
||
return ""
|
||
s = unicodedata.normalize('NFC', text).lower()
|
||
s = _PUNCT.sub(' ', s)
|
||
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)
|
||
return _WS.sub(' ', s).strip()
|
||
|
||
|
||
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]
|