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

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

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