lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

fix(matching): Tier 2b alias sync + collapse double SAVEPOINT (#849) #870

Merged
bot-reviewer merged 1 commit from feat/849-matching-sync-savepoint into main 2026-05-31 08:05:49 +00:00
Conversation 0 Commits 1 Files changed 6 +867 -314
6 changed files with 867 additions and 314 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

286
tradein-mvp/backend/app/services/matching/houses.py
View file

@ -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,
},
)

179
tradein-mvp/backend/app/services/matching/listings.py
View file

@ -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,
},
)

42
tradein-mvp/backend/app/services/scrapers/base.py
View file

@ -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,

334
tradein-mvp/backend/tests/test_849_matching_savepoint.py Normal file
View file

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

31
tradein-mvp/backend/tests/test_base_save_listings.py
View file

@ -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 ───────────────────────

309
tradein-mvp/backend/tests/test_matching.py
View file

@ -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': <hex>}.
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