fix(tradein): cross-source matching — address PR #470 BLOCK verdict (NOT NULL / geom type / abbreviation order / determinism)

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.

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",
 ]

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")

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,

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,

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()
 
 

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())