feat(tradein): hook matching в save_listings — fill listing_sources (PR I) (#595)

tradein-mvp/backend/app/services/scrapers/base.py

@@ -6,6 +6,8 @@
 - Sleep между запросами (anti-ban)
 - ScrapedLot — единая Pydantic схема всех источников
 - Запись в `listings` Postgres с дедупом по dedup_hash
+- После INSERT/UPDATE — hook в matching service для регистрации в
+  `listing_sources` и привязки к каноническому `houses` через `house_sources`.
 
 Каждый конкретный парсер (avito.py, cian.py, ...) наследуется от BaseScraper
 и реализует `_fetch_page()` и `_parse_listings()`.
@@ -27,21 +29,24 @@ from sqlalchemy import text
 from sqlalchemy.orm import Session
 from tenacity import retry, stop_after_attempt, wait_exponential
 
+from app.services.matching import match_or_create_house, upsert_listing_source
+
 logger = logging.getLogger(__name__)
 
 
 # ── Realistic browser User-Agents (Apr 2026 versions) ──────────────────────
+
 _USER_AGENTS: list[str] = [
     # Chrome on macOS
-    "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36",
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36",  # noqa: E501
     # Chrome on Windows
-    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36",
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36",  # noqa: E501
     # Firefox on macOS
     "Mozilla/5.0 (Macintosh; Intel Mac OS X 14.5; rv:127.0) Gecko/20100101 Firefox/127.0",
     # Safari on macOS
-    "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_5) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 Safari/605.1.15",
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_5) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 Safari/605.1.15",  # noqa: E501
     # Edge on Windows
-    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36 Edg/132.0.0.0",
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36 Edg/132.0.0.0",  # noqa: E501
 ]
 
 
@@ -207,6 +212,8 @@ def save_listings(db: Session, lots: list[ScrapedLot]) -> tuple[int, int]:
 
     inserted = 0
     updated = 0
+    matched = 0
+    match_failures = 0
 
     for lot in lots:
         ppm2 = lot.price_per_m2 or lot.compute_price_per_m2()
@@ -264,7 +271,7 @@ def save_listings(db: Session, lots: list[ScrapedLot]) -> tuple[int, int]:
                         bargain_allowed = EXCLUDED.bargain_allowed,
                         sale_type = EXCLUDED.sale_type,
                         metro_stations = EXCLUDED.metro_stations
