feat(tradein/domclick): production Layer B detail-backfill orchestrator (#2000) (#2436)
All checks were successful
Deploy Trade-In / changes (push) Successful in 10s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-browser (push) Has been skipped
Deploy Trade-In / test (push) Successful in 47s
Deploy Trade-In / build-backend (push) Successful in 1m30s
Deploy Trade-In / deploy (push) Successful in 1m32s

This commit is contained in:
lekss361 2026-07-04 19:49:07 +00:00
parent b2d42a631a
commit f93ef5374d
5 changed files with 810 additions and 4 deletions

View file

@ -251,6 +251,15 @@ async def _job_yandex_detail_backfill(
await run_yandex_detail_backfill(db, run_id=run_id, params=params) 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) ───── # ── geoportal_coords_backfill — sync local exact match в executor (#1967) ─────
async def _job_geoportal_coords_backfill( async def _job_geoportal_coords_backfill(
db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext 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. """Реестр НЕ-sweep продуктовых source→Handler для kit build_registry.
Kit-native sweeps (avito/yandex/cian/domclick city/full-load/newbuilding) НЕ здесь 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. (deactivate_stale_*), покрывающие каждый НЕ-sweep source боевого scheduler-dispatch.
`ctx` принят для симметрии контракта; сами Handler-job'ы получают ctx во время `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"), "avito_detail_backfill": Handler(_job_avito_detail_backfill, "avito_detail_backfill"),
"yandex_detail_backfill": Handler(_job_yandex_detail_backfill, "yandex_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"), "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"), "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"), "house_imv_backfill": Handler(_job_house_imv_backfill, "house_imv_backfill"),

View file

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

View file

@ -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;

View file

@ -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()

View file

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