fix(avito): reconnect-retry on detail 403/firewall under backconnect proxy #1768
2 changed files with 292 additions and 28 deletions
|
|
@ -19,6 +19,7 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
|
|
@ -43,6 +44,15 @@ logger = logging.getLogger(__name__)
|
|||
|
||||
AVITO_BASE = "https://www.avito.ru"
|
||||
|
||||
# Reconnect-retry на detail-странице под backconnect-прокси — зеркало SERP-фикса
|
||||
# (#1765, avito.py): 403/firewall = залочен лишь текущий exit-IP backconnect-прокси;
|
||||
# пересоздание curl_cffi-сессии = новый CONNECT-туннель = свежий exit-IP → escape.
|
||||
# 429 = transient conn-limit, лечится коротким retry на той же сессии (без reconnect).
|
||||
_AVITO_DETAIL_403_MAX_RETRIES = 5
|
||||
_AVITO_DETAIL_403_BACKOFF_SEC = 2.0
|
||||
_AVITO_DETAIL_429_MAX_RETRIES = 4
|
||||
_AVITO_DETAIL_429_BACKOFF_SEC = 1.5
|
||||
|
||||
# ── Русские месяцы ─────────────────────────────────────────────────────────
|
||||
RUS_MONTHS: dict[str, int] = {
|
||||
"января": 1,
|
||||
|
|
@ -234,6 +244,35 @@ class DetailEnrichment:
|
|||
raw_html_meta: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
def _build_detail_session() -> AsyncSession:
|
||||
"""Создать curl_cffi-сессию для detail-fetch (impersonate chrome120, через прокси
|
||||
если задан settings.scraper_proxy_url).
|
||||
|
||||
Вынесено из own-session-ветки fetch_detail чтобы reconnect-retry мог пересоздать
|
||||
эфемерную сессию (новый CONNECT-туннель = свежий backconnect exit-IP) под 403.
|
||||
proxy=None → прямое подключение.
|
||||
"""
|
||||
from app.core.config import settings
|
||||
|
||||
proxy_url = settings.scraper_proxy_url
|
||||
proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None
|
||||
return AsyncSession(
|
||||
impersonate="chrome120",
|
||||
timeout=25,
|
||||
proxies=proxies,
|
||||
headers={
|
||||
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
||||
"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
|
||||
"Cache-Control": "max-age=0",
|
||||
"Sec-Fetch-Dest": "document",
|
||||
"Sec-Fetch-Mode": "navigate",
|
||||
"Sec-Fetch-Site": "none",
|
||||
"Sec-Fetch-User": "?1",
|
||||
"Upgrade-Insecure-Requests": "1",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
# ── fetch_detail ──────────────────────────────────────────────────────────────
|
||||
async def fetch_detail(
|
||||
item_url: str,
|
||||
|
|
@ -256,47 +295,81 @@ async def fetch_detail(
|
|||
raise AvitoBlockedError(f"Avito detail firewall (browser-mode) for {full_url}")
|
||||
return parse_detail_html(html, full_url)
|
||||
|
||||
from app.core.config import settings
|
||||
|
||||
own_session = cffi_session is None
|
||||
if cffi_session is None:
|
||||
# Mobile proxy wiring (#806 follow-up): own-session path (no shared session passed,
|
||||
# e.g. admin endpoint). Caller-provided sessions (scrape_pipeline) already have proxy
|
||||
# from AvitoScraper.__aenter__ — don't override. proxy=None → прямое подключение.
|
||||
from app.core.config import settings
|
||||
|
||||
_proxy_url = settings.scraper_proxy_url
|
||||
_proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
|
||||
session = AsyncSession(
|
||||
impersonate="chrome120",
|
||||
timeout=25,
|
||||
proxies=_proxies,
|
||||
headers={
|
||||
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
||||
"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
|
||||
"Cache-Control": "max-age=0",
|
||||
"Sec-Fetch-Dest": "document",
|
||||
"Sec-Fetch-Mode": "navigate",
|
||||
"Sec-Fetch-Site": "none",
|
||||
"Sec-Fetch-User": "?1",
|
||||
"Upgrade-Insecure-Requests": "1",
|
||||
},
|
||||
)
|
||||
session = _build_detail_session()
|
||||
else:
|
||||
session = cffi_session
|
||||
|
||||
full_url = item_url if item_url.startswith("http") else urljoin(AVITO_BASE, item_url)
|
||||
# curl_cffi egress всегда через scraper_proxy_url (backconnect mproxy). 403/firewall =
|
||||
# залочен лишь текущий exit-IP → reconnect новой сессией = свежий IP. Гейтим reconnect
|
||||
# ТОЛЬКО на наличие proxy (зеркало SERP-фикса #1765). Без proxy — старое поведение
|
||||
# (raise на первом 403). Caller-provided сессию НЕ трогаем: reconnect использует
|
||||
# эфемерную retry_session, которую сами закрываем в finally.
|
||||
backconnect = bool(settings.scraper_proxy_url)
|
||||
r403 = 0
|
||||
r429 = 0
|
||||
attempt_session = session
|
||||
retry_session: AsyncSession | None = None
|
||||
try:
|
||||
response = await session.get(full_url)
|
||||
if response.status_code == 403:
|
||||
raise AvitoBlockedError(f"Avito detail HTTP 403 for {full_url}")
|
||||
if response.status_code == 429:
|
||||
raise AvitoRateLimitedError(f"Avito detail HTTP 429 for {full_url}")
|
||||
if response.status_code != 200:
|
||||
raise ValueError(f"avito detail HTTP {response.status_code} for {full_url}")
|
||||
enrichment = parse_detail_html(response.text, full_url)
|
||||
return enrichment
|
||||
while True:
|
||||
response = await attempt_session.get(full_url)
|
||||
sc = response.status_code
|
||||
is_firewall = sc == 200 and _is_firewall_page(response.text)
|
||||
|
||||
# 429 через backconnect = transient conn-limit, НЕ IP-ban: короткий retry той
|
||||
# же сессии без reconnect (зеркало SERP). Только при исчерпании — raise.
|
||||
if sc == 429 and r429 < _AVITO_DETAIL_429_MAX_RETRIES:
|
||||
r429 += 1
|
||||
logger.info(
|
||||
"avito detail HTTP 429 (backconnect conn-limit) — short retry %d/%d",
|
||||
r429,
|
||||
_AVITO_DETAIL_429_MAX_RETRIES,
|
||||
)
|
||||
await asyncio.sleep(_AVITO_DETAIL_429_BACKOFF_SEC)
|
||||
continue
|
||||
|
||||
if sc == 403 or is_firewall:
|
||||
if backconnect and r403 < _AVITO_DETAIL_403_MAX_RETRIES:
|
||||
r403 += 1
|
||||
# Эфемерная свежая сессия (новый CONNECT-туннель = свежий exit-IP).
|
||||
# Caller-provided/own session НЕ пересоздаём — только retry_session.
|
||||
if retry_session is not None:
|
||||
try:
|
||||
await retry_session.close()
|
||||
except Exception:
|
||||
pass
|
||||
retry_session = _build_detail_session()
|
||||
attempt_session = retry_session
|
||||
logger.info(
|
||||
"avito detail 403/firewall %s — backconnect reconnect retry %d/%d",
|
||||
full_url,
|
||||
r403,
|
||||
_AVITO_DETAIL_403_MAX_RETRIES,
|
||||
)
|
||||
await asyncio.sleep(_AVITO_DETAIL_403_BACKOFF_SEC)
|
||||
continue
|
||||
raise AvitoBlockedError(f"Avito detail HTTP 403 for {full_url}")
|
||||
|
||||
if sc == 429:
|
||||
raise AvitoRateLimitedError(f"Avito detail HTTP 429 for {full_url}")
|
||||
if sc != 200:
|
||||
raise ValueError(f"avito detail HTTP {sc} for {full_url}")
|
||||
return parse_detail_html(response.text, full_url)
|
||||
finally:
|
||||
if own_session:
|
||||
await session.close()
|
||||
if retry_session is not None and retry_session is not session:
|
||||
try:
|
||||
await retry_session.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
# ── parse_detail_html ─────────────────────────────────────────────────────────
|
||||
|
|
|
|||
|
|
@ -0,0 +1,191 @@
|
|||
"""curl_cffi 403/firewall reconnect-retry на detail-странице под backconnect-прокси.
|
||||
|
||||
fetch_detail ронял весь detail-backfill на первом 403/429 через AvitoBlockedError /
|
||||
AvitoRateLimitedError (run 191 обогатил 48 → zombie). Зеркало SERP-фикса (#1765): под
|
||||
backconnect rotating-прокси (mproxy.site) каждое НОВОЕ соединение даёт свежий exit-IP,
|
||||
поэтому 403/firewall ретраим пересозданием эфемерной curl_cffi-сессии до
|
||||
_AVITO_DETAIL_403_MAX_RETRIES раз. 429 — короткий retry той же сессии без reconnect.
|
||||
|
||||
Ownership-правило: caller-provided cffi_session НИКОГДА не закрывается/не пересоздаётся —
|
||||
retries идут через свою эфемерную retry_session (закрываем в finally).
|
||||
|
||||
Включаем backconnect через scraper_proxy_url_env (scraper_proxy_url — read-only property,
|
||||
его не monkeypatch'ишь). Зеркало fixture-стиля test_avito_403_backconnect_reconnect.py.
|
||||
"""
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.core.config import settings
|
||||
from app.services.scrapers import avito_detail as detail_mod
|
||||
from app.services.scrapers.avito_detail import fetch_detail
|
||||
from app.services.scrapers.avito_exceptions import AvitoBlockedError
|
||||
|
||||
_SENTINEL = object()
|
||||
_FIREWALL_HTML = (
|
||||
"<html><head><title>Доступ ограничен: проблема с IP</title></head>"
|
||||
'<body><div class="firewall-container">...</div></body></html>'
|
||||
)
|
||||
|
||||
|
||||
def _resp(status_code: int, text: str = "") -> MagicMock:
|
||||
r = MagicMock()
|
||||
r.status_code = status_code
|
||||
r.text = text
|
||||
return r
|
||||
|
||||
|
||||
def _enable_backconnect(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(settings, "scraper_proxy_url_env", "http://u:p@mproxy.site:14619")
|
||||
|
||||
|
||||
def _disable_backconnect(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(settings, "scraper_proxy_url_env", None)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _patch_common() -> object:
|
||||
"""asyncio.sleep → no-op; parse_detail_html → sentinel (нет нужды в реальном HTML)."""
|
||||
with (
|
||||
patch("app.services.scrapers.avito_detail.asyncio.sleep", AsyncMock()),
|
||||
patch(
|
||||
"app.services.scrapers.avito_detail.parse_detail_html", return_value=_SENTINEL
|
||||
) as parse,
|
||||
):
|
||||
yield parse
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_403_403_200_caller_session_reconnect_succeeds(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""caller-provided session: 403 → 403 → 200 → reconnect-retry возвращает enrichment.
|
||||
|
||||
_build_detail_session дёрнут для retries; caller-session НЕ закрыта.
|
||||
"""
|
||||
_enable_backconnect(monkeypatch)
|
||||
|
||||
caller = AsyncMock()
|
||||
caller.get = AsyncMock(return_value=_resp(403))
|
||||
|
||||
# Каждый retry → свежая эфемерная сессия. Первая отдаёт 403 (для 2-го reconnect),
|
||||
# вторая — 200.
|
||||
retry1 = AsyncMock()
|
||||
retry1.get = AsyncMock(return_value=_resp(403))
|
||||
retry2 = AsyncMock()
|
||||
retry2.get = AsyncMock(return_value=_resp(200, "<html/>"))
|
||||
build = MagicMock(side_effect=[retry1, retry2])
|
||||
|
||||
with patch.object(detail_mod, "_build_detail_session", build):
|
||||
result = await fetch_detail("/ekaterinburg/kvartiry/test-123", cffi_session=caller)
|
||||
|
||||
assert result is _SENTINEL
|
||||
assert build.call_count == 2 # 403, 403 → два reconnect перед 200
|
||||
caller.close.assert_not_called() # caller-provided session НЕ закрывается
|
||||
assert caller.get.await_count == 1 # первый attempt
|
||||
retry1.close.assert_awaited() # промежуточная retry-session закрыта
|
||||
retry2.close.assert_awaited() # последняя retry-session закрыта в finally
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_firewall_200_reconnect_succeeds(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""firewall(HTTP 200) → 200 → новый HTTP-200 firewall-detect триггерит reconnect."""
|
||||
_enable_backconnect(monkeypatch)
|
||||
|
||||
caller = AsyncMock()
|
||||
caller.get = AsyncMock(return_value=_resp(200, _FIREWALL_HTML))
|
||||
retry1 = AsyncMock()
|
||||
retry1.get = AsyncMock(return_value=_resp(200, "<html/>"))
|
||||
build = MagicMock(side_effect=[retry1])
|
||||
|
||||
with patch.object(detail_mod, "_build_detail_session", build):
|
||||
result = await fetch_detail("/x", cffi_session=caller)
|
||||
|
||||
assert result is _SENTINEL
|
||||
assert build.call_count == 1
|
||||
caller.close.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_403_exhausted_raises_blocked(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""403 _AVITO_DETAIL_403_MAX_RETRIES+1 раз → AvitoBlockedError."""
|
||||
_enable_backconnect(monkeypatch)
|
||||
|
||||
caller = AsyncMock()
|
||||
caller.get = AsyncMock(return_value=_resp(403))
|
||||
|
||||
def _make_retry() -> AsyncMock:
|
||||
s = AsyncMock()
|
||||
s.get = AsyncMock(return_value=_resp(403))
|
||||
return s
|
||||
|
||||
build = MagicMock(side_effect=lambda: _make_retry())
|
||||
|
||||
with patch.object(detail_mod, "_build_detail_session", build):
|
||||
with pytest.raises(AvitoBlockedError, match="HTTP 403"):
|
||||
await fetch_detail("/x", cffi_session=caller)
|
||||
|
||||
assert build.call_count == detail_mod._AVITO_DETAIL_403_MAX_RETRIES
|
||||
caller.close.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_429_short_retry_then_200_same_session(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""429 → 429 → 200: short-retry той же сессии, без reconnect."""
|
||||
_enable_backconnect(monkeypatch)
|
||||
|
||||
caller = AsyncMock()
|
||||
caller.get = AsyncMock(side_effect=[_resp(429), _resp(429), _resp(200, "<html/>")])
|
||||
build = MagicMock()
|
||||
|
||||
with patch.object(detail_mod, "_build_detail_session", build):
|
||||
result = await fetch_detail("/x", cffi_session=caller)
|
||||
|
||||
assert result is _SENTINEL
|
||||
build.assert_not_called() # 429 не дёргает reconnect
|
||||
assert caller.get.await_count == 3 # та же сессия переиспользована
|
||||
caller.close.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_proxy_single_403_raises_immediately(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""scraper_proxy_url unset → нет reconnect-пути → первый 403 сразу raise."""
|
||||
_disable_backconnect(monkeypatch)
|
||||
|
||||
caller = AsyncMock()
|
||||
caller.get = AsyncMock(return_value=_resp(403))
|
||||
build = MagicMock()
|
||||
|
||||
with patch.object(detail_mod, "_build_detail_session", build):
|
||||
with pytest.raises(AvitoBlockedError, match="HTTP 403"):
|
||||
await fetch_detail("/x", cffi_session=caller)
|
||||
|
||||
build.assert_not_called()
|
||||
assert caller.get.await_count == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_own_session_403_200_closes_both(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""own-session path (cffi_session=None): 403 → 200; own session + retry session
|
||||
обе закрыты в finally."""
|
||||
_enable_backconnect(monkeypatch)
|
||||
|
||||
own = AsyncMock()
|
||||
own.get = AsyncMock(return_value=_resp(403))
|
||||
retry1 = AsyncMock()
|
||||
retry1.get = AsyncMock(return_value=_resp(200, "<html/>"))
|
||||
# Первый _build_detail_session → own (own_session-ветка), второй → retry1 (reconnect).
|
||||
build = MagicMock(side_effect=[own, retry1])
|
||||
|
||||
with patch.object(detail_mod, "_build_detail_session", build):
|
||||
result = await fetch_detail("/x", cffi_session=None)
|
||||
|
||||
assert result is _SENTINEL
|
||||
own.close.assert_awaited() # own session закрыта (own_session=True)
|
||||
retry1.close.assert_awaited() # retry session закрыта в finally
|
||||
assert build.call_count == 2
|
||||
Loading…
Add table
Reference in a new issue