-                RETURNING (xmax = 0) AS inserted
+                RETURNING id, (xmax = 0) AS inserted
                 """
             ),
             {
@@ -311,17 +318,44 @@ def save_listings(db: Session, lots: list[ScrapedLot]) -> tuple[int, int]:
             },
         ).fetchone()
 
-        if result is not None and result.inserted:
+        listing_id: int | None = None
-            inserted += 1
+        if result is not None:
-        else:
+            listing_id = int(result.id)
-            updated += 1
+            if result.inserted:
+                inserted += 1
+            else:
+                updated += 1
+
+        # ── Hook: link listing → house via matching service ─────────────
+        # Idempotent (UPSERT on listing_sources UNIQUE(ext_source, ext_id))
+        # and fault-tolerant: failure here MUST NOT abort the listings batch.
+        # Wrapped in SAVEPOINT so a failed match only rolls back its own work,
+        # not the surrounding INSERT (see .claude/rules/backend.md SAVEPOINT).
+        if listing_id is not None:
+            try:
+                with db.begin_nested():
+                    _link_listing_to_house(db, listing_id, lot)
+                matched += 1
+            except Exception as e:
+                # Best-effort hook: log and continue so the listings batch isn't aborted.
+                match_failures += 1
+                logger.warning(
+                    "save_listings:match_failed source=%s source_id=%s listing_id=%s: %s",
+                    lot.source,
+                    lot.source_id,
+                    listing_id,
+                    e,
+                )
 
     db.commit()
     logger.info(
-        "save_listings: source=%s inserted=%d updated=%d (total %d)",
+        "save_listings: source=%s inserted=%d updated=%d matched=%d "
+        "match_failures=%d (total %d)",
         lots[0].source if lots else "?",
         inserted,
         updated,
+        matched,
+        match_failures,
         len(lots),
     )
     return inserted, updated
@@ -332,3 +366,75 @@ def _to_json(value: Any) -> str:
     import json
 
     return json.dumps(value, ensure_ascii=False, default=str)
+
+
+def _link_listing_to_house(db: Session, listing_id: int, lot: ScrapedLot) -> None:
+    """Hook scraped listing into matching service: resolve house, upsert listing_sources.
+
+    Steps:
+      1. match_or_create_house() — find or create canonical houses row for this
+         listing's address/coords. Uses Tier 0-3 matching (cadastr → source_exact →
+         fingerprint → geo_proximity → new).
+      2. upsert_listing_source() — register (ext_source, ext_id) → listing_id in
+         listing_sources. Idempotent via UNIQUE (ext_source, ext_id).
+
+    The matching method recorded is 'source_link': scraper already deduped by
+    `dedup_hash` (INSERT … ON CONFLICT DO UPDATE), so this is a direct registration
+    rather than a fuzzy listing-level match.
+
+    ext_id source: lot.source_id if present, else dedup_hash (Yandex without
+    stable source_id falls back to URL-based dedup_hash — same hash on re-scrape).
+
+    Skips silently if:
+      - lot has no source_id AND no address/lat/lon (cannot match house anyway)
+
+    Raises on DB errors — caller wraps in try/except + SAVEPOINT.
+    """
+    ext_id = lot.source_id or lot.compute_dedup_hash()
+
+    # House resolution: needs at least address or (lat, lon). Skip otherwise —
+    # listing_sources requires listing_id but not house linkage, so still upsert.
+    house_id: int | None = None
+    if lot.address or (lot.lat is not None and lot.lon is not None):
+        # Use house_source/house_ext_id when scraper extracted them (Avito Houses
+        # catalog link, Cian newbuilding). Falls back to per-listing identity
+        # 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
+
+    upsert_listing_source(
+        db,
+        listing_id=listing_id,
+        ext_source=lot.source,
+        ext_id=str(ext_id),
+        method='source_link',
+        confidence=1.0,
+        price_rub=lot.price_rub,
+        area_m2=lot.area_m2,
+        floor=lot.floor,
+        rooms_count=lot.rooms,
+        source_url=lot.source_url,
+        source_data={'house_id': house_id} if house_id else None,
+    )

tradein-mvp/backend/tests/test_base_save_listings.py

@@ -5,19 +5,42 @@
 """
 
 import json
-from unittest.mock import MagicMock, call, patch
+from contextlib import contextmanager
+from unittest.mock import MagicMock, patch
 
 import pytest
 
 from app.services.scrapers.base import ScrapedLot, save_listings
 
+# ── Autouse fixture: patch matching service so all save_listings tests are deterministic ──
+
+
+@pytest.fixture(autouse=True)
+def _patch_matching():
+    """Stub matching service for all tests in this module.
+
+    save_listings() now calls match_or_create_house + upsert_listing_source after
+    each INSERT. Without patching, those would issue extra db.execute() calls with
+    real SQL (and MagicMock returns truthy mappings() that would simulate Tier 0
+    matches with fake house_ids). Tests assert via call_args_list[0] (the listings
+    INSERT) and call_count on the patched stubs to verify the hook fires.
+    """
+    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")  # (house_id, conf, method)
+        m_link.return_value = None
+        yield {"house": m_house, "link": m_link}
+
 
 # ── Helpers ──────────────────────────────────────────────────────────────────
 
 
-def _mock_db(inserted: bool = True) -> MagicMock:
+def _mock_db(inserted: bool = True, listing_id: int = 42) -> MagicMock:
-    """Создать mock Session с fetchone() → (inserted=<bool>,)."""
+    """Mock Session: INSERT listings RETURNING id, inserted=<bool>; SAVEPOINTs no-op."""
     row = MagicMock()
+    row.id = listing_id
     row.inserted = inserted
 
     result = MagicMock()
@@ -25,6 +48,13 @@ def _mock_db(inserted: bool = True) -> MagicMock:
 
     db = MagicMock()
     db.execute.return_value = result
+
+    # begin_nested() returns a context manager; SAVEPOINT semantics — no-op here
+    @contextmanager
+    def _nested():
+        yield MagicMock()
+
+    db.begin_nested.side_effect = _nested
     return db
 
 
@@ -80,12 +110,20 @@ def _cian_lot(**overrides) -> ScrapedLot:
 
 
 def _get_execute_params(db: MagicMock) -> dict:
-    """Получить params dict из первого db.execute() вызова."""
+    """Получить params dict из первого db.execute() вызова (INSERT listings)."""
     assert db.execute.called, "db.execute was not called"
-    _sql, params = db.execute.call_args.args
+    # First call is INSERT INTO listings ... RETURNING id, inserted.
+    # Subsequent calls are matching hook (match_or_create_house, listing_sources).
+    _sql, params = db.execute.call_args_list[0].args
     return params
 
 
+def _get_listings_insert_sql(db: MagicMock) -> str:
+    """Получить SQL-текст INSERT INTO listings (первый db.execute() вызов)."""
+    assert db.execute.called, "db.execute was not called"
+    return str(db.execute.call_args_list[0].args[0])
+
+
 # ── Test 1: Cian lot → все новые колонки попадают в params ──────────────────
 
 
@@ -133,7 +171,7 @@ def test_save_listings_avito_compat_cian_cols_are_null():
     db = _mock_db(inserted=True)
     lot = _base_lot(source="avito", source_id="987654")
 
-    inserted, updated = save_listings(db, [lot])
+    inserted, _updated = save_listings(db, [lot])
 
     assert inserted == 1
 
@@ -178,8 +216,7 @@ def test_save_listings_on_conflict_updates_cian_cols():
     assert updated == 1
 
     # Проверяем SQL-текст на наличие DO UPDATE SET с новыми колонками
-    sql_obj = db.execute.call_args.args[0]
+    sql_text = _get_listings_insert_sql(db)
-    sql_text = str(sql_obj)
 
     assert "cadastral_number = EXCLUDED.cadastral_number" in sql_text
     assert "description_minhash = EXCLUDED.description_minhash" in sql_text
@@ -240,7 +277,7 @@ def test_save_listings_sql_contains_all_new_columns():
 
     save_listings(db, [lot])
 
-    sql_text = str(db.execute.call_args.args[0])
+    sql_text = _get_listings_insert_sql(db)
 
     for col in (
         "living_area_m2",
@@ -258,3 +295,181 @@ def test_save_listings_sql_contains_all_new_columns():
         "metro_stations",
     ):
         assert col in sql_text, f"Column '{col}' missing from INSERT SQL"
