fix(tradein/avito-detail): живой прогрев сессии (yandex-referer→search) + батч-на-сессию вместо холодного per-card fetch — снимает 403 (#1551)

This commit is contained in:
bot-backend 2026-06-29 00:54:21 +03:00
parent a96af4c19c
commit cb1f9e9514
4 changed files with 256 additions and 24 deletions

View file

@ -22,11 +22,12 @@ from __future__ import annotations
import asyncio
import json
import logging
import random
import re
from dataclasses import dataclass, field
from datetime import date
from typing import TYPE_CHECKING, Any
from urllib.parse import urljoin, urlparse
from urllib.parse import quote, urljoin, urlparse
from curl_cffi.requests import AsyncSession
from selectolax.parser import HTMLParser, Node
@ -49,6 +50,18 @@ logger = logging.getLogger(__name__)
AVITO_BASE = "https://www.avito.ru"
# Живой прогрев сессии (#1551): GET Yandex SERP (органический referer-источник) -> GET
# Avito EKB search-страница (Referer=yandex) сеет антибот-куки, после чего прогретая
# curl_cffi-сессия держит батч detail-GET'ов без 403 (доказано пробами на проде: 22/22
# карточек подряд, 0 блоков). referer на detail-GET = URL avito-search страницы.
_AVITO_WARM_SEARCH_URL = "https://www.avito.ru/ekaterinburg/kvartiry/prodam?q=" + quote(
"купить квартиру"
)
_AVITO_WARM_YANDEX_URL = "https://yandex.ru/search/?text=" + quote(
"купить квартиру екатеринбург авито"
)
_AVITO_WARM_YANDEX_REFERER = "https://yandex.ru/"
# Reconnect-retry на detail-странице под backconnect-прокси — зеркало SERP-фикса
# (#1765/#1769, avito.py): 403/firewall = залочен лишь текущий exit-IP backconnect-прокси;
# пересоздание curl_cffi-сессии = новый CONNECT-туннель = свежий exit-IP → escape.
@ -283,6 +296,38 @@ def _build_detail_session() -> AsyncSession:
)
async def warm_up_session(session: AsyncSession) -> bool:
"""Органический прогрев сессии: Yandex SERP (best-effort) -> Avito EKB search
(Referer=yandex). Сеет антибот-куки, после чего сессия держит батч detail-GET'ов
без 403. Возвращает True если search-страница загрузилась (не firewall).
"""
try:
await session.get(_AVITO_WARM_YANDEX_URL) # органический referer-источник, best-effort
except Exception:
pass
await asyncio.sleep(random.uniform(2.0, 3.5))
try:
r = await session.get(
_AVITO_WARM_SEARCH_URL, headers={"Referer": _AVITO_WARM_YANDEX_REFERER}
)
except Exception as exc:
logger.warning("avito warm-up: search fetch error: %s", exc)
return False
ok = r.status_code == 200 and not _is_firewall_page(r.text)
if not ok:
logger.warning("avito warm-up: search page blocked sc=%s", r.status_code)
return ok
async def build_warmed_session() -> AsyncSession:
"""Построить detail-сессию (_build_detail_session) и прогреть её. Возвращает
прогретую сессию (даже если warm-up soft-failed caller может всё равно пробовать).
"""
session = _build_detail_session()
await warm_up_session(session)
return session
# ── Soft-block detection (#1967 / #1950) ──────────────────────────────────────
# Под нагрузкой Avito иногда вместо detail-страницы отдаёт HTTP 200 с ОБЩЕЙ
# витриной (<title>Авито — Объявления на сайте Авито</title>) вместо запрошенного
@ -327,11 +372,19 @@ async def fetch_detail(
*,
cffi_session: AsyncSession | None = None,
browser_fetcher: BrowserFetcher | None = None,
referer: str | None = None,
reconnect_on_block: bool = True,
) -> DetailEnrichment:
"""GET <avito>/{item_url} → parse HTML via selectolax → DetailEnrichment.
Если browser_fetcher передан использует браузерный fetch (browser mode).
Если cffi_session не передана создаёт новую (impersonate='chrome120').
referer если задан, шлётся в Referer-заголовке detail-GET'а (warm-batch #1551:
referer = URL avito-search страницы, на которой прогрелась сессия).
reconnect_on_block если False, 403/firewall поднимает AvitoBlockedError СРАЗУ
(без холодных reconnect'ов — ими займётся outer-loop через rebuild+rewarm прогретой
сессии); 429 после short-retry поднимает AvitoRateLimitedError. Дефолт True
сохраняет старое поведение own-session / scrape_pipeline путей.
Raises:
httpx.HTTPError если status != 200 (через raise_for_status-like).
ValueError если item_id не извлечён из HTML.
@ -387,7 +440,12 @@ async def fetch_detail(
# ТОЛЬКО на наличие proxy (зеркало SERP-фикса #1765). Без proxy — старое поведение
# (raise на первом 403). Caller-provided сессию НЕ трогаем: reconnect использует
# эфемерную retry_session, которую сами закрываем в finally.
backconnect = bool(settings.scraper_proxy_url)
# reconnect_on_block=False (warm-batch #1551): не делаем холодных per-card
# reconnect'ов — блок поднимает AvitoBlockedError сразу, outer-loop пересоздаёт+
# прогревает сессию (свежий exit-IP). reconnect_on_block=True (own-session /
# scrape_pipeline) — старое поведение: backconnect-reconnect эфемерной сессией.
backconnect = bool(settings.scraper_proxy_url) and reconnect_on_block
req_headers = {"Referer": referer} if referer else None
r403 = 0
r429 = 0
r429_reconnect = 0
@ -395,7 +453,7 @@ async def fetch_detail(
retry_session: AsyncSession | None = None
try:
while True:
response = await attempt_session.get(full_url)
response = await attempt_session.get(full_url, headers=req_headers)
sc = response.status_code
is_firewall = sc == 200 and (
_is_firewall_page(response.text) or _is_detail_soft_block(response.text)

View file

@ -17,6 +17,7 @@ from __future__ import annotations
import asyncio
import logging
import random
import time
from dataclasses import dataclass, field
from urllib.parse import urlparse
@ -30,7 +31,12 @@ 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.scrapers.avito import AvitoScraper
from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment
from app.services.scrapers.avito_detail import (
_AVITO_WARM_SEARCH_URL,
build_warmed_session,
fetch_detail,
save_detail_enrichment,
)
from app.services.scrapers.avito_exceptions import (
AvitoBlockedError,
AvitoListingGoneError,
@ -89,6 +95,7 @@ async def run_avito_detail_backfill(
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", 20))
# #1950: hard-timeout'ы на блокирующие await'ы внутри loop'а. Без них один
# зависший fetch_detail/rotate ронял весь run в zombie (run 423 завис 7.7ч).
@ -129,6 +136,13 @@ async def run_avito_detail_backfill(
await browser_fetcher.__aenter__()
own_browser = True
scraper._browser = browser_fetcher
elif use_curl:
# use_curl=True (#1551 warm-batch): прогретая shared-сессия — yandex-referer ->
# avito-search сеет антибот-куки, сессия держит батч detail без 403 (доказано
# пробами). Ротация каждые warm_batch карточек И на блоке = свежий exit-IP +
# свежий органический прогрев.
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.
@ -140,8 +154,6 @@ async def run_avito_detail_backfill(
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)
@ -197,6 +209,7 @@ async def run_avito_detail_backfill(
consecutive_blocks = 0
do_sleep = False
items_since_warm = 0
for idx, row in enumerate(snapshot):
# Budget guard
@ -227,9 +240,10 @@ async def run_avito_detail_backfill(
)
break
# Delay before each request except the first
# Delay before each request except the first (джиттер ±30% — менее
# роботизированный паттерн на органически прогретой сессии).
if do_sleep:
await asyncio.sleep(request_delay_sec)
await asyncio.sleep(request_delay_sec * random.uniform(0.7, 1.4))
do_sleep = True
source_url: str = row["source_url"]
@ -238,6 +252,17 @@ async def run_avito_detail_backfill(
# 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): ротация прогретой сессии каждые warm_batch
# карточек — свежий exit-IP + свежий органический прогрев, щадит IP-пул
# (1 exit-IP на батч вместо 1 на карточку).
if use_curl and session is not None and items_since_warm >= warm_batch:
try:
await session.close()
except Exception:
pass
session = await build_warmed_session()
items_since_warm = 0
try:
# #1950: hard-timeout — зависший fetch (browser hang / curl-stall) не
# должен блокировать loop навсегда (иначе budget-guard/heartbeat молчат
@ -247,11 +272,15 @@ async def run_avito_detail_backfill(
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,
),
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:
@ -299,18 +328,31 @@ async def run_avito_detail_backfill(
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,
)
# use_curl warm-batch (#1551): блок = текущий exit-IP залочен → закрываем
# прогретую сессию и строим новую (свежий exit-IP + новый органический
# прогрев) вместо changeip-ротации. Иначе legacy auv/browser-путь —
# bound rotate backconnect-IP.
if use_curl:
if session is not None:
try:
await session.close()
except Exception:
pass
session = await build_warmed_session()
items_since_warm = 0
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,
)
if consecutive_blocks >= max_consecutive_blocks:
logger.error(
"avito_detail_backfill: run_id=%d ABORT -- %d consecutive blocks, "

View file

@ -0,0 +1,106 @@
"""Живой прогрев сессии + батч-на-сессию (#1551).
warm_up_session: GET Yandex SERP (best-effort) -> GET Avito EKB search-страница с
Referer=yandex сеет антибот-куки. Прогретая сессия потом держит батч detail-GET'ов
без 403 (доказано пробами на проде: 22/22 карточек подряд, 0 блоков).
fetch_detail(referer=..., reconnect_on_block=False): шлёт Referer detail-GET'а и при
403/firewall поднимает AvitoBlockedError СРАЗУ без холодных per-card reconnect'ов
(их делает outer-loop через rebuild+rewarm прогретой сессии в avito_detail_backfill).
Стиль фикстур/моков зеркалит test_avito_detail_403_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, warm_up_session
from app.services.scrapers.avito_exceptions import AvitoBlockedError
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")
# ── warm_up_session ───────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_warm_up_session_search_with_yandex_referer_true_on_200() -> None:
"""warm_up_session шлёт GET на _AVITO_WARM_SEARCH_URL с Referer=yandex и
возвращает True на 200/non-firewall."""
session = AsyncMock()
session.get = AsyncMock(return_value=_resp(200, "<html>ok</html>"))
with (
patch("app.services.scrapers.avito_detail.asyncio.sleep", AsyncMock()),
patch("app.services.scrapers.avito_detail._is_firewall_page", return_value=False),
):
ok = await warm_up_session(session)
assert ok is True
# search-GET (на avito-search URL) ушёл ровно один раз и с Referer=yandex.
search_calls = [
c
for c in session.get.await_args_list
if c.args and c.args[0] == detail_mod._AVITO_WARM_SEARCH_URL
]
assert len(search_calls) == 1
assert search_calls[0].kwargs["headers"]["Referer"] == detail_mod._AVITO_WARM_YANDEX_REFERER
# Yandex SERP — органический referer-источник — тоже фетчился (best-effort, до search).
assert any(
c.args and c.args[0] == detail_mod._AVITO_WARM_YANDEX_URL
for c in session.get.await_args_list
)
@pytest.mark.asyncio
async def test_warm_up_session_false_on_403() -> None:
"""search-страница вернула 403 → warm_up_session возвращает False."""
session = AsyncMock()
session.get = AsyncMock(return_value=_resp(403))
with (
patch("app.services.scrapers.avito_detail.asyncio.sleep", AsyncMock()),
patch("app.services.scrapers.avito_detail._is_firewall_page", return_value=False),
):
ok = await warm_up_session(session)
assert ok is False
# ── fetch_detail referer + reconnect_on_block=False ───────────────────────────
@pytest.mark.asyncio
async def test_fetch_detail_referer_forwarded_and_block_raises_immediately(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""fetch_detail(referer="X", reconnect_on_block=False): Referer="X" уходит в
detail-GET, а 403 поднимает AvitoBlockedError СРАЗУ без холодного reconnect
(даже когда backconnect-proxy включён гейт = reconnect_on_block, не наличие proxy).
"""
_enable_backconnect(monkeypatch)
caller = AsyncMock()
caller.get = AsyncMock(return_value=_resp(403))
build = MagicMock() # _build_detail_session НЕ должна дёргаться (нет reconnect'а)
with patch.object(detail_mod, "_build_detail_session", build):
with pytest.raises(AvitoBlockedError, match="HTTP 403"):
await fetch_detail(
"/ekaterinburg/kvartiry/test-123",
cffi_session=caller,
referer="X",
reconnect_on_block=False,
)
build.assert_not_called() # никаких холодных reconnect'ов
assert caller.get.await_count == 1 # один detail-GET → сразу raise
assert caller.get.await_args.kwargs["headers"] == {"Referer": "X"}

View file

@ -27,6 +27,19 @@ def _reset_shutdown() -> None:
_sd.reset_shutdown()
@pytest.fixture(autouse=True)
def _patch_build_warmed() -> object:
"""use_curl-ветка (#1551) строит прогретую сессию через build_warmed_session
(реальный yandex/avito warm-up + sleep). fetch_detail в этих тестах всё равно
замокан мокаем build_warmed_session не-сетевой AsyncMock-сессией, чтобы setup
и rebuild-on-block не ходили в сеть."""
with patch(
"app.tasks.avito_detail_backfill.build_warmed_session",
AsyncMock(return_value=AsyncMock()),
) as p:
yield p
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
@ -129,7 +142,9 @@ async def test_backfill_blocked_abort_after_max_consecutive() -> None:
mock_fetch = AsyncMock(side_effect=blocked_exc)
mock_scraper = MagicMock()
mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
fake_settings = MagicMock(scraper_fetch_mode="cffi")
# use_curl=False: legacy block→_rotate_ip путь (#1551 warm-batch rebuild только при
# use_curl=True). MagicMock без явного флага сделал бы use_curl truthy.
fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False)
with (
patch(_SETTINGS, fake_settings),
patch(_SESSION, return_value=AsyncMock()),
@ -249,7 +264,9 @@ async def test_backfill_rotate_ip_called_on_each_block() -> None:
mock_fetch = AsyncMock(side_effect=[blocked_exc, mock_enrichment])
mock_scraper = MagicMock()
mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
fake_settings = MagicMock(scraper_fetch_mode="cffi")
# use_curl=False: legacy block→_rotate_ip путь (#1551 warm-batch rebuild только при
# use_curl=True). MagicMock без явного флага сделал бы use_curl truthy.
fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False)
with (
patch(_SETTINGS, fake_settings),
patch(_SESSION, return_value=AsyncMock()),
@ -342,7 +359,14 @@ async def test_backfill_fetch_timeout_skips_and_continues() -> None:
call_urls: list[str] = []
async def _fetch(url: str, *, cffi_session: object = None, browser_fetcher: object = None):
async def _fetch(
url: str,
*,
cffi_session: object = None,
browser_fetcher: object = None,
referer: object = None,
reconnect_on_block: bool = True,
):
call_urls.append(url)
if len(call_urls) == 1:
await asyncio.Event().wait() # висит вечно → wait_for отменит по timeout
@ -394,7 +418,9 @@ async def test_backfill_listing_gone_marks_inactive_no_breaker() -> None:
mock_fetch = AsyncMock(side_effect=AvitoListingGoneError("404 gone"))
mock_scraper = MagicMock()
mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
fake_settings = MagicMock(scraper_fetch_mode="cffi")
# use_curl=False: legacy block→_rotate_ip путь (#1551 warm-batch rebuild только при
# use_curl=True). MagicMock без явного флага сделал бы use_curl truthy.
fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False)
with (
patch(_SETTINGS, fake_settings),
patch(_SESSION, return_value=AsyncMock()),