lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

feat(yandex): route gate-API fetch through camoufox (sustainable, no proxy burn) #1699

Merged
lekss361 merged 1 commit from feat/yandex-camoufox-transport into main 2026-06-17 18:28:51 +00:00
Conversation 0 Commits 1 Files changed 3 +165 -198
3 changed files with 165 additions and 198 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

174
tradein-mvp/backend/app/services/scrapers/yandex_realty.py
View file

@ -1,7 +1,13 @@
"""Yandex.Nedvizhimost scraper (realty.yandex.ru) -- gate-API JSON parser.
Transport: system curl --compressed -x <socks5 proxy> (subprocess).
Proven to fetch gate-API payload in ~17s at 129 KB on prod (2026-06-17).
Transport: BrowserFetcher (camoufox real-browser fingerprint via tradein-browser service).
camoufox's fingerprint is not tarpitted by yandex and does not burn the mobile-proxy IP;
raw system-curl over SOCKS5 gets the proxy IP flagged. One BrowserFetcher context is opened
per scraper session (__aenter__) and reused across all page fetches.
camoufox returns the gate-API JSON wrapped in HTML (verified prod 2026-06-17):
<html><head>...</head><body><pre>{"response":...}</pre></body></html>
JSON is extracted via _extract_json_from_content() (handles the <pre> wrapper).
API path: GET https://realty.yandex.ru/gate/react-page/get/
_pageType=search&_providers=react-search-results-data
@ -24,6 +30,7 @@ import asyncio
import json
import logging
import math
import re
from dataclasses import dataclass
from typing import Any
from urllib.parse import urlencode
@ -32,18 +39,17 @@ from curl_cffi.requests import AsyncSession as _CurlCffiSession
from app.services.scraper_settings import get_scraper_delay
from app.services.scrapers.base import BaseScraper, ScrapedLot
from app.services.scrapers.browser_fetcher import BrowserFetcher
logger = logging.getLogger(__name__)
# camoufox renders JSON endpoints as <pre>{"response":...}</pre> inside an HTML wrapper.
_RE_PRE_JSON = re.compile("<pre[^>]*>(\\{.*?)</pre>", re.DOTALL | re.IGNORECASE)
DEFAULT_CITY = "ekaterinburg"
_GATE_MAX_PAGES_CAP = 50
_EKB_RGID = 559132
_GATE_URL = "https://realty.yandex.ru/gate/react-page/get/"
_CURL_UA = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36"
)
_CURL_STATUS_MARKER = "\n###CURL_HTTP_STATUS:"
ROOM_GATE_PARAM: dict[str, str | int] = {
"studio": "STUDIO",
@ -100,6 +106,29 @@ def _extract_gate_data(
return None
def _extract_json_from_content(content: str) -> str | None:
"""Extract raw JSON text from camoufox HTML-wrapped content.
camoufox renders a JSON endpoint as:
<html><head>...</head><body><pre>{"response":...}</pre></body></html>
Strategy:
1. Try <pre>...</pre> regex (primary -- matches prod output).
2. Fallback: slice from the first '{' (handles raw-JSON response, no wrapper).
Returns the JSON string, or None if no '{' is found.
"""
m = _RE_PRE_JSON.search(content)
if m:
return m.group(1)
idx = content.find("{")
if idx != -1:
return content[idx:]
return None
def _parse_gate_json(payload: dict[str, Any], page_param: int = 1) -> list[ScrapedLot]:
"""Parse gate-API JSON payload into a list of ScrapedLot.
@ -257,114 +286,61 @@ class YandexRealtyScraper(BaseScraper):
super().__init__()
self.city = city
self.request_delay_sec = get_scraper_delay(self.name)
self._browser: BrowserFetcher | None = None
# _cffi_session retained only for _rotate_ip (changeip call).
self._cffi_session: _CurlCffiSession | None = None
self._cookies: dict[str, str] = {}
async def __aenter__(self) -> YandexRealtyScraper: # type: ignore[override]
"""Create curl_cffi session as fallback transport (dev / non-socks5 proxy).
"""Open ONE BrowserFetcher (camoufox) session, reused across all page fetches.
Primary transport: system curl subprocess over SOCKS5 (see _http_get).
gate-API does not require cookies or heavy anti-bot headers.
camoufox's real-browser fingerprint avoids the mobile-proxy IP tarpit
escalation that raw system-curl over SOCKS5 triggers. Opening per-page is
expensive and unnecessary.
"""
from app.core.config import settings
_proxy_url = settings.yandex_proxy_url
_proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
self._cffi_session = _CurlCffiSession(
impersonate="chrome120",
cookies={},
proxies=_proxies,
headers={
"Accept": "application/json",
"X-Requested-With": "XMLHttpRequest",
"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
"Accept-Encoding": "gzip, deflate, br",
},
)
self._browser = BrowserFetcher()
await self._browser.__aenter__()
return self
async def __aexit__(self, *args: object) -> None: # type: ignore[override]
if self._browser is not None:
await self._browser.__aexit__(*args)
self._browser = None
if self._cffi_session is not None:
await self._cffi_session.close()
self._cffi_session = None
async def _curl_subprocess_get(self, url: str, proxy: str, timeout: int) -> _CurlResponse:
"""Fetch via system curl --compressed -x <socks5 proxy>.
async def _http_get(self, url: str, **kwargs: object) -> _CurlResponse: # type: ignore[override]
"""Fetch gate-API URL via the shared BrowserFetcher (camoufox).
Appends HTTP status via -w marker and strips it from body.
curl exit 28 (timeout) -> status_code=0.
camoufox returns HTML-wrapped JSON; the JSON is extracted and returned as
the body of a _CurlResponse so existing callers (json.loads(resp.text),
resp.status_code checks) keep working unchanged.
Maps outcomes onto the curl-style status contract:
- JSON extracted -> status_code=200, text=<json>
- fetch raised / no JSON found -> status_code=0 (treated as transient
failure by the caller retry logic)
"""
args = [
"curl",
"-sS",
"-L",
"--compressed",
"-x",
proxy,
"--max-time",
str(timeout),
"-A",
_CURL_UA,
"-H",
"Accept: application/json",
"-H",
"X-Requested-With: XMLHttpRequest",
"-H",
"Accept-Language: ru-RU,ru;q=0.9,en;q=0.8",
"-w",
_CURL_STATUS_MARKER + "%{http_code}",
url,
]
try:
proc = await asyncio.create_subprocess_exec(
*args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
out, err = await asyncio.wait_for(proc.communicate(), timeout=timeout + 10)
except TimeoutError:
logger.warning("yandex curl: subprocess timeout url=%s", url)
return _CurlResponse(status_code=0, text="")
except Exception:
logger.exception("yandex curl: subprocess failed url=%s", url)
return _CurlResponse(status_code=0, text="")
body = out.decode("utf-8", "replace")
status = 0
marker = _CURL_STATUS_MARKER
idx = body.rfind(marker)
if idx != -1:
try:
status = int(body[idx + len(marker) :].strip() or "0")
except ValueError:
status = 0
body = body[:idx]
if proc.returncode == 28:
status = 0
elif proc.returncode not in (0, None) and status == 0:
logger.warning(
"yandex curl: rc=%s url=%s err=%s",
proc.returncode,
url,
err.decode("utf-8", "replace")[:200],
)
return _CurlResponse(status_code=status, text=body)
async def _http_get(self, url: str, **kwargs: object) -> object: # type: ignore[override]
"""Route GET to system curl (SOCKS5 proxy) or curl_cffi fallback."""
from app.core.config import settings
proxy = settings.yandex_proxy_url
timeout = int(kwargs.get("timeout", 60) or 60)
if proxy and proxy.startswith(("socks5://", "socks5h://")):
return await self._curl_subprocess_get(url, proxy, timeout)
if self._cffi_session is None:
if self._browser is None:
raise RuntimeError("YandexRealtyScraper must be used as async context manager")
kwargs.setdefault("timeout", 60)
return await self._cffi_session.get(url, **kwargs)
try:
content = await self._browser.fetch(url)
except Exception:
logger.warning("yandex gate: BrowserFetcher.fetch failed url=%s", url, exc_info=True)
return _CurlResponse(status_code=0, text="")
json_text = _extract_json_from_content(content)
if not json_text:
logger.warning(
"yandex gate: could not extract JSON from content url=%s first200=%r",
url,
content[:200],
)
return _CurlResponse(status_code=0, text="")
return _CurlResponse(status_code=200, text=json_text)
async def _rotate_ip(self) -> bool:
"""Rotate mobile proxy IP via changeip URL. Returns True on success."""

53
tradein-mvp/backend/tests/test_scraper_proxy.py
View file

@ -174,52 +174,25 @@ _EXPECTED_PROXIES = {"http": _PROXY_URL, "https": _PROXY_URL}
@pytest.mark.asyncio
async def test_yandex_realty_session_receives_proxies(monkeypatch):
"""YandexRealtyScraper.__aenter__ forwards scraper_proxy_url as proxies= kwarg.
async def test_yandex_realty_opens_browser_fetcher(monkeypatch):
"""YandexRealtyScraper.__aenter__ opens ONE BrowserFetcher (camoufox) session.
settings is imported deferred inside __aenter__ via `from app.core.config import settings`,
so we patch app.core.config.settings (module attribute) which the deferred import reads.
Transport moved off the proxied curl_cffi session to BrowserFetcher (camoufox);
the mobile proxy is applied server-side by the tradein-browser service, not via a
proxies= kwarg here.
"""
from app.services.scrapers.yandex_realty import YandexRealtyScraper
mock_session = MagicMock()
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock(return_value=None)
mock_session.close = AsyncMock()
mock_browser = MagicMock()
mock_browser.__aenter__ = AsyncMock(return_value=mock_browser)
mock_browser.__aexit__ = AsyncMock(return_value=None)
captured_kwargs: dict = {}
monkeypatch.setattr("app.services.scrapers.yandex_realty.BrowserFetcher", lambda: mock_browser)
scraper = YandexRealtyScraper()
await scraper.__aenter__()
def _fake_session(*args, **kwargs):
captured_kwargs.update(kwargs)
return mock_session
monkeypatch.setattr("app.services.scrapers.yandex_realty._CurlCffiSession", _fake_session)
with patch("app.core.config.settings", _mock_settings(_PROXY_URL)):
scraper = YandexRealtyScraper()
await scraper.__aenter__()
assert captured_kwargs.get("proxies") == _EXPECTED_PROXIES
@pytest.mark.asyncio
async def test_yandex_realty_session_no_proxies_when_none(monkeypatch):
"""YandexRealtyScraper.__aenter__: proxies=None when scraper_proxy_url is None."""
from app.services.scrapers.yandex_realty import YandexRealtyScraper
mock_session = MagicMock()
mock_session.close = AsyncMock()
captured_kwargs: dict = {}
def _fake_session(*args, **kwargs):
captured_kwargs.update(kwargs)
return mock_session
monkeypatch.setattr("app.services.scrapers.yandex_realty._CurlCffiSession", _fake_session)
with patch("app.core.config.settings", _mock_settings(None)):
scraper = YandexRealtyScraper()
await scraper.__aenter__()
assert captured_kwargs.get("proxies") is None
mock_browser.__aenter__.assert_awaited_once()
assert scraper._browser is mock_browser
# ── YandexValuationScraper ───────────────────────────────────────────────────

136
tradein-mvp/backend/tests/test_yandex_realty_serp.py
View file

@ -11,7 +11,6 @@ import pytest
from app.services.scrapers.base import ScrapedLot
from app.services.scrapers.yandex_realty import (
_CURL_STATUS_MARKER,
_GATE_MAX_PAGES_CAP,
DEFAULT_CITY,
DEFAULT_PRICE_RANGES,
@ -21,6 +20,7 @@ from app.services.scrapers.yandex_realty import (
_CurlResponse,
_entity_to_lot,
_extract_gate_data,
_extract_json_from_content,
_is_gate_error,
_parse_gate_json,
)
@ -389,92 +389,110 @@ def test_combo_label_different_ranges_differ():
# ---------------------------------------------------------------------------
# camoufox renders the gate-API JSON wrapped in an HTML <pre> (verified prod 2026-06-17):
_CAMOUFOX_HTML_PREFIX = (
'<html><head><link rel="stylesheet" '
'href="resource://content-accessible/plaintext.css"></head><body><pre>'
)
_CAMOUFOX_HTML_SUFFIX = "</pre></body></html>"
def _wrap_camoufox(json_text: str) -> str:
"""Wrap a JSON string the way camoufox renders a JSON endpoint."""
return f"{_CAMOUFOX_HTML_PREFIX}{json_text}{_CAMOUFOX_HTML_SUFFIX}"
def test_extract_json_from_content_unwraps_camoufox_pre():
"""_extract_json_from_content must unwrap camoufox's <pre>{json}</pre> HTML.
Feeds the real prod-observed wrapper and asserts the extracted text parses
back to the original gate-API payload.
"""
payload = _make_gate_payload([_ENTITY_FULL])
wrapped = _wrap_camoufox(json.dumps(payload))
extracted = _extract_json_from_content(wrapped)
assert extracted is not None
assert extracted.startswith('{"response"')
assert json.loads(extracted) == payload
def test_extract_json_from_content_raw_json_fallback():
"""Raw JSON (no HTML wrapper) is returned via the first-'{' fallback."""
raw = json.dumps(_make_gate_payload([]))
assert _extract_json_from_content(raw) == raw
def test_extract_json_from_content_returns_none_when_no_json():
assert _extract_json_from_content("<html><body>nope</body></html>") is None
@pytest.mark.asyncio
async def test_aexit_closes_session():
async def test_aexit_closes_browser_and_session():
s = YandexRealtyScraper()
mock_browser = AsyncMock()
mock_session = AsyncMock()
s._browser = mock_browser
s._cffi_session = mock_session
await s.__aexit__(None, None, None)
mock_browser.__aexit__.assert_awaited_once()
mock_session.close.assert_awaited_once()
assert s._browser is None
assert s._cffi_session is None
@pytest.mark.asyncio
async def test_http_get_raises_when_no_context():
"""_http_get must raise RuntimeError outside async context manager (cffi path)."""
async def test_http_get_raises_when_no_browser():
"""_http_get must raise RuntimeError outside async context manager (no browser)."""
s = YandexRealtyScraper()
mock_settings = MagicMock()
mock_settings.yandex_proxy_url = None
with patch.dict("sys.modules", {"app.core.config": MagicMock(settings=mock_settings)}):
with pytest.raises(RuntimeError, match="async context manager"):
await s._http_get("https://realty.yandex.ru/test/")
with pytest.raises(RuntimeError, match="async context manager"):
await s._http_get("https://realty.yandex.ru/test/")
@pytest.mark.asyncio
async def test_http_get_routes_socks5_to_curl_subprocess():
async def test_http_get_unwraps_camoufox_html_to_json_body():
"""_http_get fetches via BrowserFetcher and returns extracted JSON with status 200."""
s = YandexRealtyScraper()
sentinel = _CurlResponse(status_code=200, text="{}")
payload = _make_gate_payload([_ENTITY_FULL])
wrapped = _wrap_camoufox(json.dumps(payload))
mock_settings = MagicMock()
mock_settings.yandex_proxy_url = "socks5h://x:y@h:1"
mock_browser = AsyncMock()
mock_browser.fetch = AsyncMock(return_value=wrapped)
s._browser = mock_browser
cffi_mock = AsyncMock()
s._cffi_session = cffi_mock
async def fake_curl_subprocess(url: str, proxy: str, timeout: int) -> object:
return sentinel
with patch.dict("sys.modules", {"app.core.config": MagicMock(settings=mock_settings)}):
with patch.object(s, "_curl_subprocess_get", side_effect=fake_curl_subprocess) as mock_curl:
result = await s._http_get("https://realty.yandex.ru/test/", timeout=30)
mock_curl.assert_awaited_once()
assert result is sentinel
cffi_mock.get.assert_not_called()
@pytest.mark.asyncio
async def test_curl_subprocess_get_parses_status_and_strips_marker():
s = YandexRealtyScraper()
fake_body = b"{" + b'"response":{}}'
fake_stdout = fake_body + _CURL_STATUS_MARKER.encode() + b"200"
fake_proc = MagicMock()
fake_proc.returncode = 0
fake_proc.communicate = AsyncMock(return_value=(fake_stdout, b""))
with patch("asyncio.create_subprocess_exec", return_value=fake_proc):
resp = await s._curl_subprocess_get(
"https://realty.yandex.ru/test/", "socks5h://x:y@h:1", 30
)
resp = await s._http_get("https://realty.yandex.ru/test/", timeout=30)
assert isinstance(resp, _CurlResponse)
assert resp.status_code == 200
assert resp.text == "{" + '"response":{}}'
assert json.loads(resp.text) == payload
mock_browser.fetch.assert_awaited_once_with("https://realty.yandex.ru/test/")
@pytest.mark.asyncio
async def test_http_get_http_proxy_uses_cffi_not_curl():
"""When proxy is http://, must NOT call _curl_subprocess_get."""
async def test_http_get_status_0_on_fetch_exception():
"""BrowserFetcher.fetch raising -> _CurlResponse(status_code=0) (transient failure)."""
s = YandexRealtyScraper()
sentinel_resp = MagicMock()
sentinel_resp.status_code = 200
sentinel_resp.text = "{}"
mock_browser = AsyncMock()
mock_browser.fetch = AsyncMock(side_effect=RuntimeError("browser down"))
s._browser = mock_browser
cffi_mock = AsyncMock()
cffi_mock.get = AsyncMock(return_value=sentinel_resp)
s._cffi_session = cffi_mock
resp = await s._http_get("https://realty.yandex.ru/test/")
mock_settings = MagicMock()
mock_settings.yandex_proxy_url = "http://proxy.example.com:8080"
assert resp.status_code == 0
assert resp.text == ""
with patch.dict("sys.modules", {"app.core.config": MagicMock(settings=mock_settings)}):
with patch.object(s, "_curl_subprocess_get") as mock_curl:
result = await s._http_get("https://realty.yandex.ru/test/", timeout=30)
mock_curl.assert_not_called()
cffi_mock.get.assert_awaited_once()
assert result is sentinel_resp
@pytest.mark.asyncio
async def test_http_get_status_0_when_no_json_in_content():
"""Content with no JSON blob -> _CurlResponse(status_code=0)."""
s = YandexRealtyScraper()
mock_browser = AsyncMock()
mock_browser.fetch = AsyncMock(return_value="<html><body>blocked</body></html>")
s._browser = mock_browser
resp = await s._http_get("https://realty.yandex.ru/test/")
assert resp.status_code == 0
# ---------------------------------------------------------------------------