+
+
+# ── Test 7: INSERT RETURNING id — used downstream by matching hook ──────────
+
+
+def test_save_listings_returning_id_for_matching_hook():
+    """INSERT … RETURNING id, (xmax = 0) — needs both columns for the hook."""
+    db = _mock_db(inserted=True, listing_id=12345)
+    lot = _base_lot()
+
+    save_listings(db, [lot])
+
+    sql_text = _get_listings_insert_sql(db)
+    assert "RETURNING id, (xmax = 0) AS inserted" in sql_text
+
+
+# ── Test 8: matching hook fires exactly once per saved listing ──────────────
+
+
+def test_save_listings_invokes_matching_hook_once_per_lot(_patch_matching):
+    """Each saved lot triggers exactly one upsert_listing_source call.
+
+    match_or_create_house may or may not be called (depends on address/lat/lon
+    presence). upsert_listing_source MUST always be called.
+    """
+    db = _mock_db(inserted=True, listing_id=777)
+    lots = [
+        _base_lot(source_id="a1"),
+        _base_lot(source_id="a2", address="Екатеринбург, Ленина 1", lat=56.84, lon=60.6),
+        _cian_lot(source_id="c1"),
+    ]
+
+    save_listings(db, lots)
+
+    # Hook fires once per lot
+    assert _patch_matching["link"].call_count == 3
+
+    # House resolution attempted only when address/coords present (2 of 3)
+    assert _patch_matching["house"].call_count == 2
+
+    # All upsert_listing_source calls use ext_source = listing.source
+    for idx, call_args in enumerate(_patch_matching["link"].call_args_list):
+        kwargs = call_args.kwargs
+        assert kwargs["listing_id"] == 777
+        assert kwargs["ext_source"] == lots[idx].source
+        assert kwargs["ext_id"] == lots[idx].source_id
+        assert kwargs["method"] == "source_link"
+        assert kwargs["confidence"] == 1.0
+
+
+# ── Test 9: matching hook receives full listing context ─────────────────────
+
+
+def test_save_listings_matching_hook_passes_listing_context(_patch_matching):
+    """upsert_listing_source receives price/area/floor/rooms/url from the lot."""
+    db = _mock_db(inserted=True, listing_id=555)
+    lot = _cian_lot(
+        source_id="cian-999",
+        price_rub=7_500_000,
+        area_m2=54.5,
+        floor=3,
+        rooms=2,
+    )
+
+    save_listings(db, [lot])
+
+    kwargs = _patch_matching["link"].call_args.kwargs
+    assert kwargs["listing_id"] == 555
+    assert kwargs["ext_source"] == "cian"
+    assert kwargs["ext_id"] == "cian-999"
+    assert kwargs["price_rub"] == 7_500_000
+    assert kwargs["area_m2"] == 54.5
+    assert kwargs["floor"] == 3
+    assert kwargs["rooms_count"] == 2
+    assert kwargs["source_url"] == lot.source_url
+
+
+# ── Test 10: matching hook idempotent on re-scrape (ON CONFLICT updated path) ─
+
+
+def test_save_listings_matching_hook_fires_on_update_path(_patch_matching):
+    """ON CONFLICT DO UPDATE (re-scrape) — hook still fires so last_seen_at refreshes."""
+    db = _mock_db(inserted=False, listing_id=12)  # updated, not new
+    lot = _base_lot()
+
+    save_listings(db, [lot])
+
+    assert _patch_matching["link"].call_count == 1
+
+
+# ── Test 11: matching failure logs warning but does not abort the batch ─────
+
+
+def test_save_listings_matching_failure_does_not_abort_batch(_patch_matching, caplog):
+    """When upsert_listing_source raises, the loop logs and continues."""
+    import logging
+
+    _patch_matching["link"].side_effect = RuntimeError("simulated DB blip")
+
+    db = _mock_db(inserted=True, listing_id=99)
+    lots = [_base_lot(source_id="a1"), _base_lot(source_id="a2")]
+
+    with caplog.at_level(logging.WARNING, logger="app.services.scrapers.base"):
+        inserted, updated = save_listings(db, lots)
+
+    # Both listings still INSERTed; only the hook failed
+    assert inserted == 2
+    assert updated == 0
+
+    # 2 warnings — one per lot
+    warns = [r for r in caplog.records if "match_failed" in r.getMessage()]
+    assert len(warns) == 2
+
+    # Hook attempted on both
+    assert _patch_matching["link"].call_count == 2
+
+
+# ── Test 12: lot without source_id falls back to dedup_hash as ext_id ───────
+
+
+def test_save_listings_matching_hook_dedup_hash_fallback(_patch_matching):
+    """Yandex-style lot without stable source_id → ext_id := dedup_hash."""
+    db = _mock_db(inserted=True, listing_id=44)
+    lot = _base_lot(
+        source="yandex",
+        source_url="https://realty.yandex.ru/offer/some-offer/",
+        source_id=None,
+        address="Екатеринбург, Малышева 51",
+        lat=56.84,
+        lon=60.6,
+    )
+
+    save_listings(db, [lot])
+
+    kwargs = _patch_matching["link"].call_args.kwargs
+    assert kwargs["ext_source"] == "yandex"
+    # ext_id should be the dedup_hash (sha256 hex), not None
+    assert kwargs["ext_id"]
+    assert len(kwargs["ext_id"]) == 64  # sha256 hex
+    assert kwargs["ext_id"] == lot.compute_dedup_hash()
+
+
+# ── Test 13: house_match failure (e.g. address normalization) still upserts source ─
+
+
+def test_save_listings_house_match_failure_still_upserts_listing_source(_patch_matching):
+    """If match_or_create_house raises, listing_sources still gets upserted."""
+    _patch_matching["house"].side_effect = RuntimeError("address parse boom")
+
+    db = _mock_db(inserted=True, listing_id=88)
+    lot = _base_lot(
+        source_id="x1",
+        address="Bad addr",
+        lat=56.84,
+        lon=60.6,
+    )
+
+    inserted, _ = save_listings(db, [lot])
+
+    assert inserted == 1
+    # House match attempted but failed — listing_source still 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
+
+
+# ── Test 14: empty list → matching service not called ───────────────────────
+
+
+def test_save_listings_empty_lots_skips_matching(_patch_matching):
+    """Empty input → no hook calls (verifies early return preserves behaviour)."""
+    db = _mock_db()
+    save_listings(db, [])
+
+    assert _patch_matching["house"].call_count == 0
+    assert _patch_matching["link"].call_count == 0