From 03e43f12b554eb8d1d37cc0d23fa80c1fa2257ab Mon Sep 17 00:00:00 2001 From: bot-backend Date: Tue, 16 Jun 2026 13:58:20 +0300 Subject: [PATCH] =?UTF-8?q?feat(tradein):=20yandex=20detail=20backfill=20?= =?UTF-8?q?=E2=80=94=20save=5Fdetail=5Fenrichment=20+=20nightly=20task=20(?= =?UTF-8?q?#1553)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - yandex_detail.py: добавить save_detail_enrichment(db, listing_id, enrichment) → bool; UPDATE listings COALESCE по 17 полям (rooms/area_m2/floor/total_floors/address/ description/repair_state/publish_date/views_total_yandex/publish_date_relative/ agency_name/agency_founded_year/agency_objects_count/metro_stations/photo_urls/ newbuilding_url/newbuilding_id); metro/photos → jsonb; detail_enriched_at = NOW() - app/tasks/yandex_detail_backfill.py: новый task run_yandex_detail_backfill; snapshot SELECT source='yandex' AND detail_enriched_at IS NULL; YandexDetailScraper async context manager; consecutive None abort (max_consecutive_blocks=5); budget guard; run lifecycle update_heartbeat/mark_done/mark_failed - scheduler.py: trigger_yandex_detail_backfill_run + elif source=='yandex_detail_backfill' в scheduler_loop; window 12-15 UTC (после avito_detail 09-12 UTC) - migration 113: ADD COLUMN IF NOT EXISTS photo_urls jsonb, newbuilding_url text, newbuilding_id text; seed scrape_schedules yandex_detail_backfill enabled=true - tests: 10 тестов (8 backfill + 2 save_detail_enrichment); 1819 passed 0 failed Closes #1553 --- tradein-mvp/backend/app/services/scheduler.py | 44 +++ .../app/services/scrapers/yandex_detail.py | 106 ++++++ .../app/tasks/yandex_detail_backfill.py | 232 ++++++++++++ .../data/sql/113_yandex_detail_backfill.sql | 49 +++ .../tasks/test_yandex_detail_backfill.py | 357 ++++++++++++++++++ 5 files changed, 788 insertions(+) create mode 100644 tradein-mvp/backend/app/tasks/yandex_detail_backfill.py create mode 100644 tradein-mvp/backend/data/sql/113_yandex_detail_backfill.sql create mode 100644 tradein-mvp/backend/tests/tasks/test_yandex_detail_backfill.py 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