diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 18070bb7..f24f29d1 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -47,6 +47,7 @@ from app.services.scrapers.avito_houses import fetch_house_catalog, save_house_c from app.services.scrapers.base import ScrapedLot, save_listings from app.services.scrapers.browser_fetcher import BrowserFetcher from app.services.scrapers.yandex_realty import YandexRealtyScraper +from app.services.yandex_price_history import record_yandex_price_history logger = logging.getLogger(__name__) @@ -1114,6 +1115,7 @@ class YandexCitySweepCounters: lots_fetched: int = 0 lots_inserted: int = 0 lots_updated: int = 0 + price_history_rows: int = 0 address_attempted: int = 0 address_enriched: int = 0 address_failed: int = 0 @@ -1288,6 +1290,20 @@ async def run_yandex_city_sweep( inserted, updated = save_listings(db, anchor_lots, run_id=run_id) counters.lots_inserted += inserted counters.lots_updated += updated + # Price-history из gate price.previous/trend (graceful — провал + # истории НЕ должен ронять сейв листингов). + try: + counters.price_history_rows += record_yandex_price_history(db, anchor_lots) + except Exception as ph_exc: + logger.warning( + "yandex-sweep run_id=%d: price-history failed: %s", + run_id, + ph_exc, + ) + try: + db.rollback() + except Exception: + pass consecutive_failures = 0 logger.info( "yandex-sweep run_id=%d center-combos %s: fetched=%d ins=%d upd=%d", @@ -2167,6 +2183,7 @@ class YandexFullLoadCounters: unique_fetched: int = 0 saved_inserted: int = 0 saved_updated: int = 0 + price_history_rows: int = 0 errors_count: int = 0 def to_dict(self) -> dict[str, int]: @@ -2240,6 +2257,21 @@ async def run_yandex_full_load( counters.saved_inserted += inserted counters.saved_updated += updated counters.unique_fetched += len(lots) + # Price-history из gate price.previous/trend (graceful — провал истории + # НЕ должен ронять сейв листингов/бакета). + try: + counters.price_history_rows += record_yandex_price_history(db, lots) + except Exception as ph_exc: + logger.warning( + "yandex-full-load run_id=%d: price-history failed bucket=%s: %s", + run_id, + bucket_key, + ph_exc, + ) + try: + db.rollback() + except Exception: + pass done.add(bucket_key) scrape_runs.update_heartbeat( db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)} diff --git a/tradein-mvp/backend/app/services/yandex_price_history.py b/tradein-mvp/backend/app/services/yandex_price_history.py new file mode 100644 index 00000000..7665aad2 --- /dev/null +++ b/tradein-mvp/backend/app/services/yandex_price_history.py @@ -0,0 +1,184 @@ +"""Yandex offer price-history capture из gate-ответа. + +В отличие от Cian (отдельный detail-fetch priceChanges, см. cian_price_history.py), +Yandex gate-ответ уже несёт `price.previous` + `price.trend`, которые парсятся в +`ScrapedLot.price_previous_rub` / `price_trend`. Otsenka-скрапер НЕ нужен — gate достаточно. + +Эта функция вызывается из scrape_pipeline СРАЗУ ПОСЛЕ save_listings(те же lots): резолвит +listing_id по (source='yandex', source_id), читает последнюю по change_time точку +offer_price_history и идемпотентно дописывает наблюдение, если цена изменилась. + +Idempotent-логика (на каждый yandex-lot): + - история ПУСТА: + * price_previous задан И != price_rub → seed previous-точку (change_time = now-1d, + timestamp синтетический — gate реального времени смены не несёт), затем current-точку; + * иначе → только current-точку (now). + - история ЕСТЬ и latest_price != price_rub → одна current-точка (now) — зафиксировали смену. + - история ЕСТЬ и latest_price == price_rub → skip (цена не менялась). + +Per-lot SAVEPOINT: один битый lot не валит весь батч. db.commit() в конце. +psycopg v3 discipline: только CAST(:x AS type), никаких :x::type. Все timestamp'ы +считаются в Python и биндятся как параметры (DB-agnostic, без Postgres-only interval). +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime, timedelta + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.services.scrapers.base import ScrapedLot + +logger = logging.getLogger(__name__) + +_SOURCE = "yandex" + +# Резолв listing_id по стабильному (source, source_id) — у Yandex source_id = offer_id. +_RESOLVE_SQL = text( + """ + SELECT id + FROM listings + WHERE source = 'yandex' + AND source_id = CAST(:source_id AS text) + LIMIT 1 + """ +) + +# Последняя зафиксированная цена для листинга (по change_time DESC — индекс oph_listing_time_idx). +_LATEST_SQL = text( + """ + SELECT price_rub + FROM offer_price_history + WHERE listing_id = CAST(:listing_id AS bigint) + AND source = 'yandex' + ORDER BY change_time DESC + LIMIT 1 + """ +) + +# change_time bound как Python datetime — psycopg v3 адаптирует к timestamptz нативно, +# CAST не нужен (и был бы вреден: SQLite numeric-affinity CAST манглит ISO-строку). +_INSERT_SQL = text( + """ + INSERT INTO offer_price_history (listing_id, change_time, price_rub, source) + VALUES ( + CAST(:listing_id AS bigint), + :change_time, + CAST(:price_rub AS numeric), + 'yandex' + ) + """ +) + + +def _insert_point( + db: Session, + *, + listing_id: int, + price_rub: float, + change_time: datetime, +) -> None: + db.execute( + _INSERT_SQL, + { + "listing_id": listing_id, + "change_time": change_time, + "price_rub": price_rub, + }, + ) + + +def record_yandex_price_history(db: Session, lots: list[ScrapedLot]) -> int: + """Дописать offer_price_history по yandex-lots из gate price.previous/trend. + + Args: + db: SQLAlchemy session (caller-owned; commit делается здесь в конце). + lots: те же ScrapedLot, что были переданы в save_listings. Не-yandex и + lot'ы без source_id / без резолвящегося листинга — пропускаются. + + Returns: + Число фактически вставленных строк offer_price_history. + """ + if not lots: + return 0 + + now = datetime.now(UTC) + prev_ts = now - timedelta(days=1) + + inserted = 0 + skipped = 0 + errors = 0 + + for lot in lots: + if lot.source != _SOURCE: + continue + if not lot.source_id: + skipped += 1 + continue + + try: + with db.begin_nested(): + row = db.execute( + _RESOLVE_SQL, + {"source_id": lot.source_id}, + ).fetchone() + if row is None: + skipped += 1 + continue + listing_id = int(row[0]) + + latest = db.execute( + _LATEST_SQL, + {"listing_id": listing_id}, + ).fetchone() + + if latest is None: + # История пуста — seed (+ опционально previous-точка). + prev = lot.price_previous_rub + if prev is not None and prev != lot.price_rub: + _insert_point( + db, + listing_id=listing_id, + price_rub=prev, + change_time=prev_ts, + ) + inserted += 1 + _insert_point( + db, + listing_id=listing_id, + price_rub=lot.price_rub, + change_time=now, + ) + inserted += 1 + else: + latest_price = float(latest[0]) + if latest_price != float(lot.price_rub): + _insert_point( + db, + listing_id=listing_id, + price_rub=lot.price_rub, + change_time=now, + ) + inserted += 1 + else: + skipped += 1 + except Exception as exc: + # Per-lot SAVEPOINT откатился — продолжаем батч, не роняем остальные lot'ы. + errors += 1 + logger.warning( + "yandex_price_history: lot failed source_id=%s: %s", + lot.source_id, + exc, + ) + + db.commit() + logger.info( + "yandex_price_history: lots=%d inserted=%d skipped=%d errors=%d", + len(lots), + inserted, + skipped, + errors, + ) + return inserted diff --git a/tradein-mvp/backend/tests/test_yandex_price_history.py b/tradein-mvp/backend/tests/test_yandex_price_history.py new file mode 100644 index 00000000..9a6af431 --- /dev/null +++ b/tradein-mvp/backend/tests/test_yandex_price_history.py @@ -0,0 +1,238 @@ +"""Tests for record_yandex_price_history (gate price.previous/trend → offer_price_history). + +Behavioural tests run against an in-memory SQLite DB with a minimal listings + +offer_price_history schema. The service keeps all timestamp arithmetic in Python and +uses only CAST(:x AS type) (no :x::type, no Postgres-only interval), so the same SQL +exercises correctly on SQLite — letting us assert the stateful idempotent logic +(seed previous-point, current-point, no-op on unchanged price) without a live Postgres. + +Plus static checks that the emitted SQL keeps psycopg-v3 discipline. +""" + +from __future__ import annotations + +import inspect +import os +import re + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +import pytest +from sqlalchemy import create_engine, text +from sqlalchemy.orm import Session, sessionmaker + +from app.services import yandex_price_history as yph +from app.services.scrapers.base import ScrapedLot + +_SERVICE_SRC = inspect.getsource(yph) +# Concatenated text of every actual SQL statement the service emits (not the docstring, +# which may mention the :x::type anti-pattern in prose). +_SQL_TEXT = "\n".join( + str(getattr(stmt, "text", "")) for stmt in (yph._RESOLVE_SQL, yph._LATEST_SQL, yph._INSERT_SQL) +) + + +# ── Static: psycopg-v3 discipline ───────────────────────────────────────────── + + +def test_service_uses_cast_not_colon_colon() -> None: + """No :x::type shorthand in emitted SQL — only CAST(:x AS type) (psycopg-v3 rule).""" + # No bind-param cast shorthand (:param::type) in the real SQL statements. + assert not re.search(r":\w+::", _SQL_TEXT) + assert "CAST(:listing_id AS bigint)" in _SQL_TEXT + assert "CAST(:price_rub AS numeric)" in _SQL_TEXT + + +def test_service_inserts_with_yandex_source_tag() -> None: + assert "INSERT INTO offer_price_history" in _SERVICE_SRC + assert "'yandex'" in _SERVICE_SRC + + +# ── Behavioural: in-memory SQLite ───────────────────────────────────────────── + + +@pytest.fixture() +def db() -> Session: + """In-memory SQLite session with minimal listings + offer_price_history schema.""" + engine = create_engine("sqlite://", future=True) + with engine.begin() as conn: + conn.execute( + text( + """ + CREATE TABLE listings ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source TEXT NOT NULL, + source_id TEXT + ) + """ + ) + ) + conn.execute( + text( + """ + CREATE TABLE offer_price_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + listing_id INTEGER NOT NULL, + change_time TEXT NOT NULL, + price_rub NUMERIC NOT NULL, + source TEXT + ) + """ + ) + ) + sess = sessionmaker(bind=engine, future=True)() + try: + yield sess + finally: + sess.close() + + +def _seed_listing(db: Session, *, source: str, source_id: str) -> int: + db.execute( + text("INSERT INTO listings (source, source_id) VALUES (:s, :sid)"), + {"s": source, "sid": source_id}, + ) + db.commit() + row = db.execute( + text("SELECT id FROM listings WHERE source = :s AND source_id = :sid"), + {"s": source, "sid": source_id}, + ).fetchone() + assert row is not None + return int(row[0]) + + +def _lot( + *, + source_id: str, + price_rub: int, + price_previous_rub: int | None = None, + source: str = "yandex", +) -> ScrapedLot: + return ScrapedLot( + source=source, + source_url=f"https://realty.yandex.ru/offer/{source_id}/", + source_id=source_id, + price_rub=price_rub, + price_previous_rub=price_previous_rub, + ) + + +def _history(db: Session, listing_id: int) -> list[tuple[float, str]]: + rows = db.execute( + text( + "SELECT price_rub, change_time FROM offer_price_history " + "WHERE listing_id = :lid ORDER BY change_time ASC" + ), + {"lid": listing_id}, + ).fetchall() + return [(float(r[0]), str(r[1])) for r in rows] + + +def test_first_capture_with_previous_inserts_two_points(db: Session) -> None: + """price_previous задан и != price_rub → previous@-1d + current@now (2 строки).""" + lid = _seed_listing(db, source="yandex", source_id="111") + n = yph.record_yandex_price_history( + db, [_lot(source_id="111", price_rub=5_000_000, price_previous_rub=5_500_000)] + ) + assert n == 2 + hist = _history(db, lid) + assert len(hist) == 2 + # Chronological: previous-point first (earlier change_time), current second. + assert hist[0][0] == 5_500_000.0 # previous + assert hist[1][0] == 5_000_000.0 # current + assert hist[0][1] < hist[1][1] # previous change_time strictly earlier + + +def test_first_capture_without_previous_inserts_one_point(db: Session) -> None: + """price_previous отсутствует → только current-точка (1 строка).""" + lid = _seed_listing(db, source="yandex", source_id="222") + n = yph.record_yandex_price_history(db, [_lot(source_id="222", price_rub=4_000_000)]) + assert n == 1 + hist = _history(db, lid) + assert len(hist) == 1 + assert hist[0][0] == 4_000_000.0 + + +def test_first_capture_previous_equals_current_inserts_one_point(db: Session) -> None: + """price_previous == price_rub → previous-seed пропущена, только current (1 строка).""" + lid = _seed_listing(db, source="yandex", source_id="223") + n = yph.record_yandex_price_history( + db, [_lot(source_id="223", price_rub=4_000_000, price_previous_rub=4_000_000)] + ) + assert n == 1 + assert len(_history(db, lid)) == 1 + + +def test_recapture_same_price_is_noop(db: Session) -> None: + """Повторный захват с той же ценой → 0 строк (idempotent).""" + _seed_listing(db, source="yandex", source_id="333") + first = yph.record_yandex_price_history(db, [_lot(source_id="333", price_rub=6_000_000)]) + assert first == 1 + second = yph.record_yandex_price_history(db, [_lot(source_id="333", price_rub=6_000_000)]) + assert second == 0 + + +def test_recapture_new_price_inserts_one_point(db: Session) -> None: + """Повторный захват с новой ценой → 1 строка (зафиксировали смену).""" + lid = _seed_listing(db, source="yandex", source_id="444") + yph.record_yandex_price_history(db, [_lot(source_id="444", price_rub=6_000_000)]) + n = yph.record_yandex_price_history(db, [_lot(source_id="444", price_rub=5_800_000)]) + assert n == 1 + hist = _history(db, lid) + assert len(hist) == 2 + assert {h[0] for h in hist} == {6_000_000.0, 5_800_000.0} + + +def test_non_yandex_lot_skipped(db: Session) -> None: + """Не-yandex lot пропускается (даже если листинг с таким source_id есть).""" + _seed_listing(db, source="cian", source_id="555") + n = yph.record_yandex_price_history( + db, [_lot(source_id="555", price_rub=3_000_000, source="cian")] + ) + assert n == 0 + + +def test_unknown_source_id_skipped(db: Session) -> None: + """source_id без листинга в БД → skip (листинг не найден).""" + n = yph.record_yandex_price_history(db, [_lot(source_id="does-not-exist", price_rub=1_000_000)]) + assert n == 0 + + +def test_lot_without_source_id_skipped(db: Session) -> None: + lot = ScrapedLot( + source="yandex", + source_url="https://realty.yandex.ru/offer/x/", + source_id=None, + price_rub=2_000_000, + ) + assert yph.record_yandex_price_history(db, [lot]) == 0 + + +def test_empty_lots_returns_zero(db: Session) -> None: + assert yph.record_yandex_price_history(db, []) == 0 + + +def test_broken_lot_does_not_abort_batch(db: Session, monkeypatch: pytest.MonkeyPatch) -> None: + """SAVEPOINT: один битый lot (INSERT падает) не валит остальные.""" + good_lid = _seed_listing(db, source="yandex", source_id="aaa") + _seed_listing(db, source="yandex", source_id="bbb") + + real_insert = yph._insert_point + + def _flaky(db_: Session, **kw: object) -> None: + # Ломаем только для листинга 'bbb' (его listing_id заранее не знаем — + # рвём по price_rub-значению, привязанному к bbb). + if kw.get("price_rub") == 999: + raise RuntimeError("boom") + real_insert(db_, **kw) # type: ignore[arg-type] + + monkeypatch.setattr(yph, "_insert_point", _flaky) + + lots = [ + _lot(source_id="aaa", price_rub=7_000_000), + _lot(source_id="bbb", price_rub=999), + ] + n = yph.record_yandex_price_history(db, lots) + # 'aaa' inserted (1), 'bbb' rolled back via SAVEPOINT → batch survives. + assert n == 1 + assert len(_history(db, good_lid)) == 1