feat(tradein/domclick): production Layer B detail-backfill orchestrator (#2000)

Ports the cookie-injection + organic SERP-origin session->cookies->fetch_detail
wiring (PR #2430, PR #2433), previously only reachable via the manual debug
endpoint, into a scheduled orchestrator mirroring avito_detail_backfill.py's
shape: snapshot-once-then-loop, budget guard, SIGTERM-drain, jittered delay,
heartbeat every 25 attempts.

DomClickBlockedError (QRATOR challenge / browser-fetch failure) increments a
consecutive-block counter and aborts via mark_done (not mark_failed) past
max_consecutive_blocks -- a block is an expected operational outcome for a
QRATOR reputation-based anti-bot, not a task failure. DomClickParseError
(__SSR_STATE__ schema drift) is a plain failed++, neutral to the breaker.
Unlike Avito there is no curl fallback (one BrowserFetcher per run) and no
IP-rotation recovery (single dedicated residential proxy).

Migration 175 seeds the schedule with enabled=false -- activation is a
deliberate manual step after smoke-testing against a live cookie session.
default_params are more conservative than Avito's (batch_size=200,
request_delay_sec=12, max_consecutive_blocks=3) since a burned QRATOR
reputation/session is costlier to recover than an Avito IP-block.

Also fixes a stale docstring in scraper_kit's domclick/detail.py that
incorrectly named DataDome as the anti-bot mechanism -- it's QRATOR
(exit-IP reputation-based), per the extensive live verification in #2000.

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

@@ -251,6 +251,15 @@ async def _job_yandex_detail_backfill(
     await run_yandex_detail_backfill(db, run_id=run_id, params=params)
 
 
+# ── domclick_detail_backfill — async, owns lifecycle (issue #2000 Layer B) ───
+async def _job_domclick_detail_backfill(
+    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
+) -> None:
+    from app.tasks.domclick_detail_backfill import run_domclick_detail_backfill
+
+    await run_domclick_detail_backfill(db, run_id=run_id, params=params)
+
+
 # ── geoportal_coords_backfill — sync local exact match в executor (#1967) ─────
 async def _job_geoportal_coords_backfill(
     db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
@@ -363,7 +372,7 @@ def build_product_handlers(ctx: SchedulerContext) -> dict[str, Handler]:
     """Реестр НЕ-sweep продуктовых source→Handler для kit build_registry.
 
     Kit-native sweeps (avito/yandex/cian/domclick city/full-load/newbuilding) НЕ здесь —
-    их даёт build_registry(_default_kit_handlers). Здесь — 17 именованных + 1 wildcard
+    их даёт build_registry(_default_kit_handlers). Здесь — 18 именованных + 1 wildcard
     (deactivate_stale_*), покрывающие каждый НЕ-sweep source боевого scheduler-dispatch.
 
     `ctx` — принят для симметрии контракта; сами Handler-job'ы получают ctx во время
@@ -397,6 +406,9 @@ def build_product_handlers(ctx: SchedulerContext) -> dict[str, Handler]:
         ),
         "avito_detail_backfill": Handler(_job_avito_detail_backfill, "avito_detail_backfill"),
         "yandex_detail_backfill": Handler(_job_yandex_detail_backfill, "yandex_detail_backfill"),
+        "domclick_detail_backfill": Handler(
+            _job_domclick_detail_backfill, "domclick_detail_backfill"
+        ),
         "cadastral_geo_match": Handler(_job_cadastral_geo_match, "cadastral_geo_match"),
         "osm_poi_ekb_refresh": Handler(_job_osm_poi_ekb_refresh, "osm_poi_ekb_refresh"),
         "house_imv_backfill": Handler(_job_house_imv_backfill, "house_imv_backfill"),

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

@@ -0,0 +1,320 @@
+"""Scheduled backfill: detail-enrichment (Layer B) for domklik listings (issue #2000).
+
+Nightly window 15:00-18:00 UTC (migration 175, source=domclick_detail_backfill).
+Offset from avito_detail_backfill (09-12 UTC) and yandex_detail_backfill (12-15 UTC)
+to avoid parallel egress across scraper sources sharing the tradein-browser pool.
+
+Problem: DomClick SERP layer (Layer A, scraper_kit.providers.domclick.serp) discovers
+listings via the BFF JSON API, but detail-enrichment (Layer B — repair_state,
+living/kitchen area, year_built, owners_count, price history) requires a per-card
+browser-fetch that is QRATOR-guarded (issue #2000). Two prior PRs solved the block at
+the single-fetch level:
+  - PR #2430 -- organic same-site SERP-origin navigation before the card fetch.
+  - PR #2433 -- cookie-injection MVP (authenticated test-account session bypasses the
+    QRATOR reputation-block even on an already-suspicious proxy-IP).
+Both are wired together in the debug endpoint `POST /scrape/domclick/debug/detail-fetch`
+(app/api/v1/admin.py). This module ports that same session -> cookies -> fetch_detail
+wiring into the production scheduled orchestrator (previously only reachable manually).
+
+Solution: single snapshot SELECT at start (guarantees termination) + one BrowserFetcher
+per run (async context manager, source="domclick" -- dedicated residential proxy pool,
+see 173_scrape_proxies_add_domclick_affinity.sql) + cookies loaded ONCE via
+domclick_session.load_session(db) and threaded into every fetch_detail() call.
+
+NAMING TRAP (verified live against prod DB 2026-07-04, do NOT "fix" this anywhere):
+listings.source value for DomClick rows is 'domklik' (with a "k") -- that is the
+DATA/business identifier used across listings.source, DomClickScraper.name/.source,
+schemas, PDF exporter labels. 'domclick' (with a "c") is the newer INFRA/transport
+identifier -- BrowserFetcher(source="domclick"), scrape_proxies.provider_affinity,
+the SSRF host-guard, cookie-session tables. Both strings are correct in their own
+context; this split is intentional and pre-existing, not a bug to unify here.
+
+No curl fallback: unlike Avito's dual-mode (curl_cffi backconnect OR browser),
+DomClick only has a BrowserFetcher path -- scraper_kit.providers.domclick.detail.
+fetch_detail() raises DomClickBlockedError if the browser fetch itself fails. Exactly
+one BrowserFetcher is constructed per run.
+
+Exception triad differs from Avito:
+  - DomClickBlockedError (QRATOR challenge page OR any browser-fetch failure) --
+    increments consecutive_blocks, abort via mark_done (NOT mark_failed) once
+    max_consecutive_blocks is hit -- a block-abort is an expected operational
+    outcome (QRATOR reputation burn), not a task failure. Mirrors Avito's
+    AvitoBlockedError handling. No IP-rotation/cooldown recovery step exists here
+    (DomClick uses one dedicated residential proxy, not a rotating pool) -- an
+    aborted run simply retries the remaining backlog next window.
+  - DomClickParseError (__SSR_STATE__ missing/malformed -- schema drift, NOT a
+    block) -- counted as failed++, logged, does NOT touch consecutive_blocks and
+    does NOT abort the run (neutral to the block-breaker, mirrors how Avito's
+    AvitoListingGoneError is neutral to its breaker for a different reason).
+  - There is no "listing gone / 404" exception type for DomClick yet -- not
+    invented here.
+
+Cookie injection is mandatory wiring, not optional: cookies are loaded ONCE per run.
+If None (no valid session uploaded / expired) -- the run still proceeds (cookie-
+injection is a QRATOR-defeat mechanism, not a hard requirement; organic SERP-origin
+navigation from PR #2430 still applies) but a warning is logged once at run start so
+operators notice the test-account session needs refreshing via
+`POST /scrape/domclick/upload-cookies` (no auto-login -- documented MVP limitation,
+see app/services/domclick_session.py module docstring).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import random
+import time
+from dataclasses import dataclass, field
+
+from scraper_kit.browser_fetcher import BrowserFetcher
+from scraper_kit.domclick_exceptions import DomClickBlockedError, DomClickParseError
+from scraper_kit.providers.domclick.detail import fetch_detail, save_detail_enrichment
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.core.config import settings
+from app.core.shutdown import shutdown_requested
+from app.services import domclick_session as domclick_session_svc
+from app.services import scrape_runs as runs_mod
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+    "DomClickDetailBackfillResult",
+    "run_domclick_detail_backfill",
+]
+
+
+@dataclass
+class DomClickDetailBackfillResult:
+    """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_domclick_detail_backfill(
+    db: Session,
+    *,
+    run_id: int,
+    params: dict,
+) -> DomClickDetailBackfillResult:
+    """Backfill detail_enriched_at for domklik listings via cookie-injected browser fetch.
+
+    Params (from default_params jsonb in scrape_schedules):
+        batch_size: int -- snapshot size (SELECT LIMIT), default 200.
+        budget_sec: float -- wall-clock budget per run, default 3600s.
+        request_delay_sec: float -- delay between listings, default 12.0s.
+        max_consecutive_blocks: int -- abort threshold, default 3.
+
+    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", 200))
+    budget_sec = float(params.get("budget_sec", 3600))
+    request_delay_sec = float(params.get("request_delay_sec", 12.0))
+    max_consecutive_blocks = int(params.get("max_consecutive_blocks", 3))
+
+    counters = DomClickDetailBackfillResult()
+    current_counters: dict[str, int] = counters.to_dict()
+    start = time.monotonic()
+
+    logger.info("domclick_detail_backfill: run_id=%d starting", run_id)
+
+    try:
+        # Cookie injection (#2000 PR #2433) -- loaded ONCE per run, threaded into every
+        # fetch_detail() call below. None is a valid (degraded) state, not an error.
+        cookies = domclick_session_svc.load_session(db)
+        if cookies is None:
+            logger.warning(
+                "domclick_detail_backfill: run_id=%d -- no valid DomClick session cookies "
+                "in DB; proceeding WITHOUT cookie-injection (QRATOR-defeat degraded to "
+                "organic SERP-origin navigation only, PR #2430). Refresh test-account "
+                "session via POST /scrape/domclick/upload-cookies.",
+                run_id,
+            )
+
+        runs_mod.update_heartbeat(db, run_id, current_counters)
+
+        # SNAPSHOT: single SELECT at start -- NOT re-selected in loop.
+        # NAMING TRAP (see module docstring): source='domklik' here (data identifier) is
+        # NOT the same string as BrowserFetcher(source="domclick") below (infra
+        # identifier) -- both are correct, do not "fix" one to match the other.
+        snapshot = (
+            db.execute(
+                text(
+                    """
+                    SELECT id, source_url
+                    FROM listings
+                    WHERE source = 'domklik'
+                      AND detail_enriched_at IS NULL
+                      AND source_url IS NOT NULL
+                      AND is_active = TRUE
+                    ORDER BY (lat IS NULL) DESC, scraped_at DESC NULLS LAST
+                    LIMIT CAST(:batch_size AS int)
+                    """
+                ),
+                {"batch_size": batch_size},
+            )
+            .mappings()
+            .all()
+        )
+
+        if not snapshot:
+            logger.info(
+                "domclick_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(
+            "domclick_detail_backfill: run_id=%d snapshot=%d (budget=%.0fs "
+            "delay=%.1fs max_blocks=%d cookies=%s)",
+            run_id,
+            len(snapshot),
+            budget_sec,
+            request_delay_sec,
+            max_consecutive_blocks,
+            "yes" if cookies is not None else "no",
+        )
+
+        consecutive_blocks = 0
+        do_sleep = False
+
+        # Exactly ONE BrowserFetcher per run (no curl fallback for DomClick, see
+        # module docstring). source="domclick" -- infra identifier, dedicated
+        # residential proxy (scrape_proxies.provider_affinity='domclick',
+        # 173_scrape_proxies_add_domclick_affinity.sql).
+        async with BrowserFetcher(source="domclick", endpoint=settings.browser_http_endpoint) as bf:
+            for idx, row in enumerate(snapshot):
+                # Budget guard
+                elapsed = time.monotonic() - start
+                if elapsed > budget_sec:
+                    logger.info(
+                        "domclick_detail_backfill: run_id=%d -- budget %.0fs exhausted "
+                        "(elapsed=%.1fs), stopping at #%d/%d",
+                        run_id,
+                        budget_sec,
+                        elapsed,
+                        idx,
+                        len(snapshot),
+                    )
+                    break
+
+                # #1182 Phase 2: кооперативный SIGTERM-drain (деплой recreate scraper).
+                if shutdown_requested():
+                    logger.info(
+                        "domclick_detail_backfill: run_id=%d SIGTERM-drain — stopping at " "#%d/%d",
+                        run_id,
+                        idx,
+                        len(snapshot),
+                    )
+                    break
+
+                # Jittered delay (±30%) -- organic pacing, matches the request_delay_sec
+                # spirit of PR #2430's same-site navigation (less robotic cadence).
+                if do_sleep:
+                    await asyncio.sleep(request_delay_sec * random.uniform(0.7, 1.4))
+                do_sleep = True
+
+                source_url: str = row["source_url"]
+                listing_id: int = row["id"]
+                counters.attempted += 1
+
+                try:
+                    enrichment = await fetch_detail(source_url, browser_fetcher=bf, cookies=cookies)
+                    if save_detail_enrichment(db, listing_id, enrichment):
+                        counters.enriched += 1
+                    consecutive_blocks = 0
+
+                except DomClickParseError as e:
+                    # Schema drift, not a block -- neutral to the block-breaker (does
+                    # NOT touch consecutive_blocks, does NOT abort the run).
+                    counters.failed += 1
+                    logger.warning(
+                        "domclick_detail_backfill: run_id=%d listing %s PARSE-ERROR "
+                        "(schema drift, not a block): %s",
+                        run_id,
+                        source_url,
+                        e,
+                    )
+
+                except DomClickBlockedError as e:
+                    consecutive_blocks += 1
+                    counters.blocked += 1
+                    logger.warning(
+                        "domclick_detail_backfill: run_id=%d BLOCKED #%d/%d "
+                        "(consecutive=%d): %s",
+                        run_id,
+                        idx + 1,
+                        len(snapshot),
+                        consecutive_blocks,
+                        e,
+                    )
+                    if consecutive_blocks >= max_consecutive_blocks:
+                        logger.error(
+                            "domclick_detail_backfill: run_id=%d ABORT -- %d consecutive "
+                            "blocks, QRATOR reputation likely burned for the session/proxy. "
+                            "enriched=%d attempted=%d",
+                            run_id,
+                            consecutive_blocks,
+                            counters.enriched,
+                            counters.attempted,
+                        )
+                        break
+
+                except Exception as e:
+                    counters.failed += 1
+                    logger.warning(
+                        "domclick_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(
+            "domclick_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(
+            "domclick_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

tradein-mvp/backend/data/sql/175_scrape_schedules_seed_domclick_detail_backfill.sql

@@ -0,0 +1,80 @@
+-- 175_scrape_schedules_seed_domclick_detail_backfill.sql
+-- Nightly detail-enrichment backfill (Layer B) for domklik listings (issue #2000).
+--
+-- Root cause (#2000):
+--   DomClick SERP layer (Layer A) discovers listings via the BFF JSON API but does not
+--   fetch card-detail pages (repair_state, living/kitchen area, year_built, owners_count,
+--   price history). Detail-enrichment requires a per-card browser-fetch that is
+--   QRATOR-guarded -- solved at the single-fetch level by PR #2430 (organic same-site
+--   SERP-origin navigation) + PR #2433 (cookie-injection MVP, authenticated test-account
+--   session bypasses the QRATOR reputation-block even on an already-suspicious proxy-IP).
+--   This schedule wires that session->cookies->fetch_detail path into the production
+--   scheduled orchestrator (app.tasks.domclick_detail_backfill), previously only
+--   reachable via the manual debug endpoint (POST /scrape/domclick/debug/detail-fetch).
+--
+-- Solution:
+--   Nightly backfill task (domclick_detail_backfill) takes a SNAPSHOT of pending
+--   listings (single SELECT LIMIT batch_size at start, source='domklik' -- see NAMING
+--   TRAP note in app/tasks/domclick_detail_backfill.py module docstring), then iterates
+--   with budget guard through ONE BrowserFetcher(source="domclick") per run with cookies
+--   loaded once via domclick_session.load_session(). DomClickBlockedError (QRATOR
+--   challenge) increments a consecutive-block counter, aborting the run (mark_done, NOT
+--   mark_failed -- block is an expected operational outcome) once max_consecutive_blocks
+--   is hit. DomClickParseError (schema drift) is a plain failed++, does not abort.
+--
+-- Window 15:00-18:00 UTC (18:00-21:00 MSK):
+--   - Offset from avito_detail_backfill (09-12 UTC) and yandex_detail_backfill (12-15
+--     UTC) -- avoids simultaneous egress across scraper sources sharing the
+--     tradein-browser pool.
+--
+-- default_params (more conservative than Avito -- DomClick is QRATOR
+-- reputation-based, single dedicated residential proxy, no IP-rotation recovery path
+-- on block; a burned session/proxy reputation is costlier to recover than an Avito
+-- IP-block, which can rotate):
+--   batch_size  -- 200 (vs Avito's 800): smaller batch keeps a bad run's blast radius
+--                  small while the pipeline is still ramping up post-#2000.
+--   budget_sec  -- 3600 = 1 hour (same as Avito, fits well within the 3h window).
+--   request_delay_sec -- 12s (vs Avito's 6s): slower organic pacing matches the
+--                  same-site SERP-origin navigation cadence from PR #2430 -- QRATOR is
+--                  reputation-based, not purely rate-based, but a more human cadence
+--                  reduces the odds of tripping it in the first place.
+--   max_consecutive_blocks -- 3 (vs Avito's 5): DomClick has no IP-rotation recovery
+--                  on block (one dedicated residential proxy, not a pool) -- aborting
+--                  sooner avoids hammering an already-reputation-damaged proxy/session
+--                  and preserves it for the next scheduled window.
+--
+-- Seeded with enabled=false: activation is a deliberate, separate manual step after
+-- this migration deploys and the orchestrator is smoke-tested against a live cookie
+-- session (see app/services/domclick_session.py -- MVP requires a manually uploaded
+-- test-account session, no auto-login).
+--
+-- DEPENDENCIES: 052_scrape_schedules.sql (table + UNIQUE(source)),
+--   173_scrape_proxies_add_domclick_affinity.sql (source="domclick" proxy pool),
+--   174_domclick_session_cookies.sql (cookie storage).
+-- 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
+(
+    'domclick_detail_backfill',
+    false,
+    15,
+    18,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 15)) AT TIME ZONE 'UTC',
+    '{"batch_size": 200, "budget_sec": 3600, "request_delay_sec": 12, "max_consecutive_blocks": 3}'::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), domclick_detail_backfill (#2000: nightly Layer B detail-enrichment backfill for domklik listings, cookie-injection + QRATOR-aware, disabled by default until smoke-tested).';
+
+COMMIT;

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

@@ -0,0 +1,390 @@
+"""Tests для DomClick detail-backfill orchestrator (issue #2000 Layer B).
+
+Зеркалит конвенции test_avito_detail_backfill.py: module-level patch-target строки,
+_mock_db(snapshot) helper, runs = MagicMock() assertions. DomClick-специфика:
+единственный BrowserFetcher (нет curl-fallback), cookies из domclick_session_svc
+(мокается целиком как модуль), эксепшн-триада DomClickBlockedError/DomClickParseError
+вместо Avito's Blocked/RateLimited/ListingGone.
+"""
+
+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 scraper_kit.domclick_exceptions import DomClickBlockedError, DomClickParseError  # noqa: E402
+
+from app.core import shutdown as _sd  # noqa: E402
+from app.tasks.domclick_detail_backfill import (  # noqa: E402
+    DomClickDetailBackfillResult,
+    run_domclick_detail_backfill,
+)
+
+
+@pytest.fixture(autouse=True)
+def _reset_shutdown() -> None:
+    """shutdown — module-global Event: чистим вокруг каждого теста (изоляция #1182)."""
+    _sd.reset_shutdown()
+    yield
+    _sd.reset_shutdown()
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_snapshot(n: int) -> list[dict]:
+    return [
+        {
+            "id": i + 1,
+            "source_url": f"https://ekaterinburg.domclick.ru/card/sale__flat__{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
+
+
+def _mock_browser_fetcher_cls() -> MagicMock:
+    """MagicMock class whose instance is a working async context manager."""
+    instance = AsyncMock()
+    instance.__aenter__ = AsyncMock(return_value=instance)
+    instance.__aexit__ = AsyncMock(return_value=False)
+    return MagicMock(return_value=instance)
+
+
+_FETCH = "app.tasks.domclick_detail_backfill.fetch_detail"
+_SAVE = "app.tasks.domclick_detail_backfill.save_detail_enrichment"
+_RUNS = "app.tasks.domclick_detail_backfill.runs_mod"
+_SLEEP = "app.tasks.domclick_detail_backfill.asyncio.sleep"
+_SETTINGS = "app.tasks.domclick_detail_backfill.settings"
+_SHUTDOWN = "app.tasks.domclick_detail_backfill.shutdown_requested"
+_BROWSER_FETCHER = "app.tasks.domclick_detail_backfill.BrowserFetcher"
+_SESSION_SVC = "app.tasks.domclick_detail_backfill.domclick_session_svc"
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_backfill_empty_snapshot_marks_done() -> None:
+    """Empty snapshot -> mark_done immediately, no fetch calls, no BrowserFetcher open."""
+    db = _mock_db([])
+    runs = MagicMock()
+    fake_settings = MagicMock(browser_http_endpoint="http://browser:9000")
+    mock_svc = MagicMock()
+    mock_svc.load_session.return_value = {"CAS_ID": "123"}
+    mock_bf_cls = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION_SVC, mock_svc),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH) as mock_fetch,
+    ):
+        result = await run_domclick_detail_backfill(
+            db, run_id=1, params={"batch_size": 10, "budget_sec": 60}
+        )
+
+    assert isinstance(result, DomClickDetailBackfillResult)
+    assert result.attempted == 0
+    assert result.enriched == 0
+    mock_fetch.assert_not_called()
+    mock_bf_cls.assert_not_called()
+    runs.mark_done.assert_called_once()
+    runs.mark_failed.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_backfill_processes_snapshot_with_cookies_threaded() -> None:
+    """3 listings -> all fetched+enriched, cookies from load_session() threaded into
+    every fetch_detail() call kwargs, 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(browser_http_endpoint="http://browser:9000")
+    fake_cookies = {"CAS_ID": "999", "qrator_jsid2": "abc"}
+    mock_svc = MagicMock()
+    mock_svc.load_session.return_value = fake_cookies
+    mock_bf_cls = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION_SVC, mock_svc),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SAVE, mock_save),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        result = await run_domclick_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
+    mock_bf_cls.assert_called_once_with(
+        source="domclick", endpoint=fake_settings.browser_http_endpoint
+    )
+    for call in mock_fetch.call_args_list:
+        _, kwargs = call
+        assert kwargs.get("cookies") == fake_cookies
+        assert kwargs.get("browser_fetcher") is not None
+    runs.mark_done.assert_called_once()
+    runs.mark_failed.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_backfill_cookies_none_still_proceeds_with_warning(caplog) -> None:
+    """No valid session (load_session()->None) -> run still proceeds (fetch_detail
+    called with cookies=None), but a warning is logged so operators refresh the session.
+    """
+    snapshot = _make_snapshot(1)
+    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(browser_http_endpoint="http://browser:9000")
+    mock_svc = MagicMock()
+    mock_svc.load_session.return_value = None
+    mock_bf_cls = _mock_browser_fetcher_cls()
+    with (
+        caplog.at_level("WARNING"),
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION_SVC, mock_svc),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SAVE, mock_save),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        result = await run_domclick_detail_backfill(
+            db, run_id=3, params={"batch_size": 10, "budget_sec": 3600}
+        )
+
+    assert result.enriched == 1
+    _, kwargs = mock_fetch.call_args
+    assert kwargs.get("cookies") is None
+    assert "no valid DomClick session cookies" in caplog.text
+    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:
+    """3 consecutive DomClickBlockedError -> abort, mark_done (NOT mark_failed).
+
+    No IP-rotation recovery step exists for DomClick (single dedicated proxy) --
+    abort happens on the SAME iteration the threshold is hit, no extra recovery calls.
+    """
+    snapshot = _make_snapshot(10)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    blocked_exc = DomClickBlockedError("QRATOR challenge page detected")
+    mock_fetch = AsyncMock(side_effect=blocked_exc)
+    fake_settings = MagicMock(browser_http_endpoint="http://browser:9000")
+    mock_svc = MagicMock()
+    mock_svc.load_session.return_value = {"CAS_ID": "123"}
+    mock_bf_cls = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION_SVC, mock_svc),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        result = await run_domclick_detail_backfill(
+            db,
+            run_id=4,
+            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 3},
+        )
+
+    assert result.blocked == 3
+    assert result.attempted == 3
+    assert result.enriched == 0
+    runs.mark_done.assert_called_once()
+    runs.mark_failed.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_backfill_parse_error_counts_failed_no_abort() -> None:
+    """DomClickParseError (schema drift, not a block) -> failed++, consecutive-block
+    breaker NOT touched, run does NOT abort, continues to next listing.
+    """
+    snapshot = _make_snapshot(2)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    mock_enrichment = MagicMock()
+    mock_fetch = AsyncMock(
+        side_effect=[DomClickParseError("__SSR_STATE__ not found"), mock_enrichment]
+    )
+    mock_save = MagicMock(return_value=True)
+    fake_settings = MagicMock(browser_http_endpoint="http://browser:9000")
+    mock_svc = MagicMock()
+    mock_svc.load_session.return_value = {"CAS_ID": "123"}
+    mock_bf_cls = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION_SVC, mock_svc),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SAVE, mock_save),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        result = await run_domclick_detail_backfill(
+            db,
+            run_id=5,
+            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 3},
+        )
+
+    assert result.failed == 1
+    assert result.enriched == 1
+    assert result.attempted == 2
+    assert result.blocked == 0
+    runs.mark_done.assert_called_once()
+    runs.mark_failed.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_backfill_sigterm_drain_breaks_and_marks_done_partial() -> None:
+    """#1182 Phase 2: shutdown_requested() True -> loop выходит на границе карточки,
+    mark_done вызывается с ЧАСТИЧНЫМИ счётчиками (не mark_failed).
+    """
+    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(browser_http_endpoint="http://browser:9000")
+    mock_svc = MagicMock()
+    mock_svc.load_session.return_value = {"CAS_ID": "123"}
+    mock_bf_cls = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION_SVC, mock_svc),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SAVE, mock_save),
+        patch(_SLEEP, new_callable=AsyncMock),
+        patch(_SHUTDOWN, side_effect=[False, True]),
+    ):
+        result = await run_domclick_detail_backfill(
+            db, run_id=6, params={"batch_size": 10, "budget_sec": 3600}
+        )
+
+    assert result.attempted == 1
+    assert result.enriched == 1
+    assert mock_fetch.call_count == 1
+    runs.mark_done.assert_called_once()
+    runs.mark_failed.assert_not_called()
+    runs.mark_cancelled.assert_not_called()
+    done_counters = runs.mark_done.call_args.args[2]
+    assert done_counters["attempted"] == 1
+
+
+@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(browser_http_endpoint="http://browser:9000")
+    mock_svc = MagicMock()
+    mock_svc.load_session.return_value = {"CAS_ID": "123"}
+    mock_bf_cls = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION_SVC, mock_svc),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch("app.tasks.domclick_detail_backfill.time.monotonic", side_effect=mono_values),
+        patch(_FETCH) as mock_fetch,
+    ):
+        await run_domclick_detail_backfill(db, run_id=7, 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(browser_http_endpoint="http://browser:9000")
+    mock_svc = MagicMock()
+    mock_svc.load_session.return_value = {"CAS_ID": "123"}
+    mock_bf_cls = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION_SVC, mock_svc),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+    ):
+        with pytest.raises(RuntimeError, match="DB connection lost"):
+            await run_domclick_detail_backfill(
+                db, run_id=8, 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_generic_exception_continues_and_rolls_back() -> None:
+    """Unexpected exception on one listing -> failed++, db.rollback(), loop continues."""
+    snapshot = _make_snapshot(2)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    mock_enrichment = MagicMock()
+    mock_fetch = AsyncMock(side_effect=[RuntimeError("unexpected"), mock_enrichment])
+    fake_settings = MagicMock(browser_http_endpoint="http://browser:9000")
+    mock_svc = MagicMock()
+    mock_svc.load_session.return_value = {"CAS_ID": "123"}
+    mock_bf_cls = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION_SVC, mock_svc),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SAVE, return_value=True),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        result = await run_domclick_detail_backfill(
+            db, run_id=9, 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()

tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/domclick/detail.py

@@ -482,9 +482,13 @@ async def fetch_detail(
     в БД) → fetch без инъекции, остаётся только SERP-warmup ниже.
 
     Raises:
-        DomClickBlockedError: anti-bot challenge ИЛИ ошибка браузерного fetch
-            (нет curl-fallback — DataDome даёт 401 на curl). Оркестратор считает
-            это блоком и помечает listing failed.
+        DomClickBlockedError: anti-bot challenge ИЛИ ошибка браузерного fetch (нет
+            curl-fallback — анти-бот DomClick это QRATOR, reputation-based блок
+            конкретного exit-IP, а не DataDome-челлендж; повторные непрошедшие
+            зонды с одного IP жгут его репутацию, свежий/чистый IP проходит с
+            первого раза, инъекция cookies валидной аутентифицированной сессии
+            обходит блок даже на уже подозрительном IP — issue #2000). Оркестратор
+            считает это блоком и помечает listing failed.
         DomClickParseError: HTTP 200, но SSR-стейт не парсится (дрейф схемы).
     """
     parsed = urlsplit(card_url)