feat(admin): unified scrape/runs + scraper/health + rotate-ip API (scrapers UI Phase 1) #1818

Merged
lekss361 merged 1 commit from feat/scraper-admin-apis into main 2026-06-20 06:40:36 +00:00
3 changed files with 582 additions and 0 deletions

View file

@ -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)

View file

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

View file

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