bot-backend
eef23071c5
ПАРКОВКА (не мержить до прокси): код готов — parser на реальных __SSR_STATE__ путях (productCard.objectInfo/priceInfo.priceHistory/egrnData/houseInfo.info + pricePrediction AVM), orchestrator зеркало avito_detail_backfill, schedule 139 enabled=false. Деплоить нельзя: DomClick card/offer-card за QRATOR reputation-403 на прод mobile-прокси (нужен dedicated residential RU IP). Детали/аудит транспорта — Forgejo issue.
272 lines
11 KiB
Python
272 lines
11 KiB
Python
"""Scheduled backfill: DomClick Layer B detail-enrichment (#1846 follow-up).
|
||
|
||
Nightly window 15:00-18:00 UTC (migration 139, source=domclick_detail_backfill).
|
||
Offset after avito (09-12) / yandex (12-15) detail backfills + the city sweeps to
|
||
avoid parallel egress on the shared mobile proxy.
|
||
|
||
Problem: Layer A (domclick.py BFF sweep) собирает только базовые SERP-поля. Поля
|
||
карточки (renovation/repair_state, living/kitchen area, balconies, sale_type,
|
||
owners_count, encumbrances, year_built, views, priceHistory) живут только на
|
||
странице карточки (.../card/sale__flat__<id>), за DataDome.
|
||
|
||
Solution: single snapshot SELECT at start (guarantees termination). Транспорт —
|
||
BrowserFetcher(source="cian"): card-хосты DomClick заблокированы QRATOR через
|
||
domclick/avito-прокси, но чисты через cian-прокси (verified live 2026-06-27).
|
||
Одна браузер-сессия открывается на весь прогон и передаётся в fetch_detail.
|
||
Block-handling: DomClickBlockedError транзиентен (anti-bot / fetch-fail)
|
||
→ abort после max_consecutive_blocks с mark_done (НЕ mark_failed; retry следующей ночью
|
||
через NULL detail_enriched_at). DomClickParseError (HTTP 200, но SSR не парсится) →
|
||
листинг остаётся необогащённым (rollback), переотбирается на следующем прогоне.
|
||
|
||
Schedule ships DORMANT (seed 139, enabled=false) — enable manually after prod-smoke.
|
||
Depends on Layer A (#1979 / domclick_city_sweep) having populated source='domklik'
|
||
listings first.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.services import scrape_runs as runs_mod
|
||
from app.services.scrapers.browser_fetcher import BrowserFetcher
|
||
from app.services.scrapers.domclick_detail import fetch_detail, save_detail_enrichment
|
||
from app.services.scrapers.domclick_exceptions import DomClickBlockedError, DomClickParseError
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
__all__ = [
|
||
"DomClickDetailBackfillResult",
|
||
"run_domclick_detail_backfill",
|
||
]
|
||
|
||
|
||
@dataclass
|
||
class DomClickDetailBackfillResult:
|
||
"""Counters for one domclick detail 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 legacy domclick listings via BrowserFetcher.
|
||
|
||
Params (from default_params jsonb in scrape_schedules):
|
||
batch_size: int -- snapshot size (SELECT LIMIT), default 300.
|
||
budget_sec: float -- wall-clock budget per run, default 3600s.
|
||
request_delay_sec: float -- delay between listings, default 8.0s.
|
||
max_consecutive_blocks: int -- consecutive blocks before abort, default 5.
|
||
|
||
Transport: ONE BrowserFetcher(source="cian") for the whole run (DomClick card
|
||
hosts are QRATOR-clean only via the cian proxy). fetch_detail receives the
|
||
already-open fetcher.
|
||
|
||
Lifecycle: update_heartbeat -> cancel-check -> snapshot -> loop with budget guard ->
|
||
mark_done (incl. partial / block-abort) / mark_failed (outer exception only).
|
||
"""
|
||
batch_size = int(params.get("batch_size", 300))
|
||
budget_sec = float(params.get("budget_sec", 3600))
|
||
request_delay_sec = float(params.get("request_delay_sec", 8.0))
|
||
max_consecutive_blocks = int(params.get("max_consecutive_blocks", 5))
|
||
|
||
counters = DomClickDetailBackfillResult()
|
||
current_counters: dict[str, int] = counters.to_dict()
|
||
|
||
start = time.monotonic()
|
||
|
||
try:
|
||
runs_mod.update_heartbeat(db, run_id, current_counters)
|
||
|
||
# Cooperative cancel before any browser launch / SELECT. The run is already in
|
||
# terminal 'cancelled' state (that is how is_cancelled detects it) — do NOT
|
||
# mark_done (would overwrite 'cancelled'), just return early.
|
||
if runs_mod.is_cancelled(db, run_id):
|
||
logger.info(
|
||
"domclick_detail_backfill: run_id=%d cancelled before loop -- returning early",
|
||
run_id,
|
||
)
|
||
return counters
|
||
|
||
# SNAPSHOT: single SELECT at start -- NOT re-selected in loop (termination guarantee).
|
||
snapshot = (
|
||
db.execute(
|
||
text(
|
||
"""
|
||
SELECT id, source_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 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)",
|
||
run_id,
|
||
len(snapshot),
|
||
budget_sec,
|
||
request_delay_sec,
|
||
max_consecutive_blocks,
|
||
)
|
||
|
||
consecutive_blocks = 0
|
||
do_sleep = False
|
||
|
||
# ONE browser session for the whole run (mirror Layer A domclick.py fetch_city).
|
||
# DomClick card/offer-card hosts are QRATOR-blocked via the domclick/avito
|
||
# proxies but clean via the cian proxy (verified live 2026-06-27); route detail
|
||
# fetches through it.
|
||
async with BrowserFetcher(source="cian") 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
|
||
|
||
# Delay before each request except the first
|
||
if do_sleep:
|
||
await asyncio.sleep(request_delay_sec)
|
||
do_sleep = True
|
||
|
||
listing_id: int = row["id"]
|
||
source_url: str = row["source_url"]
|
||
counters.attempted += 1
|
||
|
||
try:
|
||
enrichment = await fetch_detail(source_url, browser_fetcher=bf)
|
||
if save_detail_enrichment(db, listing_id, enrichment):
|
||
counters.enriched += 1
|
||
consecutive_blocks = 0
|
||
|
||
except DomClickBlockedError as exc:
|
||
# Anti-bot challenge / browser fetch failure. Transient — keep the
|
||
# inter-request delay (no IP rotation here, unlike avito) and abort
|
||
# the whole run after too many in a row (retried next night).
|
||
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,
|
||
exc,
|
||
)
|
||
if consecutive_blocks >= max_consecutive_blocks:
|
||
logger.error(
|
||
"domclick_detail_backfill: run_id=%d ABORT -- %d consecutive blocks "
|
||
"(anti-bot). enriched=%d attempted=%d",
|
||
run_id,
|
||
consecutive_blocks,
|
||
counters.enriched,
|
||
counters.attempted,
|
||
)
|
||
break
|
||
|
||
except DomClickParseError as exc:
|
||
# HTTP 200 but SSR unparseable (schema drift). Do NOT mark the listing
|
||
# enriched -> rollback so it is retried next run.
|
||
counters.failed += 1
|
||
logger.warning(
|
||
"domclick_detail_backfill: run_id=%d listing_id=%d parse failure: %s",
|
||
run_id,
|
||
listing_id,
|
||
exc,
|
||
)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
|
||
except Exception as exc:
|
||
counters.failed += 1
|
||
logger.warning(
|
||
"domclick_detail_backfill: run_id=%d listing_id=%d failed: %s",
|
||
run_id,
|
||
listing_id,
|
||
exc,
|
||
)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
|
||
if counters.attempted % 25 == 0:
|
||
current_counters = counters.to_dict()
|
||
runs_mod.update_heartbeat(db, run_id, current_counters)
|
||
|
||
counters.duration_sec = time.monotonic() - start
|
||
current_counters = counters.to_dict()
|
||
runs_mod.update_heartbeat(db, run_id, current_counters)
|
||
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
|