diff --git a/tradein-mvp/backend/app/services/matching/houses.py b/tradein-mvp/backend/app/services/matching/houses.py index b640ec7f..7d91f822 100644 --- a/tradein-mvp/backend/app/services/matching/houses.py +++ b/tradein-mvp/backend/app/services/matching/houses.py @@ -62,60 +62,101 @@ def match_or_create_house( # pg_advisory_xact_lock takes (int4, int4) — namespace 42 = "tradein house matching". # Lock is released automatically on caller's COMMIT/ROLLBACK. db.execute( - text('SELECT pg_advisory_xact_lock(42, hashtext(:fp))'), - {'fp': fp}, + text("SELECT pg_advisory_xact_lock(42, hashtext(:fp))"), + {"fp": fp}, ) # 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, + row = ( + db.execute( + text("SELECT id FROM houses WHERE cadastral_number = :cad ORDER BY id ASC LIMIT 1"), + {"cad": cad}, ) - logger.info('house match cadastr_exact house_id=%s cad=%s', house_id, cad) - return (house_id, 1.0, 'cadastr_exact') + .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') - - # 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, + 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)}, ) - logger.info('house match fingerprint house_id=%s fp=%s', house_id, fp) - return (house_id, 0.9, 'fingerprint') + .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") + + # Tier 2: address fingerprint lookup. + # Two sub-tiers to handle coord drift across scrapers: + # 2a. exact fingerprint match (address + rounded coords) + # 2b. normalized_address-only match — same street/number, different provider coords + # Without 2b, two scrapers for the same house with slightly different lat/lon (beyond + # the 4-decimal rounding tolerance) would produce distinct fingerprints, miss Tier 2a, + # and each potentially create a duplicate house row. + norm_addr = normalize_address(address) + row = ( + db.execute( + text("SELECT house_id FROM house_address_aliases " "WHERE fingerprint = :fp LIMIT 1"), + {"fp": fp}, + ) + .mappings() + .first() + ) + if row is None and norm_addr: + # Tier 2b: same normalized address, possibly different coords fingerprint + row = ( + db.execute( + text( + "SELECT house_id FROM house_address_aliases " + "WHERE normalized_address = :na LIMIT 1" + ), + {"na": norm_addr}, + ) + .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, + ) + # Ensure this fingerprint is also registered so future calls hit Tier 2a directly. + _insert_alias(db, house_id=house_id, address=address, fp=fp, source=ext_source) + 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(""" + row = ( + db.execute( + text(""" SELECT id, ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) AS dist FROM houses @@ -124,28 +165,38 @@ def match_or_create_house( ORDER BY dist ASC LIMIT 1 """), - {'lat': lat, 'lon': lon}, - ).mappings().first() + {"lat": lat, "lon": lon}, + ) + .mappings() + .first() + ) if row: - house_id = int(row['id']) + 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, + 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, + "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') + 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(""" + 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 ( @@ -160,27 +211,34 @@ def match_or_create_house( 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']) + { + "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, + 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') + logger.info("house new house_id=%s addr=%r src=%s", house_id, address, ext_source) + return (house_id, 1.0, "new") def match_house_readonly( @@ -210,32 +268,41 @@ def match_house_readonly( """ # Tier 0: cadastral number exact match if cadastral_number: - row = db.execute( - text('SELECT id FROM houses WHERE cadastral_number = :cad ORDER BY id ASC LIMIT 1'), - {'cad': cadastral_number}, - ).mappings().first() - if row: - house_id = int(row['id']) - logger.info( - 'house readonly match cadastr_exact house_id=%s cad=%s', house_id, cadastral_number + row = ( + db.execute( + text("SELECT id FROM houses WHERE cadastral_number = :cad ORDER BY id ASC LIMIT 1"), + {"cad": cadastral_number}, ) - return (house_id, 1.0, 'cadastr_exact') + .mappings() + .first() + ) + if row: + house_id = int(row["id"]) + logger.info( + "house readonly match cadastr_exact house_id=%s cad=%s", house_id, cadastral_number + ) + return (house_id, 1.0, "cadastr_exact") # Tier 1: address fingerprint lookup fp = address_fingerprint(address, lat, lon) - row = db.execute( - text('SELECT house_id FROM house_address_aliases WHERE fingerprint = :fp LIMIT 1'), - {'fp': fp}, - ).mappings().first() + 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']) - logger.info('house readonly match fingerprint house_id=%s fp=%s', house_id, fp) - return (house_id, 0.9, 'fingerprint') + house_id = int(row["house_id"]) + logger.info("house readonly match fingerprint house_id=%s fp=%s", house_id, fp) + return (house_id, 0.9, "fingerprint") # Tier 2: geo-proximity — within 50 m if lat is not None and lon is not None: - row = db.execute( - text(""" + row = ( + db.execute( + text(""" SELECT id, ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) AS dist FROM houses @@ -244,15 +311,19 @@ def match_house_readonly( ORDER BY dist ASC LIMIT 1 """), - {'lat': lat, 'lon': lon}, - ).mappings().first() - if row: - house_id = int(row['id']) - logger.info( - 'house readonly match geo_proximity house_id=%s dist=%.1f', - house_id, float(row['dist']), + {"lat": lat, "lon": lon}, ) - return (house_id, 0.7, 'geo_proximity') + .mappings() + .first() + ) + if row: + house_id = int(row["id"]) + logger.info( + "house readonly match geo_proximity house_id=%s dist=%.1f", + house_id, + float(row["dist"]), + ) + return (house_id, 0.7, "geo_proximity") return None @@ -279,7 +350,7 @@ def _upsert_house_source( 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}, + {"hid": house_id, "s": ext_source, "e": str(ext_id), "c": confidence, "m": method}, ) @@ -291,17 +362,26 @@ def _insert_alias( fp: str, source: str, ) -> None: - """Register normalized_address alias for this house (idempotent on normalized_address).""" + """Register normalized_address alias for this house (idempotent on normalized_address). + + ON CONFLICT DO UPDATE keeps the fingerprint in sync with the most recent call so that + two scrapers providing the same address with slightly different coords (beyond the + 4-decimal rounding boundary) do NOT spawn duplicate alias rows — they converge to one + row with the latest fingerprint, which is then found by Tier 2a on the next scrape. + house_id is not updated on conflict: the first writer wins canonical ownership. + """ 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 + ON CONFLICT (normalized_address) DO UPDATE SET + fingerprint = EXCLUDED.fingerprint, + source = EXCLUDED.source """), { - 'hid': house_id, - 'na': normalize_address(address), - 'fp': fp, - 'src': source, + "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 index 733759ad..f2b94f72 100644 --- a/tradein-mvp/backend/app/services/matching/listings.py +++ b/tradein-mvp/backend/app/services/matching/listings.py @@ -53,65 +53,95 @@ 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 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, + row = ( + db.execute( + text( + "SELECT id FROM listings WHERE cadastral_number = :cad ORDER BY id ASC LIMIT 1" + ), + {"cad": cadastral_number}, ) - logger.info('listing match cadastr_exact id=%s cad=%s', listing_id, cadastral_number) - return (listing_id, 1.0, 'cadastr_exact') + .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() + 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') + 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(""" + 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, + {"hid": house_id, "hash": description_minhash}, ) - logger.info('listing match minhash id=%s hash=%s', listing_id, description_minhash) - return (listing_id, 0.85, '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(""" + row = ( + db.execute( + text(""" SELECT id, ABS(area_m2 - CAST(:area AS numeric)) AS diff FROM listings @@ -123,24 +153,47 @@ def match_or_create_listing( ORDER BY diff ASC LIMIT 1 """), - {'hid': house_id, 'fl': floor, 'rc': rooms_count, 'area': area_m2}, - ).mappings().first() + {"hid": house_id, "fl": floor, "rc": rooms_count, "area": area_m2}, + ) + .mappings() + .first() + ) if row: - listing_id = int(row['id']) + 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, + 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, + "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') + return (listing_id, 0.75, "composite") # No match — signal caller to create new listing - return (0, 1.0, 'new') + # NOTE (Part 4 / #849): match_or_create_listing returns (listing_id, confidence, method) + # but the main scraper ingestion path (_link_listing_to_house in scrapers/base.py) does NOT + # call this function — it uses match_or_create_house + upsert_listing_source('source_link') + # directly. Wiring the matcher confidence/method from this function into listing_sources on + # the source_link path would require restructuring the ingestion flow non-trivially. + # listing_sources already has confidence + matched_method columns (028_matching_tables.sql), + # but they are written as confidence=1.0 / method='source_link' by the scraper path. + # Accurate per-tier confidence wiring is deferred to #774-3. + return (0, 1.0, "new") def upsert_listing_source( @@ -220,18 +273,18 @@ def _upsert_listing_source( 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, + "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, + "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/scrapers/base.py b/tradein-mvp/backend/app/services/scrapers/base.py index 5c6f1c1c..af28cd02 100644 --- a/tradein-mvp/backend/app/services/scrapers/base.py +++ b/tradein-mvp/backend/app/services/scrapers/base.py @@ -451,30 +451,24 @@ def _link_listing_to_house(db: Session, listing_id: int, lot: ScrapedLot) -> Non # so each unique listing creates its own row if no canonical house exists. h_src = lot.house_source or lot.source h_ext = lot.house_ext_id or ext_id - try: - with db.begin_nested(): - house_id, _conf, _method = match_or_create_house( - db, - ext_source=h_src, - ext_id=h_ext, - address=lot.address, - lat=lot.lat, - lon=lot.lon, - year_built=lot.year_built, - building_cadastral_number=lot.building_cadastral_number, - cadastral_number=lot.cadastral_number or lot.kadastr_num, - source_url=lot.house_url or lot.source_url, - ) - except Exception as e: - # Fall through to listing-only upsert: listing_sources row still useful - # even without house linkage (e.g. for later backfill). - logger.warning( - "save_listings:house_match_failed source=%s ext_id=%s: %s", - lot.source, - ext_id, - e, - ) - house_id = None + # No inner begin_nested here — the caller (save_listings) already wraps this + # entire function in a per-row SAVEPOINT (backend.md SAVEPOINT pattern). + # A second nested SAVEPOINT inside would be redundant: if match_or_create_house + # raises a DB error, that error propagates out of _link_listing_to_house, + # the caller's SAVEPOINT rolls back only the hook work (not the listings INSERT), + # and the caller logs a match_failed warning. One SAVEPOINT level is sufficient. + house_id, _conf, _method = match_or_create_house( + db, + ext_source=h_src, + ext_id=h_ext, + address=lot.address, + lat=lot.lat, + lon=lot.lon, + year_built=lot.year_built, + building_cadastral_number=lot.building_cadastral_number, + cadastral_number=lot.cadastral_number or lot.kadastr_num, + source_url=lot.house_url or lot.source_url, + ) upsert_listing_source( db, diff --git a/tradein-mvp/backend/tests/test_849_matching_savepoint.py b/tradein-mvp/backend/tests/test_849_matching_savepoint.py new file mode 100644 index 00000000..297ae453 --- /dev/null +++ b/tradein-mvp/backend/tests/test_849_matching_savepoint.py @@ -0,0 +1,334 @@ +"""Tests for issue #849 Parts 2 and 3. + +Part 2: same normalized_address → exactly 1 house and 1 alias key. +Part 3: batch save_listings with one failing lot → siblings persist, + no orphan SAVEPOINT / no whole-batch rollback. + +All tests use mock DB sessions — no real Postgres required. +""" + +from __future__ import annotations + +import logging +from contextlib import contextmanager +from unittest.mock import MagicMock, patch + +import pytest + +from app.services.matching.houses import _insert_alias, match_or_create_house +from app.services.matching.normalize import address_fingerprint, normalize_address +from app.services.scrapers.base import ScrapedLot, save_listings + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_db(rows: list[dict | None]) -> MagicMock: + """Build a Session mock returning rows sequentially for each 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 _mock_save_db(inserted: bool = True, listing_id: int = 42) -> MagicMock: + """Mock Session for save_listings: INSERT RETURNING + SAVEPOINTs as no-op.""" + row = MagicMock() + row.id = listing_id + row.inserted = inserted + + result = MagicMock() + result.fetchone.return_value = row + + db = MagicMock() + db.execute.return_value = result + + @contextmanager + def _nested(): + yield MagicMock() + + db.begin_nested.side_effect = _nested + return db + + +def _base_lot(**overrides) -> ScrapedLot: + defaults = { + "source": "avito", + "source_url": "https://www.avito.ru/ekaterinburg/kvartiry/1-k_11111", + "source_id": "11111", + "price_rub": 4_000_000, + } + defaults.update(overrides) + return ScrapedLot(**defaults) + + +# --------------------------------------------------------------------------- +# Part 2: fingerprint↔alias sync +# --------------------------------------------------------------------------- + + +class TestFingerprintAliasSync: + """Two listings with the same address → exactly 1 house and 1 alias key. + + Scenario: two scrapers submit the same street address but with coords that + differ beyond the 4-decimal rounding boundary — they produce different + fingerprints. The matcher must still resolve to the same canonical house + and NOT create a second alias for the same normalized_address. + """ + + def test_same_normalized_address_resolves_to_same_house_via_tier2b(self): + """First call misses all tiers → creates house + alias (fp=fp1). + Second call with same address but different coords: + - Tier 2a: fp2 ≠ fp1 → miss + - Tier 2b: normalized_address hit → returns existing house_id + Result: same house_id, NOT a new house row. + """ + address = "Екатеринбург, ул. Ленина, 5" + lat1, lon1 = 56.838000, 60.594600 # yields fp1 + lat2, lon2 = 56.839100, 60.594600 # 4dp rounds to 56.8391 ≠ 56.8380 → fp2 ≠ fp1 + + fp1 = address_fingerprint(address, lat1, lon1) + fp2 = address_fingerprint(address, lat2, lon2) + assert fp1 != fp2, "precondition: coords produce different fingerprints" + + # ── First call: all tiers miss → INSERT new house ─────────────────── + db1 = _make_db( + [ + None, # pg_advisory_xact_lock + None, # house_sources miss (Tier 1) + None, # fingerprint miss (Tier 2a) + None, # normalized_address miss (Tier 2b) — first time + None, # geo miss (Tier 3; no lat/lon provided here) + {"id": 7}, # INSERT RETURNING id + None, # _upsert_house_source + None, # _insert_alias + ] + ) + house_id_1, _, method_1 = match_or_create_house( + db1, + "avito", + "ext-001", + address=address, + lat=lat1, + lon=lon1, + ) + assert method_1 == "new" + assert house_id_1 == 7 + + # ── Second call (same address, different coords) ──────────────────── + # Tier 2a misses (fp2 ≠ stored fp1), Tier 2b hits on normalized_address. + db2 = _make_db( + [ + None, # pg_advisory_xact_lock + None, # house_sources miss (Tier 1) + None, # fingerprint miss (Tier 2a — fp2 not registered yet) + {"house_id": 7}, # normalized_address hit (Tier 2b) + None, # _upsert_house_source + None, # _insert_alias (register fp2) + ] + ) + house_id_2, _, method_2 = match_or_create_house( + db2, + "cian", + "ext-002", + address=address, + lat=lat2, + lon=lon2, + ) + assert ( + method_2 == "fingerprint" + ), f"Expected 'fingerprint' (Tier 2b match), got {method_2!r}" + assert ( + house_id_2 == 7 + ), f"Expected same house_id=7, got {house_id_2} — second scraper created a duplicate" + + def test_same_normalized_address_tier2a_hits_directly_when_fp_matches(self): + """When the fingerprint IS already registered (same address + coords within 11 m), + Tier 2a hits directly and normalized_address fallback is not reached.""" + address = "ул Мира 10" + lat, lon = 56.84, 60.59 + + db = _make_db( + [ + None, # pg_advisory_xact_lock + None, # house_sources miss + {"house_id": 99}, # fingerprint hit (Tier 2a) + None, # _upsert_house_source + None, # _insert_alias + ] + ) + house_id, conf, method = match_or_create_house( + db, + "yandex", + "ext-yz", + address=address, + lat=lat, + lon=lon, + ) + assert method == "fingerprint" + assert house_id == 99 + assert conf == 0.9 + + def test_insert_alias_upsert_keeps_fingerprint_in_sync(self): + """_insert_alias with ON CONFLICT DO UPDATE ensures the stored fingerprint + is refreshed, not left stale from the first writer. + + Verify the emitted SQL contains 'DO UPDATE SET' (not 'DO NOTHING'). + """ + db = MagicMock() + result = MagicMock() + db.execute.return_value = result + + _insert_alias( + db, + house_id=5, + address="ул. Ленина 5", + fp="aabbccdd" * 4, + source="avito", + ) + + assert db.execute.called + sql_str = str(db.execute.call_args[0][0]) + assert ( + "DO UPDATE SET" in sql_str + ), "_insert_alias must use ON CONFLICT DO UPDATE SET fingerprint, not DO NOTHING" + assert "fingerprint" in sql_str + + def test_two_listings_same_address_exactly_one_alias_key(self): + """Integration check: normalize_address produces identical key for two scrapers + submitting the same address in different formats, so at most one alias row can + exist (UNIQUE normalized_address constraint holds). + + This test verifies the normalization layer — the DB constraint is structural. + addr_a uses the abbreviated form 'ул.' which expands to 'улица'; + addr_b already uses the full form. Both should normalize identically. + """ + addr_a = "ул. Ленина 5" # abbreviated 'ул.' → expands to 'улица' + addr_b = "улица ленина 5" # already expanded, lowercase + assert normalize_address(addr_a) == normalize_address(addr_b), ( + "normalize_address must produce identical key for both address forms " + "so the UNIQUE(normalized_address) constraint prevents duplicate aliases" + ) + + +# --------------------------------------------------------------------------- +# Part 3: single SAVEPOINT level — batch isolation +# --------------------------------------------------------------------------- + + +class TestSingleSavepointBatchIsolation: + """save_listings: one failing lot → siblings persist; no orphan SAVEPOINT.""" + + @pytest.fixture(autouse=True) + def _patch_matching(self): + """Stub matching so save_listings tests don't hit real DB queries.""" + with ( + patch("app.services.scrapers.base.match_or_create_house") as m_house, + patch("app.services.scrapers.base.upsert_listing_source") as m_link, + ): + m_house.return_value = (101, 1.0, "new") + m_link.return_value = None + self.m_house = m_house + self.m_link = m_link + yield + + def test_failing_lot_does_not_abort_batch(self, caplog): + """Lot whose match_or_create_house raises → that lot's hook is rolled back + (outer SAVEPOINT), but other lots in the batch are fully inserted. + + No whole-batch rollback, no orphan SAVEPOINTs. + """ + # First lot: house match succeeds; second lot: house match raises + self.m_house.side_effect = [ + (10, 1.0, "new"), + RuntimeError("simulated house-match DB error"), + (12, 1.0, "new"), + ] + + db = _mock_save_db(inserted=True, listing_id=50) + + lots = [ + _base_lot( + source_id="ok-1", + address="Екатеринбург, Ленина 5", + lat=56.84, + lon=60.59, + ), + _base_lot( + source_id="fail-1", + address="Екатеринбург, Мира 10", + lat=56.85, + lon=60.60, + ), + _base_lot( + source_id="ok-2", + address="Екатеринбург, Куйбышева 3", + lat=56.83, + lon=60.58, + ), + ] + + with caplog.at_level(logging.WARNING, logger="app.services.scrapers.base"): + inserted, _updated = save_listings(db, lots) + + # All 3 listings INSERT'd (listing INSERT is outside any SAVEPOINT) + assert inserted == 3 + + # House match attempted on all 3 (all have address+coords) + assert self.m_house.call_count == 3 + + # upsert_listing_source: lots ok-1 and ok-2 succeed; fail-1's hook was aborted + # by the outer SAVEPOINT so link not called for it. + assert self.m_link.call_count == 2 + + # Warning logged for the failing lot + warns = [r for r in caplog.records if "match_failed" in r.getMessage()] + assert len(warns) == 1 + + def test_no_begin_nested_inside_link_listing_to_house(self): + """Verify that _link_listing_to_house does NOT open its own begin_nested. + + The outer per-row SAVEPOINT (one call per lot in save_listings) is the only + SAVEPOINT level allowed. If begin_nested is called more than N lots times, + that indicates a double-nesting. + """ + db = _mock_save_db(inserted=True, listing_id=77) + lots = [_base_lot(source_id=str(i)) for i in range(3)] + + save_listings(db, lots) + + # save_listings calls begin_nested once per lot for the snapshot + # and once per lot for the hook — so 2 * len(lots) total. + # If _link_listing_to_house opens an INNER begin_nested too, + # the count would be > 2 * len(lots). + # (snapshot SAVEPOINT + hook SAVEPOINT = 2 per lot, no extra inner) + assert db.begin_nested.call_count == 2 * len(lots), ( + f"Expected exactly 2 begin_nested per lot (snapshot + hook outer), " + f"got {db.begin_nested.call_count} — inner SAVEPOINT may still be present" + ) + + def test_whole_batch_not_rolled_back_on_single_hook_failure(self, caplog): + """When one lot's hook fails, db.commit() is still called (batch persisted).""" + self.m_house.side_effect = RuntimeError("boom") + + db = _mock_save_db(inserted=True, listing_id=10) + lots = [_base_lot(source_id="only-one", address="x", lat=56.8, lon=60.5)] + + with caplog.at_level(logging.WARNING, logger="app.services.scrapers.base"): + save_listings(db, lots) + + # db.commit() called despite the hook failure — no whole-batch rollback + db.commit.assert_called_once() + db.rollback.assert_not_called() diff --git a/tradein-mvp/backend/tests/test_base_save_listings.py b/tradein-mvp/backend/tests/test_base_save_listings.py index 440c7386..9d2d8a26 100644 --- a/tradein-mvp/backend/tests/test_base_save_listings.py +++ b/tradein-mvp/backend/tests/test_base_save_listings.py @@ -437,11 +437,23 @@ def test_save_listings_matching_hook_dedup_hash_fallback(_patch_matching): assert kwargs["ext_id"] == lot.compute_dedup_hash() -# ── Test 13: house_match failure (e.g. address normalization) still upserts source ─ +# ── Test 13: house_match failure aborts the hook (single SAVEPOINT level) ───── -def test_save_listings_house_match_failure_still_upserts_listing_source(_patch_matching): - """If match_or_create_house raises, listing_sources still gets upserted.""" +def test_save_listings_house_match_failure_aborts_hook(_patch_matching, caplog): + """If match_or_create_house raises, the outer per-row SAVEPOINT rolls back the + entire hook call (_link_listing_to_house). upsert_listing_source is NOT called + because there is no inner SAVEPOINT to shelter it — a single SAVEPOINT level + (per backend.md) means one failure = one hook roll-back. The listings INSERT + that happened before the SAVEPOINT is unaffected (inserted counter still 1). + + This is a behaviour change from the old double-SAVEPOINT design where house + failure was silently swallowed and listing_sources was still upserted. With a + single SAVEPOINT the outer except at save_listings:386 logs match_failed and + continues to the next lot. (#849 Part 3) + """ + import logging + _patch_matching["house"].side_effect = RuntimeError("address parse boom") db = _mock_db(inserted=True, listing_id=88) @@ -452,15 +464,16 @@ def test_save_listings_house_match_failure_still_upserts_listing_source(_patch_m lon=60.6, ) - inserted, _ = save_listings(db, [lot]) + with caplog.at_level(logging.WARNING, logger="app.services.scrapers.base"): + inserted, _ = save_listings(db, [lot]) assert inserted == 1 - # House match attempted but failed — listing_source still recorded + # House match attempted — hook aborted, listing_source NOT recorded assert _patch_matching["house"].call_count == 1 - assert _patch_matching["link"].call_count == 1 - kwargs = _patch_matching["link"].call_args.kwargs - # source_data should not carry a house_id since match failed - assert kwargs["source_data"] is None + assert _patch_matching["link"].call_count == 0 + # Outer except logs the failure + warns = [r for r in caplog.records if "match_failed" in r.getMessage()] + assert len(warns) == 1 # ── Test 14: empty list → matching service not called ─────────────────────── diff --git a/tradein-mvp/backend/tests/test_matching.py b/tradein-mvp/backend/tests/test_matching.py index 51c38bd9..ae527f0c 100644 --- a/tradein-mvp/backend/tests/test_matching.py +++ b/tradein-mvp/backend/tests/test_matching.py @@ -38,14 +38,14 @@ def test_normalize_expands_pr_abbreviation(): assert "100" in result -def test_normalize_expands_prkT_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() + assert "кт" not in result.split() -def test_normalize_prkT_no_leftover_kt(): +def test_normalize_prkt_no_leftover_kt(): """пр-кт must expand to 'проспект' cleanly — no leftover 'кт' token.""" result = normalize_address("пр-кт Ленина 10") tokens = result.split() @@ -147,19 +147,25 @@ 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([ - None, # pg_advisory_xact_lock - {'id': 42}, # cadastral match - None, # _upsert_house_source (INSERT ... ON CONFLICT) - ]) + db = _make_db( + [ + None, # pg_advisory_xact_lock + {"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', + 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' + assert method == "cadastr_exact" def test_match_house_tier1_source_exact(): @@ -171,78 +177,115 @@ def test_match_house_tier1_source_exact(): """ from app.services.matching.houses import match_or_create_house - db = _make_db([ - None, # pg_advisory_xact_lock - {'house_id': 7}, # house_sources hit - ]) - house_id, conf, method = match_or_create_house( - db, 'avito', 'ext-999', - address='пр Мира 10', lat=56.8, lon=60.5, + db = _make_db( + [ + None, # pg_advisory_xact_lock + {"house_id": 7}, # house_sources hit + ] + ) + 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' + assert method == "source_exact" def test_match_house_tier2_fingerprint(): - """Tier 2: fingerprint match in house_address_aliases.""" + """Tier 2a: fingerprint match in house_address_aliases (exact fp hit). + + After Tier 2a hit, _insert_alias is called to ensure fp is kept in sync + (#849 Part 2 — alias sync). + """ from app.services.matching.houses import match_or_create_house - db = _make_db([ - None, # pg_advisory_xact_lock - None, # house_sources miss - {'house_id': 15}, # fingerprint hit - None, # _upsert_house_source - ]) + db = _make_db( + [ + None, # pg_advisory_xact_lock + None, # house_sources miss + {"house_id": 15}, # fingerprint hit (Tier 2a) + None, # _upsert_house_source + None, # _insert_alias (sync fingerprint — Part 2) + ] + ) house_id, conf, method = match_or_create_house( - db, 'yandex', 'ext-123', - address='Тестовая 1', lat=56.84, lon=60.60, + db, + "yandex", + "ext-123", + address="Тестовая 1", + lat=56.84, + lon=60.60, ) assert house_id == 15 assert conf == 0.9 - assert method == 'fingerprint' + assert method == "fingerprint" def test_match_house_tier3_geo(): - """Tier 3: geo-proximity match.""" + """Tier 3: geo-proximity match. + + Tier 2b (normalized_address fallback) is tried after Tier 2a miss. + """ from app.services.matching.houses import match_or_create_house - db = _make_db([ - None, # pg_advisory_xact_lock - None, # house_sources miss - None, # fingerprint miss - {'id': 22, 'dist': 15.5}, # geo hit - None, # _upsert_house_source - None, # _insert_alias - ]) + db = _make_db( + [ + None, # pg_advisory_xact_lock + None, # house_sources miss + None, # fingerprint miss (Tier 2a) + None, # normalized_address miss (Tier 2b — Part 2) + {"id": 22, "dist": 15.5}, # geo hit (Tier 3) + 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, + db, + "n1", + "ext-456", + address="Новая 3", + lat=56.83, + lon=60.59, ) assert house_id == 22 assert conf == 0.7 - assert method == 'geo_proximity' + assert method == "geo_proximity" def test_match_house_new(): - """All tiers miss → new house created.""" + """All tiers miss → new house created. + + Tier 2b (normalized_address fallback) is tried after Tier 2a miss. + """ from app.services.matching.houses import match_or_create_house - db = _make_db([ - None, # pg_advisory_xact_lock - None, # house_sources miss - None, # fingerprint miss - None, # geo miss - {'id': 99}, # INSERT RETURNING id - None, # _upsert_house_source - None, # _insert_alias - ]) + db = _make_db( + [ + None, # pg_advisory_xact_lock + None, # house_sources miss + None, # fingerprint miss (Tier 2a) + None, # normalized_address miss (Tier 2b — Part 2) + 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, + db, + "avito", + "new-ext", + address="Новостройка 1", + lat=56.85, + lon=60.61, ) assert house_id == 99 assert conf == 1.0 - assert method == 'new' + assert method == "new" def test_match_house_advisory_lock_called_first(): @@ -251,27 +294,33 @@ def test_match_house_advisory_lock_called_first(): (finding #1 from 2026-05-24 audit). PR #501.""" from app.services.matching.houses import match_or_create_house - db = _make_db([ - None, # pg_advisory_xact_lock - {'house_id': 7}, # house_sources Tier 1 hit (short path) - ]) + db = _make_db( + [ + None, # pg_advisory_xact_lock + {"house_id": 7}, # house_sources Tier 1 hit (short path) + ] + ) match_or_create_house( - db, 'avito', 'regress-001', - address='Тестовая 1', lat=56.8, lon=60.5, + db, + "avito", + "regress-001", + address="Тестовая 1", + lat=56.8, + lon=60.5, ) # First execute MUST be the lock acquisition, with bind {'fp': }. first_call = db.execute.call_args_list[0] - sql_obj = first_call[0][0] # TextClause - bind = first_call[0][1] # dict + sql_obj = first_call[0][0] # TextClause + bind = first_call[0][1] # dict - assert 'pg_advisory_xact_lock' in str(sql_obj), ( - f'first execute must be advisory lock, got: {sql_obj}' - ) - assert 'fp' in bind, f"lock bind must include fp, got: {bind}" - assert isinstance(bind['fp'], str) and len(bind['fp']) == 32, ( - f"fp must be 32-char sha256 hex, got: {bind.get('fp')!r}" - ) + assert "pg_advisory_xact_lock" in str( + sql_obj + ), f"first execute must be advisory lock, got: {sql_obj}" + assert "fp" in bind, f"lock bind must include fp, got: {bind}" + assert ( + isinstance(bind["fp"], str) and len(bind["fp"]) == 32 + ), f"fp must be 32-char sha256 hex, got: {bind.get('fp')!r}" # --------------------------------------------------------------------------- @@ -282,48 +331,66 @@ def test_match_house_advisory_lock_called_first(): 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', + 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' + 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, + 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' + 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 - ]) + 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', + db, + "cian", + "cian-999", + house_id=10, + description_minhash="abc123deadbeef", ) assert listing_id == 55 assert conf == 0.85 - assert method == 'minhash' + assert method == "minhash" def test_match_listing_tier3_composite(): @@ -332,32 +399,44 @@ def test_match_listing_tier3_composite(): """ 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 - ]) + 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, + 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' + 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, + 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' + assert method == "new" # --------------------------------------------------------------------------- @@ -380,13 +459,13 @@ def test_field_priority_sources_are_lists_or_known_string_rules(): 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}' - ) + 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}' - ) + 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(): @@ -397,5 +476,5 @@ def test_update_canonical_fields_is_callable(): """ db = MagicMock() # Should not raise — no-op stub returns None - result = update_canonical_fields(db, listing_id=1, ext_source='cian', lot_data=object()) + result = update_canonical_fields(db, listing_id=1, ext_source="cian", lot_data=object()) assert result is None