tradein-mvp/backend/app/services/scheduler.py

@@ -1170,43 +1170,6 @@ def import_rosreestr_dkp(
         raise
 
 
-async def trigger_avito_detail_backfill_run(
-    db: Session, schedule_row: dict[str, Any]
-) -> int | None:
-    """Create scrape_run + launch run_avito_detail_backfill in asyncio.create_task.
-
-    Nightly backfill detail_enriched_at for legacy avito listings (#1551).
-    ~15243/15917 avito listings have detail_enriched_at IS NULL -- never enriched
-    by city_sweep ENRICH_DETAIL (only top-N proximate per anchor are touched).
-
-    Mirrors trigger_geocode_missing_listings_run: claim run -> create_task ->
-    mark_done/failed delegated to tasks/avito_detail_backfill.run_avito_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.avito_detail_backfill import run_avito_detail_backfill
-
-            await run_avito_detail_backfill(run_db, run_id=run_id, params=params)
-        except Exception:
-            logger.exception("scheduler: run_avito_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 avito_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 = (
@@ -1274,8 +1237,6 @@ async def scheduler_loop() -> None:
                         await trigger_yandex_newbuilding_sweep_run(db, sch)
                     elif source == "geocode_missing_listings":
                         await trigger_geocode_missing_listings_run(db, sch)
-                    elif source == "avito_detail_backfill":
-                        await trigger_avito_detail_backfill_run(db, sch)
                     else:
                         logger.warning("scheduler: unknown source=%s, skip", source)
             finally:

tradein-mvp/backend/app/tasks/avito_detail_backfill.py

@@ -1,270 +0,0 @@
-"""Scheduled backfill: detail-enrichment for legacy avito listings (#1551).
-
-Nightly window 09:00-12:00 UTC (migration 112, source=avito_detail_backfill).
-Offset from avito_city_sweep (~03-04 UTC) to avoid sharing proxy IP simultaneously.
-
-Problem: ~15243/15917 avito listings have detail_enriched_at IS NULL.
-City sweep ENRICH_DETAIL only enriches top-N proximate listings per anchor.
-Legacy listings (older than 2h or outside radius) are never enriched.
-
-Solution: single snapshot SELECT at start (guarantees termination), same proxy
-session path as scrape_pipeline.py step 5. Block handling mirrors step 5:
-rotate IP on every block, abort after max_consecutive_blocks (mark_done not
-mark_failed -- block is temporary, retry next night via NULL detail_enriched_at).
-"""
-
-from __future__ import annotations
-
-import asyncio
-import logging
-import time
-from dataclasses import dataclass, field
-from urllib.parse import urlparse
-
-from curl_cffi.requests import AsyncSession
-from sqlalchemy import text
-from sqlalchemy.orm import Session
-
-from app.core.config import settings
-from app.services import scrape_runs as runs_mod
-from app.services.scrape_pipeline import _CHROME_HEADERS, _avito_proxies
-from app.services.scrapers.avito import AvitoScraper
-from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment
-from app.services.scrapers.avito_exceptions import (
-    AvitoBlockedError,
-    AvitoRateLimitedError,
-)
-from app.services.scrapers.browser_fetcher import BrowserFetcher
-
-logger = logging.getLogger(__name__)
-
-__all__ = [
-    "AvitoDetailBackfillResult",
-    "run_avito_detail_backfill",
-]
-
-
-@dataclass
-class AvitoDetailBackfillResult:
-    """Counters for one backfill run."""
-
-    attempted: int = 0
-    enriched: int = 0
-    blocked: 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,
-            "blocked": self.blocked,
-            "failed": self.failed,
-            "duration_sec": int(self.duration_sec),
-        }
-
-
-async def run_avito_detail_backfill(
-    db: Session,
-    *,
-    run_id: int,
-    params: dict,
-) -> AvitoDetailBackfillResult:
-    """Backfill detail_enriched_at for legacy avito listings via mobile proxy.
-
-    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, default 6.0s.
-        max_consecutive_blocks: int -- abort threshold, default 5.
-
-    Lifecycle: update_heartbeat -> snapshot -> loop with budget guard ->
-    mark_done (incl. partial/block-abort) / mark_failed (exception only).
-    """
-    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", 6.0))
-    max_consecutive_blocks = int(params.get("max_consecutive_blocks", 5))
-
-    counters = AvitoDetailBackfillResult()
-    current_counters: dict[str, int] = counters.to_dict()
-
-    browser_mode = settings.scraper_fetch_mode == "browser"
-    session: AsyncSession | None = None
-    browser_fetcher: BrowserFetcher | None = None
-    own_session = False
-    own_browser = False
-
-    scraper = AvitoScraper()
-    start = time.monotonic()
-
-    try:
-        # Setup session (mirrors run_avito_pipeline lines 161-183)
-        if browser_mode:
-            browser_fetcher = BrowserFetcher()
-            await browser_fetcher.__aenter__()
-            own_browser = True
-            scraper._browser = browser_fetcher
-        else:
-            own_session = True
-            session = AsyncSession(
-                impersonate="chrome120",
-                timeout=25,
-                headers=_CHROME_HEADERS,
-                proxies=_avito_proxies(),
-            )
-            scraper._cffi = session
-
-        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 = 'avito'
-                      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(
-                "avito_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(
-            "avito_detail_backfill: run_id=%d snapshot=%d (budget=%.0fs "
-            "delay=%.1fs max_blocks=%d browser=%s)",
-            run_id,
-            len(snapshot),
-            budget_sec,
-            request_delay_sec,
-            max_consecutive_blocks,
-            browser_mode,
-        )
-
-        consecutive_blocks = 0
-        do_sleep = False
-
-        for idx, row in enumerate(snapshot):
-            # Budget guard
-            elapsed = time.monotonic() - start
-            if elapsed > budget_sec:
-                logger.info(
-                    "avito_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"]
-            counters.attempted += 1
-
-            # Normalise URL -> path (mirrors scrape_pipeline.py line 311)
-            item_url = urlparse(source_url).path if source_url.startswith("http") else source_url
-
-            try:
-                enrichment = await fetch_detail(
-                    item_url,
-                    cffi_session=session,
-                    browser_fetcher=browser_fetcher,
-                )
-                if save_detail_enrichment(db, enrichment):
-                    counters.enriched += 1
-                consecutive_blocks = 0
-
-            except (AvitoBlockedError, AvitoRateLimitedError) as e:
-                consecutive_blocks += 1
-                counters.blocked += 1
-                do_sleep = False
-                logger.warning(
-                    "avito_detail_backfill: run_id=%d BLOCKED #%d/%d (consecutive=%d): %s",
-                    run_id,
-                    idx + 1,
-                    len(snapshot),
-                    consecutive_blocks,
-                    e,
-                )
-                await scraper._rotate_ip()
-                if consecutive_blocks >= max_consecutive_blocks:
-                    logger.error(
-                        "avito_detail_backfill: run_id=%d ABORT -- %d consecutive blocks, "
-                        "IP rate-limited. enriched=%d attempted=%d",
-                        run_id,
-                        consecutive_blocks,
-                        counters.enriched,
-                        counters.attempted,
-                    )
-                    break
-
-            except Exception as e:
-                counters.failed += 1
-                logger.warning(
-                    "avito_detail_backfill: run_id=%d listing %s failed: %s",
-                    run_id,
-                    source_url,
-                    e,
-                )
-                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(
-            "avito_detail_backfill: run_id=%d DONE -- attempted=%d enriched=%d "
-            "blocked=%d failed=%d duration=%.1fs",
-            run_id,
-            counters.attempted,
-            counters.enriched,
-            counters.blocked,
-            counters.failed,
-            counters.duration_sec,
-        )
-        return counters
-
-    except Exception as exc:
-        counters.duration_sec = time.monotonic() - start
-        logger.exception(
-            "avito_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
-
-    finally:
-        if own_session and session is not None:
-            await session.close()
-        if own_browser and browser_fetcher is not None:
-            await browser_fetcher.__aexit__(None, None, None)

tradein-mvp/backend/data/sql/112_scrape_schedules_seed_avito_detail_backfill.sql

@@ -1,54 +0,0 @@
--- 112_scrape_schedules_seed_avito_detail_backfill.sql
--- Nightly detail-enrichment backfill for legacy avito listings (detail_enriched_at IS NULL).
---
--- Root cause (#1551):
---   City sweep ENRICH_DETAIL step only enriches top-N listings proximate to each anchor.
---   Legacy avito listings (older than 2h or outside radius) are never enriched.
---   On prod: 15243/15917 avito listings with detail_enriched_at IS NULL.
---   These carry no kitchen_area_m2 / ceiling_height_m / finishing / views / description.
---
--- Solution:
---   Nightly backfill task (avito_detail_backfill) takes a SNAPSHOT of pending listings
---   (single SELECT LIMIT batch_size at start, same proxy path as scrape_pipeline step 5),
---   then iterates with budget guard. Block handling: rotate IP on every AvitoBlockedError,
---   abort after max_consecutive_blocks (mark_done, NOT mark_failed -- block is temporary).
---
--- Window 09:00-12:00 UTC (12:00-15:00 MSK):
---   - After city sweeps (~03-04 UTC, depositing new listings).
---   - After geocode_missing_listings backfill (06:00-09:00 UTC).
---   - Offset from avito_city_sweep to avoid sharing proxy IP simultaneously.
---
--- default_params:
---   batch_size  -- listings per run (800 ~ 1.3h at 6s/listing, fits in 3h window).
---   budget_sec  -- wall-clock budget; 3600 = 1 hour (safe within 3h window).
---   request_delay_sec -- anti-bot delay between requests (6s = ~600 req/hour).
---   max_consecutive_blocks -- abort threshold for consecutive IP blocks (5).
---
--- DEPENDENCIES: 052_scrape_schedules.sql (table + UNIQUE(source)).
--- Idempotent: ON CONFLICT (source) DO NOTHING.
-
-BEGIN;
-
-INSERT INTO scrape_schedules (
-    source,
-    enabled,
-    window_start_hour,
-    window_end_hour,
-    next_run_at,
-    default_params
-)
-VALUES
-(
-    'avito_detail_backfill',
-    true,
-    9,
-    12,
-    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 9)) AT TIME ZONE 'UTC',
-    '{"batch_size": 800, "budget_sec": 3600, "request_delay_sec": 6, "max_consecutive_blocks": 5}'::jsonb
-)
-ON CONFLICT (source) DO NOTHING;
-
-COMMENT ON TABLE scrape_schedules IS
-    'In-app scheduler config (replaces cron-script setup). Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), asking_to_sold_ratio_refresh (#648), refresh_search_matview (#769), yandex_address_backfill (#855, EKB pilot), sber_index_pull (#887, monthly), rosreestr_quarter_poll (#888, monthly), cian_city_sweep (dormant, #973), yandex_newbuilding_sweep (dormant, #974), geocode_missing_listings (#1: listings geom backfill, all sources), avito_detail_backfill (#1551: nightly detail-enrichment backfill for legacy avito listings).';
-
-COMMIT;

tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py

@@ -1,244 +0,0 @@
-from __future__ import annotations
-
-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.avito_detail_backfill import (  # noqa: E402
-    AvitoDetailBackfillResult,
-    run_avito_detail_backfill,
-)
-
-# ---------------------------------------------------------------------------
-# Helpers
-# ---------------------------------------------------------------------------
-
-
-def _make_snapshot(n: int) -> list[dict]:
-    return [{"id": i + 1, "source_url": f"/items/{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
-
-
-_FETCH = "app.tasks.avito_detail_backfill.fetch_detail"
-_SAVE = "app.tasks.avito_detail_backfill.save_detail_enrichment"
-_RUNS = "app.tasks.avito_detail_backfill.runs_mod"
-_SLEEP = "app.tasks.avito_detail_backfill.asyncio.sleep"
-_SESSION = "app.tasks.avito_detail_backfill.AsyncSession"
-_SCRAPER = "app.tasks.avito_detail_backfill.AvitoScraper"
-_SETTINGS = "app.tasks.avito_detail_backfill.settings"
-
-# ---------------------------------------------------------------------------
-# Tests
-# ---------------------------------------------------------------------------
-
-
-@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()
-    fake_session = AsyncMock()
-    fake_settings = MagicMock(scraper_fetch_mode="cffi")
-    with (
-        patch(_SETTINGS, fake_settings),
-        patch(_SESSION, return_value=fake_session),
-        patch(_SCRAPER),
-        patch(_RUNS, runs),
-        patch(_FETCH) as mock_fetch,
-    ):
-        result = await run_avito_detail_backfill(
-            db, run_id=1, params={"batch_size": 10, "budget_sec": 60}
-        )
-
-    assert isinstance(result, AvitoDetailBackfillResult)
-    assert result.attempted == 0
-    assert result.enriched == 0
-    mock_fetch.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_fetch = AsyncMock(return_value=mock_enrichment)
-    mock_save = MagicMock(return_value=True)
-    fake_settings = MagicMock(scraper_fetch_mode="cffi")
-    with (
-        patch(_SETTINGS, fake_settings),
-        patch(_SESSION, return_value=AsyncMock()),
-        patch(_SCRAPER),
-        patch(_RUNS, runs),
-        patch(_FETCH, mock_fetch),
-        patch(_SAVE, mock_save),
-        patch(_SLEEP, new_callable=AsyncMock),
-    ):
-        result = await run_avito_detail_backfill(
-            db, run_id=2, params={"batch_size": 10, "budget_sec": 3600}
-        )
-
-    assert result.attempted == 3
-    assert result.enriched == 3
-    assert result.blocked == 0
-    assert result.failed == 0
-    assert mock_fetch.call_count == 3
-    runs.mark_done.assert_called_once()
-    runs.mark_failed.assert_not_called()
-
-
-@pytest.mark.asyncio
-async def test_backfill_blocked_abort_after_max_consecutive() -> None:
-    """5 consecutive AvitoBlockedError -> abort, mark_done (NOT mark_failed), rotate_ip x5."""
-    from app.services.scrapers.avito_exceptions import AvitoBlockedError
-
-    snapshot = _make_snapshot(10)
-    db = _mock_db(snapshot)
-    runs = MagicMock()
-    blocked_exc = AvitoBlockedError("ip blocked")
-    mock_fetch = AsyncMock(side_effect=blocked_exc)
-    mock_scraper = MagicMock()
-    mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
-    fake_settings = MagicMock(scraper_fetch_mode="cffi")
-    with (
-        patch(_SETTINGS, fake_settings),
-        patch(_SESSION, return_value=AsyncMock()),
-        patch(_SCRAPER, mock_scraper),
-        patch(_RUNS, runs),
-        patch(_FETCH, mock_fetch),
-        patch(_SLEEP, new_callable=AsyncMock),
-    ):
-        result = await run_avito_detail_backfill(
-            db, run_id=3, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5}
-        )
-
-    assert result.blocked == 5
-    assert result.attempted == 5
-    assert result.enriched == 0
-    runs.mark_done.assert_called_once()
-    runs.mark_failed.assert_not_called()
-    assert mock_scraper.return_value._rotate_ip.call_count == 5
-
-
-@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()
-    mono_values = iter([0.0, 999.0, 999.0])
-    fake_settings = MagicMock(scraper_fetch_mode="cffi")
-    with (
-        patch(_SETTINGS, fake_settings),
-        patch(_SESSION, return_value=AsyncMock()),
-        patch(_SCRAPER),
-        patch(_RUNS, runs),
-        patch("app.tasks.avito_detail_backfill.time.monotonic", side_effect=mono_values),
-        patch(_FETCH) as mock_fetch,
-    ):
-        await run_avito_detail_backfill(db, run_id=4, params={"batch_size": 5, "budget_sec": 1})
-
-    mock_fetch.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()
-    fake_settings = MagicMock(scraper_fetch_mode="cffi")
-    with (
-        patch(_SETTINGS, fake_settings),
-        patch(_SESSION, return_value=AsyncMock()),
-        patch(_SCRAPER),
-        patch(_RUNS, runs),
-    ):
-        with pytest.raises(RuntimeError, match="DB connection lost"):
-            await run_avito_detail_backfill(
-                db, run_id=5, 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_rotate_ip_called_on_each_block() -> None:
-    """1 block + 1 success -> rotate_ip called once, enriched=1."""
-    from app.services.scrapers.avito_exceptions import AvitoBlockedError
-
-    snapshot = _make_snapshot(2)
-    db = _mock_db(snapshot)
-    runs = MagicMock()
-    mock_enrichment = MagicMock()
-    blocked_exc = AvitoBlockedError("x")
-    mock_fetch = AsyncMock(side_effect=[blocked_exc, mock_enrichment])
-    mock_scraper = MagicMock()
-    mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
-    fake_settings = MagicMock(scraper_fetch_mode="cffi")
-    with (
-        patch(_SETTINGS, fake_settings),
-        patch(_SESSION, return_value=AsyncMock()),
-        patch(_SCRAPER, mock_scraper),
-        patch(_RUNS, runs),
-        patch(_FETCH, mock_fetch),
-        patch(_SAVE, return_value=True),
-        patch(_SLEEP, new_callable=AsyncMock),
-    ):
-        result = await run_avito_detail_backfill(
-            db, run_id=6, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5}
-        )
-
-    assert result.enriched == 1
-    assert result.blocked == 1
-    assert mock_scraper.return_value._rotate_ip.call_count == 1
-    runs.mark_done.assert_called_once()
-
-
-@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_fetch = AsyncMock(side_effect=[RuntimeError("parse error"), mock_enrichment])
-    fake_settings = MagicMock(scraper_fetch_mode="cffi")
-    with (
-        patch(_SETTINGS, fake_settings),
-        patch(_SESSION, return_value=AsyncMock()),
-        patch(_SCRAPER),
-        patch(_RUNS, runs),
-        patch(_FETCH, mock_fetch),
-        patch(_SAVE, return_value=True),
-        patch(_SLEEP, new_callable=AsyncMock),
-    ):
-        result = await run_avito_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()