From bfe56e44d6631cc9d259b5978b9f9754c0487222 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sat, 23 May 2026 18:25:40 +0300 Subject: [PATCH] feat(tradein): global scraper delay setting (applies across all scrapers) - SQL migrations 053 (scraper_settings table) + 054 (seed global + per-source rows) - scraper_settings.py: get_scraper_delay() returns max(per_source, global); in-memory cache TTL=60s; invalidate_cache() for immediate effect on PUT - avito/cian/n1 scrapers load delay from DB in __init__ (mirrors yandex pattern) - Admin API: GET /scraper-settings (list all), PUT /scraper-settings/{source} with cache invalidation on update; CAST(:d AS numeric) per psycopg v3 rules - 10 unit tests: global>per_source, per_source>global, global=0, DB error fallback, cache invalidation, API endpoint smoke --- tradein-mvp/backend/app/api/v1/admin.py | 69 ++++++ .../backend/app/services/scraper_settings.py | 130 ++++++++++++ .../backend/app/services/scrapers/avito.py | 5 +- .../backend/app/services/scrapers/cian.py | 3 + .../backend/app/services/scrapers/n1.py | 13 +- .../backend/data/sql/053_scraper_settings.sql | 19 ++ .../data/sql/054_scraper_settings_global.sql | 25 +++ .../backend/tests/test_scraper_settings.py | 196 ++++++++++++++++++ 8 files changed, 455 insertions(+), 5 deletions(-) create mode 100644 tradein-mvp/backend/app/services/scraper_settings.py create mode 100644 tradein-mvp/backend/data/sql/053_scraper_settings.sql create mode 100644 tradein-mvp/backend/data/sql/054_scraper_settings_global.sql create mode 100644 tradein-mvp/backend/tests/test_scraper_settings.py diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index 9fa51729..3f767eb1 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/tradein-mvp/backend/app/api/v1/admin.py @@ -22,6 +22,7 @@ from app.services import cian_session as cian_session_svc from app.services import scrape_runs as runs_mod from app.services.geocoder import geocode from app.services.scrape_pipeline import run_avito_city_sweep +from app.services.scraper_settings import invalidate_cache from app.services.scrapers.avito import AvitoScraper from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment from app.services.scrapers.avito_houses import fetch_house_catalog, save_house_catalog_enrichment @@ -545,6 +546,74 @@ def cancel_avito_city_sweep( return {"ok": True, "run_id": run_id, "cancelled": cancelled} +# ── Scraper settings: global + per-source delay management ─────────────────── + + +class ScraperSettingPayload(BaseModel): + source: str = Field(..., min_length=1, max_length=64) + request_delay_sec: float = Field(..., ge=0.0, le=60.0) + description: str | None = None + + +@router.get("/scraper-settings", status_code=200) +def list_scraper_settings( + db: Annotated[Session, Depends(get_db)], +) -> dict: + """Список всех настроек задержки (per-source + global).""" + rows = db.execute( + text(""" + SELECT source, request_delay_sec, description, updated_at + FROM scraper_settings + ORDER BY source ASC + """) + ).mappings().all() + return {"settings": [dict(r) for r in rows]} + + +@router.put("/scraper-settings/{source}", status_code=200) +def update_scraper_setting( + source: str, + payload: ScraperSettingPayload, + db: Annotated[Session, Depends(get_db)], +) -> dict: + """Обновить задержку для source (или 'global'). Path source используется как ключ. + + Эффект применяется немедленно — invalidate_cache() сбрасывает кеш для source, + следующий get_scraper_delay() перечитает из БД. + + source='global' — нижняя планка для всех парсеров (0 = отключено). + """ + if payload.source != source: + raise HTTPException( + status_code=400, + detail=f"Path source={source!r} does not match body source={payload.source!r}", + ) + + row = db.execute( + text(""" + INSERT INTO scraper_settings (source, request_delay_sec, description, updated_at) + VALUES (:s, CAST(:d AS numeric), :desc, NOW()) + ON CONFLICT (source) DO UPDATE SET + request_delay_sec = EXCLUDED.request_delay_sec, + description = COALESCE(EXCLUDED.description, scraper_settings.description), + updated_at = NOW() + RETURNING source, request_delay_sec, description, updated_at + """), + { + "s": source, + "d": payload.request_delay_sec, + "desc": payload.description, + }, + ).mappings().one() + db.commit() + + invalidate_cache(source) + logger.info( + "scraper-settings: updated source=%s delay=%.1f", source, payload.request_delay_sec + ) + return dict(row) + + # ── In-app scheduler endpoints (Stage 4e) ──────────────────────────────────── diff --git a/tradein-mvp/backend/app/services/scraper_settings.py b/tradein-mvp/backend/app/services/scraper_settings.py new file mode 100644 index 00000000..6f1bbf8a --- /dev/null +++ b/tradein-mvp/backend/app/services/scraper_settings.py @@ -0,0 +1,130 @@ +"""scraper_settings.py — загрузка задержек парсеров из таблицы scraper_settings. + +Кеш в памяти с TTL 60 секунд — не нагружает БД при каждом запросе. +invalidate_cache() вызывается из admin API PUT для немедленного применения. + +Ключевая логика: + get_scraper_delay(source) = max(per_source_delay, global_delay) + Строка source='global' задаёт нижнюю планку для ВСЕХ парсеров. + Если global=0 — используется только per-source значение. +""" + +from __future__ import annotations + +import logging +import threading +import time +from collections.abc import Generator +from contextlib import contextmanager + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.core.db import SessionLocal + +logger = logging.getLogger(__name__) + +# Специальный ключ глобальной задержки (строка в scraper_settings с этим source). +_GLOBAL_KEY = "global" + +# TTL кеша в секундах — после истечения перечитывается из БД. +_CACHE_TTL_SEC = 60.0 + +# Дефолтные задержки если строка в БД отсутствует. +_DEFAULT_DELAY_BY_SOURCE: dict[str, float] = { + "avito": 7.0, + "cian": 5.0, + "n1": 5.0, + "yandex": 5.0, + "domrf": 5.0, + "rosreestr": 5.0, +} + +# Fallback для неизвестного источника. +_GLOBAL_DEFAULT_DELAY = 5.0 + +# Алиасы источников: ключ → canonical source (можно расширять без изменения вызывающего кода). +_KEY_ALIASES: dict[str, str] = {} + +# Кеш: source → (delay_value, timestamp). +_CACHE: dict[str, tuple[float, float]] = {} +_CACHE_LOCK = threading.Lock() + + +@contextmanager +def _open_session() -> Generator[Session, None, None]: + """Открыть сессию БД с гарантированным закрытием.""" + db = SessionLocal() + try: + yield db + finally: + db.close() + + +def _get_setting_cached(key: str) -> float: + """Кеш-aware чтение одного source из scraper_settings. + + Возвращает request_delay_sec из БД (с кешем TTL=60s). + При отсутствии строки — фолбек на _DEFAULT_DELAY_BY_SOURCE или 0.0 для global. + """ + now = time.time() + with _CACHE_LOCK: + cached = _CACHE.get(key) + if cached is not None and now - cached[1] < _CACHE_TTL_SEC: + return cached[0] + + try: + with _open_session() as db: + row = db.execute( + text("SELECT request_delay_sec FROM scraper_settings WHERE source = :s"), + {"s": key}, + ).first() + if row is not None: + value = float(row[0]) + elif key == _GLOBAL_KEY: + # Глобальная строка ещё не создана — не применяем floor. + value = 0.0 + else: + value = _DEFAULT_DELAY_BY_SOURCE.get(key, _GLOBAL_DEFAULT_DELAY) + except Exception as exc: + logger.warning( + "scraper_settings: load failed for %s — using default: %s", key, exc + ) + value = 0.0 if key == _GLOBAL_KEY else _DEFAULT_DELAY_BY_SOURCE.get( + key, _GLOBAL_DEFAULT_DELAY + ) + + with _CACHE_LOCK: + _CACHE[key] = (value, now) + return value + + +def get_scraper_delay(source: str) -> float: + """Вернуть эффективную задержку для source = max(per_source, global). + + Строка source='global' задаёт нижнюю планку для всех парсеров. + Если global=0 — используется только per-source значение. + """ + key = _KEY_ALIASES.get(source, source) + per_source = _get_setting_cached(key) + global_value = _get_setting_cached(_GLOBAL_KEY) + return max(per_source, global_value) + + +def invalidate_cache(source: str | None = None) -> None: + """Сбросить кеш для source (или весь кеш если source=None). + + Вызывается из admin API PUT /scraper-settings/{source} для немедленного + применения нового значения без ожидания TTL. + """ + with _CACHE_LOCK: + if source is None: + _CACHE.clear() + logger.info("scraper_settings: cache cleared (all sources)") + else: + _CACHE.pop(source, None) + # Если source — алиас, сбрасываем canonical key тоже. + canonical = _KEY_ALIASES.get(source) + if canonical: + _CACHE.pop(canonical, None) + logger.info("scraper_settings: cache cleared for source=%s", source) diff --git a/tradein-mvp/backend/app/services/scrapers/avito.py b/tradein-mvp/backend/app/services/scrapers/avito.py index f6571b23..6530ab22 100644 --- a/tradein-mvp/backend/app/services/scrapers/avito.py +++ b/tradein-mvp/backend/app/services/scrapers/avito.py @@ -23,6 +23,7 @@ from urllib.parse import urlencode, urljoin from curl_cffi.requests import AsyncSession from selectolax.parser import HTMLParser +from app.services.scraper_settings import get_scraper_delay from app.services.scrapers.base import BaseScraper, ScrapedLot logger = logging.getLogger(__name__) @@ -36,11 +37,13 @@ class AvitoScraper(BaseScraper): name = "avito" base_url = "https://www.avito.ru" - # Avito жёстко мониторит — спим долго между запросами + # Avito жёстко мониторит — спим долго между запросами. + # Класс-дефолт; реальное значение загружается из scraper_settings при создании экземпляра. request_delay_sec = 7.0 def __init__(self) -> None: super().__init__() + self.request_delay_sec = get_scraper_delay(self.name) self._cffi: AsyncSession | None = None async def __aenter__(self) -> AvitoScraper: diff --git a/tradein-mvp/backend/app/services/scrapers/cian.py b/tradein-mvp/backend/app/services/scrapers/cian.py index 8ad29df3..3dbd4a38 100644 --- a/tradein-mvp/backend/app/services/scrapers/cian.py +++ b/tradein-mvp/backend/app/services/scrapers/cian.py @@ -23,6 +23,7 @@ from urllib.parse import urlencode from curl_cffi.requests import AsyncSession +from app.services.scraper_settings import get_scraper_delay from app.services.scrapers.base import BaseScraper, ScrapedLot from app.services.scrapers.cian_state_parser import extract_state @@ -46,10 +47,12 @@ class CianScraper(BaseScraper): name = "cian" # ekb.cian.ru — city-specific subdomain для ЕКБ (per Schema sec 13, closed Q4) base_url = "https://ekb.cian.ru" + # Класс-дефолт; реальное значение загружается из scraper_settings при создании экземпляра. request_delay_sec = 5.0 # консервативно: Cian менее агрессивен чем Avito, но 5s безопасно def __init__(self) -> None: super().__init__() + self.request_delay_sec = get_scraper_delay(self.name) self._cffi: AsyncSession | None = None async def __aenter__(self) -> CianScraper: diff --git a/tradein-mvp/backend/app/services/scrapers/n1.py b/tradein-mvp/backend/app/services/scrapers/n1.py index bcf5b153..907756f6 100644 --- a/tradein-mvp/backend/app/services/scrapers/n1.py +++ b/tradein-mvp/backend/app/services/scrapers/n1.py @@ -25,6 +25,7 @@ from typing import Any from curl_cffi.requests import AsyncSession from selectolax.parser import HTMLParser +from app.services.scraper_settings import get_scraper_delay from app.services.scrapers.base import BaseScraper, ScrapedLot logger = logging.getLogger(__name__) @@ -35,13 +36,15 @@ class N1Scraper(BaseScraper): name = "n1" base_url = "https://ekaterinburg.n1.ru" + # Класс-дефолт; реальное значение загружается из scraper_settings при создании экземпляра. request_delay_sec = 5.0 def __init__(self) -> None: super().__init__() + self.request_delay_sec = get_scraper_delay(self.name) self._cffi: AsyncSession | None = None - async def __aenter__(self) -> "N1Scraper": + async def __aenter__(self) -> N1Scraper: await super().__aenter__() self._cffi = AsyncSession( impersonate="chrome120", @@ -74,7 +77,7 @@ class N1Scraper(BaseScraper): try: assert self._cffi is not None response = await self._cffi.get(url) - except Exception: # noqa: BLE001 + except Exception: logger.exception("n1 fetch failed for %s", url) return [] if response.status_code != 200: @@ -153,7 +156,9 @@ class N1Scraper(BaseScraper): title = link.text(strip=True) or "" rooms = _extract_rooms(title) # Адрес = title без префикса комнат - address = re.sub(r"^\s*(\d+-к|студи[яи])\s*,?\s*", "", title, flags=re.IGNORECASE).strip() + address = re.sub( + r"^\s*(\d+-к|студи[яи])\s*,?\s*", "", title, flags=re.IGNORECASE + ).strip() if not address: address = "Екатеринбург (N1)" @@ -192,7 +197,7 @@ class N1Scraper(BaseScraper): photo_urls=photos, raw_payload={"title": title}, ) - except Exception: # noqa: BLE001 + except Exception: logger.exception("n1 card parse failed") return None diff --git a/tradein-mvp/backend/data/sql/053_scraper_settings.sql b/tradein-mvp/backend/data/sql/053_scraper_settings.sql new file mode 100644 index 00000000..71fd4bef --- /dev/null +++ b/tradein-mvp/backend/data/sql/053_scraper_settings.sql @@ -0,0 +1,19 @@ +-- 053_scraper_settings.sql +-- Таблица настроек парсеров: задержки между запросами (anti-ban), extensible. +-- Одна строка на источник (source); UNIQUE(source). +BEGIN; + +CREATE TABLE IF NOT EXISTS scraper_settings ( + id SERIAL PRIMARY KEY, + source TEXT NOT NULL, + request_delay_sec NUMERIC(6,2) NOT NULL DEFAULT 5.0, + description TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT scraper_settings_source_uq UNIQUE (source) +); + +COMMENT ON TABLE scraper_settings IS + 'Per-source scraper delay settings. Special source=''global'' acts as floor across all scrapers.'; + +COMMIT; diff --git a/tradein-mvp/backend/data/sql/054_scraper_settings_global.sql b/tradein-mvp/backend/data/sql/054_scraper_settings_global.sql new file mode 100644 index 00000000..da09aefa --- /dev/null +++ b/tradein-mvp/backend/data/sql/054_scraper_settings_global.sql @@ -0,0 +1,25 @@ +-- 054_scraper_settings_global.sql +-- Add global delay row + per-source aliases for all scrapers. +-- Idempotent — ON CONFLICT DO NOTHING preserves user-edited values. +BEGIN; + +-- 'global' row: when > 0, acts as floor (max with per-source delay). +-- Default 0 = use per-source defaults. +INSERT INTO scraper_settings (source, request_delay_sec, description) +VALUES ( + 'global', + 0.0, + 'Global delay applied across ALL scrapers (max with per-source). 0 = use per-source defaults.' +) +ON CONFLICT (source) DO NOTHING; + +-- Per-source rows for all scrapers (individual control). +INSERT INTO scraper_settings (source, request_delay_sec, description) VALUES + ('avito', 7.0, 'Avito SERP delay (anti-ban)'), + ('cian', 5.0, 'Cian SERP delay'), + ('n1', 5.0, 'N1 SERP delay'), + ('domrf', 5.0, 'DOM.RF SERP delay'), + ('rosreestr', 5.0, 'Rosreestr API delay') +ON CONFLICT (source) DO NOTHING; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_scraper_settings.py b/tradein-mvp/backend/tests/test_scraper_settings.py new file mode 100644 index 00000000..e754e2ad --- /dev/null +++ b/tradein-mvp/backend/tests/test_scraper_settings.py @@ -0,0 +1,196 @@ +"""Offline tests for scraper_settings service (no DB required). + +Тестируем логику max(per_source, global) через монкей-патч _get_setting_cached. +""" + +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from unittest.mock import MagicMock, patch + +import pytest + +import app.services.scraper_settings as ss_mod +from app.services.scraper_settings import get_scraper_delay, invalidate_cache + + +# ── helpers ────────────────────────────────────────────────────────────────── + + +def _make_mock_session(values: dict[str, float | None]) -> MagicMock: + """Создать mock сессии БД с фиксированными значениями per source key.""" + + mock_db = MagicMock() + + def fake_execute(sql, params): + key = params["s"] + result = MagicMock() + val = values.get(key) + result.first.return_value = (val,) if val is not None else None + return result + + mock_db.execute = fake_execute + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + return mock_db + + +# ── tests ───────────────────────────────────────────────────────────────────── + + +def test_global_delay_takes_precedence_when_higher(monkeypatch): + """global=10 > avito=7 → возвращает 10.""" + invalidate_cache() + + mock_db = _make_mock_session({"avito": 7.0, "global": 10.0}) + monkeypatch.setattr(ss_mod, "_open_session", lambda: mock_db) + + delay = get_scraper_delay("avito") + assert delay == 10.0 + + +def test_per_source_takes_precedence_when_higher(monkeypatch): + """avito=7 > global=2 → возвращает 7.""" + invalidate_cache() + + mock_db = _make_mock_session({"avito": 7.0, "global": 2.0}) + monkeypatch.setattr(ss_mod, "_open_session", lambda: mock_db) + + delay = get_scraper_delay("avito") + assert delay == 7.0 + + +def test_global_zero_means_no_floor(monkeypatch): + """global=0 → используется только per-source значение.""" + invalidate_cache() + + mock_db = _make_mock_session({"cian": 5.0, "global": 0.0}) + monkeypatch.setattr(ss_mod, "_open_session", lambda: mock_db) + + delay = get_scraper_delay("cian") + assert delay == 5.0 + + +def test_missing_per_source_row_uses_class_default(monkeypatch): + """Если строки source в БД нет — фолбек на _DEFAULT_DELAY_BY_SOURCE.""" + invalidate_cache() + + # n1 строки нет в БД, global=0 + mock_db = _make_mock_session({"n1": None, "global": 0.0}) + monkeypatch.setattr(ss_mod, "_open_session", lambda: mock_db) + + delay = get_scraper_delay("n1") + # Дефолт для n1 = 5.0 + assert delay == ss_mod._DEFAULT_DELAY_BY_SOURCE["n1"] + + +def test_unknown_source_uses_global_default(monkeypatch): + """Неизвестный source без строки в БД — _GLOBAL_DEFAULT_DELAY.""" + invalidate_cache() + + mock_db = _make_mock_session({"some_new_source": None, "global": 0.0}) + monkeypatch.setattr(ss_mod, "_open_session", lambda: mock_db) + + delay = get_scraper_delay("some_new_source") + assert delay == ss_mod._GLOBAL_DEFAULT_DELAY + + +def test_db_error_falls_back_to_default(monkeypatch): + """Ошибка БД → graceful fallback на дефолт, не crash.""" + invalidate_cache() + + def broken_open(): + raise RuntimeError("DB is down") + + monkeypatch.setattr(ss_mod, "_open_session", broken_open) + + # Не должно кидать исключение + delay = get_scraper_delay("avito") + # При ошибке: per_source → _DEFAULT_DELAY_BY_SOURCE['avito'] = 7.0, global → 0.0 + assert delay == 7.0 + + +def test_invalidate_cache_single_source(monkeypatch): + """invalidate_cache(source) удаляет только указанный source.""" + invalidate_cache() + + mock_db = _make_mock_session({"cian": 5.0, "avito": 7.0, "global": 0.0}) + monkeypatch.setattr(ss_mod, "_open_session", lambda: mock_db) + + # Прогреть кеш для cian и avito + get_scraper_delay("cian") + get_scraper_delay("avito") + + # Сбросить только cian + invalidate_cache("cian") + + with ss_mod._CACHE_LOCK: + assert "cian" not in ss_mod._CACHE + assert "avito" in ss_mod._CACHE + + +def test_invalidate_cache_all(monkeypatch): + """invalidate_cache() без аргумента очищает весь кеш.""" + invalidate_cache() + + mock_db = _make_mock_session({"avito": 7.0, "global": 0.0}) + monkeypatch.setattr(ss_mod, "_open_session", lambda: mock_db) + + get_scraper_delay("avito") + + invalidate_cache() + + with ss_mod._CACHE_LOCK: + assert len(ss_mod._CACHE) == 0 + + +def test_global_missing_row_treated_as_zero(monkeypatch): + """Если строки 'global' нет в БД — фолбек 0.0 (не применяем floor).""" + invalidate_cache() + + mock_db = _make_mock_session({"avito": 7.0, "global": None}) + monkeypatch.setattr(ss_mod, "_open_session", lambda: mock_db) + + delay = get_scraper_delay("avito") + assert delay == 7.0 + + +# ── Admin API integration smoke ─────────────────────────────────────────────── + + +def test_list_scraper_settings_endpoint(monkeypatch): + """GET /scraper-settings — эндпоинт монтируется, возвращает список.""" + import sys + from unittest.mock import MagicMock as MM + + # Stub weasyprint (нет на CI) + _wp = MM() + sys.modules.setdefault("weasyprint", _wp) + + from fastapi.testclient import TestClient + + from app.api.v1 import admin as admin_module + from app.core.db import get_db + from fastapi import FastAPI + + app = FastAPI() + app.include_router(admin_module.router, prefix="/api/v1/admin") + + fake_rows = [ + {"source": "avito", "request_delay_sec": 7.0, "description": "Avito", "updated_at": None}, + {"source": "global", "request_delay_sec": 0.0, "description": "Global", "updated_at": None}, + ] + + mock_db = MM() + mock_db.execute.return_value.mappings.return_value.all.return_value = fake_rows + app.dependency_overrides[get_db] = lambda: mock_db + + client = TestClient(app) + r = client.get("/api/v1/admin/scraper-settings") + assert r.status_code == 200 + data = r.json() + assert "settings" in data + sources = [s["source"] for s in data["settings"]] + assert "global" in sources + assert "avito" in sources -- 2.45.3