feat(tradein-imv): route IMV backfill через BrowserFetcher.fetch_json за dormant-флагом (#915 Stage 3) #1916

Merged
bot-backend merged 2 commits from feat/915-stage3-imv-browser into main 2026-06-26 09:54:29 +00:00
8 changed files with 583 additions and 57 deletions

View file

@ -21,6 +21,16 @@ YANDEX_GEOCODER_API_KEY=
DADATA_API_TOKEN=
DADATA_API_SECRET=
# === Scrapers ===
# Транспорт скраперов: "browser" (default — через камуфокс-сайдкар tradein-browser,
# обходит anti-bot/datacenter-IP) или "curl_cffi" (legacy fallback).
# SCRAPER_FETCH_MODE=browser
#
# #915 Stage 3: гонять avito IMV-backfill через /fetch-json сайдкара (in-page fetch
# на avito.ru — same-origin + прокси + warmed-cookies обходят datacenter-403, #562).
# Dormant по умолчанию; включать после прод-валидации IMV-через-браузер.
# AVITO_IMV_USE_BROWSER_FETCHER=false
# === Frontend (build-time) ===
# Backend URL для Next.js rewrites (browser обращается к /api/* — Next.js проксирует на BACKEND_URL/api/*).
# В docker compose это уже выставлено в http://backend:8000 — менять не нужно.

View file

@ -228,6 +228,9 @@ class Settings(BaseSettings):
# avito_imv_thin_market_threshold: если market_count < порога — IMV-оценка
# на тонком рынке (thin_market=True в AvitoImvSummary) + warning.
avito_imv_thin_market_threshold: int = 10
# #915 Stage 3: route IMV backfill через /fetch-json sidecar (обходит
# datacenter-403, #562). Dormant по умолчанию (ENV: AVITO_IMV_USE_BROWSER_FETCHER).
avito_imv_use_browser_fetcher: bool = False
# ── #764: per-cadastral-quarter price index correction ───────────────────
# Gap-correction: квартальный индекс применяется ТОЛЬКО в pure-radius пути

View file

@ -28,6 +28,7 @@ from typing import Literal
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.core.config import settings
from app.services.scrapers.avito_imv import (
IMVAddressNotFoundError,
IMVAuthError,
@ -35,6 +36,7 @@ from app.services.scrapers.avito_imv import (
IMVTransientError,
evaluate_via_imv,
)
from app.services.scrapers.browser_fetcher import BrowserFetcher
logger = logging.getLogger(__name__)
@ -444,9 +446,10 @@ async def backfill_house_imv(
request_delay_sec,
)
async def _run_loop(_bf: BrowserFetcher | None) -> None:
for i, house in enumerate(rows):
hid: int = house["id"]
status_str = await _process_one_house(db, dict(house))
status_str = await _process_one_house(db, dict(house), browser_fetcher=_bf)
result.status_counts[status_str] = result.status_counts.get(status_str, 0) + 1
if status_str == "ok":
@ -476,6 +479,16 @@ async def backfill_house_imv(
else:
await asyncio.sleep(request_delay_sec)
# #915 Stage 3: при флаге открываем ОДИН browser на весь батч (warmed-сессия
# + прокси переиспользуются всеми домами; обходит datacenter-403, #562/#853).
# Флаг OFF → _bf=None → evaluate_via_imv делает свою curl-сессию как раньше
# (поведение байт-в-байт идентично доспринтовому).
if settings.avito_imv_use_browser_fetcher:
async with BrowserFetcher(source="avito") as _bf:
await _run_loop(_bf)
else:
await _run_loop(None)
result.duration_sec = time.time() - t0
logger.info(
"house_imv_backfill done: checked=%d saved=%d skipped=%d errors=%d %.1fs %s",
@ -604,8 +617,15 @@ async def process_houses_imv_batch(
return result
async def _process_one_house(db: Session, house: dict) -> str:
"""Process a single house. Returns final status string."""
async def _process_one_house(
db: Session, house: dict, *, browser_fetcher: BrowserFetcher | None = None
) -> str:
"""Process a single house. Returns final status string.
browser_fetcher: при #915 Stage 3 backfill пробрасывает один общий
BrowserFetcher на весь батч evaluate_via_imv ходит через /fetch-json
sidecar (обходит datacenter-403). None curl_cffi-путь (как раньше).
"""
hid: int = house["id"]
params = pick_lot_params(db, hid)
@ -623,7 +643,9 @@ async def _process_one_house(db: Session, house: dict) -> str:
logger.debug("house_imv: enriched address house=%d %r -> %r", hid, address, enriched)
try:
eval_result = await evaluate_via_imv(address=enriched, **params)
eval_result = await evaluate_via_imv(
address=enriched, browser_fetcher=browser_fetcher, **params
)
except IMVAddressNotFoundError as exc:
_mark_status(db, hid, "not_found", str(exc)[:200])
return "not_found"

View file

@ -25,13 +25,19 @@ import logging
import re
from dataclasses import dataclass, field
from datetime import date, datetime
from typing import Any
from typing import TYPE_CHECKING, Any
from urllib.parse import urlencode
from uuid import UUID
from sqlalchemy import text
from sqlalchemy.orm import Session
if TYPE_CHECKING:
# Импорт под TYPE_CHECKING (как в avito_detail.py/avito_houses.py) — избегаем
# hard import cycle browser_fetcher → config → ... и лишней зависимости в
# offline-тестах парсинга, которым browser не нужен.
from app.services.scrapers.browser_fetcher import BrowserFetcher
logger = logging.getLogger(__name__)
AVITO_BASE = "https://www.avito.ru"
@ -360,6 +366,65 @@ def _parse_suggestions(data: dict[str, Any]) -> tuple[list[IMVSuggestion], str |
return suggestions, suggestions_block.get("searchSimilarUrl")
# ---------------------------------------------------------------------------
# Browser-transport adapter (#915 Stage 3)
# ---------------------------------------------------------------------------
class _BrowserResponse:
"""Мимикрия под curl_cffi Response поверх ответа BrowserFetcher.fetch_json.
IMV-флоу читает только .status_code / .text / .json() / .raise_for_status()
адаптируем ровно эти четыре (подтверждено по avito_imv.py: warm-up, _geocode,
_imv_evaluate, _raise_for_status_categorized).
"""
def __init__(self, status: int, body: str) -> None:
self.status_code = status
self.text = body
def json(self) -> Any:
return json.loads(self.text)
def raise_for_status(self) -> None:
if self.status_code >= 400:
# _raise_for_status_categorized уже обрабатывает 401/403/5xx ДО вызова
# raise_for_status(); сюда попадают только "прочие 4xx" (fall-through).
# Зеркалим curl_cffi/httpx-семантику ровно настолько, чтобы caller
# получил исключение.
raise RuntimeError(f"HTTP {self.status_code}")
class _BrowserSessionAdapter:
"""Адаптер: IMV-флоу думает что это curl_cffi-сессия, а ходит через камуфокс
in-page fetch (same-origin avito.ru). Куки/JWT живут в контексте браузера
провайдера, поэтому 4 запроса IMV переиспользуют warmed-сессию. Прокси и
fingerprint из sidecar (обходит datacenter-403, #562/#853)."""
def __init__(self, browser_fetcher: BrowserFetcher, *, origin: str) -> None:
self._bf = browser_fetcher
self._origin = origin
async def get(self, url: str, *, headers: dict | None = None) -> _BrowserResponse:
r = await self._bf.fetch_json(url, method="GET", headers=headers, origin=self._origin)
return _BrowserResponse(r["status"], r["body"])
async def post(
self, url: str, *, headers: dict | None = None, json: Any = None
) -> _BrowserResponse:
import json as _json
body = _json.dumps(json) if json is not None else None
r = await self._bf.fetch_json(
url, method="POST", headers=headers, body=body, origin=self._origin
)
return _BrowserResponse(r["status"], r["body"])
async def close(self) -> None:
# browser принадлежит вызывающему (backfill открывает один на батч) — no-op.
return None
# ---------------------------------------------------------------------------
# Main async function
# ---------------------------------------------------------------------------
@ -377,6 +442,7 @@ async def evaluate_via_imv(
has_loggia: bool,
*,
cffi_session: Any | None = None,
browser_fetcher: BrowserFetcher | None = None,
) -> IMVEvaluation:
"""Avito IMV — 3 HTTP requests (current contract 2026-05+):
@ -392,6 +458,16 @@ async def evaluate_via_imv(
IMVAuthError на 401/403
IMVTransientError на 5xx / network errors
"""
# #915 Stage 3: browser-транспорт. Если передан browser_fetcher — ходим через
# камуфокс in-page fetch (same-origin avito.ru), а не curl_cffi. Прокси +
# warmed-cookies + реальный fingerprint из sidecar обходят datacenter-403
# (#562/#853). _own_session=True → finally дёрнет adapter.close() (no-op,
# браузер принадлежит caller'у — backfill открывает один на батч). Этот путь
# НЕ требует curl_cffi, поэтому импорт пакета остаётся только в else-ветке.
if browser_fetcher is not None:
cffi_session = _BrowserSessionAdapter(browser_fetcher, origin=WARMUP_URL)
_own_session = True
else:
# Импортируем здесь чтобы избежать циклических зависимостей при тестах без сети
try:
from curl_cffi.requests import AsyncSession as CffiAsyncSession

View file

@ -216,6 +216,15 @@ class BrowserFetcher:
)
resp.raise_for_status()
data: dict = resp.json()
# Defensive: контракт сервера — {"status": int, "body": str} (зеркалит как
# _post_fetch читает data["html"]). Если ключи пропали (несовместимый сервер /
# прокинутый error-payload) — падаем с понятной ошибкой, а не KeyError ниже по
# стеку у адаптера, который ждёт r["status"]/r["body"].
if "status" not in data or "body" not in data:
raise RuntimeError(
f"BrowserFetcher: /fetch-json вернул некорректный ответ "
f"(нет 'status'/'body'): keys={sorted(data)} url={url!r}"
)
logger.debug(
"BrowserFetcher: fetch-json OK url=%r status=%s body_len=%d",
url,

View file

@ -0,0 +1,256 @@
"""test_avito_imv_browser_transport.py — #915 Stage 3.
Юниты browser-транспорта для IMV-флоу (без сети, без камуфокса):
1. _BrowserResponse мимикрия под curl_cffi Response (.status_code/.text/.json/
.raise_for_status).
2. _BrowserSessionAdapter .get/.post вызывают BrowserFetcher.fetch_json с
корректными (url, method, headers, body, origin); .post сериализует json.
3. evaluate_via_imv(..., browser_fetcher=mock) happy path через 4 канонических
ответа (warm-up GET coords position get-data) populated IMVEvaluation;
403 на get-data IMVAuthError.
browser_fetcher мокается AsyncMock'ом — реальный HTTP/камуфокс не поднимается.
"""
from __future__ import annotations
import asyncio
import json
from typing import Any
from unittest.mock import AsyncMock
import pytest
from app.services.scrapers.avito_imv import (
AVITO_BASE,
WARMUP_URL,
IMVAuthError,
_BrowserResponse,
_BrowserSessionAdapter,
evaluate_via_imv,
)
# JWT-заглушка (структура неважна — парсер берёт строку как есть в geo.geo_hash).
_FAKE_JWT = "eyJhbGciOiJIUzI1NiJ9.eyJsYXQiOjU2Ljg0fQ.sig"
# ── _BrowserResponse ──────────────────────────────────────────────────────────
def test_browser_response_status_and_text() -> None:
r = _BrowserResponse(200, '{"ok": true}')
assert r.status_code == 200
assert r.text == '{"ok": true}'
def test_browser_response_json_parses_body() -> None:
r = _BrowserResponse(200, '{"price": 7370000, "nested": {"a": 1}}')
assert r.json() == {"price": 7370000, "nested": {"a": 1}}
def test_browser_response_raise_for_status_noop_under_400() -> None:
# < 400 → не бросает (200/302/399).
_BrowserResponse(200, "ok").raise_for_status()
_BrowserResponse(399, "ok").raise_for_status()
def test_browser_response_raise_for_status_raises_on_4xx() -> None:
with pytest.raises(RuntimeError, match="HTTP 404"):
_BrowserResponse(404, "not found").raise_for_status()
def test_browser_response_403_status_code() -> None:
# 403 как status_code (категоризацию в IMVAuthError делает _raise_for_status_categorized).
r = _BrowserResponse(403, "forbidden")
assert r.status_code == 403
# ── _BrowserSessionAdapter ────────────────────────────────────────────────────
def _make_adapter(return_value: dict) -> tuple[_BrowserSessionAdapter, AsyncMock]:
bf = AsyncMock()
bf.fetch_json = AsyncMock(return_value=return_value)
adapter = _BrowserSessionAdapter(bf, origin=WARMUP_URL)
return adapter, bf
def test_adapter_get_calls_fetch_json() -> None:
adapter, bf = _make_adapter({"status": 200, "body": '{"x": 1}'})
async def _run() -> _BrowserResponse:
return await adapter.get(
"https://www.avito.ru/api/x", headers={"Accept": "application/json"}
)
resp = asyncio.run(_run())
bf.fetch_json.assert_awaited_once_with(
"https://www.avito.ru/api/x",
method="GET",
headers={"Accept": "application/json"},
origin=WARMUP_URL,
)
assert resp.status_code == 200
assert resp.json() == {"x": 1}
def test_adapter_post_serializes_body() -> None:
adapter, bf = _make_adapter({"status": 201, "body": "{}"})
payload = {"address": "ЕКБ", "latitude": 56.84, "isRadius": False}
resp = asyncio.run(
adapter.post(
"https://www.avito.ru/js/v2/geo/position",
headers={"Content-Type": "application/json"},
json=payload,
)
)
bf.fetch_json.assert_awaited_once()
_, kwargs = bf.fetch_json.await_args
assert kwargs["method"] == "POST"
assert kwargs["headers"] == {"Content-Type": "application/json"}
assert kwargs["origin"] == WARMUP_URL
# body должен быть JSON-строкой ровно как json.dumps(payload).
assert kwargs["body"] == json.dumps(payload)
assert json.loads(kwargs["body"]) == payload
assert resp.status_code == 201
def test_adapter_post_none_body_passes_none() -> None:
adapter, bf = _make_adapter({"status": 200, "body": "{}"})
asyncio.run(adapter.post("https://www.avito.ru/api/x"))
_, kwargs = bf.fetch_json.await_args
assert kwargs["body"] is None
def test_adapter_403_propagates_status_code() -> None:
adapter, _ = _make_adapter({"status": 403, "body": "forbidden"})
resp = asyncio.run(adapter.get("https://www.avito.ru/api/x"))
assert resp.status_code == 403
def test_adapter_close_is_noop() -> None:
adapter, _ = _make_adapter({"status": 200, "body": "{}"})
# close() не должен ничего ронять и ничего возвращает.
assert asyncio.run(adapter.close()) is None
# ── evaluate_via_imv(browser_fetcher=...) — full flow ─────────────────────────
def _canned_responses() -> list[dict]:
"""4 ответа в порядке вызова: warm-up GET → coords (A) → position (B) → get-data (C)."""
warmup = {"status": 200, "body": "<html>warm</html>"}
coords_a = {
"status": 200,
"body": json.dumps(
{
"normalizedAddress": "Свердловская область, Екатеринбург, ул. Чайковского, 66А",
"point": {"latitude": 56.840872, "longitude": 60.654077},
}
),
}
position_b = {
"status": 200,
"body": json.dumps(
{
"address": "Свердловская область, Екатеринбург, ул. Чайковского, 66А",
"addressId": 22319477,
"districtId": 123,
"geoFieldsHash": _FAKE_JWT,
"latitude": 56.840872,
"longitude": 60.654077,
"locationId": 654070,
"metroId": None,
}
),
}
getdata_c = {
"status": 200,
"body": json.dumps(
{
"price": {
"recommendedPrice": 7370000,
"lowerPrice": 7148900,
"higherPrice": 7738500,
"count": 800,
},
"placementHistory": {"items": []},
"suggestions": {"items": [], "searchSimilarUrl": "/ekb/kvartiry"},
}
),
}
return [warmup, coords_a, position_b, getdata_c]
def test_evaluate_via_imv_browser_happy_path() -> None:
bf = AsyncMock()
bf.fetch_json = AsyncMock(side_effect=_canned_responses())
evaluation = asyncio.run(
evaluate_via_imv(
address="Свердловская область, Екатеринбург, ул. Чайковского, 66А",
rooms=2,
area_m2=54.0,
floor=4,
floor_at_home=9,
house_type="panel",
renovation_type="cosmetic",
has_balcony=True,
has_loggia=False,
browser_fetcher=bf,
)
)
assert evaluation.recommended_price == 7370000
assert evaluation.lower_price == 7148900
assert evaluation.higher_price == 7738500
assert evaluation.market_count == 800
assert evaluation.geo.geo_hash == _FAKE_JWT
assert evaluation.geo.avito_address_id == 22319477
assert evaluation.geo.avito_location_id == 654070
assert evaluation.geo.lat == 56.840872
assert len(evaluation.cache_key) == 64 # sha256 hex
# 4 вызова fetch_json: warm-up, coords, position, get-data.
assert bf.fetch_json.await_count == 4
# Каждый ходил с origin=WARMUP_URL (same-origin якорь avito.ru).
for call in bf.fetch_json.await_args_list:
assert call.kwargs["origin"] == WARMUP_URL
# Последний (get-data) — POST на правильный endpoint.
last_call = bf.fetch_json.await_args_list[-1]
assert last_call.args[0] == f"{AVITO_BASE}/web/1/realty-imv/get-data"
assert last_call.kwargs["method"] == "POST"
# body несёт geoHash из step B.
sent_body: dict[str, Any] = json.loads(last_call.kwargs["body"])
assert sent_body["geoHash"] == _FAKE_JWT
assert sent_body["rooms"] == 2
def test_evaluate_via_imv_browser_403_on_getdata_raises_auth() -> None:
responses = _canned_responses()
responses[-1] = {"status": 403, "body": "forbidden"} # get-data → 403
bf = AsyncMock()
bf.fetch_json = AsyncMock(side_effect=responses)
with pytest.raises(IMVAuthError):
asyncio.run(
evaluate_via_imv(
address="ЕКБ, ул. Чайковского, 66А",
rooms=2,
area_m2=54.0,
floor=4,
floor_at_home=9,
house_type="panel",
renovation_type="cosmetic",
has_balcony=True,
has_loggia=False,
browser_fetcher=bf,
)
)

View file

@ -0,0 +1,98 @@
"""test_house_imv_backfill_browser_flag.py — #915 Stage 3.
Проверяет проводку флага avito_imv_use_browser_fetcher в backfill_house_imv:
- флаг ON конструируется один BrowserFetcher(source="avito") на весь батч,
и он прокидывается в _process_one_house(browser_fetcher=...).
- флаг OFF BrowserFetcher НЕ конструируется, browser_fetcher=None (старый путь).
DB и _process_one_house мокаются реальный камуфокс/curl/Postgres не нужны.
"""
from __future__ import annotations
import os
from typing import Any, ClassVar
from unittest.mock import MagicMock
import pytest
# settings нужен DSN на импорте (как в соседних тестах) — fake-db, статический.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
from app.services import house_imv_backfill
class _FakeBrowserFetcher:
"""Поддельный async-CM: считает конструкции/входы, возвращает себя из __aenter__."""
instances: ClassVar[list[_FakeBrowserFetcher]] = []
def __init__(self, source: str = "avito", **_: Any) -> None:
self.source = source
self.entered = False
_FakeBrowserFetcher.instances.append(self)
async def __aenter__(self) -> _FakeBrowserFetcher:
self.entered = True
return self
async def __aexit__(self, *_: object) -> None:
return None
def _fake_db_with_one_row() -> MagicMock:
"""MagicMock-Session, чей execute(...).mappings().all() отдаёт один дом."""
db = MagicMock()
row = {"id": 42, "address": "ЕКБ, ул. X, 1", "full_address": None, "lat": 56.8, "lon": 60.6}
db.execute.return_value.mappings.return_value.all.return_value = [row]
return db
async def test_backfill_flag_on_constructs_browser_and_threads_it(
monkeypatch: pytest.MonkeyPatch,
) -> None:
_FakeBrowserFetcher.instances.clear()
monkeypatch.setattr(house_imv_backfill.settings, "avito_imv_use_browser_fetcher", True)
monkeypatch.setattr(house_imv_backfill, "BrowserFetcher", _FakeBrowserFetcher)
captured: dict[str, Any] = {}
async def _fake_process(db: Any, house: dict, *, browser_fetcher: Any = None) -> str:
captured["browser_fetcher"] = browser_fetcher
return "ok"
monkeypatch.setattr(house_imv_backfill, "_process_one_house", _fake_process)
db = _fake_db_with_one_row()
await house_imv_backfill.backfill_house_imv(db, batch_size=10)
# Ровно один BrowserFetcher(source="avito"), вошли в контекст.
assert len(_FakeBrowserFetcher.instances) == 1
bf = _FakeBrowserFetcher.instances[0]
assert bf.source == "avito"
assert bf.entered is True
# И он прокинут в _process_one_house.
assert captured["browser_fetcher"] is bf
async def test_backfill_flag_off_no_browser_and_none_threaded(
monkeypatch: pytest.MonkeyPatch,
) -> None:
_FakeBrowserFetcher.instances.clear()
monkeypatch.setattr(house_imv_backfill.settings, "avito_imv_use_browser_fetcher", False)
monkeypatch.setattr(house_imv_backfill, "BrowserFetcher", _FakeBrowserFetcher)
captured: dict[str, Any] = {}
async def _fake_process(db: Any, house: dict, *, browser_fetcher: Any = None) -> str:
captured["browser_fetcher"] = browser_fetcher
return "ok"
monkeypatch.setattr(house_imv_backfill, "_process_one_house", _fake_process)
db = _fake_db_with_one_row()
await house_imv_backfill.backfill_house_imv(db, batch_size=10)
# Флаг OFF → BrowserFetcher не конструируется, browser_fetcher=None.
assert _FakeBrowserFetcher.instances == []
assert captured["browser_fetcher"] is None

View file

@ -266,3 +266,55 @@ def test_fetch_json_500_on_evaluate_error(monkeypatch: pytest.MonkeyPatch) -> No
assert "RuntimeError" in _json_body(response)["error"]
# Даже при ошибке evaluate страница должна закрыться (finally).
assert page.closed == 1
# ── crash-retry (#1915 deep-review nice-to-have) ──────────────────────────────────
class TargetClosedError(Exception):
"""Имя класса распознаётся _is_browser_crash (по type(exc).__name__) как краш."""
def test_do_fetch_json_relaunch_on_browser_crash(monkeypatch: pytest.MonkeyPatch) -> None:
"""_do_fetch_json: первый _fetch_json_once крашит браузер → relaunch + второй успех.
Проверяем именно crash-retry ветку: первый page.evaluate бросает
TargetClosedError (его type().__name__ ловит _is_browser_crash), что триггерит
_relaunch_browser + повторный _fetch_json_once на свежей странице. Результат
должен прийти из ВТОРОЙ попытки.
"""
crashing_page = _FakePage({"status": 0, "body": ""})
crashing_page.evaluate = AsyncMock(side_effect=TargetClosedError("Target page crashed"))
healthy_page = _FakePage({"status": 200, "body": '{"recovered": true}'})
# Первый инстанс — крашащийся; после relaunch подставляем здоровый.
server._browsers["avito"] = _FakeBrowser(crashing_page)
monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000)
relaunched: list[str] = []
async def _fake_relaunch(provider: str) -> None:
relaunched.append(provider)
server._browsers[provider] = _FakeBrowser(healthy_page)
monkeypatch.setattr(server, "_relaunch_browser", _fake_relaunch)
result = asyncio.run(
server._do_fetch_json(
"avito",
"https://www.avito.ru/api/x",
method="GET",
headers={},
body=None,
origin="https://www.avito.ru/",
)
)
# Результат — из ВТОРОЙ (после relaunch) попытки, не из крашнувшей.
assert result == {"status": 200, "body": '{"recovered": true}'}
assert relaunched == ["avito"]
# Обе страницы успели открыться/закрыться (crashing — в finally, healthy — норм).
crashing_page.evaluate.assert_awaited_once()
healthy_page.evaluate.assert_awaited_once()
assert crashing_page.closed == 1
assert healthy_page.closed == 1