feat(tradein): hook matching в save_listings — fill listing_sources (PR I) #595
2 changed files with 340 additions and 19 deletions
|
|
@ -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:
|
||||
inserted += 1
|
||||
else:
|
||||
updated += 1
|
||||
listing_id: int | None = None
|
||||
if result is not None:
|
||||
listing_id = int(result.id)
|
||||
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,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
"""Создать mock Session с fetchone() → (inserted=<bool>,)."""
|
||||
def _mock_db(inserted: bool = True, listing_id: int = 42) -> MagicMock:
|
||||
"""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 = str(sql_obj)
|
||||
sql_text = _get_listings_insert_sql(db)
|
||||
|
||||
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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue