feat(admin): unified scrape/runs + scraper/health + rotate-ip API для единой scrapers-страницы

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)

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]

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"]