bot-backend
e3adcd47be
Prod run 423 завис на 7.7ч (healthy ~2.4ч) → reaped как zombie → detail-backlog не уменьшился (21456 листингов ждут detail, 29.3% enriched). Корень: fetch_detail(...) в loop'е без hard-timeout. Один зависший fetch (camoufox/browser hang или curl-stall) блокировал loop навсегда → budget-guard (раз в итерацию) не срабатывал → heartbeat (раз в 25 attempts) не обновлялся → run zombie. Вторичный hang-вектор: _rotate_ip() на блок — тоже без timeout. Fix: - asyncio.wait_for(fetch_detail, timeout=avito_detail_fetch_timeout_s=90s): зависший fetch отменяется → TimeoutError → листинг failed, loop идёт дальше. TimeoutError ловится ДО общего Exception (TimeoutError ⊂ OSError ⊂ Exception), порядок: AvitoBlocked/RateLimited → TimeoutError → Exception. - bound rotate: asyncio.wait_for(_rotate_ip, timeout=settle+30s) + try/except — зависший changeip не роняет run. - config: avito_detail_fetch_timeout_s (ENV AVITO_DETAIL_FETCH_TIMEOUT_S). - admin: POST /scrape/avito-detail-backfill (batch_size, budget_sec) — on-demand прогон/верификация (основной schedule window-gated 09-12 UTC). Восстанавливает throughput detail-очереди (точные coords #1967 + house_type/kitchen).
335 lines
14 KiB
Python
335 lines
14 KiB
Python
"""Scheduled backfill: detail-enrichment for legacy avito listings (#1551).
|
||
|
||
Nightly window 09:00-12:00 UTC (migration 112, source=avito_detail_backfill).
|
||
Offset from avito_city_sweep (~03-04 UTC) to avoid sharing proxy IP simultaneously.
|
||
|
||
Problem: ~15243/15917 avito listings have detail_enriched_at IS NULL.
|
||
City sweep ENRICH_DETAIL only enriches top-N proximate listings per anchor.
|
||
Legacy listings (older than 2h or outside radius) are never enriched.
|
||
|
||
Solution: single snapshot SELECT at start (guarantees termination), same proxy
|
||
session path as scrape_pipeline.py step 5. Block handling mirrors step 5:
|
||
rotate IP on every block, abort after max_consecutive_blocks (mark_done not
|
||
mark_failed -- block is temporary, retry next night via NULL detail_enriched_at).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
from urllib.parse import urlparse
|
||
|
||
from curl_cffi.requests import AsyncSession
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import settings
|
||
from app.services import scrape_runs as runs_mod
|
||
from app.services.scrape_pipeline import _CHROME_HEADERS, _avito_proxies
|
||
from app.services.scrapers.avito import AvitoScraper
|
||
from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment
|
||
from app.services.scrapers.avito_exceptions import (
|
||
AvitoBlockedError,
|
||
AvitoRateLimitedError,
|
||
)
|
||
from app.services.scrapers.browser_fetcher import BrowserFetcher
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
__all__ = [
|
||
"AvitoDetailBackfillResult",
|
||
"run_avito_detail_backfill",
|
||
]
|
||
|
||
|
||
@dataclass
|
||
class AvitoDetailBackfillResult:
|
||
"""Counters for one backfill run."""
|
||
|
||
attempted: int = 0
|
||
enriched: int = 0
|
||
blocked: int = 0
|
||
failed: int = 0
|
||
duration_sec: float = field(default=0.0)
|
||
|
||
def to_dict(self) -> dict[str, int]:
|
||
return {
|
||
"attempted": self.attempted,
|
||
"enriched": self.enriched,
|
||
"blocked": self.blocked,
|
||
"failed": self.failed,
|
||
"duration_sec": int(self.duration_sec),
|
||
}
|
||
|
||
|
||
async def run_avito_detail_backfill(
|
||
db: Session,
|
||
*,
|
||
run_id: int,
|
||
params: dict,
|
||
) -> AvitoDetailBackfillResult:
|
||
"""Backfill detail_enriched_at for legacy avito listings via mobile proxy.
|
||
|
||
Params (from default_params jsonb in scrape_schedules):
|
||
batch_size: int -- snapshot size (SELECT LIMIT), default 800.
|
||
budget_sec: float -- wall-clock budget per run, default 3600s.
|
||
request_delay_sec: float -- delay between listings, default 6.0s.
|
||
max_consecutive_blocks: int -- abort threshold, default 5.
|
||
|
||
Lifecycle: update_heartbeat -> snapshot -> loop with budget guard ->
|
||
mark_done (incl. partial/block-abort) / mark_failed (exception only).
|
||
"""
|
||
batch_size = int(params.get("batch_size", 800))
|
||
budget_sec = float(params.get("budget_sec", 3600))
|
||
request_delay_sec = float(params.get("request_delay_sec", 6.0))
|
||
max_consecutive_blocks = int(params.get("max_consecutive_blocks", 5))
|
||
|
||
# #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
|
||
|
||
scraper = AvitoScraper()
|
||
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")
|
||
await browser_fetcher.__aenter__()
|
||
own_browser = True
|
||
scraper._browser = browser_fetcher
|
||
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
|
||
# use_curl=True: ничего не строим — fetch_detail вызывает _build_detail_session()
|
||
# (scraper_proxy_url = backconnect mproxy) на каждый запрос.
|
||
|
||
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/%'
|
||
ORDER BY 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
|
||
|
||
for idx, row in enumerate(snapshot):
|
||
# Budget guard
|
||
elapsed = time.monotonic() - start
|
||
if elapsed > budget_sec:
|
||
logger.info(
|
||
"avito_detail_backfill: run_id=%d -- budget %.0fs exhausted "
|
||
"(elapsed=%.1fs), stopping at #%d/%d",
|
||
run_id,
|
||
budget_sec,
|
||
elapsed,
|
||
idx,
|
||
len(snapshot),
|
||
)
|
||
break
|
||
|
||
# Delay before each request except the first
|
||
if do_sleep:
|
||
await asyncio.sleep(request_delay_sec)
|
||
do_sleep = True
|
||
|
||
source_url: str = row["source_url"]
|
||
counters.attempted += 1
|
||
|
||
# Normalise URL -> path (mirrors scrape_pipeline.py line 311)
|
||
item_url = urlparse(source_url).path if source_url.startswith("http") else source_url
|
||
|
||
try:
|
||
# #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,
|
||
),
|
||
timeout=fetch_timeout_s,
|
||
)
|
||
if save_detail_enrichment(db, enrichment):
|
||
counters.enriched += 1
|
||
consecutive_blocks = 0
|
||
|
||
except (AvitoBlockedError, AvitoRateLimitedError) as e:
|
||
consecutive_blocks += 1
|
||
counters.blocked += 1
|
||
do_sleep = False
|
||
logger.warning(
|
||
"avito_detail_backfill: run_id=%d BLOCKED #%d/%d (consecutive=%d): %s",
|
||
run_id,
|
||
idx + 1,
|
||
len(snapshot),
|
||
consecutive_blocks,
|
||
e,
|
||
)
|
||
# #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,
|
||
)
|
||
if consecutive_blocks >= max_consecutive_blocks:
|
||
logger.error(
|
||
"avito_detail_backfill: run_id=%d ABORT -- %d consecutive blocks, "
|
||
"IP rate-limited. enriched=%d attempted=%d",
|
||
run_id,
|
||
consecutive_blocks,
|
||
counters.enriched,
|
||
counters.attempted,
|
||
)
|
||
break
|
||
|
||
except 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 failed=%d duration=%.1fs",
|
||
run_id,
|
||
counters.attempted,
|
||
counters.enriched,
|
||
counters.blocked,
|
||
counters.failed,
|
||
counters.duration_sec,
|
||
)
|
||
return counters
|
||
|
||
except Exception as exc:
|
||
counters.duration_sec = time.monotonic() - start
|
||
logger.exception(
|
||
"avito_detail_backfill: run_id=%d FAILED after %.1fs",
|
||
run_id,
|
||
counters.duration_sec,
|
||
)
|
||
runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters.to_dict())
|
||
raise
|
||
|
||
finally:
|
||
if own_session and session is not None:
|
||
await session.close()
|
||
if own_browser and browser_fetcher is not None:
|
||
await browser_fetcher.__aexit__(None, None, None)
|