From 8e4dea74dd7f4444260bda80f9ea55b7e073bdd5 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sat, 23 May 2026 17:10:07 +0300 Subject: [PATCH] =?UTF-8?q?feat(tradein):=20estimator=20=E2=80=94=20Yandex?= =?UTF-8?q?=20Valuation=20as=20on-demand=206th=20source?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage 8 of YandexRealtyScraper v1 (Wave 6). Adds Yandex Valuation tool (anonymous, no cookies) as on-demand enrichment source in estimate_quality. Mirrors the Avito IMV pattern with 24h cache, graceful degradation, and idempotent history persistence. estimator.py: - _yandex_valuation_cache_key(address, category, type) -> sha256 - _get_or_fetch_yandex_valuation_cached(db, address, ...) -> cache via external_valuations(source='yandex_valuation'). Any error -> None + log warning + estimator continues. - _save_yandex_history_items(db, result) -> INSERT into house_placement_history(source='yandex_valuation') with synthesized ext_item_id (sha256 of address+date+area+floor+prices, first 32 chars). Idempotent via UNIQUE (source, ext_item_id). house_id stays NULL - estimator doesn't compute target_house_id yet (matching pipeline TBD). - estimate_quality: call after IMV block, BEFORE deals fetch. Adds 'yandex_valuation' to sources_used if Yandex returned a result. Tests: 11 unit tests (cache key determinism, hit/miss, fetch error, empty result, history item save / dedup / partial error). All pass. Ruff clean (no new errors introduced). Constants: YANDEX_VALUATION_CACHE_TTL_HOURS=24, default category=APARTMENT, default type=SELL. --- tradein-mvp/backend/app/services/estimator.py | 207 ++++++++++++++++++ .../test_estimator_yandex_integration.py | 197 +++++++++++++++++ 2 files changed, 404 insertions(+) create mode 100644 tradein-mvp/backend/tests/test_estimator_yandex_integration.py diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index 8a053c14..7b2827a3 100644 --- a/tradein-mvp/backend/app/services/estimator.py +++ b/tradein-mvp/backend/app/services/estimator.py @@ -19,6 +19,7 @@ from __future__ import annotations +import hashlib import json import logging from datetime import UTC, datetime, timedelta @@ -38,6 +39,10 @@ from app.services.scrapers.avito_imv import ( evaluate_via_imv, save_imv_evaluation, ) +from app.services.scrapers.yandex_valuation import ( + YandexValuationResult, + YandexValuationScraper, +) logger = logging.getLogger(__name__) @@ -94,6 +99,10 @@ def _repair_coefficient(repair_state: str | None) -> float: # ── Avito IMV cache lookup (Stage 3) ──────────────────────────────────────── IMV_CACHE_TTL_HOURS = 24 +YANDEX_VALUATION_CACHE_TTL_HOURS = 24 +YANDEX_VALUATION_DEFAULT_CATEGORY = "APARTMENT" +YANDEX_VALUATION_DEFAULT_TYPE = "SELL" + async def _get_or_fetch_imv_cached( db: Session, @@ -200,6 +209,187 @@ async def _get_or_fetch_imv_cached( return None +# ── Yandex Valuation cache lookup (Stage 8) ───────────────────────────────── + +def _yandex_valuation_cache_key( + address: str, offer_category: str, offer_type: str +) -> str: + """SHA256 cache key for Yandex Valuation lookups.""" + payload = f"{address}|{offer_category}|{offer_type}" + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +async def _get_or_fetch_yandex_valuation_cached( + db: Session, + *, + address: str, + offer_category: str = YANDEX_VALUATION_DEFAULT_CATEGORY, + offer_type: str = YANDEX_VALUATION_DEFAULT_TYPE, +) -> YandexValuationResult | None: + """Cached Yandex Valuation lookup. TTL 24h via external_valuations table. + + Returns None on any error / cache miss + fetch failure — caller continues + without Yandex enrichment (graceful degradation). + """ + cache_key = _yandex_valuation_cache_key(address, offer_category, offer_type) + + # Cache lookup + try: + cached = db.execute( + text( + """ + SELECT raw_payload, fetched_at + FROM external_valuations + WHERE source = 'yandex_valuation' + AND cache_key = :ck + AND expires_at > NOW() + ORDER BY fetched_at DESC + LIMIT 1 + """ + ), + {"ck": cache_key}, + ).mappings().first() + except Exception as e: + logger.warning("yandex_valuation: cache lookup failed: %s", e) + cached = None + + if cached is not None and cached.get("raw_payload"): + try: + payload_dict = ( + cached["raw_payload"] + if isinstance(cached["raw_payload"], dict) + else json.loads(cached["raw_payload"]) + ) + logger.info( + "yandex_valuation: cache HIT key=%s items=%d", + cache_key[:8], + len(payload_dict.get("history_items", [])), + ) + return YandexValuationResult.model_validate(payload_dict) + except Exception as e: + logger.warning("yandex_valuation: cache deserialize failed — refetching: %s", e) + + # Fresh fetch + try: + async with YandexValuationScraper() as scraper: + result = await scraper.fetch_house_history( + address=address, + offer_category=offer_category, + offer_type=offer_type, + ) + except Exception as e: + logger.warning( + "yandex_valuation: fetch failed — estimator продолжает без Yandex: %s", e + ) + return None + + if result is None: + logger.info("yandex_valuation: empty result for address=%s", address[:60]) + return None + + # Save to cache (UPSERT on (source, cache_key)) + try: + db.execute( + text( + """ + INSERT INTO external_valuations ( + source, cache_key, address, + raw_payload, + fetched_at, expires_at + ) VALUES ( + 'yandex_valuation', :ck, :addr, + CAST(:payload AS jsonb), + NOW(), NOW() + (:ttl_hours || ' hours')::interval + ) + ON CONFLICT (source, cache_key) DO UPDATE + SET raw_payload = EXCLUDED.raw_payload, + fetched_at = NOW(), + expires_at = NOW() + (:ttl_hours || ' hours')::interval + """ + ), + { + "ck": cache_key, + "addr": address, + "payload": json.dumps(result.model_dump(mode="json"), ensure_ascii=False), + "ttl_hours": YANDEX_VALUATION_CACHE_TTL_HOURS, + }, + ) + db.commit() + logger.info( + "yandex_valuation: fresh fetch saved key=%s items=%d", + cache_key[:8], + len(result.history_items), + ) + except Exception as e: + logger.warning("yandex_valuation: cache save failed (continuing): %s", e) + db.rollback() + + return result + + +def _save_yandex_history_items( + db: Session, + result: YandexValuationResult, +) -> int: + """Persist history items to house_placement_history. Returns saved count. + + house_id stays NULL — estimator doesn't compute target_house_id yet. + Idempotent via UNIQUE (source, ext_item_id); we synthesize ext_item_id from + (address|date|area|floor) hash since Yandex history items don't carry an + explicit ID. + """ + saved = 0 + for item in result.history_items: + # Synthesize stable ext_item_id (no native ID in valuation page) + ext_seed = ( + f"{result.address}|{item.publish_date}|{item.area_m2}|{item.floor}|" + f"{item.start_price}|{item.last_price}" + ) + ext_item_id = hashlib.sha256(ext_seed.encode("utf-8")).hexdigest()[:32] + try: + db.execute( + text( + """ + INSERT INTO house_placement_history ( + source, ext_item_id, + rooms, area_m2, floor, + start_price, start_price_date, + last_price, last_price_date, + exposure_days, + raw_payload + ) VALUES ( + 'yandex_valuation', :ext_id, + :rooms, :area, :floor, + :start_price, :publish_date, + :last_price, :publish_date, + :exposure, + CAST(:raw AS jsonb) + ) + ON CONFLICT (source, ext_item_id) DO NOTHING + """ + ), + { + "ext_id": ext_item_id, + "rooms": item.rooms, + "area": item.area_m2, + "floor": item.floor, + "start_price": item.start_price, + "last_price": item.last_price, + "publish_date": item.publish_date, + "exposure": item.exposure_days, + "raw": json.dumps(item.model_dump(mode="json"), ensure_ascii=False), + }, + ) + saved += 1 + except Exception as e: + logger.warning("yandex_valuation: failed to save history item: %s", e) + db.rollback() + continue + if saved: + db.commit() + return saved + + # ── Public ─────────────────────────────────────────────────────────────────── async def estimate_quality( payload: TradeInEstimateInput, db: Session @@ -341,6 +531,21 @@ async def estimate_quality( if imv_eval is not None: sources_used_pre = sorted(set(sources_used_pre) | {"avito_imv"}) + # ── Stage 8: Yandex Valuation as on-demand source (anonymous, cached 24h) ── + yandex_val: YandexValuationResult | None = None + if geo is not None and geo.full_address: + yandex_val = await _get_or_fetch_yandex_valuation_cached( + db, address=geo.full_address, + ) + if yandex_val is not None: + sources_used_pre = sorted(set(sources_used_pre) | {"yandex_valuation"}) + saved_hist = _save_yandex_history_items(db, yandex_val) + logger.info( + "yandex_valuation: history items processed=%d saved=%d" + " (house_id=NULL — matching deferred)", + len(yandex_val.history_items), saved_hist, + ) + # 5. Deals — фактические сделки за период deals = _fetch_deals( db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2, @@ -454,6 +659,8 @@ async def estimate_quality( sources_used = sorted({lot.source for lot in analogs_lots if lot.source}) if imv_eval is not None: sources_used = sorted(set(sources_used) | {"avito_imv"}) + if yandex_val is not None: + sources_used = sorted(set(sources_used) | {"yandex_valuation"}) freshness_min = _compute_freshness_minutes(listings_clean) return AggregatedEstimate( diff --git a/tradein-mvp/backend/tests/test_estimator_yandex_integration.py b/tradein-mvp/backend/tests/test_estimator_yandex_integration.py new file mode 100644 index 00000000..d5c4db23 --- /dev/null +++ b/tradein-mvp/backend/tests/test_estimator_yandex_integration.py @@ -0,0 +1,197 @@ +"""Tests for Yandex Valuation integration in estimator.py.""" +import os + +# Settings requires DATABASE_URL at init time. Set dummy DSN before any app import. +os.environ.setdefault("DATABASE_URL", "postgresql://test:test@localhost/test_db") + +from datetime import date +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from app.services.estimator import ( + YANDEX_VALUATION_CACHE_TTL_HOURS, + _get_or_fetch_yandex_valuation_cached, + _save_yandex_history_items, + _yandex_valuation_cache_key, +) +from app.services.scrapers.yandex_valuation import ( + ValuationHistoryItem, + ValuationHouseMeta, + YandexValuationResult, +) + + +def _sample_result(address: str = "Екатеринбург, ул. Учителей, 18") -> YandexValuationResult: + return YandexValuationResult( + address=address, + offer_category="APARTMENT", + offer_type="SELL", + page=1, + source_url=( + f"https://realty.yandex.ru/otsenka-kvartiry-po-adresu-onlayn/?address={address}" + ), + house=ValuationHouseMeta( + year_built=1981, + total_floors=9, + house_type="panel", + ceiling_height=2.5, + has_lift=True, + total_objects=12, + ), + history_items=[ + ValuationHistoryItem( + area_m2=36.8, + rooms=1, + floor=8, + start_price=4_500_000, + start_price_per_m2=122_283, + last_price=4_400_000, + last_price_per_m2=119_566, + publish_date=date(2026, 2, 17), + exposure_days=96, + status="В продаже", + ), + ValuationHistoryItem( + area_m2=64.4, + rooms=2, + floor=1, + start_price=6_580_000, + start_price_per_m2=102_174, + last_price=6_100_000, + last_price_per_m2=94_721, + publish_date=date(2026, 2, 2), + exposure_days=111, + status="В продаже", + ), + ], + ) + + +def test_cache_key_deterministic(): + k1 = _yandex_valuation_cache_key("addr X", "APARTMENT", "SELL") + k2 = _yandex_valuation_cache_key("addr X", "APARTMENT", "SELL") + assert k1 == k2 + assert len(k1) == 64 + + +def test_cache_key_different_for_different_inputs(): + a = _yandex_valuation_cache_key("addr X", "APARTMENT", "SELL") + b = _yandex_valuation_cache_key("addr Y", "APARTMENT", "SELL") + c = _yandex_valuation_cache_key("addr X", "HOUSE", "SELL") + d = _yandex_valuation_cache_key("addr X", "APARTMENT", "RENT") + assert len({a, b, c, d}) == 4 + + +def test_ttl_default_24h(): + assert YANDEX_VALUATION_CACHE_TTL_HOURS == 24 + + +@pytest.mark.asyncio +async def test_cache_hit_returns_deserialized_result(): + """When cache row exists and hasn't expired, return parsed YandexValuationResult.""" + db = MagicMock() + result = _sample_result() + db.execute.return_value.mappings.return_value.first.return_value = { + "raw_payload": result.model_dump(mode="json"), + "fetched_at": None, + } + with patch("app.services.estimator.YandexValuationScraper") as mock_scraper_cls: + out = await _get_or_fetch_yandex_valuation_cached(db, address=result.address) + mock_scraper_cls.assert_not_called() # no fetch on hit + assert out is not None + assert out.address == result.address + assert len(out.history_items) == 2 + + +@pytest.mark.asyncio +async def test_cache_miss_triggers_fetch_and_persist(): + """On cache miss, fetch via scraper + INSERT into external_valuations.""" + db = MagicMock() + db.execute.return_value.mappings.return_value.first.return_value = None # miss + fresh = _sample_result() + fake_scraper = MagicMock() + fake_scraper.__aenter__ = AsyncMock(return_value=fake_scraper) + fake_scraper.__aexit__ = AsyncMock(return_value=False) + fake_scraper.fetch_house_history = AsyncMock(return_value=fresh) + with patch("app.services.estimator.YandexValuationScraper", return_value=fake_scraper): + out = await _get_or_fetch_yandex_valuation_cached(db, address=fresh.address) + assert out is fresh + fake_scraper.fetch_house_history.assert_awaited_once() + # 2 DB calls: 1 SELECT (miss) + 1 INSERT/UPSERT + assert db.execute.call_count >= 2 + + +@pytest.mark.asyncio +async def test_fetch_error_returns_none_gracefully(): + db = MagicMock() + db.execute.return_value.mappings.return_value.first.return_value = None + fake_scraper = MagicMock() + fake_scraper.__aenter__ = AsyncMock(return_value=fake_scraper) + fake_scraper.__aexit__ = AsyncMock(return_value=False) + fake_scraper.fetch_house_history = AsyncMock(side_effect=RuntimeError("network")) + with patch("app.services.estimator.YandexValuationScraper", return_value=fake_scraper): + out = await _get_or_fetch_yandex_valuation_cached(db, address="addr") + assert out is None + + +@pytest.mark.asyncio +async def test_fetch_returns_none_propagates_none(): + db = MagicMock() + db.execute.return_value.mappings.return_value.first.return_value = None + fake_scraper = MagicMock() + fake_scraper.__aenter__ = AsyncMock(return_value=fake_scraper) + fake_scraper.__aexit__ = AsyncMock(return_value=False) + fake_scraper.fetch_house_history = AsyncMock(return_value=None) + with patch("app.services.estimator.YandexValuationScraper", return_value=fake_scraper): + out = await _get_or_fetch_yandex_valuation_cached(db, address="addr") + assert out is None + + +def test_save_history_items_inserts_each(): + db = MagicMock() + result = _sample_result() + saved = _save_yandex_history_items(db, result) + assert saved == 2 + # 2 INSERTs + 1 commit + assert db.execute.call_count == 2 + db.commit.assert_called_once() + + +def test_save_history_items_empty_no_commit(): + db = MagicMock() + result = YandexValuationResult( + address="x", + offer_category="APARTMENT", + offer_type="SELL", + page=1, + source_url="https://x", + house=ValuationHouseMeta(), + history_items=[], + ) + saved = _save_yandex_history_items(db, result) + assert saved == 0 + db.execute.assert_not_called() + db.commit.assert_not_called() + + +def test_save_history_items_ext_id_stable_across_calls(): + """Same result → same ext_item_id sequence → idempotent UPSERT.""" + db1 = MagicMock() + db2 = MagicMock() + result = _sample_result() + _save_yandex_history_items(db1, result) + _save_yandex_history_items(db2, result) + # Both got same call params (ext_id derived from same seed) + ext_ids_1 = [c.args[1]["ext_id"] for c in db1.execute.call_args_list] + ext_ids_2 = [c.args[1]["ext_id"] for c in db2.execute.call_args_list] + assert ext_ids_1 == ext_ids_2 + + +def test_save_history_items_db_error_continues(): + """One item failing doesn't abort the others.""" + db = MagicMock() + db.execute.side_effect = [RuntimeError("first row fails"), None] + result = _sample_result() + saved = _save_yandex_history_items(db, result) + assert saved == 1 # second item still saved