From 2f88a719151b462b836e8beb68bff99e9578add8 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sat, 20 Jun 2026 09:25:57 +0300 Subject: [PATCH] =?UTF-8?q?feat(admin):=20unified=20scrape/runs=20+=20scra?= =?UTF-8?q?per/health=20+=20rotate-ip=20API=20=D0=B4=D0=BB=D1=8F=20=D0=B5?= =?UTF-8?q?=D0=B4=D0=B8=D0=BD=D0=BE=D0=B9=20scrapers-=D1=81=D1=82=D1=80?= =?UTF-8?q?=D0=B0=D0=BD=D0=B8=D1=86=D1=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tradein-mvp/backend/app/api/v1/admin.py | 237 ++++++++++++++ .../backend/app/services/scrape_runs.py | 52 ++++ .../backend/tests/test_scraper_admin_apis.py | 293 ++++++++++++++++++ 3 files changed, 582 insertions(+) create mode 100644 tradein-mvp/backend/tests/test_scraper_admin_apis.py diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index 981c5157..0570cbe7 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/tradein-mvp/backend/app/api/v1/admin.py @@ -5,12 +5,14 @@ Auth: Caddy basic_auth gate'ит /trade-in/api/v1/admin/* — application layer from __future__ import annotations +import asyncio import json import logging import time from typing import Annotated, Any, Literal from urllib.parse import urlparse +import httpx from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query from pydantic import BaseModel, Field from sqlalchemy import text @@ -1871,3 +1873,238 @@ async def scrape_house_imv_backfill( duration_sec=round(result.duration_sec, 2), status_counts=result.status_counts, ) + + +# ── Единая scrapers-страница: unified runs + health + rotate-ip (epic) ──────── + + +class UnifiedScrapeRunRow(BaseModel): + """Строка scrape_runs для unified-таблицы (все source'ы в одной выдаче).""" + + run_id: int + source: str + run_type: str | None = None + status: str + params: dict | None = None + counters: dict | None = None + total_seen: int | None = None + new_count: int | None = None + started_at: str | None = None + finished_at: str | None = None + heartbeat_at: str | None = None + error_text: str | None = None + + +class UnifiedScrapeRunsResponse(BaseModel): + total: int + rows: list[UnifiedScrapeRunRow] + + +class BrowserHealth(BaseModel): + reachable: bool + browsers: dict[str, bool] = Field(default_factory=dict) + + +class ProviderHealth(BaseModel): + source: str + proxy_host: str | None = None + proxy_port: int | None = None + rotate_supported: bool = False + current_ip: str | None = None + + +class ScraperHealthResponse(BaseModel): + fetch_mode: str + browser: BrowserHealth + providers: list[ProviderHealth] + + +class RotateIpResponse(BaseModel): + ok: bool + new_ip: str | None = None + reason: str | None = None + + +_ROTATABLE_SOURCES = ("avito", "cian", "yandex") + + +def _provider_proxy_url(source: str) -> str | None: + """Effective proxy URL для source (учитывает property-fallback в settings).""" + return { + "avito": settings.avito_proxy_url, + "cian": settings.cian_proxy_url, + "yandex": settings.yandex_proxy_url, + }.get(source) + + +def _provider_rotate_url(source: str) -> str | None: + """changeip-URL для source (None → auto-rotate прокси без ручной ротации).""" + return { + "avito": settings.avito_proxy_rotate_url, + "cian": settings.cian_proxy_rotate_url, + "yandex": settings.yandex_proxy_rotate_url, + }.get(source) + + +def _parse_proxy_host_port(proxy_url: str | None) -> tuple[str | None, int | None]: + """Распарсить host/port из proxy URL (схема http(s)://user:pass@host:port).""" + if not proxy_url: + return None, None + try: + parsed = urlparse(proxy_url) + return parsed.hostname, parsed.port + except Exception: + logger.warning("scraper/health: cannot parse proxy url", exc_info=True) + return None, None + + +@router.get("/scrape/runs", response_model=UnifiedScrapeRunsResponse) +def list_scrape_runs_unified( + db: Annotated[Session, Depends(get_db)], + source: Annotated[str | None, Query()] = None, + status: Annotated[str | None, Query()] = None, + limit: Annotated[int, Query(ge=1, le=200)] = 50, + offset: Annotated[int, Query(ge=0)] = 0, +) -> UnifiedScrapeRunsResponse: + """Unified-список scrape_runs по всем source'ам (для единой scrapers-страницы). + + Замена 3 per-source /runs (avito/cian/yandex-city-sweep) единым эндпоинтом. + Старые per-source эндпоинты сохранены для обратной совместимости. + + Query: + source — опц. фильтр по source (avito_city_sweep / cian_city_sweep / ...). + status — опц. фильтр (done/running/banned/zombie/failed/cancelled). + limit — default 50, max 200. + offset — default 0. + """ + total, rows = runs_mod.list_all(db, source=source, status=status, limit=limit, offset=offset) + + def _iso(v: Any) -> str | None: + return v.isoformat() if v is not None else None + + return UnifiedScrapeRunsResponse( + total=total, + rows=[ + UnifiedScrapeRunRow( + run_id=r["run_id"], + source=r["source"], + run_type=r.get("run_type"), + status=r["status"], + params=r.get("params"), + counters=r.get("counters"), + total_seen=r.get("total_seen"), + new_count=r.get("new_count"), + started_at=_iso(r.get("started_at")), + finished_at=_iso(r.get("finished_at")), + heartbeat_at=_iso(r.get("heartbeat_at")), + error_text=r.get("error_text"), + ) + for r in rows + ], + ) + + +async def _probe_browser_health() -> BrowserHealth: + """GET tradein-browser /health (timeout 5с). reachable=False при ошибке.""" + url = f"{settings.browser_http_endpoint.rstrip('/')}/health" + try: + async with httpx.AsyncClient(timeout=5.0) as client: + resp = await client.get(url) + resp.raise_for_status() + data = resp.json() + browsers = data.get("browsers") or {} + if not isinstance(browsers, dict): + browsers = {} + return BrowserHealth(reachable=True, browsers={k: bool(v) for k, v in browsers.items()}) + except Exception: + logger.warning("scraper/health: browser /health unreachable", exc_info=True) + return BrowserHealth(reachable=False, browsers={}) + + +async def _probe_current_ip(proxy_url: str | None) -> str | None: + """Best-effort: текущий exit-IP через прокси (ipify, timeout 8с). None при ошибке.""" + if not proxy_url: + return None + try: + async with httpx.AsyncClient(proxy=proxy_url, timeout=8.0) as client: + resp = await client.get("https://api.ipify.org", params={"format": "json"}) + resp.raise_for_status() + ip = resp.json().get("ip") + return str(ip) if ip else None + except Exception: + logger.warning("scraper/health: ipify probe failed", exc_info=True) + return None + + +@router.get("/scraper/health", response_model=ScraperHealthResponse) +async def scraper_health() -> ScraperHealthResponse: + """Сводный health для единой scrapers-страницы: fetch_mode + browser + провайдеры. + + - fetch_mode: settings.scraper_fetch_mode (curl_cffi / browser). + - browser: GET tradein-browser /health (reachable + per-browser ready-флаги). + - providers: для avito/cian/yandex — proxy host/port, rotate_supported, + best-effort current_ip (параллельный пробинг через прокси на ipify). + + Все пробинги параллельны (asyncio.gather) и time-boxed — суммарно ≤10с. + """ + proxy_urls = {s: _provider_proxy_url(s) for s in _ROTATABLE_SOURCES} + + browser, *ips = await asyncio.gather( + _probe_browser_health(), + *[_probe_current_ip(proxy_urls[s]) for s in _ROTATABLE_SOURCES], + ) + ip_by_source = dict(zip(_ROTATABLE_SOURCES, ips, strict=True)) + + providers: list[ProviderHealth] = [] + for source in _ROTATABLE_SOURCES: + host, port = _parse_proxy_host_port(proxy_urls[source]) + providers.append( + ProviderHealth( + source=source, + proxy_host=host, + proxy_port=port, + rotate_supported=bool(_provider_rotate_url(source)), + current_ip=ip_by_source[source], + ) + ) + + return ScraperHealthResponse( + fetch_mode=settings.scraper_fetch_mode, + browser=browser, + providers=providers, + ) + + +@router.post("/scraper/{source}/rotate-ip", response_model=RotateIpResponse) +async def rotate_proxy_ip( + source: Literal["avito", "cian", "yandex"], +) -> RotateIpResponse: + """Сменить мобильный exit-IP провайдера через changeip-ссылку (mobileproxy). + + Зеркалит логику AvitoScraper._rotate_ip (GET rotate_url + &format=json), но БЕЗ + settle-sleep — API сразу возвращает ответ changeip. Если rotate_url не задан — + прокси с авто-ротацией (свежий IP на новое соединение), ручная ротация не нужна. + """ + rotate_url = _provider_rotate_url(source) + if not rotate_url: + return RotateIpResponse(ok=False, reason="no rotate url (auto-rotate proxy)") + + sep = "&" if "?" in rotate_url else "?" + try: + async with httpx.AsyncClient(timeout=20.0) as client: + resp = await client.get(f"{rotate_url}{sep}format=json") + resp.raise_for_status() + try: + data = resp.json() + except Exception: + data = {} + except Exception as exc: + logger.warning("rotate-ip: changeip failed source=%s", source, exc_info=True) + return RotateIpResponse(ok=False, reason=f"changeip error: {exc}") + + # changeip отдаёт новый IP в одном из полей (формат провайдер-зависимый). + new_ip = None + if isinstance(data, dict): + new_ip = data.get("new_ip") or data.get("ip") or data.get("proxy_ip") + logger.info("rotate-ip: source=%s new_ip=%s", source, new_ip) + return RotateIpResponse(ok=True, new_ip=str(new_ip) if new_ip else None) diff --git a/tradein-mvp/backend/app/services/scrape_runs.py b/tradein-mvp/backend/app/services/scrape_runs.py index 66451ee0..62b3ed63 100644 --- a/tradein-mvp/backend/app/services/scrape_runs.py +++ b/tradein-mvp/backend/app/services/scrape_runs.py @@ -255,3 +255,55 @@ def list_recent(db: Session, *, source: str, limit: int = 10) -> list[dict[str, .all() ) return [dict(r) for r in rows] + + +def list_all( + db: Session, + *, + source: str | None = None, + status: str | None = None, + limit: int = 50, + offset: int = 0, +) -> tuple[int, list[dict[str, Any]]]: + """Unified-список runs по всем source'ам с опц. фильтрами source/status. + + Возвращает (total, rows): total — кол-во строк под фильтром (без limit/offset), + rows — текущая страница (ORDER BY started_at DESC, LIMIT/OFFSET). + + Используется единой scrapers-страницей вместо 3 per-source /runs. + """ + where = ["1=1"] + params: dict[str, Any] = {"limit": limit, "offset": offset} + if source is not None: + where.append("source = :source") + params["source"] = source + if status is not None: + where.append("status = :status") + params["status"] = status + where_sql = " AND ".join(where) + + total_row = db.execute( + text(f"SELECT COUNT(*) AS n FROM scrape_runs WHERE {where_sql}"), + params, + ).fetchone() + total = int(total_row.n) if total_row is not None else 0 + + rows = ( + db.execute( + text( + f""" + SELECT id AS run_id, source, run_type, status, params, counters, + total_seen, new_count, started_at, finished_at, + heartbeat_at, error AS error_text + FROM scrape_runs + WHERE {where_sql} + ORDER BY started_at DESC NULLS LAST + LIMIT CAST(:limit AS int) OFFSET CAST(:offset AS int) + """ + ), + params, + ) + .mappings() + .all() + ) + return total, [dict(r) for r in rows] diff --git a/tradein-mvp/backend/tests/test_scraper_admin_apis.py b/tradein-mvp/backend/tests/test_scraper_admin_apis.py new file mode 100644 index 00000000..c4f3b7a9 --- /dev/null +++ b/tradein-mvp/backend/tests/test_scraper_admin_apis.py @@ -0,0 +1,293 @@ +"""Offline tests для unified scrapers-страницы admin-API (epic консолидации UI). + +Покрытие 3 эндпоинтов (db/httpx мокаются, NO live network/DB): + - GET /api/v1/admin/scrape/runs — unified runs + total + фильтры + - GET /api/v1/admin/scraper/health — fetch_mode + browser + providers + - POST /api/v1/admin/scraper/{src}/rotate-ip — changeip + no-rotate-url кейс +""" + +from __future__ import annotations + +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from datetime import UTC, datetime +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + + +@pytest.fixture +def client() -> TestClient: + from app.api.v1 import admin as admin_module + from app.core.db import get_db + + app = FastAPI() + app.include_router(admin_module.router, prefix="/api/v1/admin") + + def fake_db(): + yield MagicMock() + + app.dependency_overrides[get_db] = fake_db + return TestClient(app) + + +# ── API 1: GET /scrape/runs ────────────────────────────────────────────────── + + +def test_unified_runs_returns_rows_and_total(client: TestClient) -> None: + """list_all → total + rows; ISO-форматирование дат + error_text/run_type.""" + fake_rows = [ + { + "run_id": 5, + "source": "avito_city_sweep", + "run_type": "city_sweep", + "status": "done", + "params": {"pages_per_anchor": 3}, + "counters": {"lots_fetched": 120}, + "total_seen": 120, + "new_count": 30, + "started_at": datetime(2026, 6, 1, 10, tzinfo=UTC), + "finished_at": datetime(2026, 6, 1, 11, tzinfo=UTC), + "heartbeat_at": datetime(2026, 6, 1, 11, tzinfo=UTC), + "error_text": None, + } + ] + with patch("app.services.scrape_runs.list_all", return_value=(42, fake_rows)) as m: + r = client.get("/api/v1/admin/scrape/runs") + assert r.status_code == 200 + body = r.json() + assert body["total"] == 42 + assert len(body["rows"]) == 1 + row = body["rows"][0] + assert row["run_id"] == 5 + assert row["source"] == "avito_city_sweep" + assert row["run_type"] == "city_sweep" + assert row["total_seen"] == 120 + assert row["new_count"] == 30 + assert row["error_text"] is None + assert "2026-06-01" in row["started_at"] + # defaults передаются в сервис + _, kwargs = m.call_args + assert kwargs["source"] is None + assert kwargs["status"] is None + assert kwargs["limit"] == 50 + assert kwargs["offset"] == 0 + + +def test_unified_runs_passes_filters(client: TestClient) -> None: + """source/status/limit/offset прокидываются в list_all.""" + with patch("app.services.scrape_runs.list_all", return_value=(0, [])) as m: + r = client.get( + "/api/v1/admin/scrape/runs", + params={ + "source": "cian_city_sweep", + "status": "banned", + "limit": 10, + "offset": 20, + }, + ) + assert r.status_code == 200 + _, kwargs = m.call_args + assert kwargs["source"] == "cian_city_sweep" + assert kwargs["status"] == "banned" + assert kwargs["limit"] == 10 + assert kwargs["offset"] == 20 + + +def test_unified_runs_limit_over_max_422(client: TestClient) -> None: + """limit=999 > max=200 → 422.""" + r = client.get("/api/v1/admin/scrape/runs", params={"limit": 999}) + assert r.status_code == 422 + + +def test_unified_runs_negative_offset_422(client: TestClient) -> None: + """offset=-1 < 0 → 422.""" + r = client.get("/api/v1/admin/scrape/runs", params={"offset": -1}) + assert r.status_code == 422 + + +# ── API 2: GET /scraper/health ─────────────────────────────────────────────── + + +def test_health_assembles_browser_and_providers(client: TestClient) -> None: + """browser /health + ipify пробинг → fetch_mode + browser + 3 провайдера.""" + from app.api.v1 import admin as admin_module + + async def fake_browser_health() -> admin_module.BrowserHealth: + return admin_module.BrowserHealth( + reachable=True, + browsers={"avito": False, "cian": True, "yandex": False, "generic": True}, + ) + + async def fake_current_ip(proxy_url: str | None) -> str | None: + return "1.2.3.4" if proxy_url else None + + proxy_urls = { + "avito": "http://u:p@host.a:10049", + "cian": None, + "yandex": "http://u:p@host.y:10051", + } + rotate_urls = {"avito": "http://ch/changeip", "cian": None, "yandex": None} + + with ( + patch.object(admin_module, "_probe_browser_health", fake_browser_health), + patch.object(admin_module, "_probe_current_ip", fake_current_ip), + patch.object(admin_module, "_provider_proxy_url", lambda s: proxy_urls[s]), + patch.object(admin_module, "_provider_rotate_url", lambda s: rotate_urls[s]), + patch.object(admin_module.settings, "scraper_fetch_mode", "browser"), + ): + r = client.get("/api/v1/admin/scraper/health") + + assert r.status_code == 200 + body = r.json() + assert body["fetch_mode"] == "browser" + assert body["browser"]["reachable"] is True + assert body["browser"]["browsers"]["cian"] is True + assert body["browser"]["browsers"]["avito"] is False + + by_source = {p["source"]: p for p in body["providers"]} + assert set(by_source) == {"avito", "cian", "yandex"} + assert by_source["avito"]["proxy_host"] == "host.a" + assert by_source["avito"]["proxy_port"] == 10049 + assert by_source["avito"]["rotate_supported"] is True + assert by_source["avito"]["current_ip"] == "1.2.3.4" + # cian без proxy → host/port/ip null, rotate_supported False + assert by_source["cian"]["proxy_host"] is None + assert by_source["cian"]["current_ip"] is None + assert by_source["cian"]["rotate_supported"] is False + # yandex имеет proxy но нет rotate_url + assert by_source["yandex"]["proxy_host"] == "host.y" + assert by_source["yandex"]["rotate_supported"] is False + + +def test_health_browser_unreachable(client: TestClient) -> None: + """httpx-ошибка на /health → reachable=False, browsers пуст.""" + from app.api.v1 import admin as admin_module + + async def fake_current_ip(proxy_url: str | None) -> str | None: + return None + + class _BoomClient: + def __init__(self, *a: Any, **k: Any) -> None: + pass + + async def __aenter__(self) -> _BoomClient: + return self + + async def __aexit__(self, *a: Any) -> None: + return None + + async def get(self, *a: Any, **k: Any) -> Any: + raise RuntimeError("browser down") + + with ( + patch.object(admin_module.httpx, "AsyncClient", _BoomClient), + patch.object(admin_module, "_probe_current_ip", fake_current_ip), + patch.object(admin_module, "_provider_proxy_url", lambda s: None), + patch.object(admin_module, "_provider_rotate_url", lambda s: None), + ): + r = client.get("/api/v1/admin/scraper/health") + + assert r.status_code == 200 + body = r.json() + assert body["browser"]["reachable"] is False + assert body["browser"]["browsers"] == {} + + +# ── API 3: POST /scraper/{source}/rotate-ip ────────────────────────────────── + + +def test_rotate_ip_calls_changeip(client: TestClient) -> None: + """rotate_url задан → GET changeip + format=json, возврат new_ip из ответа.""" + from app.api.v1 import admin as admin_module + + captured: dict[str, str] = {} + + class _FakeResp: + def raise_for_status(self) -> None: + return None + + def json(self) -> dict[str, str]: + return {"new_ip": "9.9.9.9"} + + class _FakeClient: + def __init__(self, *a: Any, **k: Any) -> None: + pass + + async def __aenter__(self) -> _FakeClient: + return self + + async def __aexit__(self, *a: Any) -> None: + return None + + async def get(self, url: str, *a: Any, **k: Any) -> _FakeResp: + captured["url"] = url + return _FakeResp() + + with ( + patch.object(admin_module.httpx, "AsyncClient", _FakeClient), + patch.object(admin_module.settings, "avito_proxy_rotate_url", "http://ch/changeip?key=x"), + ): + r = client.post("/api/v1/admin/scraper/avito/rotate-ip") + + assert r.status_code == 200 + body = r.json() + assert body["ok"] is True + assert body["new_ip"] == "9.9.9.9" + assert "format=json" in captured["url"] + # уже был '?' в URL → должен использовать '&' + assert captured["url"].endswith("&format=json") + + +def test_rotate_ip_no_rotate_url(client: TestClient) -> None: + """rotate_url не задан → ok=False + reason про auto-rotate, changeip НЕ дёргается.""" + from app.api.v1 import admin as admin_module + + with patch.object(admin_module.settings, "cian_proxy_rotate_url", None): + r = client.post("/api/v1/admin/scraper/cian/rotate-ip") + + assert r.status_code == 200 + body = r.json() + assert body["ok"] is False + assert body["new_ip"] is None + assert "auto-rotate" in body["reason"] + + +def test_rotate_ip_invalid_source_422(client: TestClient) -> None: + """source вне {avito,cian,yandex} → 422 (Literal-валидация path).""" + r = client.post("/api/v1/admin/scraper/domclick/rotate-ip") + assert r.status_code == 422 + + +def test_rotate_ip_changeip_error(client: TestClient) -> None: + """httpx-ошибка changeip → ok=False + reason.""" + from app.api.v1 import admin as admin_module + + class _BoomClient: + def __init__(self, *a: Any, **k: Any) -> None: + pass + + async def __aenter__(self) -> _BoomClient: + return self + + async def __aexit__(self, *a: Any) -> None: + return None + + async def get(self, *a: Any, **k: Any) -> Any: + raise RuntimeError("changeip 500") + + with ( + patch.object(admin_module.httpx, "AsyncClient", _BoomClient), + patch.object(admin_module.settings, "yandex_proxy_rotate_url", "http://ch/changeip"), + ): + r = client.post("/api/v1/admin/scraper/yandex/rotate-ip") + + assert r.status_code == 200 + body = r.json() + assert body["ok"] is False + assert "changeip error" in body["reason"]