diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py
index e33861ec..f47dd6f5 100644
--- a/tradein-mvp/backend/app/services/scheduler.py
+++ b/tradein-mvp/backend/app/services/scheduler.py
@@ -56,6 +56,11 @@ Sources:
всех источников с NULL coords; address-dedup batch loop с
wall-clock budget; window 06:00-09:00 UTC — после city sweeps
~03-04 UTC и deals OS-cron 05:30/05:45 UTC)
+ - yandex_detail_backfill → run_yandex_detail_backfill
+ (tasks/yandex_detail_backfill.py, #1553; nightly backfill of
+ yandex detail_enriched_at for ~3952 listings never enriched by
+ city_sweep; YandexDetailScraper httpx, no proxy layer;
+ window 12:00-15:00 UTC — после avito_detail_backfill 09-12 UTC)
"""
from __future__ import annotations
@@ -1207,6 +1212,43 @@ async def trigger_avito_detail_backfill_run(
return run_id
+async def trigger_yandex_detail_backfill_run(
+ db: Session, schedule_row: dict[str, Any]
+) -> int | None:
+ """Create scrape_run + launch run_yandex_detail_backfill in asyncio.create_task.
+
+ Nightly backfill detail_enriched_at for legacy yandex listings (#1553).
+ ~3952 yandex listings have detail_enriched_at IS NULL -- SERP-only data,
+ never enriched by YandexDetailScraper.
+
+ Mirrors trigger_avito_detail_backfill_run: claim run -> create_task ->
+ mark_done/failed delegated to tasks/yandex_detail_backfill.run_yandex_detail_backfill.
+
+ Returns run_id or None (skip -- already running).
+ """
+ run_id = _claim_run(db, schedule_row)
+ if run_id is None:
+ return None
+
+ params = schedule_row.get("default_params") or {}
+
+ async def _run() -> None:
+ run_db = SessionLocal()
+ try:
+ from app.tasks.yandex_detail_backfill import run_yandex_detail_backfill
+
+ await run_yandex_detail_backfill(run_db, run_id=run_id, params=params)
+ except Exception:
+ logger.exception("scheduler: run_yandex_detail_backfill crashed run_id=%d", run_id)
+ finally:
+ run_db.close()
+
+ task = asyncio.create_task(_run())
+ task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
+ logger.info("scheduler: triggered yandex_detail_backfill run_id=%d", run_id)
+ return run_id
+
+
def get_due_schedules(db: Session) -> list[dict[str, Any]]:
"""SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW())."""
rows = (
@@ -1276,6 +1318,8 @@ async def scheduler_loop() -> None:
await trigger_geocode_missing_listings_run(db, sch)
elif source == "avito_detail_backfill":
await trigger_avito_detail_backfill_run(db, sch)
+ elif source == "yandex_detail_backfill":
+ await trigger_yandex_detail_backfill_run(db, sch)
else:
logger.warning("scheduler: unknown source=%s, skip", source)
finally:
diff --git a/tradein-mvp/backend/app/services/scrapers/yandex_detail.py b/tradein-mvp/backend/app/services/scrapers/yandex_detail.py
index 603ae403..2850ee30 100644
--- a/tradein-mvp/backend/app/services/scrapers/yandex_detail.py
+++ b/tradein-mvp/backend/app/services/scrapers/yandex_detail.py
@@ -7,6 +7,7 @@ a DetailEnrichment Pydantic model. Used by enrichment pipeline
from __future__ import annotations
+import json
import logging
import re
from datetime import date
@@ -14,6 +15,8 @@ from typing import Any
from pydantic import BaseModel, Field
from selectolax.parser import HTMLParser, Node
+from sqlalchemy import text
+from sqlalchemy.orm import Session
from app.services.scraper_settings import get_scraper_delay
from app.services.scrapers.base import BaseScraper
@@ -364,3 +367,106 @@ def _extract_seller_name(summary_text: str, agency_name: str | None) -> str | No
# last short token sequence (likely "Имя Фамилия")
m = re.findall(r"([А-ЯЁ][а-яё]+(?:\s+[А-ЯЁ][а-яё]+){1,2})", head)
return m[-1] if m else None
+
+
+# ── Save helper ───────────────────────────────────────────────────────────────
+
+
+def save_detail_enrichment(db: Session, listing_id: int, e: DetailEnrichment) -> bool:
+ """Persist Yandex DetailEnrichment to listings row.
+
+ UPDATE listings SET = COALESCE(:val, ), ..., detail_enriched_at = NOW()
+ WHERE id = listing_id.
+
+ COALESCE semantics: keeps existing non-NULL value if new value is NULL (never
+ overwrites a populated column with NULL). area_m2 from detail is more accurate
+ than SERP, but COALESCE preserves SERP value if detail returns NULL — acceptable.
+
+ Returns True if the UPDATE touched at least one row (listing_id found in DB).
+ """
+ metro_json: str | None = None
+ if e.metro_stations:
+ metro_json = json.dumps(
+ [s.model_dump(exclude_none=True) for s in e.metro_stations],
+ ensure_ascii=False,
+ )
+
+ photo_json: str | None = None
+ if e.photo_urls:
+ photo_json = json.dumps(e.photo_urls, ensure_ascii=False)
+
+ result = db.execute(
+ text("""
+ UPDATE listings SET
+ rooms = COALESCE(CAST(:rooms AS int), rooms),
+ area_m2 = COALESCE(CAST(:area_m2 AS numeric), area_m2),
+ floor = COALESCE(CAST(:floor AS int), floor),
+ total_floors = COALESCE(CAST(:total_floors AS int), total_floors),
+ address = COALESCE(CAST(:address AS text), address),
+ description = COALESCE(CAST(:description AS text), description),
+ repair_state = COALESCE(CAST(:repair_state AS text),repair_state),
+ publish_date = COALESCE(CAST(:publish_date AS date),publish_date),
+ views_total_yandex = COALESCE(CAST(:views_total AS int), views_total_yandex),
+ publish_date_relative= COALESCE(
+ CAST(:pub_date_rel AS text),
+ publish_date_relative
+ ),
+ agency_name = COALESCE(CAST(:agency_name AS text), agency_name),
+ agency_founded_year = COALESCE(
+ CAST(:agency_founded_year AS int),
+ agency_founded_year
+ ),
+ agency_objects_count = COALESCE(
+ CAST(:agency_objects_count AS int),
+ agency_objects_count
+ ),
+ metro_stations = COALESCE(
+ CAST(:metro_stations AS jsonb),
+ metro_stations
+ ),
+ photo_urls = COALESCE(
+ CAST(:photo_urls AS jsonb),
+ photo_urls
+ ),
+ newbuilding_url = COALESCE(
+ CAST(:newbuilding_url AS text),
+ newbuilding_url
+ ),
+ newbuilding_id = COALESCE(
+ CAST(:newbuilding_id AS text),
+ newbuilding_id
+ ),
+ detail_enriched_at = NOW()
+ WHERE id = CAST(:listing_id AS bigint)
+ """),
+ {
+ "listing_id": listing_id,
+ "rooms": e.rooms,
+ "area_m2": e.area_m2,
+ "floor": e.floor,
+ "total_floors": e.total_floors,
+ "address": e.address,
+ "description": e.description,
+ "repair_state": e.repair_state,
+ "publish_date": e.publish_date,
+ "views_total": e.views_total,
+ "pub_date_rel": e.publish_date_relative,
+ "agency_name": e.agency_name,
+ "agency_founded_year": e.agency_founded_year,
+ "agency_objects_count": e.agency_objects_count,
+ "metro_stations": metro_json,
+ "photo_urls": photo_json,
+ "newbuilding_url": e.newbuilding_url,
+ "newbuilding_id": e.newbuilding_id,
+ },
+ )
+ db.commit()
+ saved = (result.rowcount or 0) > 0
+ logger.info(
+ "yandex detail enrichment saved listing_id=%s (metro=%d photos=%d saved=%s)",
+ listing_id,
+ len(e.metro_stations),
+ len(e.photo_urls),
+ saved,
+ )
+ return saved
diff --git a/tradein-mvp/backend/app/tasks/yandex_detail_backfill.py b/tradein-mvp/backend/app/tasks/yandex_detail_backfill.py
new file mode 100644
index 00000000..65bf213c
--- /dev/null
+++ b/tradein-mvp/backend/app/tasks/yandex_detail_backfill.py
@@ -0,0 +1,232 @@
+"""Scheduled backfill: detail-enrichment for legacy yandex listings (#1553).
+
+Nightly window 12:00-15:00 UTC (migration 113, source=yandex_detail_backfill).
+Offset from avito_detail_backfill (09-12 UTC) to avoid parallel egress on shared IP.
+
+Problem: ~3952 yandex listings have detail_enriched_at IS NULL.
+Yandex city sweep does not call YandexDetailScraper — SERP data only.
+area_m2 coverage 23%, living/kitchen 0%, repair_state 1% on prod.
+
+Solution: single snapshot SELECT at start (guarantees termination), fetch each
+offer detail page via YandexDetailScraper (httpx + BaseScraper retry), persist
+with save_detail_enrichment. No explicit block-exception from YandexDetailScraper —
+fetch_detail returns None on HTTP error or parse failure. Track consecutive None
+results; abort after max_consecutive_blocks (mark_done, not mark_failed — retry next
+night via NULL detail_enriched_at). YandexDetailScraper has no _rotate_ip: plain
+httpx with UA rotation; no proxy layer required.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from dataclasses import dataclass, field
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.services import scrape_runs as runs_mod
+from app.services.scrapers.yandex_detail import YandexDetailScraper, save_detail_enrichment
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+ "YandexDetailBackfillResult",
+ "run_yandex_detail_backfill",
+]
+
+
+@dataclass
+class YandexDetailBackfillResult:
+ """Counters for one yandex detail backfill run."""
+
+ attempted: int = 0
+ enriched: int = 0
+ failed: int = 0
+ duration_sec: float = field(default=0.0)
+
+ def to_dict(self) -> dict[str, int]:
+ return {
+ "attempted": self.attempted,
+ "enriched": self.enriched,
+ "failed": self.failed,
+ "duration_sec": int(self.duration_sec),
+ }
+
+
+async def run_yandex_detail_backfill(
+ db: Session,
+ *,
+ run_id: int,
+ params: dict,
+) -> YandexDetailBackfillResult:
+ """Backfill detail_enriched_at for legacy yandex listings via YandexDetailScraper.
+
+ Params (from default_params jsonb in scrape_schedules):
+ batch_size: int -- snapshot size (SELECT LIMIT), default 800.
+ budget_sec: float -- wall-clock budget per run, default 3600s.
+ request_delay_sec: float -- delay between listings (overrides scraper default), default 5s.
+ max_consecutive_blocks: int -- consecutive None results before abort, default 5.
+
+ Lifecycle: update_heartbeat -> snapshot -> loop with budget guard ->
+ mark_done (incl. partial / consecutive-None abort) / mark_failed (exception only).
+
+ YandexDetailScraper.fetch_detail returns None on HTTP error or parse failure
+ (no explicit block exception). Consecutive None results signal a possible
+ temporary ban or network issue; we abort after max_consecutive_blocks and let
+ the scheduler retry the next night (NULL detail_enriched_at rows reappear).
+ """
+ batch_size = int(params.get("batch_size", 800))
+ budget_sec = float(params.get("budget_sec", 3600))
+ request_delay_sec = float(params.get("request_delay_sec", 5.0))
+ max_consecutive_blocks = int(params.get("max_consecutive_blocks", 5))
+
+ counters = YandexDetailBackfillResult()
+ current_counters: dict[str, int] = counters.to_dict()
+
+ start = time.monotonic()
+
+ try:
+ runs_mod.update_heartbeat(db, run_id, current_counters)
+
+ # SNAPSHOT: single SELECT at start -- NOT re-selected in loop.
+ # Priority: is_active DESC (active first), scraped_at DESC (newest first).
+ snapshot = (
+ db.execute(
+ text(
+ """
+ SELECT id, source_url
+ FROM listings
+ WHERE source = 'yandex'
+ AND detail_enriched_at IS NULL
+ AND source_url IS NOT NULL
+ ORDER BY is_active DESC NULLS LAST, scraped_at DESC NULLS LAST
+ LIMIT CAST(:batch_size AS int)
+ """
+ ),
+ {"batch_size": batch_size},
+ )
+ .mappings()
+ .all()
+ )
+
+ if not snapshot:
+ logger.info(
+ "yandex_detail_backfill: run_id=%d -- no pending listings "
+ "(detail_enriched_at IS NULL = 0), done",
+ run_id,
+ )
+ runs_mod.mark_done(db, run_id, current_counters)
+ return counters
+
+ logger.info(
+ "yandex_detail_backfill: run_id=%d snapshot=%d (budget=%.0fs "
+ "delay=%.1fs max_consecutive_none=%d)",
+ run_id,
+ len(snapshot),
+ budget_sec,
+ request_delay_sec,
+ max_consecutive_blocks,
+ )
+
+ consecutive_none = 0
+ do_sleep = False
+
+ async with YandexDetailScraper() as scraper:
+ for idx, row in enumerate(snapshot):
+ # Budget guard
+ elapsed = time.monotonic() - start
+ if elapsed > budget_sec:
+ logger.info(
+ "yandex_detail_backfill: run_id=%d -- budget %.0fs exhausted "
+ "(elapsed=%.1fs), stopping at #%d/%d",
+ run_id,
+ budget_sec,
+ elapsed,
+ idx,
+ len(snapshot),
+ )
+ break
+
+ # Delay before each request except the first
+ if do_sleep:
+ await asyncio.sleep(request_delay_sec)
+ do_sleep = True
+
+ source_url: str = row["source_url"]
+ listing_id: int = row["id"]
+ counters.attempted += 1
+
+ try:
+ enrichment = await scraper.fetch_detail(source_url)
+
+ if enrichment is None:
+ # fetch_detail returns None on block / parse failure.
+ # Do not mark listing as done — retry next night.
+ consecutive_none += 1
+ counters.failed += 1
+ logger.warning(
+ "yandex_detail_backfill: run_id=%d listing_id=%d source_url=%s "
+ "-> None (consecutive=%d)",
+ run_id,
+ listing_id,
+ source_url,
+ consecutive_none,
+ )
+ if consecutive_none >= max_consecutive_blocks:
+ logger.error(
+ "yandex_detail_backfill: run_id=%d ABORT -- %d consecutive "
+ "None results (possible ban). enriched=%d attempted=%d",
+ run_id,
+ consecutive_none,
+ counters.enriched,
+ counters.attempted,
+ )
+ break
+ continue
+
+ consecutive_none = 0
+ if save_detail_enrichment(db, listing_id, enrichment):
+ counters.enriched += 1
+
+ except Exception as exc:
+ counters.failed += 1
+ logger.warning(
+ "yandex_detail_backfill: run_id=%d listing_id=%d failed: %s",
+ run_id,
+ listing_id,
+ exc,
+ )
+ try:
+ db.rollback()
+ except Exception:
+ pass
+
+ if counters.attempted % 25 == 0:
+ current_counters = counters.to_dict()
+ runs_mod.update_heartbeat(db, run_id, current_counters)
+
+ counters.duration_sec = time.monotonic() - start
+ current_counters = counters.to_dict()
+ runs_mod.mark_done(db, run_id, current_counters)
+ logger.info(
+ "yandex_detail_backfill: run_id=%d DONE -- attempted=%d enriched=%d "
+ "failed=%d duration=%.1fs",
+ run_id,
+ counters.attempted,
+ counters.enriched,
+ counters.failed,
+ counters.duration_sec,
+ )
+ return counters
+
+ except Exception as exc:
+ counters.duration_sec = time.monotonic() - start
+ logger.exception(
+ "yandex_detail_backfill: run_id=%d FAILED after %.1fs",
+ run_id,
+ counters.duration_sec,
+ )
+ runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters.to_dict())
+ raise
diff --git a/tradein-mvp/backend/data/sql/113_yandex_detail_backfill.sql b/tradein-mvp/backend/data/sql/113_yandex_detail_backfill.sql
new file mode 100644
index 00000000..cecc172f
--- /dev/null
+++ b/tradein-mvp/backend/data/sql/113_yandex_detail_backfill.sql
@@ -0,0 +1,49 @@
+-- Migration 113: yandex detail backfill — новые колонки + seed расписания.
+--
+-- Добавляет в listings три колонки, которых нет ни в 002_core_tables.sql ни в более ранних
+-- migrations, но которые парсит YandexDetailScraper.DetailEnrichment:
+--
+-- photo_urls jsonb -- список URL фото из Product JSON-LD image[]
+-- newbuilding_url text -- ссылка на ЖК (/kupit/novostrojka/...) если есть
+-- newbuilding_id text -- id ЖК из URL (числовая строка)
+--
+-- Уже существующие колонки (не дублируем):
+-- rooms, area_m2, floor, total_floors, address, description, repair_state,
+-- publish_date, metro_stations, detail_enriched_at -- 002/011_*.sql
+-- views_total_yandex, publish_date_relative, agency_name,
+-- agency_founded_year, agency_objects_count -- 033_listings_alter_yandex.sql
+--
+-- Seed scrape_schedules: source='yandex_detail_backfill', window 12-15 UTC,
+-- разнесено с avito_detail_backfill (09-12 UTC) чтобы не делить трафик.
+-- ON CONFLICT DO NOTHING — идемпотентно.
+
+BEGIN;
+
+-- ── listings — новые колонки для detail enrichment ────────────────────────────
+
+ALTER TABLE listings
+ ADD COLUMN IF NOT EXISTS photo_urls jsonb, -- [url, ...] из Product JSON-LD image[]
+ ADD COLUMN IF NOT EXISTS newbuilding_url text, -- ссылка на ЖК (/kupit/novostrojka/...)
+ ADD COLUMN IF NOT EXISTS newbuilding_id text; -- id ЖК из URL (числовая строка)
+
+-- ── scrape_schedules — seed yandex_detail_backfill ────────────────────────────
+
+INSERT INTO scrape_schedules
+ (source, enabled, window_start_hour, window_end_hour, next_run_at, default_params)
+VALUES (
+ 'yandex_detail_backfill',
+ true,
+ 12, -- 12:00 UTC = 15:00 МСК
+ 15, -- 15:00 UTC = 18:00 МСК
+ (NOW() + INTERVAL '1 day')::date + TIME '12:00:00' + INTERVAL '0 second',
+ '{"batch_size": 800, "budget_sec": 3600, "request_delay_sec": 5, "max_consecutive_blocks": 5}'::jsonb
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMENT ON TABLE scrape_schedules IS
+ 'Периодические задачи: avito/yandex/n1/cian city sweep, rosreestr import, '
+ 'listing snapshot, ratio refresh, geocode backfill, deactivate stale, '
+ 'sber/rosreestr quarter poll, newbuilding enrich, yandex newbuilding sweep, '
+ 'avito detail backfill, yandex detail backfill.';
+
+COMMIT;
diff --git a/tradein-mvp/backend/tests/tasks/test_yandex_detail_backfill.py b/tradein-mvp/backend/tests/tasks/test_yandex_detail_backfill.py
new file mode 100644
index 00000000..3b285ad1
--- /dev/null
+++ b/tradein-mvp/backend/tests/tasks/test_yandex_detail_backfill.py
@@ -0,0 +1,357 @@
+"""Tests for app.tasks.yandex_detail_backfill and save_detail_enrichment."""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from unittest.mock import AsyncMock, MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+_wp_mock = MagicMock()
+sys.modules.setdefault("weasyprint", _wp_mock)
+
+import pytest # noqa: E402
+
+from app.tasks.yandex_detail_backfill import ( # noqa: E402
+ YandexDetailBackfillResult,
+ run_yandex_detail_backfill,
+)
+
+# ---------------------------------------------------------------------------
+# Path constants for patching
+# ---------------------------------------------------------------------------
+
+_FETCH = "app.tasks.yandex_detail_backfill.YandexDetailScraper"
+_SAVE = "app.tasks.yandex_detail_backfill.save_detail_enrichment"
+_RUNS = "app.tasks.yandex_detail_backfill.runs_mod"
+_SLEEP = "app.tasks.yandex_detail_backfill.asyncio.sleep"
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_snapshot(n: int) -> list[dict]:
+ return [
+ {"id": i + 1, "source_url": f"https://realty.yandex.ru/offer/{i + 1}/"} for i in range(n)
+ ]
+
+
+def _mock_db(snapshot: list[dict]) -> MagicMock:
+ """Fake Session: first execute() returns snapshot via .mappings().all()."""
+ db = MagicMock()
+ sel = MagicMock()
+ sel.mappings.return_value.all.return_value = snapshot
+ db.execute.return_value = sel
+ return db
+
+
+# ---------------------------------------------------------------------------
+# Tests: run_yandex_detail_backfill
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_backfill_empty_snapshot_marks_done() -> None:
+ """Empty snapshot -> mark_done immediately, no fetch calls."""
+ db = _mock_db([])
+ runs = MagicMock()
+ mock_scraper_cls = MagicMock()
+ mock_scraper_inst = AsyncMock()
+ mock_scraper_cls.return_value.__aenter__ = AsyncMock(return_value=mock_scraper_inst)
+ mock_scraper_cls.return_value.__aexit__ = AsyncMock(return_value=None)
+
+ with (
+ patch(_FETCH, mock_scraper_cls),
+ patch(_RUNS, runs),
+ ):
+ result = await run_yandex_detail_backfill(
+ db, run_id=1, params={"batch_size": 10, "budget_sec": 60}
+ )
+
+ assert isinstance(result, YandexDetailBackfillResult)
+ assert result.attempted == 0
+ assert result.enriched == 0
+ mock_scraper_inst.fetch_detail.assert_not_called()
+ runs.mark_done.assert_called_once()
+ runs.mark_failed.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_backfill_processes_snapshot_to_completion() -> None:
+ """3 listings -> all fetched and enriched, mark_done called."""
+ snapshot = _make_snapshot(3)
+ db = _mock_db(snapshot)
+ runs = MagicMock()
+ mock_enrichment = MagicMock()
+
+ mock_scraper_cls = MagicMock()
+ mock_scraper_inst = AsyncMock()
+ mock_scraper_inst.fetch_detail = AsyncMock(return_value=mock_enrichment)
+ mock_scraper_cls.return_value.__aenter__ = AsyncMock(return_value=mock_scraper_inst)
+ mock_scraper_cls.return_value.__aexit__ = AsyncMock(return_value=None)
+
+ with (
+ patch(_FETCH, mock_scraper_cls),
+ patch(_RUNS, runs),
+ patch(_SAVE, return_value=True),
+ patch(_SLEEP, new_callable=AsyncMock),
+ ):
+ result = await run_yandex_detail_backfill(
+ db, run_id=2, params={"batch_size": 10, "budget_sec": 3600}
+ )
+
+ assert result.attempted == 3
+ assert result.enriched == 3
+ assert result.failed == 0
+ assert mock_scraper_inst.fetch_detail.call_count == 3
+ runs.mark_done.assert_called_once()
+ runs.mark_failed.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_backfill_none_abort_after_max_consecutive() -> None:
+ """5 consecutive None results -> abort, mark_done (NOT mark_failed)."""
+ snapshot = _make_snapshot(10)
+ db = _mock_db(snapshot)
+ runs = MagicMock()
+
+ mock_scraper_cls = MagicMock()
+ mock_scraper_inst = AsyncMock()
+ mock_scraper_inst.fetch_detail = AsyncMock(return_value=None)
+ mock_scraper_cls.return_value.__aenter__ = AsyncMock(return_value=mock_scraper_inst)
+ mock_scraper_cls.return_value.__aexit__ = AsyncMock(return_value=None)
+
+ with (
+ patch(_FETCH, mock_scraper_cls),
+ patch(_RUNS, runs),
+ patch(_SLEEP, new_callable=AsyncMock),
+ ):
+ result = await run_yandex_detail_backfill(
+ db,
+ run_id=3,
+ params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5},
+ )
+
+ assert result.attempted == 5
+ assert result.failed == 5
+ assert result.enriched == 0
+ runs.mark_done.assert_called_once()
+ runs.mark_failed.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_backfill_none_resets_on_success() -> None:
+ """1 None + 1 success + 1 None -> consecutive resets, no abort after 1 success."""
+ snapshot = _make_snapshot(3)
+ db = _mock_db(snapshot)
+ runs = MagicMock()
+ mock_enrichment = MagicMock()
+
+ mock_scraper_cls = MagicMock()
+ mock_scraper_inst = AsyncMock()
+ mock_scraper_inst.fetch_detail = AsyncMock(side_effect=[None, mock_enrichment, None])
+ mock_scraper_cls.return_value.__aenter__ = AsyncMock(return_value=mock_scraper_inst)
+ mock_scraper_cls.return_value.__aexit__ = AsyncMock(return_value=None)
+
+ with (
+ patch(_FETCH, mock_scraper_cls),
+ patch(_RUNS, runs),
+ patch(_SAVE, return_value=True),
+ patch(_SLEEP, new_callable=AsyncMock),
+ ):
+ result = await run_yandex_detail_backfill(
+ db,
+ run_id=4,
+ params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5},
+ )
+
+ assert result.attempted == 3
+ assert result.enriched == 1
+ assert result.failed == 2
+ runs.mark_done.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_backfill_budget_guard_stops_loop() -> None:
+ """Budget expired before first listing -> fetch_detail not called."""
+ snapshot = _make_snapshot(5)
+ db = _mock_db(snapshot)
+ runs = MagicMock()
+
+ mock_scraper_cls = MagicMock()
+ mock_scraper_inst = AsyncMock()
+ mock_scraper_cls.return_value.__aenter__ = AsyncMock(return_value=mock_scraper_inst)
+ mock_scraper_cls.return_value.__aexit__ = AsyncMock(return_value=None)
+
+ mono_values = iter([0.0, 999.0, 999.0])
+ with (
+ patch(_FETCH, mock_scraper_cls),
+ patch(_RUNS, runs),
+ patch("app.tasks.yandex_detail_backfill.time.monotonic", side_effect=mono_values),
+ ):
+ await run_yandex_detail_backfill(db, run_id=5, params={"batch_size": 5, "budget_sec": 1})
+
+ mock_scraper_inst.fetch_detail.assert_not_called()
+ runs.mark_done.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_backfill_top_level_exception_marks_failed() -> None:
+ """db.execute raises -> mark_failed called, exception re-raised."""
+ db = MagicMock()
+ db.execute.side_effect = RuntimeError("DB connection lost")
+ runs = MagicMock()
+
+ with (
+ patch(_RUNS, runs),
+ ):
+ with pytest.raises(RuntimeError, match="DB connection lost"):
+ await run_yandex_detail_backfill(
+ db, run_id=6, params={"batch_size": 5, "budget_sec": 60}
+ )
+
+ runs.mark_failed.assert_called_once()
+ runs.mark_done.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_backfill_fetch_exception_continues() -> None:
+ """RuntimeError on one listing -> failed++, db.rollback(), loop continues for next."""
+ snapshot = _make_snapshot(2)
+ db = _mock_db(snapshot)
+ runs = MagicMock()
+ mock_enrichment = MagicMock()
+
+ mock_scraper_cls = MagicMock()
+ mock_scraper_inst = AsyncMock()
+ mock_scraper_inst.fetch_detail = AsyncMock(
+ side_effect=[RuntimeError("parse error"), mock_enrichment]
+ )
+ mock_scraper_cls.return_value.__aenter__ = AsyncMock(return_value=mock_scraper_inst)
+ mock_scraper_cls.return_value.__aexit__ = AsyncMock(return_value=None)
+
+ with (
+ patch(_FETCH, mock_scraper_cls),
+ patch(_RUNS, runs),
+ patch(_SAVE, return_value=True),
+ patch(_SLEEP, new_callable=AsyncMock),
+ ):
+ result = await run_yandex_detail_backfill(
+ db, run_id=7, params={"batch_size": 10, "budget_sec": 3600}
+ )
+
+ assert result.failed == 1
+ assert result.enriched == 1
+ assert result.attempted == 2
+ db.rollback.assert_called()
+ runs.mark_done.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# Tests: save_detail_enrichment
+# ---------------------------------------------------------------------------
+
+
+def test_save_detail_enrichment_maps_fields_to_update() -> None:
+ """save_detail_enrichment calls db.execute with UPDATE and commits."""
+ from app.services.scrapers.yandex_detail import (
+ DetailEnrichment,
+ MetroStation,
+ save_detail_enrichment,
+ )
+
+ enrichment = DetailEnrichment(
+ offer_id="12345",
+ source_url="https://realty.yandex.ru/offer/12345/",
+ rooms=2,
+ area_m2=54.5,
+ floor=3,
+ total_floors=9,
+ address="Екатеринбург, ул. Ленина, 10",
+ description="Хорошая квартира",
+ repair_state="standard",
+ publish_date=None,
+ views_total=42,
+ publish_date_relative="вчера",
+ agency_name="Агентство «Тест»",
+ agency_founded_year=2005,
+ agency_objects_count=150,
+ metro_stations=[MetroStation(name="Чкаловская", walk_min=11)],
+ photo_urls=["https://example.com/1.jpg", "https://example.com/2.jpg"],
+ newbuilding_url="https://realty.yandex.ru/kupit/novostrojka/test-12345/",
+ newbuilding_id="12345",
+ )
+
+ db = MagicMock()
+ result_mock = MagicMock()
+ result_mock.rowcount = 1
+ db.execute.return_value = result_mock
+
+ saved = save_detail_enrichment(db, listing_id=99, e=enrichment)
+
+ assert saved is True
+ db.execute.assert_called_once()
+ db.commit.assert_called_once()
+
+ # Verify the SQL call contains metro_stations and photo_urls as JSON
+ call_args = db.execute.call_args
+ params = call_args[0][1] # positional second arg to execute()
+ assert params["rooms"] == 2
+ assert params["area_m2"] == 54.5
+ assert params["agency_name"] == "Агентство «Тест»"
+
+ # metro_stations must be JSON-serialised
+ metro_val = params["metro_stations"]
+ assert metro_val is not None
+ parsed_metro = json.loads(metro_val)
+ assert parsed_metro[0]["name"] == "Чкаловская"
+ assert parsed_metro[0]["walk_min"] == 11
+
+ # photo_urls must be JSON-serialised
+ photo_val = params["photo_urls"]
+ assert photo_val is not None
+ parsed_photos = json.loads(photo_val)
+ assert len(parsed_photos) == 2
+ assert parsed_photos[0] == "https://example.com/1.jpg"
+
+
+def test_save_detail_enrichment_empty_metro_and_photos() -> None:
+ """Empty metro_stations and photo_urls -> NULL passed for both jsonb columns."""
+ from app.services.scrapers.yandex_detail import DetailEnrichment, save_detail_enrichment
+
+ enrichment = DetailEnrichment(
+ offer_id="99",
+ source_url="https://realty.yandex.ru/offer/99/",
+ )
+
+ db = MagicMock()
+ result_mock = MagicMock()
+ result_mock.rowcount = 1
+ db.execute.return_value = result_mock
+
+ save_detail_enrichment(db, listing_id=1, e=enrichment)
+
+ params = db.execute.call_args[0][1]
+ assert params["metro_stations"] is None
+ assert params["photo_urls"] is None
+
+
+def test_save_detail_enrichment_rowcount_zero_returns_false() -> None:
+ """rowcount=0 (listing_id not found) -> returns False."""
+ from app.services.scrapers.yandex_detail import DetailEnrichment, save_detail_enrichment
+
+ enrichment = DetailEnrichment(
+ offer_id="404",
+ source_url="https://realty.yandex.ru/offer/404/",
+ )
+
+ db = MagicMock()
+ result_mock = MagicMock()
+ result_mock.rowcount = 0
+ db.execute.return_value = result_mock
+
+ saved = save_detail_enrichment(db, listing_id=404, e=enrichment)
+ assert saved is False