fix(tradein): system-curl subprocess transport for Yandex SERP over SOCKS5
This commit is contained in:
parent
567549c3aa
commit
4f7426b28c
2 changed files with 195 additions and 10 deletions
|
|
@ -27,6 +27,7 @@ import json
|
|||
import logging
|
||||
import math
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from datetime import date # noqa: F401 (used in type hints / future helpers)
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
|
@ -125,6 +126,22 @@ _TOTAL_COUNT_PATHS: list[list[str]] = [
|
|||
# Regex для fallback-поиска количества офферов в тексте DOM.
|
||||
_RE_TOTAL_COUNT = re.compile(r"(\d[\d ]{0,8})\s+(?:объявлен|квартир|предложен)", re.IGNORECASE)
|
||||
|
||||
# ── Subprocess curl transport (SOCKS5 proxy) ──────────────────────────────────
|
||||
# curl_cffi collapses to <1 KB/s over the SOCKS5 mobile proxy; system curl with
|
||||
# --compressed fetches the full SERP in ~17s (proven on prod 2026-06).
|
||||
|
||||
_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:"
|
||||
|
||||
|
||||
@dataclass
|
||||
class _CurlResponse:
|
||||
status_code: int
|
||||
text: str
|
||||
|
||||
|
||||
def _load_cookies_from_file(path: str | None) -> dict[str, str]:
|
||||
"""Load Yandex cookies from JSON file path (Netscape/browser export format).
|
||||
|
|
@ -298,13 +315,94 @@ class YandexRealtyScraper(BaseScraper):
|
|||
await self._cffi_session.close()
|
||||
self._cffi_session = None
|
||||
|
||||
async def _http_get(self, url: str, **kwargs: object) -> object: # type: ignore[override]
|
||||
"""curl_cffi-based GET with Chrome120 impersonation.
|
||||
async def _curl_subprocess_get(self, url: str, proxy: str, timeout: int) -> _CurlResponse:
|
||||
"""Fetch via system `curl --compressed -x <socks5 proxy>`.
|
||||
|
||||
Returns curl_cffi Response (compatible API: .status_code, .text).
|
||||
Caller must check status_code; no automatic retry (BaseScraper.retry
|
||||
decorator is per-method and not inherited cleanly when overridden).
|
||||
curl_cffi collapses to <1 KB/s over this SOCKS5 mobile proxy (Yandex anti-bot
|
||||
interaction); system curl with gzip pulls the full SERP in ~17s. Cookies from
|
||||
self._cookies are sent via a Cookie header; redirects followed (-L).
|
||||
|
||||
The HTTP status code is appended to stdout via `-w` marker and stripped from
|
||||
the body. curl exit code 28 (timeout) forces status_code=0 — an incomplete body
|
||||
from a tarpitted connection is useless.
|
||||
"""
|
||||
args = [
|
||||
"curl",
|
||||
"-sS",
|
||||
"-L",
|
||||
"--compressed",
|
||||
"-x",
|
||||
proxy,
|
||||
"--max-time",
|
||||
str(timeout),
|
||||
"-A",
|
||||
_CURL_UA,
|
||||
"-H",
|
||||
"Accept-Language: ru-RU,ru;q=0.9,en;q=0.8",
|
||||
"-w",
|
||||
_CURL_STATUS_MARKER + "%{http_code}",
|
||||
]
|
||||
if self._cookies:
|
||||
cookie_hdr = "; ".join(f"{k}={v}" for k, v in self._cookies.items())
|
||||
args += ["-b", cookie_hdr]
|
||||
args.append(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]
|
||||
|
||||
# curl exit 28 = timeout: body may be partial — treat as failure.
|
||||
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 (all other cases).
|
||||
|
||||
When settings.yandex_proxy_url is a socks5(h):// URL, uses the subprocess
|
||||
curl transport — curl_cffi collapses to <1 KB/s over the SOCKS5 mobile
|
||||
proxy; system curl fetches the full SERP in ~17s (proven on prod 2026-06).
|
||||
|
||||
All other cases (http-proxy, no proxy, dev) keep the legacy curl_cffi path.
|
||||
Callers depend only on .status_code (int) and .text (str) — both paths provide
|
||||
a compatible interface (_CurlResponse shim vs curl_cffi Response).
|
||||
"""
|
||||
from app.core.config import settings
|
||||
|
||||
proxy = settings.yandex_proxy_url
|
||||
timeout = int(kwargs.get("timeout", 30) or 30)
|
||||
if proxy and proxy.startswith(("socks5://", "socks5h://")):
|
||||
return await self._curl_subprocess_get(url, proxy, timeout)
|
||||
|
||||
# Legacy curl_cffi path — dev / no-proxy / http-proxy.
|
||||
if self._cffi_session is None:
|
||||
raise RuntimeError("YandexRealtyScraper must be used as async context manager")
|
||||
kwargs.setdefault("timeout", 30)
|
||||
|
|
|
|||
|
|
@ -8,12 +8,14 @@ import pytest
|
|||
|
||||
from app.services.scrapers.base import ScrapedLot
|
||||
from app.services.scrapers.yandex_realty import (
|
||||
_CURL_STATUS_MARKER,
|
||||
DEFAULT_CITY,
|
||||
DEFAULT_PRICE_RANGES,
|
||||
MAX_PAGES,
|
||||
ROOM_PATH,
|
||||
YandexRealtyScraper,
|
||||
_combo_label,
|
||||
_CurlResponse,
|
||||
_extract_offer_areas_from_state,
|
||||
_is_captcha,
|
||||
_load_cookies_from_file,
|
||||
|
|
@ -549,10 +551,15 @@ async def test_aexit_closes_session():
|
|||
|
||||
@pytest.mark.asyncio
|
||||
async def test_http_get_raises_when_no_context():
|
||||
"""_http_get must raise RuntimeError when called outside async context manager."""
|
||||
"""_http_get must raise RuntimeError when called outside async context manager
|
||||
when no socks5 proxy is configured (legacy cffi path)."""
|
||||
s = YandexRealtyScraper()
|
||||
with pytest.raises(RuntimeError, match="async context manager"):
|
||||
await s._http_get("https://realty.yandex.ru/test/")
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.yandex_proxy_url = None # no proxy → cffi path → RuntimeError
|
||||
with patch("app.services.scrapers.yandex_realty.settings", mock_settings, create=True):
|
||||
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/")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
@ -568,11 +575,91 @@ async def test_fetch_around_returns_empty_on_captcha():
|
|||
s._cffi_session = AsyncMock()
|
||||
s._cffi_session.get = AsyncMock(return_value=mock_resp)
|
||||
|
||||
with patch.object(s, "sleep_between_requests", new_callable=AsyncMock):
|
||||
lots = await s.fetch_around(lat=0.0, lon=0.0)
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.yandex_proxy_url = None # no socks5 → cffi path
|
||||
with patch.dict("sys.modules", {"app.core.config": MagicMock(settings=mock_settings)}):
|
||||
with patch.object(s, "sleep_between_requests", new_callable=AsyncMock):
|
||||
lots = await s.fetch_around(lat=0.0, lon=0.0)
|
||||
assert lots == []
|
||||
|
||||
|
||||
# ── PART G: subprocess curl transport ────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_http_get_routes_socks5_to_curl_subprocess():
|
||||
"""_http_get with a socks5h proxy must call _curl_subprocess_get,
|
||||
NOT the curl_cffi session."""
|
||||
s = YandexRealtyScraper()
|
||||
sentinel = _CurlResponse(status_code=200, text="<html/>")
|
||||
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.yandex_proxy_url = "socks5h://x:y@h:1"
|
||||
|
||||
cffi_mock = AsyncMock()
|
||||
s._cffi_session = cffi_mock # session present — must NOT be called
|
||||
|
||||
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():
|
||||
"""_curl_subprocess_get must parse the -w status marker and strip it from body."""
|
||||
s = YandexRealtyScraper()
|
||||
s._cookies = {}
|
||||
|
||||
fake_stdout = b"<html>cards</html>" + _CURL_STATUS_MARKER.encode() + b"200"
|
||||
fake_stderr = b""
|
||||
|
||||
fake_proc = MagicMock()
|
||||
fake_proc.returncode = 0
|
||||
fake_proc.communicate = AsyncMock(return_value=(fake_stdout, fake_stderr))
|
||||
|
||||
with patch("asyncio.create_subprocess_exec", return_value=fake_proc) as mock_exec:
|
||||
resp = await s._curl_subprocess_get(
|
||||
"https://realty.yandex.ru/test/", "socks5h://x:y@h:1", 30
|
||||
)
|
||||
|
||||
mock_exec.assert_called_once()
|
||||
assert isinstance(resp, _CurlResponse)
|
||||
assert resp.status_code == 200
|
||||
assert resp.text == "<html>cards</html>"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_http_get_http_proxy_uses_cffi_not_curl():
|
||||
"""When proxy is http://, _http_get must NOT call _curl_subprocess_get;
|
||||
it falls through to the legacy curl_cffi path."""
|
||||
s = YandexRealtyScraper()
|
||||
sentinel_resp = MagicMock()
|
||||
sentinel_resp.status_code = 200
|
||||
sentinel_resp.text = "<html/>"
|
||||
|
||||
cffi_mock = AsyncMock()
|
||||
cffi_mock.get = AsyncMock(return_value=sentinel_resp)
|
||||
s._cffi_session = cffi_mock
|
||||
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.yandex_proxy_url = "http://proxy.example.com:8080"
|
||||
|
||||
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_fetch_around_multi_room_dedup():
|
||||
"""fetch_around_multi_room deduplicates offers seen in multiple combos."""
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue