gendesign/tradein-mvp/backend/app/tasks/avito_detail_backfill.py
bot-backend eb564cfc01 feat(tradein/scraper-kit): migrate backfill-task imports to kit, Group C (#2310)
Migrates legacy app.services.scrapers.* imports to scraper_kit equivalents for
house_imv_backfill.py, avito_detail_backfill.py, cian_history_backfill.py,
ekb_geoportal_ingest.py, and yandex_detail_backfill.py, proving parity via
tests/support/parity.assert_parity per the epic's gate (#2304).

newbuilding_enrich_backfill.py and yandex_newbuilding_sweep.py are left fully
on legacy imports: both call into scraper_kit.providers.{cian,yandex}.newbuilding,
which construct BrowserFetcher(source=...) without the now-mandatory endpoint=
kwarg (issue #2322, verified still open against the actual provider source, not
just issue status) -- no caller-side fix is possible, matching Group A's (#2305)
precedent for the same bug in admin.py.

Config-gated kit function footguns found and fixed (Group B #2306 pattern):
  - avito fetch_detail's backconnect-on-403 retry silently drops when config=
    is omitted -- now passes config=RealScraperConfig() explicitly, with a
    regression test proving the gate.
  - kit AvitoScraper's constructor now requires ScraperConfig positionally --
    wired via RealScraperConfig(), which _rotate_ip() reads for
    avito_proxy_rotate_url.
  - BrowserFetcher(source=...) call sites (house_imv_backfill, avito/cian
    detail backfills) now pass the mandatory endpoint=settings.browser_http_endpoint.

New footgun discovered (NOT #2322, flagged for follow-up): kit's
build_warmed_session() builds its curl_cffi session via _build_detail_session()
with no config parameter at all, unlike fetch_detail -- migrating it would
silently drop the sticky MGTS-proxy egress on avito_detail_backfill's
warm-batch path (the prod default). Left build_warmed_session/
_AVITO_WARM_SEARCH_URL on legacy imports, documented inline.

Refs #2310
2026-07-04 01:19:34 +03:00

508 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""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 random
import time
from dataclasses import dataclass, field
from urllib.parse import urlparse
from curl_cffi.requests import AsyncSession
from scraper_kit.avito_exceptions import (
AvitoBlockedError,
AvitoListingGoneError,
AvitoRateLimitedError,
)
from scraper_kit.browser_fetcher import BrowserFetcher
from scraper_kit.providers.avito.detail import (
fetch_detail,
research_in_session,
save_detail_enrichment,
)
from scraper_kit.providers.avito.serp import AvitoScraper
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 scrape_runs as runs_mod
from app.services.scrape_pipeline import _CHROME_HEADERS, _avito_proxies
from app.services.scraper_adapters import RealScraperConfig
# _AVITO_WARM_SEARCH_URL/build_warmed_session остаются legacy (#2310): kit's
# scraper_kit.providers.avito.detail.build_warmed_session() строит сессию через
# _build_detail_session() БЕЗ config-параметра (сигнатура вообще не принимает
# ScraperConfig — в отличие от fetch_detail, здесь нет caller-side способа
# прокинуть прокси). Мигрировав этот импорт, warm-batch curl-путь (use_curl=True,
# ДЕФОЛТ прод-режима, avito_detail_backfill_use_curl=True) молча перестал бы
# использовать settings.scraper_proxy_url (sticky МГТС-прокси) — прямое
# datacenter-подключение вместо backconnect. Тот же класс бага, что #2322
# (недостающая инъекция конфига в scraper_kit provider-функции), но другая
# функция — заведён как отдельный follow-up, НЕ #2322. См. отчёт issue #2310.
from app.services.scrapers.avito_detail import (
_AVITO_WARM_SEARCH_URL,
build_warmed_session,
)
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
gone: 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,
"gone": self.gone,
"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))
warm_batch = int(params.get("warm_batch", 500))
research_every = int(params.get("research_every", 50))
block_cooldown_sec = float(params.get("block_cooldown_sec", 30.0))
warm_timeout_s = float(params.get("warm_timeout_s", 90.0))
# #1950: hard-timeout'ы на блокирующие await'ы внутри loop'а. Без них один
# зависший fetch_detail/rotate ронял весь run в zombie (run 423 завис 7.7ч).
# Локальные float'ы (а не settings.* напрямую) — чтобы wait_for/format получали
# число даже когда settings замокан MagicMock'ом в тестах.
fetch_timeout_s = float(settings.avito_detail_fetch_timeout_s)
rotate_timeout_s = float(settings.avito_proxy_rotate_settle_s) + 30.0
counters = AvitoDetailBackfillResult()
current_counters: dict[str, int] = counters.to_dict()
# Режим фетча: curl_cffi+backconnect (mproxy) когда use_curl=True; иначе браузер/auv.
# avito_detail_backfill_use_curl=True переопределяет scraper_fetch_mode — detail_backfill
# всегда идёт через backconnect чтобы не конкурировать за auv с SERP (city_sweep/full_load).
use_curl = settings.avito_detail_backfill_use_curl
browser_mode = settings.scraper_fetch_mode == "browser" and not use_curl
session: AsyncSession | None = None
browser_fetcher: BrowserFetcher | None = None
own_session = False
own_browser = False
# kit AvitoScraper требует ScraperConfig позиционно (Strangler-инжекция #2133) —
# RealScraperConfig проксирует settings.* так же, как читал legacy-конструктор без
# аргументов (avito_proxy_rotate_url и т.д. для _rotate_ip()).
scraper = AvitoScraper(RealScraperConfig())
start = time.monotonic()
logger.info(
"avito_detail_backfill: run_id=%d mode=%s",
run_id,
"curl/backconnect" if use_curl else "browser",
)
try:
# Setup session (mirrors run_avito_pipeline lines 161-183).
# use_curl=True: session=None + browser_fetcher=None → fetch_detail строит
# эфемерную _build_detail_session() через settings.scraper_proxy_url (backconnect).
# use_curl=False (legacy): browser_mode=True → BrowserFetcher/auv, как раньше.
if browser_mode:
browser_fetcher = BrowserFetcher(
source="avito", endpoint=settings.browser_http_endpoint
)
await browser_fetcher.__aenter__()
own_browser = True
scraper._browser = browser_fetcher
elif use_curl:
# use_curl=True (#1551 warm-batch): прогретая shared-сессия на sticky МГТС-IP —
# yandex-referer -> avito-search сеет антибот-куки, сессия держит батч detail без
# 403 (доказано пробами: ≥66 карточек подряд, 0 блоков). NB: МГТС sticky — один
# фикс. exit-IP, per-connection ротации нет (rebuild != новый IP); on-block —
# cooldown + in-session re-search, не дискард сессии.
own_session = True
session = await build_warmed_session()
elif not use_curl:
# curl_cffi legacy path (scraper_fetch_mode="curl_cffi", use_curl=False):
# строим shared сессию через auv, как делал scrape_pipeline.
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.
# Scope (#1814): только активные ЕКБ-листинги. region_code на insert
# хардкодится в 66 (base.py) → НЕ дискриминирует legacy не-ЕКБ; реальный
# признак региона у Avito — путь URL (/ekaterinburg/ для ЕКБ; legacy
# Москва/СПб/Тюмень — /moskva//sankt-peterburg//tyumen/). browser-fetch
# на legacy не-ЕКБ спотыкается → curl-fallback → 429-бан curl-фингерпринта.
# Не тратим фетчи на мёртвые (is_active) и не-ЕКБ.
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
AND is_active = TRUE
AND source_url LIKE '%/ekaterinburg/%'
-- сперва листинги без координат (#1967 — detail-страница даёт
-- координаты здания), затем по свежести
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(
"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 mode=%s)",
run_id,
len(snapshot),
budget_sec,
request_delay_sec,
max_consecutive_blocks,
"curl/backconnect" if use_curl else "browser",
)
consecutive_blocks = 0
do_sleep = False
items_since_warm = 0
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
# #1182 Phase 2: кооперативный SIGTERM-drain (деплой recreate scraper).
# Чисто выходим на границе карточки — каждая карточка уже закоммичена,
# mark_done с текущими счётчиками вызовется после loop'а. Snapshot —
# pending-query (detail_enriched_at IS NULL), per-item идемпотентна →
# следующий run сам до-резюмит остаток, resume_cursor не нужен.
if shutdown_requested():
logger.info(
"avito_detail_backfill: run_id=%d SIGTERM-drain — stopping at #%d/%d",
run_id,
idx,
len(snapshot),
)
break
# Delay before each request except the first (джиттер ±30% — менее
# роботизированный паттерн на органически прогретой сессии).
if do_sleep:
await asyncio.sleep(request_delay_sec * random.uniform(0.7, 1.4))
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
# use_curl warm-batch (#1551): интервалы по размеру SERP-страницы (~50).
# МГТС sticky-IP: exit-IP константа (rebuild НЕ даёт новый IP). Каждые
# warm_batch (~500) — полный re-warm (close+build) на ТОМ ЖЕ sticky exit-IP:
# session-hygiene (свежий TLS-хендшейк + свежие куки yandex->search), НЕ смена
# IP. Между ними — лёгкий in-session перепоиск каждые research_every (~50)
# (re-GET search той же сессией, освежить куки). On-block — cooldown ниже.
if use_curl and session is not None:
if warm_batch and items_since_warm >= warm_batch:
# периодический полный re-warm на том же sticky exit-IP (свежий TLS+куки).
# #1950: bound сетевой re-warm; строим НОВУЮ сессию до close старой — на
# timeout/fail сохраняем текущую рабочую сессию (graceful degrade).
try:
new_session = await asyncio.wait_for(
build_warmed_session(), timeout=warm_timeout_s
)
try:
await session.close()
except Exception:
pass
session = new_session
except Exception:
logger.warning(
"avito_detail_backfill: run_id=%d re-warm timed out/failed (>%.0fs) "
"-- keeping current session",
run_id,
warm_timeout_s,
exc_info=True,
)
items_since_warm = 0
elif (
research_every
and items_since_warm > 0
and items_since_warm % research_every == 0
):
# лёгкий in-session перепоиск: тот же exit-IP, освежить куки.
# #1950: bound сетевой re-search.
try:
await asyncio.wait_for(research_in_session(session), timeout=warm_timeout_s)
except Exception:
logger.warning(
"avito_detail_backfill: run_id=%d in-session re-search "
"timed out/failed",
run_id,
exc_info=True,
)
try:
# #1950: hard-timeout — зависший fetch (browser hang / curl-stall) не
# должен блокировать loop навсегда (иначе budget-guard/heartbeat молчат
# и run reaped как zombie). wait_for отменяет fetch → TimeoutError.
enrichment = await asyncio.wait_for(
fetch_detail(
item_url,
cffi_session=session,
browser_fetcher=browser_fetcher,
referer=_AVITO_WARM_SEARCH_URL if use_curl else None,
reconnect_on_block=not use_curl,
# config= обязателен: kit fetch_detail без него читает
# config.scraper_proxy_url=None для backconnect-gate'а
# (elif not use_curl own-session path), теряя reconnect-
# on-403 поведение legacy (settings.scraper_proxy_url
# напрямую). Не влияет на use_curl=True (прод-дефолт) —
# там reconnect_on_block=False уже гасит backconnect.
config=RealScraperConfig(),
),
timeout=fetch_timeout_s,
)
if save_detail_enrichment(db, enrichment):
counters.enriched += 1
if use_curl:
items_since_warm += 1
consecutive_blocks = 0
except AvitoListingGoneError:
# #2034: мёртвый листинг (404 / removed) — НЕ блок, НЕ failed.
# Координатные дыры в lat-null очереди в основном dead-листинги;
# browser-mode рендерит их «Ошибка 404» без item-view → раньше это
# ловилось как soft-block → consecutive-block breaker абортил run до
# live-листингов (run 458: attempted=5 enriched=0 blocked=5 → abort).
# 404 нейтрален к breaker'у: consecutive_blocks НЕ трогаем (не растим
# и не сбрасываем). Метим is_active=FALSE → листинг уходит из scope
# (snapshot SELECT фильтрует is_active = TRUE) и не тратит фетчи впредь.
counters.gone += 1
try:
with db.begin_nested():
db.execute(
text("UPDATE listings SET is_active = FALSE WHERE id = :id"),
{"id": row["id"]},
)
except Exception:
logger.warning(
"avito_detail_backfill: run_id=%d failed to mark listing %s "
"is_active=FALSE",
run_id,
row["id"],
exc_info=True,
)
logger.info(
"avito_detail_backfill: run_id=%d GONE #%d/%d listing %s -> "
"marked is_active=FALSE",
run_id,
counters.gone,
len(snapshot),
source_url,
)
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,
)
# abort FIRST — на последнем блоке не тратим cooldown+research/rotate
# (consecutive_blocks уже инкрементнут выше).
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
# МГТС sticky-IP: один фикс. exit-IP, per-connection ротации нет (проверено:
# 6/6 свежих сессий = тот же IP 109.252.125.80; ротация только вручную
# кнопкой). Уйти на свежий IP софтом нельзя → блок = rate-limit текущего IP:
# даём окну остыть (cooldown) и освежаем куки in-session, НЕ дискардим
# рабочую прогретую сессию (rebuild на том же IP бесполезен для escape +
# грузит rate-limited IP полным прогревом). items_since_warm НЕ сбрасываем.
if use_curl:
await asyncio.sleep(block_cooldown_sec)
if session is not None:
# #1950: bound сетевой re-search.
try:
await asyncio.wait_for(
research_in_session(session), timeout=warm_timeout_s
)
except Exception:
logger.warning(
"avito_detail_backfill: run_id=%d on-block re-search "
"timed out/failed",
run_id,
exc_info=True,
)
else:
# #1950: bound rotate — _rotate_ip ждёт settle + до 3 changeip-попыток;
# на блокирующем changeip (зависшее соединение) без timeout loop виснет.
try:
await asyncio.wait_for(scraper._rotate_ip(), timeout=rotate_timeout_s)
except Exception:
logger.warning(
"avito_detail_backfill: run_id=%d rotate_ip timed out/failed "
"(>%.0fs) -- continuing",
run_id,
rotate_timeout_s,
exc_info=True,
)
except TimeoutError:
# asyncio.wait_for → TimeoutError (py3.12: asyncio.TimeoutError — alias).
# Ловим ДО общего Exception (TimeoutError ⊂ OSError ⊂ Exception). Зависший
# fetch отменён → листинг failed, переходим к следующему (loop не зависает,
# run не zombie #1950). Не считаем soft-блоком: rotate не дёргаем.
counters.failed += 1
logger.warning(
"avito_detail_backfill: run_id=%d listing %s TIMEOUT (>%.0fs) -- skip",
run_id,
source_url,
fetch_timeout_s,
)
try:
db.rollback()
except Exception:
pass
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 gone=%d failed=%d duration=%.1fs",
run_id,
counters.attempted,
counters.enriched,
counters.blocked,
counters.gone,
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)