fix(tradein): avito full-load probe resilience — browser 500/timeout splits/skips instead of crashing run #1757

Merged
lekss361 merged 1 commit from fix/avito-fullload-probe-resilience into main 2026-06-18 13:11:46 +00:00
2 changed files with 291 additions and 10 deletions

View file

@ -33,6 +33,7 @@ from datetime import date, datetime, timedelta, timezone
from typing import Any
from urllib.parse import urlencode, urljoin, urlparse, urlunparse
import httpx
from curl_cffi.requests import AsyncSession
from selectolax.parser import HTMLParser
@ -637,6 +638,49 @@ class AvitoScraper(BaseScraper):
url = self._build_rooms_url(room_slug, page, min_price, max_price)
return await self._fetch_serp_html(url, page)
async def _probe_total(
self, room_slug: str, min_price: int | None, max_price: int
) -> tuple[str | None, int | None]:
"""Probe page=1 бакета и извлекает total, устойчиво к browser-ошибкам.
tradein-browser на тяжёлой выдаче (огромный диапазон, напр. студии
pmax=200M) может таймаутить на goto HTTP 500 ``httpx.HTTPStatusError``,
либо отдавать ``httpx.TimeoutException``/``ReadTimeout``. Раньше эти ошибки
пробрасывались из probe до самого верха и роняли весь full-load с 0 uniq.
Теперь они трактуются как «total недоступен» (``None``) вызывающий
``_walk_price_range`` решит: split (бакет делим) или skip (узкий бакет).
AvitoBlockedError/AvitoRateLimitedError НЕ глушатся это hard-block,
должен всплыть в pipeline (mark_banned). Возвращает (html, total); при
ошибке fetch'а html=None.
"""
try:
html = await self._fetch_rooms_page_html(room_slug, 1, min_price, max_price)
except (AvitoBlockedError, AvitoRateLimitedError):
# Hard-block — пробрасываем наверх (pipeline пометит run banned).
raise
except (httpx.HTTPStatusError, httpx.TimeoutException, httpx.HTTPError) as exc:
logger.warning(
"avito: probe browser error %s for %s [%s, %d] — treating as total=None",
type(exc).__name__,
room_slug,
min_price,
max_price,
)
return None, None
except Exception:
logger.warning(
"avito: probe unexpected error for %s [%s, %d] — treating as total=None",
room_slug,
min_price,
max_price,
exc_info=True,
)
return None, None
total = self._extract_total_count(html) if html is not None else None
return html, total
async def fetch_all_secondary(
self,
*,
@ -748,13 +792,12 @@ class AvitoScraper(BaseScraper):
_lo_param = lo if lo > 0 else None
# ── Шаг 1: probe page 1 ────────────────────────────────────────────────
html = await self._fetch_rooms_page_html(room_slug, 1, _lo_param, hi)
# _probe_total ловит browser-ошибки (HTTP 500 / timeout от tradein-browser
# на тяжёлой выдаче, напр. студии pmax=200M) → total=None вместо краха.
# AvitoBlockedError/AvitoRateLimitedError всплывают наверх (mark_banned).
html, total = await self._probe_total(room_slug, _lo_param, hi)
total: int | None = None
if html is not None:
total = self._extract_total_count(html)
# Ретрай на captcha/missing-count: rotate IP + 1 retry.
# Ретрай на captcha/missing-count/browser-ошибке: rotate IP + 1 retry.
if total is None:
logger.warning(
"avito: total=None for %s [%d, %d] depth=%d — rotating IP + retry",
@ -765,16 +808,61 @@ class AvitoScraper(BaseScraper):
)
rotated = await self._rotate_ip()
if rotated:
html = await self._fetch_rooms_page_html(room_slug, 1, _lo_param, hi)
if html is not None:
total = self._extract_total_count(html)
html, total = await self._probe_total(room_slug, _lo_param, hi)
if total is None:
bracket_size = hi - lo
# Probe не удался даже после ретрая. Если бакет ещё делим — НЕ теряем
# его целиком: тяжёлая страница (огромная выдача) скорее всего станет
# загружаемой в более узком ценовом диапазоне → принудительный split.
if bracket_size >= _AVITO_MIN_BRACKET:
logger.warning(
"avito: probe failed for %s [%d, %d] depth=%d — splitting (assume heavy)",
room_label,
lo,
hi,
_depth,
)
mid = (lo + hi) // 2
await self._walk_price_range(
room_slug=room_slug,
room_label=room_label,
lo=lo,
hi=mid,
seen=seen,
price_cap_per_bucket=price_cap_per_bucket,
max_pages_per_bucket=max_pages_per_bucket,
concurrency=concurrency,
secondary_only=secondary_only,
on_bucket=on_bucket,
skip_buckets=skip_buckets,
_depth=_depth + 1,
)
await self._walk_price_range(
room_slug=room_slug,
room_label=room_label,
lo=mid + 1,
hi=hi,
seen=seen,
price_cap_per_bucket=price_cap_per_bucket,
max_pages_per_bucket=max_pages_per_bucket,
concurrency=concurrency,
secondary_only=secondary_only,
on_bucket=on_bucket,
skip_buckets=skip_buckets,
_depth=_depth + 1,
)
return
# Бакет уже узкий (< MIN_BRACKET) — split не поможет, хвост теряем
# осознанно (как cian) и логируем.
logger.error(
"avito: skipping bucket %s [%d, %d] — total unavailable after retry",
"avito: skipping bucket %s [%d, %d] — total unavailable after retry "
"(bracket=%d < MIN_BRACKET=%d, tail loss accepted)",
room_label,
lo,
hi,
bracket_size,
_AVITO_MIN_BRACKET,
)
return

View file

@ -16,6 +16,7 @@ from unittest.mock import MagicMock, patch
# Settings requires DATABASE_URL at import time — set dummy DSN before any app import.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
import httpx
import pytest
from app.services.scrapers.avito import (
@ -24,6 +25,7 @@ from app.services.scrapers.avito import (
_AVITO_OFFERS_PER_PAGE,
AvitoScraper,
)
from app.services.scrapers.avito_exceptions import AvitoBlockedError
from app.services.scrapers.base import ScrapedLot
@ -441,3 +443,194 @@ async def test_on_bucket_cancel_stops_run(scraper: AvitoScraper) -> None:
price_cap_per_bucket=1400,
on_bucket=cancel_on_bucket,
)
# ── Probe-устойчивость к browser-ошибкам (HTTP 500 / timeout) ─────────────────
def _http_500_error() -> httpx.HTTPStatusError:
"""Сборка httpx.HTTPStatusError(500) как из browser_fetcher.raise_for_status()."""
request = httpx.Request("POST", "http://tradein-browser:3000/fetch")
response = httpx.Response(500, request=request)
return httpx.HTTPStatusError("Server error '500'", request=request, response=response)
@pytest.mark.asyncio
async def test_probe_browser_500_on_wide_bucket_splits_not_crash(
scraper: AvitoScraper,
) -> None:
"""Probe бросает HTTP 500 на ШИРОКОМ bucket → НЕ крах, происходит split.
Воспроизводит прод-инцидент run 199: студии pmax=200M таймаутят на goto
browser HTTP 500. Раньше это роняло весь full-load с 0 uniq. Теперь широкий
bucket принудительно делится пополам; под-бакеты пагинируются нормально.
"""
probe_calls: list[tuple[int | None, int]] = []
async def fake_fetch(
room_slug: str, page: int, min_price: int | None, max_price: int | None
) -> str:
if page == 1 and (min_price in (None, 0)) and max_price == _AVITO_MAX_PRICE:
probe_calls.append((min_price, max_price))
raise _http_500_error()
return f"<html>page={page} min={min_price} max={max_price}</html>"
def fake_total(html: str) -> int | None:
return 50
def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
import re
page_m = re.search(r"page=(\d+)", html)
page = int(page_m.group(1)) if page_m else 1
max_m = re.search(r"max=(\w+)", html)
max_p = max_m.group(1) if max_m else "MAX"
if page == 1:
return [_make_lot(f"lot_{max_p}_{i}") for i in range(3)]
return []
with (
patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
patch.object(scraper, "_extract_total_count", side_effect=fake_total),
patch.object(scraper, "_parse_html", side_effect=fake_parse),
patch.object(scraper, "_rotate_ip", return_value=False),
):
lots = await scraper.fetch_all_secondary(
rooms_buckets=[("студии", "studii-X")],
price_cap_per_bucket=1400,
)
assert probe_calls, "Probe полного диапазона должен был вызваться и упасть"
assert len(lots) > 0, f"Ожидались лоты из под-бакетов после split, got {len(lots)}"
@pytest.mark.asyncio
async def test_probe_timeout_on_wide_bucket_splits_not_crash(
scraper: AvitoScraper,
) -> None:
"""httpx.TimeoutException на широком bucket обрабатывается как HTTP 500 → split."""
probe_failed = False
async def fake_fetch(
room_slug: str, page: int, min_price: int | None, max_price: int | None
) -> str:
nonlocal probe_failed
if page == 1 and (min_price in (None, 0)) and max_price == _AVITO_MAX_PRICE:
probe_failed = True
raise httpx.TimeoutException("Page.goto timeout")
return f"<html>page={page} max={max_price}</html>"
def fake_total(html: str) -> int | None:
return 40
def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
import re
page_m = re.search(r"page=(\d+)", html)
page = int(page_m.group(1)) if page_m else 1
max_m = re.search(r"max=(\w+)", html)
max_p = max_m.group(1) if max_m else "MAX"
return [_make_lot(f"lot_{max_p}_{i}") for i in range(2)] if page == 1 else []
with (
patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
patch.object(scraper, "_extract_total_count", side_effect=fake_total),
patch.object(scraper, "_parse_html", side_effect=fake_parse),
patch.object(scraper, "_rotate_ip", return_value=False),
):
lots = await scraper.fetch_all_secondary(
rooms_buckets=[("студии", "studii-X")],
price_cap_per_bucket=1400,
)
assert probe_failed, "Probe должен был таймаутнуть"
assert len(lots) > 0, "После split под-бакеты должны собрать лоты"
@pytest.mark.asyncio
async def test_probe_browser_500_on_narrow_bucket_skips_not_crash(
scraper: AvitoScraper,
) -> None:
"""Probe бросает HTTP 500 на УЗКОМ bucket (< MIN_BRACKET) → skip+return, без краха.
Узкий bucket делить бесполезно хвост теряется осознанно. Прогон не падает,
остальные комнатности продолжаются.
"""
second_room_paginated = False
async def fake_fetch(
room_slug: str, page: int, min_price: int | None, max_price: int | None
) -> str:
nonlocal second_room_paginated
if room_slug == "narrow-X":
raise _http_500_error()
if room_slug == "ok-X" and page == 1:
second_room_paginated = True
return f"<html>page={page}</html>"
def fake_total(html: str) -> int | None:
return 5
def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
return [_make_lot("ok_lot")] if "page=1" in html else []
narrow_lo = 1_000_000
narrow_hi = narrow_lo + _AVITO_MIN_BRACKET - 1
seen: dict[str, ScrapedLot] = {}
with (
patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
patch.object(scraper, "_extract_total_count", side_effect=fake_total),
patch.object(scraper, "_parse_html", side_effect=fake_parse),
patch.object(scraper, "_rotate_ip", return_value=False),
):
await scraper._walk_price_range(
room_slug="narrow-X",
room_label="room_narrow",
lo=narrow_lo,
hi=narrow_hi,
seen=seen,
price_cap_per_bucket=1400,
max_pages_per_bucket=100,
)
assert seen == {}, "Узкий bucket с упавшим probe не должен дать лотов"
await scraper._walk_price_range(
room_slug="ok-X",
room_label="room_ok",
lo=0,
hi=_AVITO_MAX_PRICE,
seen=seen,
price_cap_per_bucket=1400,
max_pages_per_bucket=100,
)
assert second_room_paginated, "Вторая комнатность должна пагинироваться после skip"
assert "ok_lot" in seen
@pytest.mark.asyncio
async def test_probe_avito_blocked_propagates_not_swallowed(
scraper: AvitoScraper,
) -> None:
"""AvitoBlockedError из probe ВСПЛЫВАЕТ (hard-block), а не глушится как total=None."""
async def fake_fetch(
room_slug: str, page: int, min_price: int | None, max_price: int | None
) -> str:
raise AvitoBlockedError("Avito SERP firewall (browser-mode) — IP banned")
def fake_total(html: str) -> int | None:
return 100
with (
patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
patch.object(scraper, "_extract_total_count", side_effect=fake_total),
patch.object(scraper, "_parse_html", side_effect=lambda h, source_url_base: []),
patch.object(scraper, "_rotate_ip", return_value=False),
):
with pytest.raises(AvitoBlockedError):
await scraper.fetch_all_secondary(
rooms_buckets=[("1-комн.", "1-komnatnye-X")],
price_cap_per_bucket=1400,
)