feat(tradein): cross-source matching service (3-tier: cadastr / fingerprint / geo / composite) #470
5 changed files with 923 additions and 0 deletions
19
tradein-mvp/backend/app/services/matching/__init__.py
Normal file
19
tradein-mvp/backend/app/services/matching/__init__.py
Normal file
|
|
@ -0,0 +1,19 @@
|
|||
"""Cross-source matching service — public API.
|
||||
|
||||
Provides 3-tier deduplication for houses and listings across Avito, Cian, Yandex, etc.
|
||||
Algorithm reference: decisions/Cross_Source_Matching_Strategy.md
|
||||
"""
|
||||
|
||||
from app.services.matching.conflict_resolution import update_canonical_fields
|
||||
from app.services.matching.houses import match_or_create_house
|
||||
from app.services.matching.listings import (
|
||||
match_or_create_listing,
|
||||
upsert_listing_source, # public helper for 'new' sentinel flow
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"match_or_create_house",
|
||||
"match_or_create_listing",
|
||||
"update_canonical_fields",
|
||||
"upsert_listing_source",
|
||||
]
|
||||
224
tradein-mvp/backend/app/services/matching/houses.py
Normal file
224
tradein-mvp/backend/app/services/matching/houses.py
Normal file
|
|
@ -0,0 +1,224 @@
|
|||
"""House cross-source matching — 3-tier algorithm.
|
||||
|
||||
Tier 0 (confidence 1.0): cadastral_number exact match on houses table.
|
||||
Tier 1 (confidence 1.0): ext_source + ext_id already in house_sources.
|
||||
Tier 2 (confidence 0.9): address_fingerprint match in house_address_aliases.
|
||||
Tier 3 (confidence 0.7): geo-proximity within 30 m (PostGIS ST_DWithin).
|
||||
New (confidence 1.0): INSERT new canonical house.
|
||||
|
||||
Algorithm reference: decisions/Cross_Source_Matching_Strategy.md sec 3
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.services.matching.normalize import address_fingerprint, normalize_address
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def match_or_create_house(
|
||||
db: Session,
|
||||
ext_source: str,
|
||||
ext_id: str,
|
||||
address: str | None = None,
|
||||
lat: float | None = None,
|
||||
lon: float | None = None,
|
||||
*,
|
||||
year_built: int | None = None,
|
||||
building_cadastral_number: str | None = None,
|
||||
cadastral_number: str | None = None,
|
||||
source_url: str | None = None,
|
||||
) -> tuple[int, float, str]:
|
||||
"""Match existing house or create new canonical record.
|
||||
|
||||
⚠️ KNOWN LIMITATION: Concurrent calls for the same address can race —
|
||||
two scrapers calling simultaneously for an unknown address can both
|
||||
miss Tier 0-3 and each INSERT a new house row. Mitigation (Stage 8.x):
|
||||
gate scraper concurrency to 1 per fingerprint OR add
|
||||
pg_advisory_xact_lock(hashtext(normalized_address)) at function entry.
|
||||
|
||||
Returns:
|
||||
(house_id, confidence ∈ [0.0, 1.0], method ∈ {
|
||||
'cadastr_exact', 'source_exact', 'fingerprint', 'geo_proximity', 'new'
|
||||
})
|
||||
|
||||
Method values:
|
||||
'cadastr_exact' — matched by cadastral number (confidence 1.0)
|
||||
'source_exact' — already in house_sources for this source+ext_id (confidence 1.0)
|
||||
'fingerprint' — matched by address fingerprint (confidence 0.9)
|
||||
'geo_proximity' — matched by geo within 30 m (confidence 0.7)
|
||||
'new' — new house created (confidence 1.0)
|
||||
"""
|
||||
cad = building_cadastral_number or cadastral_number
|
||||
|
||||
# Tier 0: cadastral number exact match
|
||||
if cad:
|
||||
row = db.execute(
|
||||
text('SELECT id FROM houses WHERE cadastral_number = :cad ORDER BY id ASC LIMIT 1'),
|
||||
{'cad': cad},
|
||||
).mappings().first()
|
||||
if row:
|
||||
house_id = int(row['id'])
|
||||
_upsert_house_source(
|
||||
db, house_id=house_id, ext_source=ext_source, ext_id=ext_id,
|
||||
method='cadastr_exact', confidence=1.0,
|
||||
)
|
||||
logger.info('house match cadastr_exact house_id=%s cad=%s', house_id, cad)
|
||||
return (house_id, 1.0, 'cadastr_exact')
|
||||
|
||||
# Tier 1: source+ext_id already registered in house_sources
|
||||
row = db.execute(
|
||||
text(
|
||||
'SELECT house_id FROM house_sources '
|
||||
'WHERE ext_source = :s AND ext_id = :e LIMIT 1'
|
||||
),
|
||||
{'s': ext_source, 'e': str(ext_id)},
|
||||
).mappings().first()
|
||||
if row:
|
||||
house_id = int(row['house_id'])
|
||||
logger.info('house match source_exact house_id=%s src=%s ext_id=%s', house_id, ext_source,
|
||||
ext_id)
|
||||
return (house_id, 1.0, 'source_exact')
|
||||
|
||||
# Compute fingerprint once for Tier 2 and reuse for new house registration
|
||||
fp = address_fingerprint(address, lat, lon)
|
||||
|
||||
# Tier 2: address fingerprint lookup
|
||||
row = db.execute(
|
||||
text(
|
||||
'SELECT house_id FROM house_address_aliases '
|
||||
'WHERE fingerprint = :fp LIMIT 1'
|
||||
),
|
||||
{'fp': fp},
|
||||
).mappings().first()
|
||||
if row:
|
||||
house_id = int(row['house_id'])
|
||||
_upsert_house_source(
|
||||
db, house_id=house_id, ext_source=ext_source, ext_id=ext_id,
|
||||
method='fingerprint', confidence=0.9,
|
||||
)
|
||||
logger.info('house match fingerprint house_id=%s fp=%s', house_id, fp)
|
||||
return (house_id, 0.9, 'fingerprint')
|
||||
|
||||
# Tier 3: geo-proximity — within 30 m (PostGIS geography cast on both sides)
|
||||
if lat is not None and lon is not None:
|
||||
row = db.execute(
|
||||
text("""
|
||||
SELECT id,
|
||||
ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) AS dist
|
||||
FROM houses
|
||||
WHERE geom IS NOT NULL
|
||||
AND ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, 30)
|
||||
ORDER BY dist ASC
|
||||
LIMIT 1
|
||||
"""),
|
||||
{'lat': lat, 'lon': lon},
|
||||
).mappings().first()
|
||||
if row:
|
||||
house_id = int(row['id'])
|
||||
_upsert_house_source(
|
||||
db, house_id=house_id, ext_source=ext_source, ext_id=ext_id,
|
||||
method='geo_proximity', confidence=0.7,
|
||||
)
|
||||
# Register fingerprint alias so future calls skip geo lookup
|
||||
_insert_alias(db, house_id=house_id, address=address, fp=fp, source=ext_source)
|
||||
logger.info(
|
||||
'house match geo_proximity house_id=%s dist=%.1f src=%s',
|
||||
house_id, float(row['dist']), ext_source,
|
||||
)
|
||||
return (house_id, 0.7, 'geo_proximity')
|
||||
|
||||
# New house — INSERT canonical record.
|
||||
# geom column is auto-populated by houses_set_geom_trg BEFORE INSERT trigger from lat/lon.
|
||||
# Do NOT include geom in the INSERT column list — trigger handles it.
|
||||
url = source_url or f'matching://{ext_source}/{ext_id}'
|
||||
row = db.execute(
|
||||
text("""
|
||||
INSERT INTO houses (source, ext_house_id, url, address, lat, lon, year_built,
|
||||
cadastral_number)
|
||||
VALUES (
|
||||
:src, :eid, :url,
|
||||
:addr,
|
||||
CAST(:lat AS double precision),
|
||||
CAST(:lon AS double precision),
|
||||
CAST(:yb AS integer),
|
||||
:cad
|
||||
)
|
||||
ON CONFLICT (source, ext_house_id) DO UPDATE SET
|
||||
address = COALESCE(EXCLUDED.address, houses.address)
|
||||
RETURNING id
|
||||
"""),
|
||||
{
|
||||
'src': ext_source,
|
||||
'eid': str(ext_id),
|
||||
'url': url,
|
||||
'addr': address,
|
||||
'lat': lat,
|
||||
'lon': lon,
|
||||
'yb': year_built,
|
||||
'cad': cad,
|
||||
},
|
||||
).mappings().one()
|
||||
house_id = int(row['id'])
|
||||
|
||||
_upsert_house_source(
|
||||
db, house_id=house_id, ext_source=ext_source, ext_id=ext_id,
|
||||
method='new', confidence=1.0,
|
||||
)
|
||||
_insert_alias(db, house_id=house_id, address=address, fp=fp, source=ext_source)
|
||||
|
||||
logger.info('house new house_id=%s addr=%r src=%s', house_id, address, ext_source)
|
||||
return (house_id, 1.0, 'new')
|
||||
|
||||
|
||||
def _upsert_house_source(
|
||||
db: Session,
|
||||
*,
|
||||
house_id: int,
|
||||
ext_source: str,
|
||||
ext_id: str,
|
||||
method: str,
|
||||
confidence: float,
|
||||
) -> None:
|
||||
"""Insert or refresh house_sources row for this source+ext_id."""
|
||||
db.execute(
|
||||
text("""
|
||||
INSERT INTO house_sources (
|
||||
house_id, ext_source, ext_id, confidence, matched_method, matched_at, last_seen_at
|
||||
) VALUES (
|
||||
CAST(:hid AS bigint), :s, :e,
|
||||
CAST(:c AS real), :m, NOW(), NOW()
|
||||
)
|
||||
ON CONFLICT (ext_source, ext_id) DO UPDATE SET
|
||||
confidence = GREATEST(EXCLUDED.confidence, house_sources.confidence),
|
||||
last_seen_at = NOW()
|
||||
"""),
|
||||
{'hid': house_id, 's': ext_source, 'e': str(ext_id), 'c': confidence, 'm': method},
|
||||
)
|
||||
|
||||
|
||||
def _insert_alias(
|
||||
db: Session,
|
||||
*,
|
||||
house_id: int,
|
||||
address: str | None,
|
||||
fp: str,
|
||||
source: str,
|
||||
) -> None:
|
||||
"""Register normalized_address alias for this house (idempotent on normalized_address)."""
|
||||
db.execute(
|
||||
text("""
|
||||
INSERT INTO house_address_aliases (house_id, normalized_address, fingerprint, source)
|
||||
VALUES (CAST(:hid AS bigint), :na, :fp, :src)
|
||||
ON CONFLICT (normalized_address) DO NOTHING
|
||||
"""),
|
||||
{
|
||||
'hid': house_id,
|
||||
'na': normalize_address(address),
|
||||
'fp': fp,
|
||||
'src': source,
|
||||
},
|
||||
)
|
||||
237
tradein-mvp/backend/app/services/matching/listings.py
Normal file
237
tradein-mvp/backend/app/services/matching/listings.py
Normal file
|
|
@ -0,0 +1,237 @@
|
|||
"""Listing cross-source matching — 3-tier algorithm.
|
||||
|
||||
Tier 0 (confidence 1.0): cadastral_number exact match on listings table.
|
||||
Tier 1 (confidence 1.0): ext_source + ext_id already in listing_sources.
|
||||
Tier 2 (confidence 0.85): description_minhash exact match within same house.
|
||||
Tier 3 (confidence 0.75): composite match — house + floor + area ±2% + rooms.
|
||||
New (sentinel 0): caller is responsible for INSERT + call _upsert_listing_source.
|
||||
|
||||
Algorithm reference: decisions/Cross_Source_Matching_Strategy.md sec 4
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def match_or_create_listing(
|
||||
db: Session,
|
||||
ext_source: str,
|
||||
ext_id: str,
|
||||
house_id: int,
|
||||
floor: int | None = None,
|
||||
rooms_count: int | None = None,
|
||||
area_m2: float | None = None,
|
||||
price_rub: float | None = None,
|
||||
*,
|
||||
cadastral_number: str | None = None,
|
||||
description_minhash: str | None = None,
|
||||
source_url: str | None = None,
|
||||
source_data: dict | None = None,
|
||||
) -> tuple[int, float, str]:
|
||||
"""Match existing listing or signal that caller should create a new one.
|
||||
|
||||
Returns:
|
||||
(listing_id, confidence in [0.0, 1.0], method name)
|
||||
|
||||
Method values:
|
||||
'cadastr_exact' — matched by cadastral number (confidence 1.0)
|
||||
'source_exact' — already in listing_sources for this source+ext_id (confidence 1.0)
|
||||
'minhash' — matched by description minhash within house (confidence 0.85)
|
||||
'composite' — matched by house+floor+area±2%+rooms (confidence 0.75)
|
||||
'new' — no match; caller must INSERT listing then call upsert_listing_source()
|
||||
|
||||
When method == 'new', listing_id is 0 (sentinel). Caller flow:
|
||||
listing_id, conf, method = match_or_create_listing(db, ...)
|
||||
if method == 'new':
|
||||
listing_id = <INSERT canonical listing row>
|
||||
upsert_listing_source(db, listing_id=listing_id, ...)
|
||||
"""
|
||||
# Tier 0: cadastral number exact match
|
||||
if cadastral_number:
|
||||
row = db.execute(
|
||||
text(
|
||||
'SELECT id FROM listings WHERE cadastral_number = :cad ORDER BY id ASC LIMIT 1'
|
||||
),
|
||||
{'cad': cadastral_number},
|
||||
).mappings().first()
|
||||
if row:
|
||||
listing_id = int(row['id'])
|
||||
_upsert_listing_source(
|
||||
db, listing_id=listing_id, ext_source=ext_source, ext_id=ext_id,
|
||||
method='cadastr_exact', confidence=1.0,
|
||||
price_rub=price_rub, area_m2=area_m2, floor=floor,
|
||||
rooms_count=rooms_count, source_url=source_url, source_data=source_data,
|
||||
)
|
||||
logger.info('listing match cadastr_exact id=%s cad=%s', listing_id, cadastral_number)
|
||||
return (listing_id, 1.0, 'cadastr_exact')
|
||||
|
||||
# Tier 1: source+ext_id already registered
|
||||
row = db.execute(
|
||||
text(
|
||||
'SELECT listing_id FROM listing_sources '
|
||||
'WHERE ext_source = :s AND ext_id = :e LIMIT 1'
|
||||
),
|
||||
{'s': ext_source, 'e': str(ext_id)},
|
||||
).mappings().first()
|
||||
if row:
|
||||
listing_id = int(row['listing_id'])
|
||||
logger.info('listing match source_exact id=%s src=%s ext_id=%s', listing_id, ext_source,
|
||||
ext_id)
|
||||
return (listing_id, 1.0, 'source_exact')
|
||||
|
||||
# Tier 2: description minhash match within same house
|
||||
# Stage 8 v1: exact match on pre-computed minhash string.
|
||||
# Future (Stage 8.x): replace with datasketch MinHash LSH for approximate Jaccard.
|
||||
if description_minhash:
|
||||
row = db.execute(
|
||||
text("""
|
||||
SELECT id FROM listings
|
||||
WHERE house_id = CAST(:hid AS bigint)
|
||||
AND description_minhash = :hash
|
||||
LIMIT 1
|
||||
"""),
|
||||
{'hid': house_id, 'hash': description_minhash},
|
||||
).mappings().first()
|
||||
if row:
|
||||
listing_id = int(row['id'])
|
||||
_upsert_listing_source(
|
||||
db, listing_id=listing_id, ext_source=ext_source, ext_id=ext_id,
|
||||
method='minhash', confidence=0.85,
|
||||
price_rub=price_rub, area_m2=area_m2, floor=floor,
|
||||
rooms_count=rooms_count, source_url=source_url, source_data=source_data,
|
||||
)
|
||||
logger.info('listing match minhash id=%s hash=%s', listing_id, description_minhash)
|
||||
return (listing_id, 0.85, 'minhash')
|
||||
|
||||
# Tier 3: composite match — house + floor + area ±2% + rooms
|
||||
if house_id and floor is not None and area_m2 is not None and rooms_count is not None:
|
||||
row = db.execute(
|
||||
text("""
|
||||
SELECT id,
|
||||
ABS(area_m2 - CAST(:area AS numeric)) AS diff
|
||||
FROM listings
|
||||
WHERE house_id = CAST(:hid AS bigint)
|
||||
AND floor = :fl
|
||||
AND rooms = :rc
|
||||
AND area_m2 BETWEEN CAST(:area AS numeric) * 0.98
|
||||
AND CAST(:area AS numeric) * 1.02
|
||||
ORDER BY diff ASC
|
||||
LIMIT 1
|
||||
"""),
|
||||
{'hid': house_id, 'fl': floor, 'rc': rooms_count, 'area': area_m2},
|
||||
).mappings().first()
|
||||
if row:
|
||||
listing_id = int(row['id'])
|
||||
_upsert_listing_source(
|
||||
db, listing_id=listing_id, ext_source=ext_source, ext_id=ext_id,
|
||||
method='composite', confidence=0.75,
|
||||
price_rub=price_rub, area_m2=area_m2, floor=floor,
|
||||
rooms_count=rooms_count, source_url=source_url, source_data=source_data,
|
||||
)
|
||||
logger.info(
|
||||
'listing match composite id=%s house=%s floor=%s area=%.1f rooms=%s',
|
||||
listing_id, house_id, floor, area_m2, rooms_count,
|
||||
)
|
||||
return (listing_id, 0.75, 'composite')
|
||||
|
||||
# No match — signal caller to create new listing
|
||||
return (0, 1.0, 'new')
|
||||
|
||||
|
||||
def upsert_listing_source(
|
||||
db: Session,
|
||||
*,
|
||||
listing_id: int,
|
||||
ext_source: str,
|
||||
ext_id: str,
|
||||
method: str,
|
||||
confidence: float,
|
||||
price_rub: float | None = None,
|
||||
area_m2: float | None = None,
|
||||
floor: int | None = None,
|
||||
rooms_count: int | None = None,
|
||||
source_url: str | None = None,
|
||||
source_data: dict | None = None,
|
||||
) -> None:
|
||||
"""Public helper — register or refresh a listing_sources row.
|
||||
|
||||
Called by external code after creating a new canonical listing.
|
||||
"""
|
||||
_upsert_listing_source(
|
||||
db,
|
||||
listing_id=listing_id,
|
||||
ext_source=ext_source,
|
||||
ext_id=ext_id,
|
||||
method=method,
|
||||
confidence=confidence,
|
||||
price_rub=price_rub,
|
||||
area_m2=area_m2,
|
||||
floor=floor,
|
||||
rooms_count=rooms_count,
|
||||
source_url=source_url,
|
||||
source_data=source_data,
|
||||
)
|
||||
|
||||
|
||||
def _upsert_listing_source(
|
||||
db: Session,
|
||||
*,
|
||||
listing_id: int,
|
||||
ext_source: str,
|
||||
ext_id: str,
|
||||
method: str,
|
||||
confidence: float,
|
||||
price_rub: float | None,
|
||||
area_m2: float | None,
|
||||
floor: int | None,
|
||||
rooms_count: int | None,
|
||||
source_url: str | None,
|
||||
source_data: dict | None,
|
||||
) -> None:
|
||||
"""Insert or refresh listing_sources row for this source+ext_id."""
|
||||
raw = json.dumps(source_data) if source_data is not None else None
|
||||
db.execute(
|
||||
text("""
|
||||
INSERT INTO listing_sources (
|
||||
listing_id, ext_source, ext_id,
|
||||
confidence, matched_method, matched_at, last_seen_at,
|
||||
source_url, last_scraped_at,
|
||||
price_rub, area_m2, floor, rooms_count, raw_payload
|
||||
) VALUES (
|
||||
CAST(:lid AS bigint), :s, :e,
|
||||
CAST(:c AS real), :m, NOW(), NOW(),
|
||||
:url, NOW(),
|
||||
CAST(:p AS bigint), CAST(:a AS numeric), :fl, :rc,
|
||||
CAST(:raw AS jsonb)
|
||||
)
|
||||
ON CONFLICT (ext_source, ext_id) DO UPDATE SET
|
||||
confidence = GREATEST(EXCLUDED.confidence, listing_sources.confidence),
|
||||
last_seen_at = NOW(),
|
||||
last_scraped_at = NOW(),
|
||||
price_rub = COALESCE(EXCLUDED.price_rub, listing_sources.price_rub),
|
||||
area_m2 = COALESCE(EXCLUDED.area_m2, listing_sources.area_m2),
|
||||
floor = COALESCE(EXCLUDED.floor, listing_sources.floor),
|
||||
rooms_count = COALESCE(EXCLUDED.rooms_count, listing_sources.rooms_count),
|
||||
raw_payload = COALESCE(EXCLUDED.raw_payload, listing_sources.raw_payload)
|
||||
"""),
|
||||
{
|
||||
'lid': listing_id,
|
||||
's': ext_source,
|
||||
'e': str(ext_id),
|
||||
'c': confidence,
|
||||
'm': method,
|
||||
'url': source_url,
|
||||
# Migration 029 declares listing_sources.price_rub bigint — whole rubles only,
|
||||
# kopecks truncated. Document expectation in caller.
|
||||
'p': int(price_rub) if price_rub is not None else None,
|
||||
'a': area_m2,
|
||||
'fl': floor,
|
||||
'rc': rooms_count,
|
||||
'raw': raw,
|
||||
},
|
||||
)
|
||||
76
tradein-mvp/backend/app/services/matching/normalize.py
Normal file
76
tradein-mvp/backend/app/services/matching/normalize.py
Normal file
|
|
@ -0,0 +1,76 @@
|
|||
"""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
|
||||
|
||||
# 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+')
|
||||
|
||||
# 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 = [
|
||||
(' пр-кт ', ' проспект '),
|
||||
(' пр-кт. ', ' проспект '),
|
||||
(' б-р ', ' бульвар '),
|
||||
(' пр-д ', ' проезд '),
|
||||
(' ш ', ' шоссе '),
|
||||
(' ул ', ' улица '),
|
||||
(' ул. ', ' улица '),
|
||||
(' пр ', ' проспект '),
|
||||
(' пр. ', ' проспект '),
|
||||
(' пер ', ' переулок '),
|
||||
(' пл ', ' площадь '),
|
||||
(' наб ', ' набережная '),
|
||||
(' бул ', ' бульвар '),
|
||||
(' туп ', ' тупик '),
|
||||
(' стр ', ' строение '),
|
||||
(' корп ', ' корпус '),
|
||||
(' д ', ' дом '),
|
||||
(' к ', ' корпус '),
|
||||
]
|
||||
|
||||
|
||||
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
|
||||
"""
|
||||
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('-', ' ')
|
||||
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]
|
||||
367
tradein-mvp/backend/tests/test_matching.py
Normal file
367
tradein-mvp/backend/tests/test_matching.py
Normal file
|
|
@ -0,0 +1,367 @@
|
|||
"""Unit tests for cross-source matching service.
|
||||
|
||||
Covers:
|
||||
- normalize_address(): abbreviation expansion, punctuation stripping, edge cases
|
||||
- address_fingerprint(): stability on case variation, sensitivity to coords, None handling
|
||||
- match_or_create_house() / match_or_create_listing(): mock-DB tier routing
|
||||
|
||||
Reference: decisions/Cross_Source_Matching_Strategy.md
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from app.services.matching.conflict_resolution import (
|
||||
HOUSE_FIELD_PRIORITY,
|
||||
LISTING_FIELD_PRIORITY,
|
||||
update_canonical_fields,
|
||||
)
|
||||
from app.services.matching.normalize import address_fingerprint, normalize_address
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# normalize_address
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_normalize_strips_punctuation_and_collapses_whitespace():
|
||||
assert normalize_address(" Test, Address! ") == "test address"
|
||||
|
||||
|
||||
def test_normalize_expands_ul_abbreviation():
|
||||
result = normalize_address("ул Ленина, 5")
|
||||
assert "улица" in result
|
||||
assert "5" in result
|
||||
|
||||
|
||||
def test_normalize_expands_pr_abbreviation():
|
||||
result = normalize_address("пр Мира 100")
|
||||
assert "проспект" in result
|
||||
assert "100" in result
|
||||
|
||||
|
||||
def test_normalize_expands_prkT_abbreviation():
|
||||
result = normalize_address("пр-кт Ленина 10")
|
||||
assert "проспект" in result
|
||||
# 'кт' must NOT remain as a stray token after expansion
|
||||
assert 'кт' not in result.split()
|
||||
|
||||
|
||||
def test_normalize_prkT_no_leftover_kt():
|
||||
"""пр-кт must expand to 'проспект' cleanly — no leftover 'кт' token."""
|
||||
result = normalize_address("пр-кт Ленина 10")
|
||||
tokens = result.split()
|
||||
assert "проспект" in tokens
|
||||
assert "пр" not in tokens
|
||||
assert "кт" not in tokens
|
||||
|
||||
|
||||
def test_normalize_handles_none():
|
||||
assert normalize_address(None) == ""
|
||||
|
||||
|
||||
def test_normalize_handles_empty_string():
|
||||
assert normalize_address("") == ""
|
||||
|
||||
|
||||
def test_normalize_lowercases():
|
||||
assert normalize_address("ТЕСТ") == "тест"
|
||||
|
||||
|
||||
def test_normalize_unicode_nfc():
|
||||
# Composed vs decomposed е (U+0435 vs U+0435)
|
||||
s1 = normalize_address("улица")
|
||||
s2 = normalize_address("улица")
|
||||
assert s1 == s2
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# address_fingerprint
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_fingerprint_stable_on_case_variation():
|
||||
f1 = address_fingerprint("Ленина 5", 56.8378, 60.5946)
|
||||
f2 = address_fingerprint("ЛЕНИНА 5", 56.8378, 60.5946)
|
||||
assert f1 == f2
|
||||
|
||||
|
||||
def test_fingerprint_distinct_on_different_lat():
|
||||
f1 = address_fingerprint("Same Address", 56.8, 60.5)
|
||||
f2 = address_fingerprint("Same Address", 56.9, 60.5)
|
||||
assert f1 != f2
|
||||
|
||||
|
||||
def test_fingerprint_distinct_on_different_lon():
|
||||
f1 = address_fingerprint("Same Address", 56.8, 60.5)
|
||||
f2 = address_fingerprint("Same Address", 56.8, 60.6)
|
||||
assert f1 != f2
|
||||
|
||||
|
||||
def test_fingerprint_stable_within_11m(monkeypatch):
|
||||
# 4 decimal places → ~11 m resolution; coords differing in 5th decimal should match
|
||||
f1 = address_fingerprint("Addr", 56.83780, 60.59460)
|
||||
f2 = address_fingerprint("Addr", 56.83781, 60.59461)
|
||||
# 4dp rounding: 56.8378 == 56.8378, 60.5946 == 60.5946 → same fingerprint
|
||||
assert f1 == f2
|
||||
|
||||
|
||||
def test_fingerprint_handles_none_coords():
|
||||
fp = address_fingerprint("Some Address", None, None)
|
||||
assert len(fp) == 32 # truncated sha256 hex
|
||||
|
||||
|
||||
def test_fingerprint_handles_none_address():
|
||||
fp = address_fingerprint(None, 56.8, 60.5)
|
||||
assert len(fp) == 32
|
||||
|
||||
|
||||
def test_fingerprint_all_none():
|
||||
fp = address_fingerprint(None, None, None)
|
||||
assert len(fp) == 32
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# match_or_create_house — mock DB tier routing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_db(rows: list[dict | None]) -> MagicMock:
|
||||
"""Build a Session mock that returns rows sequentially per execute() call."""
|
||||
db = MagicMock()
|
||||
side_effects = []
|
||||
for row in rows:
|
||||
result = MagicMock()
|
||||
mappings = MagicMock()
|
||||
if row is None:
|
||||
mappings.first.return_value = None
|
||||
mappings.one.return_value = None
|
||||
else:
|
||||
mappings.first.return_value = row
|
||||
mappings.one.return_value = row
|
||||
result.mappings.return_value = mappings
|
||||
side_effects.append(result)
|
||||
db.execute.side_effect = side_effects
|
||||
return db
|
||||
|
||||
|
||||
def test_match_house_tier0_cadastr():
|
||||
"""Tier 0: cadastral_number match → returns cadastr_exact without geo query."""
|
||||
from app.services.matching.houses import match_or_create_house
|
||||
|
||||
db = _make_db([
|
||||
{'id': 42}, # cadastral match
|
||||
None, # _upsert_house_source (INSERT ... ON CONFLICT)
|
||||
])
|
||||
house_id, conf, method = match_or_create_house(
|
||||
db, 'cian', 'ext-001',
|
||||
address='ул Ленина 5', lat=56.8, lon=60.5,
|
||||
cadastral_number='66:41:0000000:999',
|
||||
)
|
||||
assert house_id == 42
|
||||
assert conf == 1.0
|
||||
assert method == 'cadastr_exact'
|
||||
|
||||
|
||||
def test_match_house_tier1_source_exact():
|
||||
"""Tier 1: ext_source+ext_id already in house_sources.
|
||||
|
||||
No cadastral_number supplied → Tier 0 skipped entirely.
|
||||
First db.execute call goes to house_sources query.
|
||||
source_exact returns immediately without calling _upsert_house_source.
|
||||
"""
|
||||
from app.services.matching.houses import match_or_create_house
|
||||
|
||||
db = _make_db([
|
||||
{'house_id': 7}, # house_sources hit (first and only execute call)
|
||||
])
|
||||
house_id, conf, method = match_or_create_house(
|
||||
db, 'avito', 'ext-999',
|
||||
address='пр Мира 10', lat=56.8, lon=60.5,
|
||||
)
|
||||
assert house_id == 7
|
||||
assert method == 'source_exact'
|
||||
|
||||
|
||||
def test_match_house_tier2_fingerprint():
|
||||
"""Tier 2: fingerprint match in house_address_aliases."""
|
||||
from app.services.matching.houses import match_or_create_house
|
||||
|
||||
db = _make_db([
|
||||
None, # house_sources miss
|
||||
{'house_id': 15}, # fingerprint hit
|
||||
None, # _upsert_house_source
|
||||
])
|
||||
house_id, conf, method = match_or_create_house(
|
||||
db, 'yandex', 'ext-123',
|
||||
address='Тестовая 1', lat=56.84, lon=60.60,
|
||||
)
|
||||
assert house_id == 15
|
||||
assert conf == 0.9
|
||||
assert method == 'fingerprint'
|
||||
|
||||
|
||||
def test_match_house_tier3_geo():
|
||||
"""Tier 3: geo-proximity match."""
|
||||
from app.services.matching.houses import match_or_create_house
|
||||
|
||||
db = _make_db([
|
||||
None, # house_sources miss
|
||||
None, # fingerprint miss
|
||||
{'id': 22, 'dist': 15.5}, # geo hit
|
||||
None, # _upsert_house_source
|
||||
None, # _insert_alias
|
||||
])
|
||||
house_id, conf, method = match_or_create_house(
|
||||
db, 'n1', 'ext-456',
|
||||
address='Новая 3', lat=56.83, lon=60.59,
|
||||
)
|
||||
assert house_id == 22
|
||||
assert conf == 0.7
|
||||
assert method == 'geo_proximity'
|
||||
|
||||
|
||||
def test_match_house_new():
|
||||
"""All tiers miss → new house created."""
|
||||
from app.services.matching.houses import match_or_create_house
|
||||
|
||||
db = _make_db([
|
||||
None, # house_sources miss
|
||||
None, # fingerprint miss
|
||||
None, # geo miss
|
||||
{'id': 99}, # INSERT RETURNING id
|
||||
None, # _upsert_house_source
|
||||
None, # _insert_alias
|
||||
])
|
||||
house_id, conf, method = match_or_create_house(
|
||||
db, 'avito', 'new-ext',
|
||||
address='Новостройка 1', lat=56.85, lon=60.61,
|
||||
)
|
||||
assert house_id == 99
|
||||
assert conf == 1.0
|
||||
assert method == 'new'
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# match_or_create_listing — mock DB tier routing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_match_listing_tier0_cadastr():
|
||||
from app.services.matching.listings import match_or_create_listing
|
||||
|
||||
db = _make_db([
|
||||
{'id': 200}, # cadastral hit
|
||||
None, # _upsert_listing_source
|
||||
])
|
||||
listing_id, conf, method = match_or_create_listing(
|
||||
db, 'cian', 'cian-555', house_id=10,
|
||||
floor=5, rooms_count=2, area_m2=48.5, price_rub=5_000_000.0,
|
||||
cadastral_number='66:41:0204016:1234',
|
||||
)
|
||||
assert listing_id == 200
|
||||
assert method == 'cadastr_exact'
|
||||
|
||||
|
||||
def test_match_listing_tier1_source_exact():
|
||||
"""Tier 1: no cadastral_number → Tier 0 skipped; first execute is listing_sources hit."""
|
||||
from app.services.matching.listings import match_or_create_listing
|
||||
|
||||
db = _make_db([
|
||||
{'listing_id': 101}, # listing_sources hit (first and only execute call)
|
||||
])
|
||||
listing_id, conf, method = match_or_create_listing(
|
||||
db, 'avito', 'avito-888', house_id=10,
|
||||
)
|
||||
assert listing_id == 101
|
||||
assert method == 'source_exact'
|
||||
|
||||
|
||||
def test_match_listing_tier2_minhash():
|
||||
from app.services.matching.listings import match_or_create_listing
|
||||
|
||||
db = _make_db([
|
||||
None, # listing_sources miss
|
||||
{'id': 55}, # minhash hit
|
||||
None, # _upsert_listing_source
|
||||
])
|
||||
listing_id, conf, method = match_or_create_listing(
|
||||
db, 'cian', 'cian-999', house_id=10,
|
||||
description_minhash='abc123deadbeef',
|
||||
)
|
||||
assert listing_id == 55
|
||||
assert conf == 0.85
|
||||
assert method == 'minhash'
|
||||
|
||||
|
||||
def test_match_listing_tier3_composite():
|
||||
"""Tier 3: no cadastral + no minhash provided → tiers 0/2 skipped.
|
||||
Execute order: listing_sources miss → composite hit → _upsert.
|
||||
"""
|
||||
from app.services.matching.listings import match_or_create_listing
|
||||
|
||||
db = _make_db([
|
||||
None, # listing_sources miss (Tier 1)
|
||||
{'id': 77, 'diff': 0.1}, # composite hit (Tier 3; Tier 2 skipped — no hash)
|
||||
None, # _upsert_listing_source
|
||||
])
|
||||
listing_id, conf, method = match_or_create_listing(
|
||||
db, 'cian', 'cian-777', house_id=10,
|
||||
floor=3, rooms_count=1, area_m2=36.0,
|
||||
)
|
||||
assert listing_id == 77
|
||||
assert conf == 0.75
|
||||
assert method == 'composite'
|
||||
|
||||
|
||||
def test_match_listing_new():
|
||||
from app.services.matching.listings import match_or_create_listing
|
||||
|
||||
db = _make_db([
|
||||
None, # listing_sources miss
|
||||
None, # composite miss (floor/area/rooms all None → tier skipped)
|
||||
])
|
||||
listing_id, conf, method = match_or_create_listing(
|
||||
db, 'avito', 'avito-new', house_id=10,
|
||||
)
|
||||
assert listing_id == 0
|
||||
assert method == 'new'
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# conflict_resolution
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_field_priority_dicts_not_empty():
|
||||
assert HOUSE_FIELD_PRIORITY
|
||||
assert LISTING_FIELD_PRIORITY
|
||||
|
||||
|
||||
VALID_STRING_RULES = {"union", "cross_validate", "first_non_null"}
|
||||
|
||||
|
||||
def test_field_priority_sources_are_lists_or_known_string_rules():
|
||||
"""Each priority entry is either a source list or a known string rule.
|
||||
|
||||
PR #471 (main) introduced 'union' / 'cross_validate' / 'first_non_null' string
|
||||
rules alongside list entries. Both are valid per LISTING_FIELD_PRIORITY type hint.
|
||||
"""
|
||||
for col, sources in HOUSE_FIELD_PRIORITY.items():
|
||||
assert isinstance(sources, list) or sources in VALID_STRING_RULES, (
|
||||
f'HOUSE_FIELD_PRIORITY[{col!r}] must be list or known rule, got {sources!r}'
|
||||
)
|
||||
for col, sources in LISTING_FIELD_PRIORITY.items():
|
||||
assert isinstance(sources, list) or sources in VALID_STRING_RULES, (
|
||||
f'LISTING_FIELD_PRIORITY[{col!r}] must be list or known rule, got {sources!r}'
|
||||
)
|
||||
|
||||
|
||||
def test_update_canonical_fields_is_callable():
|
||||
"""Stage 8 v1 — legacy stub (no-op pass). Full arbitration deferred to Stage 8.x.
|
||||
|
||||
PR #471 (main) replaced NotImplementedError stub with silent pass to allow
|
||||
backward-compat imports. Test updated accordingly.
|
||||
"""
|
||||
db = MagicMock()
|
||||
# Should not raise — no-op stub returns None
|
||||
result = update_canonical_fields(db, listing_id=1, ext_source='cian', lot_data=object())
|
||||
assert result is None
|
||||
Loading…
Add table
Reference in a new issue