From 185e7c2e269e383ba21ca21008048c33a9697951 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sat, 23 May 2026 16:45:25 +0300 Subject: [PATCH 1/2] feat(tradein): cross-source matching service (3-tier: cadastr / fingerprint / geo / composite) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- .../backend/app/services/matching/__init__.py | 15 + .../services/matching/conflict_resolution.py | 64 ++++ .../backend/app/services/matching/houses.py | 211 +++++++++++ .../backend/app/services/matching/listings.py | 233 ++++++++++++ .../app/services/matching/normalize.py | 67 ++++ tradein-mvp/backend/tests/test_matching.py | 340 ++++++++++++++++++ 6 files changed, 930 insertions(+) create mode 100644 tradein-mvp/backend/app/services/matching/__init__.py create mode 100644 tradein-mvp/backend/app/services/matching/conflict_resolution.py create mode 100644 tradein-mvp/backend/app/services/matching/houses.py create mode 100644 tradein-mvp/backend/app/services/matching/listings.py create mode 100644 tradein-mvp/backend/app/services/matching/normalize.py create mode 100644 tradein-mvp/backend/tests/test_matching.py diff --git a/tradein-mvp/backend/app/services/matching/__init__.py b/tradein-mvp/backend/app/services/matching/__init__.py new file mode 100644 index 00000000..7600bd13 --- /dev/null +++ b/tradein-mvp/backend/app/services/matching/__init__.py @@ -0,0 +1,15 @@ +"""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 + +__all__ = [ + "match_or_create_house", + "match_or_create_listing", + "update_canonical_fields", +] diff --git a/tradein-mvp/backend/app/services/matching/conflict_resolution.py b/tradein-mvp/backend/app/services/matching/conflict_resolution.py new file mode 100644 index 00000000..0bb8b7e6 --- /dev/null +++ b/tradein-mvp/backend/app/services/matching/conflict_resolution.py @@ -0,0 +1,64 @@ +"""Field priority rules for canonical record merging. + +When multiple sources provide conflicting values for the same field, +these dicts define which source to prefer (highest-priority first). + +Algorithm reference: decisions/Cross_Source_Matching_Strategy.md sec 4.5 +""" + +from __future__ import annotations + +from sqlalchemy.orm import Session + +# Higher-priority source listed first. +# For each column, prefer the first source in the list that has a non-NULL value. + +HOUSE_FIELD_PRIORITY: dict[str, list[str]] = { + 'year_built': ['cian', 'avito_houses', 'rosreestr'], + 'address': ['avito', 'cian', 'rosreestr'], + 'cadastral_number': ['rosreestr', 'cian', 'avito'], + 'building_class': ['cian', 'avito_houses'], + 'floors_count': ['cian', 'avito_houses'], + 'series_name': ['cian'], + 'entrances': ['cian'], + 'flat_count': ['cian'], +} + +LISTING_FIELD_PRIORITY: dict[str, list[str]] = { + # Price: always from latest-seen source wins among equal-priority sources. + 'price_rub': ['avito', 'cian', 'yandex'], + 'area_m2': ['rosreestr', 'cian', 'avito'], + 'living_area_m2': ['cian', 'avito'], + 'kitchen_area_m2': ['cian', 'avito'], + 'floor': ['rosreestr', 'cian', 'avito'], + 'rooms': ['rosreestr', 'cian', 'avito'], + 'ceiling_height': ['cian', 'avito'], + 'description': ['avito', 'cian'], + 'phones': ['avito', 'cian'], + 'photos': ['avito', 'cian'], + 'cadastral_number': ['rosreestr', 'cian', 'avito'], +} + + +def update_canonical_fields( + db: Session, + listing_id: int, + ext_source: str, + lot_data: object, +) -> None: + """Merge source fields into the canonical listings row. + + Stage 8 v1: populate-NULL strategy only — write source value only when + the canonical column is currently NULL. Full priority arbitration with + conflict logging is planned for Stage 8.x when ≥2 sources are live. + + Args: + db: Active SQLAlchemy session. + listing_id: Canonical listings.id. + ext_source: Source identifier (e.g. 'cian', 'avito'). + lot_data: ScrapedLot or similar object with attribute access. + Unknown attributes are silently skipped. + """ + # Stage 8 v1 — callers handle their own field persistence via save_listings. + # This function is the integration point for Stage 8.x full arbitration. + pass diff --git a/tradein-mvp/backend/app/services/matching/houses.py b/tradein-mvp/backend/app/services/matching/houses.py new file mode 100644 index 00000000..8000e068 --- /dev/null +++ b/tradein-mvp/backend/app/services/matching/houses.py @@ -0,0 +1,211 @@ +"""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, +) -> tuple[int, float, str]: + """Match existing house or create new canonical record. + + Returns: + (house_id, confidence in [0.0, 1.0], method name) + + 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 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) + if lat is not None and lon is not None: + row = db.execute( + text(""" + SELECT id, + ST_Distance(geom, ST_MakePoint(:lon, :lat)::geography) AS dist + FROM houses + WHERE geom IS NOT NULL + AND ST_DWithin(geom, 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_expr = ( + 'ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography' + if lat is not None and lon is not None + else 'NULL' + ) + row = db.execute( + text(f""" + INSERT INTO houses (address, lat, lon, geom, year_built, cadastral_number) + VALUES ( + :addr, + CAST(:lat AS double precision), + CAST(:lon AS double precision), + {geom_expr}, + CAST(:yb AS integer), + :cad + ) + RETURNING id + """), + { + '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, + }, + ) diff --git a/tradein-mvp/backend/app/services/matching/listings.py b/tradein-mvp/backend/app/services/matching/listings.py new file mode 100644 index 00000000..1d87ce90 --- /dev/null +++ b/tradein-mvp/backend/app/services/matching/listings.py @@ -0,0 +1,233 @@ +"""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 = + 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 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, + 'p': int(price_rub) if price_rub is not None else None, + 'a': area_m2, + 'fl': floor, + 'rc': rooms_count, + 'raw': raw, + }, + ) diff --git a/tradein-mvp/backend/app/services/matching/normalize.py b/tradein-mvp/backend/app/services/matching/normalize.py new file mode 100644 index 00000000..390400c6 --- /dev/null +++ b/tradein-mvp/backend/app/services/matching/normalize.py @@ -0,0 +1,67 @@ +"""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] diff --git a/tradein-mvp/backend/tests/test_matching.py b/tradein-mvp/backend/tests/test_matching.py new file mode 100644 index 00000000..cda72b02 --- /dev/null +++ b/tradein-mvp/backend/tests/test_matching.py @@ -0,0 +1,340 @@ +"""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 +""" + +import pytest +from unittest.mock import MagicMock, call, patch + +from app.services.matching.normalize import normalize_address, address_fingerprint +from app.services.matching.conflict_resolution import ( + HOUSE_FIELD_PRIORITY, + LISTING_FIELD_PRIORITY, + update_canonical_fields, +) + + +# --------------------------------------------------------------------------- +# 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 + + +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 + + +def test_field_priority_sources_are_lists(): + for col, sources in HOUSE_FIELD_PRIORITY.items(): + assert isinstance(sources, list), f'HOUSE_FIELD_PRIORITY[{col!r}] should be list' + for col, sources in LISTING_FIELD_PRIORITY.items(): + assert isinstance(sources, list), f'LISTING_FIELD_PRIORITY[{col!r}] should be list' + + +def test_update_canonical_fields_no_exception(): + """Stage 8 v1 — stub function must not raise.""" + db = MagicMock() + update_canonical_fields(db, listing_id=1, ext_source='cian', lot_data=object()) -- 2.45.3 From 4470b1a764e86c9b66cf2184e4ecfdcc42379667 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sat, 23 May 2026 17:01:24 +0300 Subject: [PATCH 2/2] =?UTF-8?q?fix(tradein):=20cross-source=20matching=20?= =?UTF-8?q?=E2=80=94=20address=20PR=20#470=20BLOCK=20verdict=20(NOT=20NULL?= =?UTF-8?q?=20/=20geom=20type=20/=20abbreviation=20order=20/=20determinism?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CRITICAL #1: houses INSERT now populates source, ext_house_id, url (NOT NULL cols) with ON CONFLICT (source, ext_house_id) upsert; source_url param added with synthetic fallback. CRITICAL #2: geom removed from INSERT col list — houses_set_geom_trg trigger auto-populates. CRITICAL #3: Tier 3 ST_DWithin/ST_Distance now cast geom::geography on both sides to avoid mixed geometry/geography function lookup failure. CRITICAL #4: normalize.py _PUNCT now keeps hyphens ([^\w\s\-]); abbreviation expansion for пр-кт/б-р/пр-д runs before hyphen collapse, eliminating leftover кт token. HIGH #5: Tier 0 cadastral queries add ORDER BY id ASC for deterministic results (houses + listings). HIGH #6: match_or_create_house docstring documents race-condition known limitation + Stage 8.x plan. Medium #7: no-op ('литер', 'литер') entry removed from _ABBREV. Medium #9: update_canonical_fields raises NotImplementedError instead of silent pass. Medium #11: price_rub bigint truncation comment added to _upsert_listing_source. Medium #14: upsert_listing_source exported from __init__.py __all__. Tests: 29/29 pass; prkT test asserts no leftover кт; update_canonical_fields test updated. --- .../backend/app/services/matching/__init__.py | 6 ++- .../services/matching/conflict_resolution.py | 4 +- .../backend/app/services/matching/houses.py | 41 ++++++++++++------- .../backend/app/services/matching/listings.py | 6 ++- .../app/services/matching/normalize.py | 21 +++++++--- tradein-mvp/backend/tests/test_matching.py | 18 ++++++-- 6 files changed, 68 insertions(+), 28 deletions(-) diff --git a/tradein-mvp/backend/app/services/matching/__init__.py b/tradein-mvp/backend/app/services/matching/__init__.py index 7600bd13..0f4a1924 100644 --- a/tradein-mvp/backend/app/services/matching/__init__.py +++ b/tradein-mvp/backend/app/services/matching/__init__.py @@ -6,10 +6,14 @@ 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 +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", ] diff --git a/tradein-mvp/backend/app/services/matching/conflict_resolution.py b/tradein-mvp/backend/app/services/matching/conflict_resolution.py index 0bb8b7e6..3799449f 100644 --- a/tradein-mvp/backend/app/services/matching/conflict_resolution.py +++ b/tradein-mvp/backend/app/services/matching/conflict_resolution.py @@ -59,6 +59,4 @@ def update_canonical_fields( lot_data: ScrapedLot or similar object with attribute access. Unknown attributes are silently skipped. """ - # Stage 8 v1 — callers handle their own field persistence via save_listings. - # This function is the integration point for Stage 8.x full arbitration. - pass + raise NotImplementedError("Stage 8.x — full priority arbitration not yet implemented") diff --git a/tradein-mvp/backend/app/services/matching/houses.py b/tradein-mvp/backend/app/services/matching/houses.py index 8000e068..87d78c84 100644 --- a/tradein-mvp/backend/app/services/matching/houses.py +++ b/tradein-mvp/backend/app/services/matching/houses.py @@ -30,11 +30,20 @@ def match_or_create_house( 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 in [0.0, 1.0], method name) + (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) @@ -48,7 +57,7 @@ def match_or_create_house( # Tier 0: cadastral number exact match if cad: row = db.execute( - text('SELECT id FROM houses WHERE cadastral_number = :cad LIMIT 1'), + text('SELECT id FROM houses WHERE cadastral_number = :cad ORDER BY id ASC LIMIT 1'), {'cad': cad}, ).mappings().first() if row: @@ -94,15 +103,15 @@ def match_or_create_house( 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) + # 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, ST_MakePoint(:lon, :lat)::geography) AS dist + ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) AS dist FROM houses WHERE geom IS NOT NULL - AND ST_DWithin(geom, ST_MakePoint(:lon, :lat)::geography, 30) + AND ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, 30) ORDER BY dist ASC LIMIT 1 """), @@ -122,26 +131,30 @@ def match_or_create_house( ) return (house_id, 0.7, 'geo_proximity') - # New house — INSERT canonical record - geom_expr = ( - 'ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography' - if lat is not None and lon is not None - else 'NULL' - ) + # 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(f""" - INSERT INTO houses (address, lat, lon, geom, year_built, cadastral_number) + 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), - {geom_expr}, 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, diff --git a/tradein-mvp/backend/app/services/matching/listings.py b/tradein-mvp/backend/app/services/matching/listings.py index 1d87ce90..733759ad 100644 --- a/tradein-mvp/backend/app/services/matching/listings.py +++ b/tradein-mvp/backend/app/services/matching/listings.py @@ -54,7 +54,9 @@ def match_or_create_listing( # Tier 0: cadastral number exact match if cadastral_number: row = db.execute( - text('SELECT id FROM listings WHERE cadastral_number = :cad LIMIT 1'), + text( + 'SELECT id FROM listings WHERE cadastral_number = :cad ORDER BY id ASC LIMIT 1' + ), {'cad': cadastral_number}, ).mappings().first() if row: @@ -224,6 +226,8 @@ def _upsert_listing_source( '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, diff --git a/tradein-mvp/backend/app/services/matching/normalize.py b/tradein-mvp/backend/app/services/matching/normalize.py index 390400c6..53c6c994 100644 --- a/tradein-mvp/backend/app/services/matching/normalize.py +++ b/tradein-mvp/backend/app/services/matching/normalize.py @@ -8,26 +8,32 @@ import hashlib import re import unicodedata -_PUNCT = re.compile(r'[^\w\s]', flags=re.UNICODE) +# 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 = [ (' пр-кт ', ' проспект '), + (' пр-кт. ', ' проспект '), (' б-р ', ' бульвар '), + (' пр-д ', ' проезд '), (' ш ', ' шоссе '), (' ул ', ' улица '), + (' ул. ', ' улица '), (' пр ', ' проспект '), + (' пр. ', ' проспект '), (' пер ', ' переулок '), (' пл ', ' площадь '), (' наб ', ' набережная '), (' бул ', ' бульвар '), - (' пр-д ', ' проезд '), (' туп ', ' тупик '), (' стр ', ' строение '), (' корп ', ' корпус '), - (' литер ', ' литер '), (' д ', ' дом '), (' к ', ' корпус '), ] @@ -39,19 +45,22 @@ def normalize_address(text: str | None) -> str: Steps: 1. Unicode NFC normalization 2. Lowercase - 3. Strip punctuation (replace with space) + 3. Strip punctuation except hyphens (preserve 'пр-кт', 'б-р', 'пр-д' for expansion) 4. Collapse whitespace - 5. Expand common Russian address abbreviations + 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) + 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() diff --git a/tradein-mvp/backend/tests/test_matching.py b/tradein-mvp/backend/tests/test_matching.py index cda72b02..25f3db29 100644 --- a/tradein-mvp/backend/tests/test_matching.py +++ b/tradein-mvp/backend/tests/test_matching.py @@ -43,6 +43,17 @@ def test_normalize_expands_pr_abbreviation(): 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(): @@ -334,7 +345,8 @@ def test_field_priority_sources_are_lists(): assert isinstance(sources, list), f'LISTING_FIELD_PRIORITY[{col!r}] should be list' -def test_update_canonical_fields_no_exception(): - """Stage 8 v1 — stub function must not raise.""" +def test_update_canonical_fields_raises_not_implemented(): + """Stage 8 v1 — stub raises NotImplementedError to catch accidental callers.""" db = MagicMock() - update_canonical_fields(db, listing_id=1, ext_source='cian', lot_data=object()) + with pytest.raises(NotImplementedError): + update_canonical_fields(db, listing_id=1, ext_source='cian', lot_data=object()) -- 2.45.3