"""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