feat(proxy-pool): P3 — kit curl-paths acquire/release from pool behind USE_PROXY_POOL_CURL, ship-dark (#2163)
All checks were successful
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-browser (push) Has been skipped
Deploy Trade-In / changes (push) Successful in 8s
Deploy Trade-In / test (push) Successful in 1m26s
Deploy Trade-In / build-backend (push) Successful in 1m18s
Deploy Trade-In / deploy (push) Successful in 1m22s
All checks were successful
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-browser (push) Has been skipped
Deploy Trade-In / changes (push) Successful in 8s
Deploy Trade-In / test (push) Successful in 1m26s
Deploy Trade-In / build-backend (push) Successful in 1m18s
Deploy Trade-In / deploy (push) Successful in 1m22s
This commit is contained in:
parent
ee47e331f3
commit
0937da4260
9 changed files with 539 additions and 93 deletions
|
|
@ -21,6 +21,7 @@ from typing import TYPE_CHECKING, Any
|
|||
|
||||
from app.core.config import settings as _settings
|
||||
from app.core.db import SessionLocal as _SessionLocal
|
||||
from app.services import proxy_pool as _proxy_pool
|
||||
from app.services.house_dedup_merge import run_house_dedup_merge as _run_house_dedup_merge
|
||||
from app.services.house_imv_backfill import backfill_house_imv as _backfill_house_imv
|
||||
from app.services.house_imv_backfill import (
|
||||
|
|
@ -40,6 +41,7 @@ from app.services.yandex_price_history import (
|
|||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from scraper_kit.contracts import ProxyLease
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.services.house_imv_backfill import HouseIMVBackfillResult
|
||||
|
|
@ -203,6 +205,59 @@ class RealScraperConfig:
|
|||
def cian_full_load_per_fetch_timeout_s(self) -> float:
|
||||
return _settings.cian_full_load_per_fetch_timeout_s
|
||||
|
||||
# ── Proxy-pool (#2163) ────────────────────────────────────────────────
|
||||
@property
|
||||
def use_proxy_pool_curl(self) -> bool:
|
||||
return _settings.use_proxy_pool_curl
|
||||
|
||||
|
||||
class RealProxyProvider:
|
||||
"""ProxyProvider-адаптер над `app.services.proxy_pool` (#2163).
|
||||
|
||||
Прячет `Session` от kit'а: открывает короткую сессию на каждую операцию
|
||||
(acquire/release/mark_health), проксирует в модульные функции proxy_pool и
|
||||
коммитит внутри них. Возвращает kit-`ProxyLease` (структурно совпадает с
|
||||
`proxy_pool.ProxyLease`), чтобы провайдеры не импортировали `app.*`.
|
||||
|
||||
Не-run вызовы (curl-путь эстиматора вне scrape_run) лизят под
|
||||
NON_RUN_LEASE_MARKER — reap_stale_leases освободит, если release не дошёл.
|
||||
"""
|
||||
|
||||
def acquire(self, provider: str) -> ProxyLease | None:
|
||||
from scraper_kit.contracts import ProxyLease as _KitProxyLease
|
||||
|
||||
db = _SessionLocal()
|
||||
try:
|
||||
lease = _proxy_pool.acquire(db, provider)
|
||||
finally:
|
||||
db.close()
|
||||
if lease is None:
|
||||
return None
|
||||
return _KitProxyLease(
|
||||
id=lease.id, url=lease.url, kind=lease.kind, rotate_url=lease.rotate_url
|
||||
)
|
||||
|
||||
def release(self, lease: ProxyLease) -> None:
|
||||
db = _SessionLocal()
|
||||
try:
|
||||
_proxy_pool.release(db, lease.id)
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
def mark_health(
|
||||
self,
|
||||
lease: ProxyLease,
|
||||
ok: bool,
|
||||
*,
|
||||
exit_ip: str | None = None,
|
||||
latency_ms: int | None = None,
|
||||
) -> None:
|
||||
db = _SessionLocal()
|
||||
try:
|
||||
_proxy_pool.mark_health(db, lease.id, ok, exit_ip=exit_ip, latency_ms=latency_ms)
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
class RealSessionFactory:
|
||||
"""SessionFactory-адаптер над `app.core.db.SessionLocal`."""
|
||||
|
|
@ -283,6 +338,7 @@ class RealEnrichmentJobs:
|
|||
__all__ = [
|
||||
"RealEnrichmentJobs",
|
||||
"RealMatcherAdapter",
|
||||
"RealProxyProvider",
|
||||
"RealScraperConfig",
|
||||
"RealSessionFactory",
|
||||
]
|
||||
|
|
|
|||
175
tradein-mvp/backend/tests/test_proxy_pool_curl_paths.py
Normal file
175
tradein-mvp/backend/tests/test_proxy_pool_curl_paths.py
Normal file
|
|
@ -0,0 +1,175 @@
|
|||
"""P3 (#2163): kit curl-пути берут прокси из пула за флагом USE_PROXY_POOL_CURL.
|
||||
|
||||
Покрывает инвариант ship-dark + fallback на уровне helper'а `curl_proxy_url` и
|
||||
class-based провайдера (YandexValuationScraper):
|
||||
- флаг off / proxy_provider=None → env-прокси, пул не трогается (golden-parity);
|
||||
- флаг on + пул пуст (acquire→None) → fallback env, не падаем;
|
||||
- флаг on + lease → fetch через lease.url, mark_health вызван, release в finally;
|
||||
- исключение внутри блока → mark_health(ok=False) + release всё равно (lease не течёт);
|
||||
- acquire кинул → fallback env (сбор не ломаем).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
|
||||
|
||||
from scraper_kit.contracts import ProxyLease
|
||||
from scraper_kit.providers._proxy import curl_proxy_url
|
||||
|
||||
# ── Фейки ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@dataclass
|
||||
class _FakeConfig:
|
||||
use_proxy_pool_curl: bool = False
|
||||
scraper_proxy_url: str | None = None
|
||||
|
||||
|
||||
class _SpyProvider:
|
||||
"""ProxyProvider-заглушка, записывающая вызовы."""
|
||||
|
||||
def __init__(self, lease: ProxyLease | None, *, acquire_raises: bool = False) -> None:
|
||||
self._lease = lease
|
||||
self._acquire_raises = acquire_raises
|
||||
self.acquire_calls: list[str] = []
|
||||
self.release_calls: list[int] = []
|
||||
self.mark_health_calls: list[tuple[int, bool]] = []
|
||||
|
||||
def acquire(self, provider: str) -> ProxyLease | None:
|
||||
self.acquire_calls.append(provider)
|
||||
if self._acquire_raises:
|
||||
raise RuntimeError("boom acquire")
|
||||
return self._lease
|
||||
|
||||
def release(self, lease: ProxyLease) -> None:
|
||||
self.release_calls.append(lease.id)
|
||||
|
||||
def mark_health(
|
||||
self, lease: ProxyLease, ok: bool, *, exit_ip: Any = None, latency_ms: Any = None
|
||||
) -> None:
|
||||
self.mark_health_calls.append((lease.id, ok))
|
||||
|
||||
|
||||
_LEASE = ProxyLease(id=7, url="http://user:pass@pool-proxy:3128", kind="http", rotate_url=None)
|
||||
|
||||
|
||||
# ── curl_proxy_url: helper-инвариант ──────────────────────────────────────────
|
||||
|
||||
|
||||
def test_flag_off_yields_env_and_never_touches_pool() -> None:
|
||||
cfg = _FakeConfig(use_proxy_pool_curl=False)
|
||||
spy = _SpyProvider(_LEASE)
|
||||
with curl_proxy_url(cfg, spy, "cian", env_fallback_url="http://env:3128") as url:
|
||||
assert url == "http://env:3128"
|
||||
# golden-parity: пул не задействован при выключенном флаге
|
||||
assert spy.acquire_calls == []
|
||||
assert spy.mark_health_calls == []
|
||||
assert spy.release_calls == []
|
||||
|
||||
|
||||
def test_provider_none_yields_env() -> None:
|
||||
cfg = _FakeConfig(use_proxy_pool_curl=True)
|
||||
with curl_proxy_url(cfg, None, "cian", env_fallback_url="http://env:3128") as url:
|
||||
assert url == "http://env:3128"
|
||||
|
||||
|
||||
def test_flag_on_empty_pool_falls_back_to_env() -> None:
|
||||
cfg = _FakeConfig(use_proxy_pool_curl=True)
|
||||
spy = _SpyProvider(None) # acquire → None (пул пуст/выключен)
|
||||
with curl_proxy_url(cfg, spy, "avito", env_fallback_url="http://env:3128") as url:
|
||||
assert url == "http://env:3128"
|
||||
assert spy.acquire_calls == ["avito"]
|
||||
# lease не выдан → mark_health/release не зовём
|
||||
assert spy.mark_health_calls == []
|
||||
assert spy.release_calls == []
|
||||
|
||||
|
||||
def test_flag_on_lease_used_and_released_on_success() -> None:
|
||||
cfg = _FakeConfig(use_proxy_pool_curl=True)
|
||||
spy = _SpyProvider(_LEASE)
|
||||
with curl_proxy_url(cfg, spy, "yandex", env_fallback_url="http://env:3128") as url:
|
||||
assert url == _LEASE.url # fetch идёт через lease.url, НЕ env
|
||||
assert spy.acquire_calls == ["yandex"]
|
||||
assert spy.mark_health_calls == [(7, True)]
|
||||
assert spy.release_calls == [7] # release в finally
|
||||
|
||||
|
||||
def test_flag_on_exception_marks_fail_and_still_releases() -> None:
|
||||
cfg = _FakeConfig(use_proxy_pool_curl=True)
|
||||
spy = _SpyProvider(_LEASE)
|
||||
with pytest.raises(RuntimeError, match="ban"):
|
||||
with curl_proxy_url(cfg, spy, "cian", env_fallback_url=None) as url:
|
||||
assert url == _LEASE.url
|
||||
raise RuntimeError("ban 403")
|
||||
# исключение → ok=False, но lease ОБЯЗАТЕЛЬНО освобождён (не течёт)
|
||||
assert spy.mark_health_calls == [(7, False)]
|
||||
assert spy.release_calls == [7]
|
||||
|
||||
|
||||
def test_acquire_raises_falls_back_to_env() -> None:
|
||||
cfg = _FakeConfig(use_proxy_pool_curl=True)
|
||||
spy = _SpyProvider(_LEASE, acquire_raises=True)
|
||||
with curl_proxy_url(cfg, spy, "cian", env_fallback_url="http://env:3128") as url:
|
||||
assert url == "http://env:3128" # ошибка acquire → env, не падаем
|
||||
assert spy.release_calls == []
|
||||
|
||||
|
||||
# ── YandexValuationScraper: lease держится на всё время сессии ─────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_yandex_scraper_flag_off_no_pool() -> None:
|
||||
from scraper_kit.providers.yandex.valuation import YandexValuationScraper
|
||||
|
||||
cfg = _FakeConfig(use_proxy_pool_curl=False, scraper_proxy_url="http://env:3128")
|
||||
spy = _SpyProvider(_LEASE)
|
||||
with patch("scraper_kit.providers.yandex.valuation._CurlCffiSession") as sess_cls:
|
||||
sess_cls.return_value = MagicMock(close=AsyncMock())
|
||||
async with YandexValuationScraper(cfg, proxy_provider=spy):
|
||||
pass
|
||||
# флаг off → env-прокси, пул не тронут
|
||||
_, kwargs = sess_cls.call_args
|
||||
assert kwargs["proxies"] == {"http": "http://env:3128", "https": "http://env:3128"}
|
||||
assert spy.acquire_calls == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_yandex_scraper_flag_on_lease_lifecycle() -> None:
|
||||
from scraper_kit.providers.yandex.valuation import YandexValuationScraper
|
||||
|
||||
cfg = _FakeConfig(use_proxy_pool_curl=True, scraper_proxy_url="http://env:3128")
|
||||
spy = _SpyProvider(_LEASE)
|
||||
with patch("scraper_kit.providers.yandex.valuation._CurlCffiSession") as sess_cls:
|
||||
sess_cls.return_value = MagicMock(close=AsyncMock())
|
||||
async with YandexValuationScraper(cfg, proxy_provider=spy):
|
||||
# сессия создана через lease.url, lease ещё удерживается
|
||||
_, kwargs = sess_cls.call_args
|
||||
assert kwargs["proxies"] == {"http": _LEASE.url, "https": _LEASE.url}
|
||||
assert spy.release_calls == []
|
||||
# выход из сессии → mark_health(ok) + release
|
||||
assert spy.acquire_calls == ["yandex"]
|
||||
assert spy.mark_health_calls == [(7, True)]
|
||||
assert spy.release_calls == [7]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_yandex_scraper_exception_releases_lease() -> None:
|
||||
from scraper_kit.providers.yandex.valuation import YandexValuationScraper
|
||||
|
||||
cfg = _FakeConfig(use_proxy_pool_curl=True, scraper_proxy_url=None)
|
||||
spy = _SpyProvider(_LEASE)
|
||||
with patch("scraper_kit.providers.yandex.valuation._CurlCffiSession") as sess_cls:
|
||||
sess_cls.return_value = MagicMock(close=AsyncMock())
|
||||
with pytest.raises(RuntimeError, match="scrape blew up"):
|
||||
async with YandexValuationScraper(cfg, proxy_provider=spy):
|
||||
raise RuntimeError("scrape blew up")
|
||||
# исключение в теле → ok=False + release (lease не течёт)
|
||||
assert spy.mark_health_calls == [(7, False)]
|
||||
assert spy.release_calls == [7]
|
||||
|
|
@ -15,6 +15,7 @@ from __future__ import annotations
|
|||
from scraper_kit.contracts import (
|
||||
EnrichmentJobs,
|
||||
HouseMatcher,
|
||||
ProxyProvider,
|
||||
ScraperConfig,
|
||||
SessionFactory,
|
||||
)
|
||||
|
|
@ -22,6 +23,7 @@ from scraper_kit.contracts import (
|
|||
from app.services.scraper_adapters import (
|
||||
RealEnrichmentJobs,
|
||||
RealMatcherAdapter,
|
||||
RealProxyProvider,
|
||||
RealScraperConfig,
|
||||
RealSessionFactory,
|
||||
)
|
||||
|
|
@ -43,6 +45,10 @@ def _check_enrichment(j: EnrichmentJobs) -> None:
|
|||
"""Статическая проверка: RealEnrichmentJobs ⊑ EnrichmentJobs."""
|
||||
|
||||
|
||||
def _check_proxy_provider(p: ProxyProvider) -> None:
|
||||
"""Статическая проверка: RealProxyProvider ⊑ ProxyProvider."""
|
||||
|
||||
|
||||
def test_matcher_adapter_satisfies_protocol() -> None:
|
||||
adapter = RealMatcherAdapter()
|
||||
_check_matcher(adapter)
|
||||
|
|
@ -60,6 +66,17 @@ def test_scraper_config_satisfies_protocol() -> None:
|
|||
assert isinstance(config.browser_http_endpoint, str)
|
||||
assert isinstance(config.avito_proxy_max_rotations, int)
|
||||
assert isinstance(config.avito_serp_ekb_only, bool)
|
||||
# #2163: новый флаг proxy-пула присутствует и bool.
|
||||
assert isinstance(config.use_proxy_pool_curl, bool)
|
||||
|
||||
|
||||
def test_proxy_provider_satisfies_protocol() -> None:
|
||||
provider = RealProxyProvider()
|
||||
_check_proxy_provider(provider)
|
||||
assert isinstance(provider, ProxyProvider)
|
||||
assert callable(provider.acquire)
|
||||
assert callable(provider.release)
|
||||
assert callable(provider.mark_health)
|
||||
|
||||
|
||||
def test_session_factory_satisfies_protocol() -> None:
|
||||
|
|
|
|||
|
|
@ -23,12 +23,29 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProxyLease:
|
||||
"""Арендованный из пула прокси (лёгкий транспорт между kit и backend).
|
||||
|
||||
Структурно совпадает с `app.services.proxy_pool.ProxyLease` (id/url/kind/rotate_url),
|
||||
но живёт в контрактах, чтобы kit не импортировал `app.*`. Провайдеры читают только
|
||||
`url` (готов для curl_cffi `proxies=`); `id`/`rotate_url` нужны backend-адаптеру для
|
||||
release / mark_health / (будущей) ротации.
|
||||
"""
|
||||
|
||||
id: int
|
||||
url: str
|
||||
kind: str
|
||||
rotate_url: str | None = None
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class HouseMatcher(Protocol):
|
||||
"""Кросс-источниковый матчинг домов/листингов (продуктовая логика matching-сервиса).
|
||||
|
|
@ -138,6 +155,48 @@ class ScraperConfig(Protocol):
|
|||
scraper_skip_seen_today: bool
|
||||
# Per-fetch watchdog-таймаут Cian full-load (секунды).
|
||||
cian_full_load_per_fetch_timeout_s: float
|
||||
# ── Proxy-pool (#2163) ────────────────────────────────────────────────────
|
||||
# Feature-флаг: брать прокси из пула (scrape_proxies) на curl_cffi-путях вместо
|
||||
# env-прокси (config.*_proxy_url). Ship-dark: дефолт False → curl-пути ходят как
|
||||
# сейчас. True + пустой пул → fallback на env (сбор не ломается). См. ProxyProvider.
|
||||
use_proxy_pool_curl: bool
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class ProxyProvider(Protocol):
|
||||
"""Источник прокси из пула для curl-путей (#2163), за флагом use_proxy_pool_curl.
|
||||
|
||||
Инкапсулирует lease/health-механику `app.services.proxy_pool` (которая требует
|
||||
`Session`), пряча БД от kit'а. Реализуется
|
||||
`app.services.scraper_adapters.RealProxyProvider` (открывает короткую сессию на
|
||||
операцию). Провайдеры вызывают `acquire` → используют `lease.url` → на выходе
|
||||
`mark_health` + `release` (в finally — lease не должен течь).
|
||||
|
||||
Контракт ship-dark/fallback: `acquire` возвращает None если свободных здоровых
|
||||
прокси нет — caller обязан откатиться на env-прокси, а НЕ падать.
|
||||
"""
|
||||
|
||||
def acquire(self, provider: str) -> ProxyLease | None:
|
||||
"""Взять прокси под провайдера (avito/cian/yandex/generic/any).
|
||||
|
||||
Returns ProxyLease или None если пул пуст/выключен → caller делает env-fallback.
|
||||
"""
|
||||
...
|
||||
|
||||
def release(self, lease: ProxyLease) -> None:
|
||||
"""Освободить lease (идемпотентно). Вызывать в finally — иначе прокси течёт."""
|
||||
...
|
||||
|
||||
def mark_health(
|
||||
self,
|
||||
lease: ProxyLease,
|
||||
ok: bool,
|
||||
*,
|
||||
exit_ip: str | None = ...,
|
||||
latency_ms: int | None = ...,
|
||||
) -> None:
|
||||
"""Записать исход использования прокси (ok=True успех, False бан/ошибка)."""
|
||||
...
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
|
|
@ -246,6 +305,8 @@ class EnrichmentJobs(Protocol):
|
|||
__all__ = [
|
||||
"EnrichmentJobs",
|
||||
"HouseMatcher",
|
||||
"ProxyLease",
|
||||
"ProxyProvider",
|
||||
"ScraperConfig",
|
||||
"SessionFactory",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -0,0 +1,83 @@
|
|||
"""Общий helper выбора прокси для curl_cffi-путей (#2163), за флагом USE_PROXY_POOL_CURL.
|
||||
|
||||
Инвариант ship-dark + fallback:
|
||||
- config.use_proxy_pool_curl=False (дефолт) ИЛИ proxy_provider=None → yield env-прокси
|
||||
(env_fallback_url) — curl-пути ходят ровно как сейчас, прод не меняется.
|
||||
- Флаг on + пул выдал lease → yield lease.url; на выходе mark_health(ok) + release(lease).
|
||||
ok=True если блок отработал без исключения, ok=False если внутри поднялось (бан/ошибка).
|
||||
- Флаг on + пул пуст/ошибка acquire → fallback на env_fallback_url, НЕ падаем (сбор цел).
|
||||
|
||||
release ВСЕГДА в finally — lease не должен течь, даже если fetch кинул. mark_health/release
|
||||
обёрнуты в best-effort try (проблема пула не должна ронять сбор).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections.abc import Iterator
|
||||
from contextlib import contextmanager
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from scraper_kit.contracts import ProxyProvider, ScraperConfig
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def curl_proxy_url(
|
||||
config: ScraperConfig | None,
|
||||
proxy_provider: ProxyProvider | None,
|
||||
provider: str,
|
||||
*,
|
||||
env_fallback_url: str | None,
|
||||
) -> Iterator[str | None]:
|
||||
"""Отдать effective proxy-url для curl_cffi (`proxies={http/https: url}`).
|
||||
|
||||
Args:
|
||||
config: ScraperConfig (читается флаг use_proxy_pool_curl). None → env-fallback.
|
||||
proxy_provider: пул прокси или None (off). None → env-fallback.
|
||||
provider: имя провайдера для affinity ("avito"/"cian"/"yandex"/...).
|
||||
env_fallback_url: прокси-url как сейчас (config.cian_proxy_url / scraper_proxy_url).
|
||||
|
||||
Yields effective url (может быть None = прямое подключение, как и раньше).
|
||||
"""
|
||||
use_pool = (
|
||||
config is not None
|
||||
and getattr(config, "use_proxy_pool_curl", False)
|
||||
and proxy_provider is not None
|
||||
)
|
||||
|
||||
lease = None
|
||||
if use_pool:
|
||||
assert proxy_provider is not None # для type-narrowing (use_pool это гарантирует)
|
||||
try:
|
||||
lease = proxy_provider.acquire(provider)
|
||||
except Exception:
|
||||
logger.warning(
|
||||
"proxy_pool: acquire(%s) failed — fallback to env proxy", provider, exc_info=True
|
||||
)
|
||||
lease = None
|
||||
|
||||
if lease is None:
|
||||
# off / пул пуст / ошибка acquire → env как сейчас (сбор не ломаем).
|
||||
yield env_fallback_url
|
||||
return
|
||||
|
||||
assert proxy_provider is not None
|
||||
ok = True
|
||||
try:
|
||||
yield lease.url
|
||||
except Exception:
|
||||
ok = False
|
||||
raise
|
||||
finally:
|
||||
# mark_health + release — best-effort: проблема пула не должна ронять сбор.
|
||||
try:
|
||||
proxy_provider.mark_health(lease, ok)
|
||||
except Exception:
|
||||
logger.warning("proxy_pool: mark_health failed for %s", provider, exc_info=True)
|
||||
try:
|
||||
proxy_provider.release(lease) # ОБЯЗАТЕЛЬНО — lease не течёт
|
||||
except Exception:
|
||||
logger.warning("proxy_pool: release failed for %s", provider, exc_info=True)
|
||||
|
|
@ -23,6 +23,7 @@ import hashlib
|
|||
import json
|
||||
import logging
|
||||
import re
|
||||
from contextlib import ExitStack
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import date
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
|
@ -32,6 +33,7 @@ from uuid import UUID
|
|||
from sqlalchemy import text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from scraper_kit.providers._proxy import curl_proxy_url
|
||||
from scraper_kit.providers.avito.shared import _unix_to_date
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
|
@ -39,7 +41,7 @@ if TYPE_CHECKING:
|
|||
# hard import cycle browser_fetcher → config → ... и лишней зависимости в
|
||||
# offline-тестах парсинга, которым browser не нужен.
|
||||
from scraper_kit.browser_fetcher import BrowserFetcher
|
||||
from scraper_kit.contracts import ScraperConfig
|
||||
from scraper_kit.contracts import ProxyProvider, ScraperConfig
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -437,6 +439,7 @@ async def evaluate_via_imv(
|
|||
cffi_session: Any | None = None,
|
||||
browser_fetcher: BrowserFetcher | None = None,
|
||||
config: ScraperConfig | None = None,
|
||||
proxy_provider: ProxyProvider | None = None,
|
||||
) -> IMVEvaluation:
|
||||
"""Avito IMV — 3 HTTP requests (current contract 2026-05+):
|
||||
|
||||
|
|
@ -458,55 +461,65 @@ async def evaluate_via_imv(
|
|||
# (#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:
|
||||
# Импортируем здесь чтобы избежать циклических зависимостей при тестах без сети
|
||||
# ExitStack держит прокси-lease пула (#2163) на всё время own-session: warm-up +
|
||||
# geocode + evaluate. Регистрируется ТОЛЬКО когда сами создаём curl_cffi-сессию;
|
||||
# для browser_fetcher / переданной cffi_session — no-op. Исключение из блока
|
||||
# (бан/ошибка) прокинется в curl_proxy_url.__exit__ → mark_health(ok=False) + release;
|
||||
# чистый выход → mark_health(ok=True) + release. Прокси-переключение не трогает
|
||||
# warm-up/cookie-последовательность (сессия создаётся один раз, до _warmup).
|
||||
with ExitStack() as _proxy_stack:
|
||||
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
|
||||
|
||||
_own_session = False
|
||||
if cffi_session is None:
|
||||
# Зеркалим production-набор из AvitoScraper.__aenter__ (avito.py:150-162):
|
||||
# chrome120 TLS + document-заголовки + timeout. Затем warm-up GET для
|
||||
# seed anti-bot cookies — bare-session XHR Avito банит на server-IP.
|
||||
# Прокси: пул за флагом use_proxy_pool_curl (#2163), иначе env
|
||||
# scraper_proxy_url. proxy=None → прямое подключение (dev).
|
||||
_env = config.scraper_proxy_url if config is not None else None
|
||||
_proxy_url = _proxy_stack.enter_context(
|
||||
curl_proxy_url(config, proxy_provider, "avito", env_fallback_url=_env)
|
||||
)
|
||||
_proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
|
||||
cffi_session = CffiAsyncSession(
|
||||
impersonate="chrome120",
|
||||
timeout=_HTTP_TIMEOUT_SEC,
|
||||
proxies=_proxies,
|
||||
headers=_DOC_HEADERS,
|
||||
)
|
||||
_own_session = True
|
||||
except ImportError as exc:
|
||||
raise RuntimeError(
|
||||
"curl_cffi не установлен. Добавь 'curl-cffi>=0.7.0' в pyproject.toml."
|
||||
) from exc
|
||||
|
||||
try:
|
||||
from curl_cffi.requests import AsyncSession as CffiAsyncSession
|
||||
|
||||
_own_session = False
|
||||
if cffi_session is None:
|
||||
# Зеркалим production-набор из AvitoScraper.__aenter__ (avito.py:150-162):
|
||||
# chrome120 TLS + document-заголовки + timeout. Затем warm-up GET для
|
||||
# seed anti-bot cookies — bare-session XHR Avito банит на server-IP.
|
||||
# Mobile proxy wiring (#806 follow-up): только для own-session (переданная
|
||||
# сессия уже с прокси от AvitoScraper). proxy=None → прямое подключение.
|
||||
_proxy_url = config.scraper_proxy_url if config is not None else None
|
||||
_proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
|
||||
cffi_session = CffiAsyncSession(
|
||||
impersonate="chrome120",
|
||||
timeout=_HTTP_TIMEOUT_SEC,
|
||||
proxies=_proxies,
|
||||
headers=_DOC_HEADERS,
|
||||
)
|
||||
_own_session = True
|
||||
except ImportError as exc:
|
||||
raise RuntimeError(
|
||||
"curl_cffi не установлен. Добавь 'curl-cffi>=0.7.0' в pyproject.toml."
|
||||
) from exc
|
||||
|
||||
try:
|
||||
if _own_session:
|
||||
await _warmup(cffi_session)
|
||||
geo = await _geocode(cffi_session, address)
|
||||
evaluation = await _imv_evaluate(
|
||||
cffi_session,
|
||||
geo=geo,
|
||||
address=address,
|
||||
rooms=rooms,
|
||||
area_m2=area_m2,
|
||||
floor=floor,
|
||||
floor_at_home=floor_at_home,
|
||||
house_type=house_type,
|
||||
renovation_type=renovation_type,
|
||||
has_balcony=has_balcony,
|
||||
has_loggia=has_loggia,
|
||||
)
|
||||
finally:
|
||||
if _own_session:
|
||||
await cffi_session.close()
|
||||
if _own_session:
|
||||
await _warmup(cffi_session)
|
||||
geo = await _geocode(cffi_session, address)
|
||||
evaluation = await _imv_evaluate(
|
||||
cffi_session,
|
||||
geo=geo,
|
||||
address=address,
|
||||
rooms=rooms,
|
||||
area_m2=area_m2,
|
||||
floor=floor,
|
||||
floor_at_home=floor_at_home,
|
||||
house_type=house_type,
|
||||
renovation_type=renovation_type,
|
||||
has_balcony=has_balcony,
|
||||
has_loggia=has_loggia,
|
||||
)
|
||||
finally:
|
||||
if _own_session:
|
||||
await cffi_session.close()
|
||||
|
||||
return evaluation
|
||||
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ from sqlalchemy import text
|
|||
from sqlalchemy.orm import Session
|
||||
|
||||
from scraper_kit.cian_state_parser import extract_all_states, extract_state
|
||||
from scraper_kit.providers._proxy import curl_proxy_url
|
||||
from scraper_kit.repair_state_normalizer import (
|
||||
infer_repair_state_from_text,
|
||||
normalize_repair_state,
|
||||
|
|
@ -30,7 +31,7 @@ from scraper_kit.snapshot_writer import upsert_listing_snapshot
|
|||
|
||||
if TYPE_CHECKING:
|
||||
from scraper_kit.browser_fetcher import BrowserFetcher
|
||||
from scraper_kit.contracts import ScraperConfig
|
||||
from scraper_kit.contracts import ProxyProvider, ScraperConfig
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -75,6 +76,7 @@ async def fetch_detail(
|
|||
config: ScraperConfig | None = None,
|
||||
session: AsyncSession | None = None,
|
||||
browser_fetcher: BrowserFetcher | None = None,
|
||||
proxy_provider: ProxyProvider | None = None,
|
||||
) -> DetailEnrichment | None:
|
||||
"""Fetch detail page and extract all containers.
|
||||
|
||||
|
|
@ -99,15 +101,22 @@ async def fetch_detail(
|
|||
except Exception as exc:
|
||||
logger.warning("Cian detail browser fetch failed %s: %s", offer_url, exc)
|
||||
return None
|
||||
elif session is not None:
|
||||
# Shared curl_cffi-сессия (прокси уже применён caller'ом) — пул не трогаем.
|
||||
resp = await session.get(offer_url, allow_redirects=True)
|
||||
if resp.status_code != 200:
|
||||
logger.warning("Cian detail fetch %s → HTTP %d", offer_url, resp.status_code)
|
||||
return None
|
||||
html = resp.text
|
||||
else:
|
||||
# curl_cffi path (legacy, back-compat).
|
||||
close_session = False
|
||||
if session is None:
|
||||
# proxies: mobile-proxy egress (#806) — без прокси datacenter-IP блокируется Cian.
|
||||
# Пусто (config не передан / env не задан) → прямое подключение (dev/no-op).
|
||||
_proxy_url = config.cian_proxy_url if config is not None else None
|
||||
# curl_cffi own-session path (legacy, back-compat).
|
||||
# proxies: mobile-proxy egress (#806) — без прокси datacenter-IP блокируется Cian.
|
||||
# Прокси: пул за флагом use_proxy_pool_curl (#2163), иначе env cian_proxy_url.
|
||||
# Пусто → прямое подключение (dev/no-op). curl_proxy_url: mark_health + release на выходе.
|
||||
_env = config.cian_proxy_url if config is not None else None
|
||||
with curl_proxy_url(config, proxy_provider, "cian", env_fallback_url=_env) as _proxy_url:
|
||||
_proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
|
||||
session = AsyncSession(
|
||||
own_session = AsyncSession(
|
||||
impersonate="chrome120",
|
||||
timeout=25.0,
|
||||
proxies=_proxies,
|
||||
|
|
@ -116,17 +125,14 @@ async def fetch_detail(
|
|||
"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
|
||||
},
|
||||
)
|
||||
close_session = True
|
||||
|
||||
try:
|
||||
resp = await session.get(offer_url, allow_redirects=True)
|
||||
if resp.status_code != 200:
|
||||
logger.warning("Cian detail fetch %s → HTTP %d", offer_url, resp.status_code)
|
||||
return None
|
||||
html = resp.text
|
||||
finally:
|
||||
if close_session:
|
||||
await session.close()
|
||||
try:
|
||||
resp = await own_session.get(offer_url, allow_redirects=True)
|
||||
if resp.status_code != 200:
|
||||
logger.warning("Cian detail fetch %s → HTTP %d", offer_url, resp.status_code)
|
||||
return None
|
||||
html = resp.text
|
||||
finally:
|
||||
await own_session.close()
|
||||
|
||||
# ── Shared parse path (identical regardless of HTML source) ──────────────
|
||||
|
||||
|
|
|
|||
|
|
@ -32,10 +32,11 @@ from sqlalchemy import text
|
|||
from sqlalchemy.orm import Session
|
||||
|
||||
from scraper_kit.cian_state_parser import extract_state
|
||||
from scraper_kit.providers._proxy import curl_proxy_url
|
||||
from scraper_kit.providers.cian.session import load_session, mark_session_invalid
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from scraper_kit.contracts import ScraperConfig
|
||||
from scraper_kit.contracts import ProxyProvider, ScraperConfig
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -142,6 +143,7 @@ async def estimate_via_cian_valuation(
|
|||
use_cache: bool = True,
|
||||
house_id: int | None = None,
|
||||
listing_id: int | None = None,
|
||||
proxy_provider: ProxyProvider | None = None,
|
||||
) -> CianValuationResult | None:
|
||||
"""Estimate value via Cian's authenticated Valuation Calculator.
|
||||
|
||||
|
|
@ -179,25 +181,30 @@ async def estimate_via_cian_valuation(
|
|||
|
||||
# 4. Fetch с TLS fingerprint (curl_cffi, impersonate chrome120)
|
||||
# Mobile proxy wiring (#806 follow-up): Cian валюация — такой же datacenter-бан риск
|
||||
# как SERP. Используем cian_proxy_url. proxy=None → прямое подключение (dev).
|
||||
_proxy_url = config.cian_proxy_url
|
||||
_proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
|
||||
try:
|
||||
async with AsyncSession(
|
||||
impersonate="chrome120",
|
||||
cookies=cookies,
|
||||
timeout=25.0,
|
||||
proxies=_proxies,
|
||||
headers={"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"},
|
||||
) as session:
|
||||
resp = await session.get(url, allow_redirects=True)
|
||||
if resp.status_code != 200:
|
||||
logger.warning("Cian valuation fetch → HTTP %d for %s", resp.status_code, url)
|
||||
return None
|
||||
html = resp.text
|
||||
except Exception as exc:
|
||||
logger.warning("Cian valuation fetch failed: %s", exc)
|
||||
return None
|
||||
# как SERP. Прокси: пул за флагом use_proxy_pool_curl (#2163), иначе env cian_proxy_url.
|
||||
# proxy=None → прямое подключение (dev). curl_proxy_url на выходе mark_health + release.
|
||||
with curl_proxy_url(
|
||||
config, proxy_provider, "cian", env_fallback_url=config.cian_proxy_url
|
||||
) as _proxy_url:
|
||||
_proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
|
||||
try:
|
||||
async with AsyncSession(
|
||||
impersonate="chrome120",
|
||||
cookies=cookies,
|
||||
timeout=25.0,
|
||||
proxies=_proxies,
|
||||
headers={
|
||||
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
|
||||
},
|
||||
) as session:
|
||||
resp = await session.get(url, allow_redirects=True)
|
||||
if resp.status_code != 200:
|
||||
logger.warning("Cian valuation fetch → HTTP %d for %s", resp.status_code, url)
|
||||
return None
|
||||
html = resp.text
|
||||
except Exception as exc:
|
||||
logger.warning("Cian valuation fetch failed: %s", exc)
|
||||
return None
|
||||
|
||||
# 5. Parse state из _cianConfig['valuation-for-agent-frontend']
|
||||
state = extract_state(html, mfe=VALUATION_MFE, key="initialState")
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ from __future__ import annotations
|
|||
import logging
|
||||
import re
|
||||
from collections.abc import Callable
|
||||
from contextlib import ExitStack
|
||||
from datetime import date
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from urllib.parse import urlencode
|
||||
|
|
@ -34,6 +35,7 @@ from pydantic import BaseModel, Field
|
|||
from selectolax.parser import HTMLParser
|
||||
|
||||
from scraper_kit.base import BaseScraper
|
||||
from scraper_kit.providers._proxy import curl_proxy_url
|
||||
from scraper_kit.yandex_helpers import (
|
||||
parse_dmy,
|
||||
parse_house_type,
|
||||
|
|
@ -41,7 +43,7 @@ from scraper_kit.yandex_helpers import (
|
|||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from scraper_kit.contracts import ScraperConfig
|
||||
from scraper_kit.contracts import ProxyProvider, ScraperConfig
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -170,6 +172,7 @@ class YandexValuationScraper(BaseScraper):
|
|||
config: ScraperConfig,
|
||||
*,
|
||||
delay_provider: Callable[[str], float] | None = None,
|
||||
proxy_provider: ProxyProvider | None = None,
|
||||
) -> None:
|
||||
super().__init__()
|
||||
# Strangler-инжекция (#2133): конфиг (прокси-egress) и провайдер задержки
|
||||
|
|
@ -179,22 +182,47 @@ class YandexValuationScraper(BaseScraper):
|
|||
if delay_provider is not None:
|
||||
self.request_delay_sec = delay_provider(self.name)
|
||||
self._cffi_session: _CurlCffiSession | None = None
|
||||
# Прокси-пул (#2163): lease держится на всё время сессии (__aenter__→__aexit__).
|
||||
self._proxy_provider = proxy_provider
|
||||
self._proxy_stack: ExitStack | None = None
|
||||
|
||||
async def __aenter__(self) -> YandexValuationScraper: # type: ignore[override]
|
||||
# Override: Yandex valuation endpoint gates SSR data on Chrome TLS
|
||||
# fingerprint. Plain httpx returns shell HTML (CSR-only).
|
||||
# Sibling: yandex_realty.py / scripts/local-sweep-ekb-yandex.py
|
||||
# Mobile proxy wiring (#806 follow-up): route via mobile proxy to avoid
|
||||
# datacenter-IP blocks. proxy=None → прямое подключение (dev).
|
||||
_proxy_url = self._config.scraper_proxy_url
|
||||
# Прокси: пул за флагом use_proxy_pool_curl (#2163), иначе env scraper_proxy_url.
|
||||
# proxy=None → прямое подключение (dev). lease освобождается в __aexit__.
|
||||
self._proxy_stack = ExitStack()
|
||||
_proxy_url = self._proxy_stack.enter_context(
|
||||
curl_proxy_url(
|
||||
self._config,
|
||||
self._proxy_provider,
|
||||
"yandex",
|
||||
env_fallback_url=self._config.scraper_proxy_url,
|
||||
)
|
||||
)
|
||||
_proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
|
||||
self._cffi_session = _CurlCffiSession(impersonate="chrome120", proxies=_proxies)
|
||||
try:
|
||||
self._cffi_session = _CurlCffiSession(impersonate="chrome120", proxies=_proxies)
|
||||
except Exception:
|
||||
# session-создание упало → не течём lease'ом
|
||||
self._proxy_stack.close()
|
||||
self._proxy_stack = None
|
||||
raise
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *args: object) -> None: # type: ignore[override]
|
||||
if self._cffi_session is not None:
|
||||
await self._cffi_session.close()
|
||||
self._cffi_session = None
|
||||
# Прокинуть инфо об исключении в curl_proxy_url.__exit__ → mark_health(ok)+release.
|
||||
if self._proxy_stack is not None:
|
||||
exc = args[1] if len(args) > 1 and isinstance(args[1], BaseException) else None
|
||||
if exc is not None:
|
||||
self._proxy_stack.__exit__(type(exc), exc, exc.__traceback__)
|
||||
else:
|
||||
self._proxy_stack.close()
|
||||
self._proxy_stack = None
|
||||
|
||||
async def _http_get(self, url: str, **kwargs: object) -> object: # type: ignore[override]
|
||||
"""curl_cffi-based GET with Chrome120 impersonation.
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue