feat(proxy-pool): scrape_proxies table + admin bulk-load/list endpoints (#2161)

tradein-mvp/backend/app/api/v1/admin.py

@@ -10,11 +10,11 @@ import json
 import logging
 import time
 from typing import Annotated, Any, Literal
-from urllib.parse import urlparse
+from urllib.parse import urlparse, urlunparse
 
 import httpx
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
@@ -2350,3 +2350,280 @@ def get_data_quality(
     )
 
     return DataQualityResponse(sources=sources, houses=houses)
+
+
+# ── Proxy pool: хранилище + bulk-загрузка / список (#2161) ───────────────────
+#
+# АДДИТИВНО: ручки лишь наполняют/читают scrape_proxies. Подключение пула к
+# боевому сбору (pick/lease/rotate вместо env) — отдельные шаги P3/P4.
+
+_PROXY_AFFINITIES = frozenset({"avito", "cian", "yandex", "generic", "any"})
+_PROXY_KINDS = frozenset({"http", "socks5"})
+
+
+def _mask_proxy_url(url: str | None) -> str | None:
+    """Маскирует пароль в proxy-URL: socks5://user:pass@host → socks5://user:***@host.
+
+    URL без userinfo/пароля возвращается как есть. Не-парсящийся URL — как есть
+    (лучше вернуть исходное, чем упасть на форматировании admin-листинга).
+    """
+    if not url:
+        return url
+    try:
+        parsed = urlparse(url)
+    except ValueError:
+        return url
+    if not parsed.password:
+        return url
+    userinfo = parsed.username or ""
+    host = parsed.hostname or ""
+    masked_netloc = f"{userinfo}:***@{host}"
+    if parsed.port:
+        masked_netloc += f":{parsed.port}"
+    return urlunparse(parsed._replace(netloc=masked_netloc))
+
+
+class ProxyIn(BaseModel):
+    """Одна запись прокси для bulk-загрузки."""
+
+    url: str = Field(..., min_length=3, max_length=500)
+    provider_affinity: str = Field(default="any")
+    kind: str = Field(default="http")
+    rotate_url: str | None = Field(default=None, max_length=500)
+    label: str | None = Field(default=None, max_length=200)
+    geo: str | None = Field(default=None, max_length=100)
+    operator: str | None = Field(default=None, max_length=100)
+
+    @field_validator("provider_affinity")
+    @classmethod
+    def _check_affinity(cls, v: str) -> str:
+        if v not in _PROXY_AFFINITIES:
+            raise ValueError(
+                f"provider_affinity must be one of {sorted(_PROXY_AFFINITIES)}, got {v!r}"
+            )
+        return v
+
+    @field_validator("kind")
+    @classmethod
+    def _check_kind(cls, v: str) -> str:
+        if v not in _PROXY_KINDS:
+            raise ValueError(f"kind must be one of {sorted(_PROXY_KINDS)}, got {v!r}")
+        return v
+
+
+class ProxyBulkRequest(BaseModel):
+    proxies: list[ProxyIn] = Field(..., min_length=1, max_length=1000)
+
+
+class ProxyBulkResponse(BaseModel):
+    inserted: int
+    updated: int
+
+
+class ProxyRow(BaseModel):
+    id: int
+    label: str | None
+    url: str  # маскированный
+    kind: str
+    provider_affinity: str
+    rotate_url: str | None  # маскированный
+    enabled: bool
+    consecutive_fails: int
+    exit_ip: str | None
+    latency_ms: int | None
+    last_check_at: str | None
+    last_ok_at: str | None
+    leased_by: int | None
+    leased_at: str | None
+    geo: str | None
+    operator: str | None
+    expires_at: str | None
+    created_at: str | None
+    updated_at: str | None
+
+
+@router.post("/proxies/bulk", response_model=ProxyBulkResponse)
+def bulk_upsert_proxies(
+    payload: ProxyBulkRequest,
+    db: Annotated[Session, Depends(get_db)],
+) -> ProxyBulkResponse:
+    """Bulk UPSERT прокси по url (#2161).
+
+    Тело: {"proxies": [{"url", "provider_affinity", "kind"?, "rotate_url"?,
+    "label"?, "geo"?, "operator"?}, ...]}.
+
+    Существующий url → DO UPDATE (affinity/kind/rotate_url + enabled=true,
+    label/geo/operator обновляются если переданы). Новый → INSERT.
+    Валидация provider_affinity/kind по whitelist на уровне Pydantic → 422.
+
+    Возвращает {inserted, updated}. Дубли по url ВНУТРИ одного запроса
+    схлопываются последним вхождением (ON CONFLICT в рамках одной транзакции).
+    """
+    inserted = 0
+    updated = 0
+    for proxy in payload.proxies:
+        # RETURNING (xmax = 0) — системный трюк Postgres: xmax=0 у только что
+        # вставленной строки, !=0 у обновлённой существующей.
+        row = db.execute(
+            text(
+                """
+                INSERT INTO scrape_proxies
+                    (url, provider_affinity, kind, rotate_url, label, geo, operator,
+                     enabled, updated_at)
+                VALUES
+                    (:url, :aff, :kind, :rotate_url, :label, :geo, :operator, true, now())
+                ON CONFLICT (url) DO UPDATE SET
+                    provider_affinity = EXCLUDED.provider_affinity,
+                    kind              = EXCLUDED.kind,
+                    rotate_url        = EXCLUDED.rotate_url,
+                    label             = COALESCE(EXCLUDED.label, scrape_proxies.label),
+                    geo               = COALESCE(EXCLUDED.geo, scrape_proxies.geo),
+                    operator          = COALESCE(EXCLUDED.operator, scrape_proxies.operator),
+                    enabled           = true,
+                    updated_at        = now()
+                RETURNING (xmax = 0) AS was_inserted
+                """
+            ),
+            {
+                "url": proxy.url,
+                "aff": proxy.provider_affinity,
+                "kind": proxy.kind,
+                "rotate_url": proxy.rotate_url,
+                "label": proxy.label,
+                "geo": proxy.geo,
+                "operator": proxy.operator,
+            },
+        ).scalar()
+        if row:
+            inserted += 1
+        else:
+            updated += 1
+    db.commit()
+    logger.info("proxies/bulk: inserted=%d updated=%d", inserted, updated)
+    return ProxyBulkResponse(inserted=inserted, updated=updated)
+
+
+@router.get("/proxies", response_model=list[ProxyRow])
+def list_proxies(
+    db: Annotated[Session, Depends(get_db)],
+    provider: str | None = Query(default=None, description="Фильтр по provider_affinity"),
+    enabled: bool | None = Query(default=None, description="Фильтр по enabled"),
+) -> list[ProxyRow]:
+    """Список прокси со статусами. Пароли в url/rotate_url маскируются.
+
+    Фильтры: provider (=provider_affinity), enabled. Без фильтров — все.
+    """
+    clauses: list[str] = []
+    params: dict[str, Any] = {}
+    if provider is not None:
+        clauses.append("provider_affinity = :provider")
+        params["provider"] = provider
+    if enabled is not None:
+        clauses.append("enabled = :enabled")
+        params["enabled"] = enabled
+    where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
+
+    rows = (
+        db.execute(
+            text(
+                f"""
+                SELECT id, label, url, kind, provider_affinity, rotate_url, enabled,
+                       consecutive_fails, exit_ip, latency_ms, last_check_at, last_ok_at,
+                       leased_by, leased_at, geo, operator, expires_at, created_at, updated_at
+                FROM scrape_proxies
+                {where}
+                ORDER BY provider_affinity, id
+                """
+                # where собран из фикс-строк, значения параметризованы (no injection)
+            ),
+            params,
+        )
+        .mappings()
+        .all()
+    )
+
+    def _iso(v: Any) -> str | None:
+        return v.isoformat() if v is not None else None
+
+    return [
+        ProxyRow(
+            id=r["id"],
+            label=r["label"],
+            url=_mask_proxy_url(r["url"]) or "",
+            kind=r["kind"],
+            provider_affinity=r["provider_affinity"],
+            rotate_url=_mask_proxy_url(r["rotate_url"]),
+            enabled=r["enabled"],
+            consecutive_fails=r["consecutive_fails"],
+            exit_ip=r["exit_ip"],
+            latency_ms=r["latency_ms"],
+            last_check_at=_iso(r["last_check_at"]),
+            last_ok_at=_iso(r["last_ok_at"]),
+            leased_by=r["leased_by"],
+            leased_at=_iso(r["leased_at"]),
+            geo=r["geo"],
+            operator=r["operator"],
+            expires_at=_iso(r["expires_at"]),
+            created_at=_iso(r["created_at"]),
+            updated_at=_iso(r["updated_at"]),
+        )
+        for r in rows
+    ]
+
+
+class ProxyPatch(BaseModel):
+    enabled: bool
+
+
+@router.patch("/proxies/{proxy_id}", response_model=ProxyRow)
+def patch_proxy(
+    proxy_id: int,
+    payload: ProxyPatch,
+    db: Annotated[Session, Depends(get_db)],
+) -> ProxyRow:
+    """Enable/disable одного прокси по id. 404 если не найден."""
+    row = (
+        db.execute(
+            text(
+                """
+                UPDATE scrape_proxies
+                SET enabled = :enabled, updated_at = now()
+                WHERE id = :id
+                RETURNING id, label, url, kind, provider_affinity, rotate_url, enabled,
+                          consecutive_fails, exit_ip, latency_ms, last_check_at, last_ok_at,
+                          leased_by, leased_at, geo, operator, expires_at, created_at, updated_at
+                """
+            ),
+            {"enabled": payload.enabled, "id": proxy_id},
+        )
+        .mappings()
+        .fetchone()
+    )
+    if row is None:
+        raise HTTPException(status_code=404, detail=f"proxy id={proxy_id} not found")
+    db.commit()
+
+    def _iso(v: Any) -> str | None:
+        return v.isoformat() if v is not None else None
+
+    return ProxyRow(
+        id=row["id"],
+        label=row["label"],
+        url=_mask_proxy_url(row["url"]) or "",
+        kind=row["kind"],
+        provider_affinity=row["provider_affinity"],
+        rotate_url=_mask_proxy_url(row["rotate_url"]),
+        enabled=row["enabled"],
+        consecutive_fails=row["consecutive_fails"],
+        exit_ip=row["exit_ip"],
+        latency_ms=row["latency_ms"],
+        last_check_at=_iso(row["last_check_at"]),
+        last_ok_at=_iso(row["last_ok_at"]),
+        leased_by=row["leased_by"],
+        leased_at=_iso(row["leased_at"]),
+        geo=row["geo"],
+        operator=row["operator"],
+        expires_at=_iso(row["expires_at"]),
+        created_at=_iso(row["created_at"]),
+        updated_at=_iso(row["updated_at"]),
+    )

tradein-mvp/backend/data/sql/157_scrape_proxies.sql

@@ -0,0 +1,64 @@
+-- 157_scrape_proxies.sql
+-- Хранилище пула прокси для скрейперов (#2161) — АДДИТИВНО.
+--
+-- WHY:
+--   Сейчас прокси задаются через env (BROWSER_PROXY_AVITO/CIAN/YANDEX и т.п.) —
+--   один статичный URL на источник, без ротации/health/lease. Таблица даёт
+--   пул с per-provider affinity, health-трекингом и lease-полями под будущий
+--   pick/rotate-механизм. Данные (сами прокси с паролями) грузятся ЧЕРЕЗ ручку
+--   POST /api/v1/admin/proxies/bulk — в миграции секретов НЕТ.
+--
+-- SCOPE (важно):
+--   Эта миграция ТОЛЬКО создаёт хранилище + индексы. Подключение к боевому
+--   сбору (pick/lease/rotate из пула вместо env) — отдельные шаги P3/P4.
+--   Ни один скрейпер здесь не трогается.
+--
+-- IDEMPOTENCY / SAFETY:
+--   - Весь файл в одной транзакции BEGIN/COMMIT.
+--   - CREATE TABLE IF NOT EXISTS + CREATE INDEX IF NOT EXISTS → повторный
+--     прогон no-op (auto-apply strict на деплое это требует).
+--   - Пустая таблица после первого прогона — нормально; прокси загружаются
+--     оператором через admin-ручку.
+
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS scrape_proxies (
+    id                bigserial PRIMARY KEY,
+    label             text,
+    url               text NOT NULL UNIQUE,
+    kind              text NOT NULL DEFAULT 'http'
+                          CHECK (kind IN ('http', 'socks5')),
+    provider_affinity text NOT NULL DEFAULT 'any'
+                          CHECK (provider_affinity IN
+                                 ('avito', 'cian', 'yandex', 'generic', 'any')),
+    rotate_url        text,
+    enabled           boolean NOT NULL DEFAULT true,
+
+    -- health
+    last_check_at     timestamptz,
+    last_ok_at        timestamptz,
+    consecutive_fails integer NOT NULL DEFAULT 0,
+    exit_ip           text,
+    latency_ms        integer,
+
+    -- lease (run_id арендатора; nullable = свободен)
+    leased_by         bigint,
+    leased_at         timestamptz,
+
+    -- meta
+    geo               text,
+    operator          text,
+    expires_at        timestamptz,
+    created_at        timestamptz NOT NULL DEFAULT now(),
+    updated_at        timestamptz NOT NULL DEFAULT now()
+);
+
+-- Подбор свободного прокси под провайдера: фильтр по affinity + enabled + свободности.
+CREATE INDEX IF NOT EXISTS idx_scrape_proxies_pick
+    ON scrape_proxies (provider_affinity, enabled, leased_by);
+
+-- Health-скан: активные с накопленными фейлами.
+CREATE INDEX IF NOT EXISTS idx_scrape_proxies_health
+    ON scrape_proxies (enabled, consecutive_fails);
+
+COMMIT;

tradein-mvp/backend/tests/test_admin_proxies.py

@@ -0,0 +1,199 @@
+"""Offline-тесты admin proxy-pool ручек (#2161).
+
+Покрытие (db мокается, NO live network/DB):
+  - POST /api/v1/admin/proxies/bulk — UPSERT-счётчики, валидация affinity/kind
+  - GET  /api/v1/admin/proxies      — маскировка пароля, фильтры
+  - PATCH /api/v1/admin/proxies/{id} — enable/disable, 404
+"""
+
+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
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture
+def db() -> MagicMock:
+    return MagicMock()
+
+
+@pytest.fixture
+def client(db: MagicMock) -> 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() -> Any:
+        yield db
+
+    app.dependency_overrides[get_db] = fake_db
+    return TestClient(app)
+
+
+def _scalar_result(value: object) -> MagicMock:
+    res = MagicMock()
+    res.scalar.return_value = value
+    return res
+
+
+# ── _mask_proxy_url unit ─────────────────────────────────────────────────────
+
+
+def test_mask_proxy_url_hides_password() -> None:
+    from app.api.v1.admin import _mask_proxy_url
+
+    assert _mask_proxy_url("socks5://user:secret@host:1080") == "socks5://user:***@host:1080"
+    assert _mask_proxy_url("http://u:p@1.2.3.4:8080") == "http://u:***@1.2.3.4:8080"
+
+
+def test_mask_proxy_url_no_password_untouched() -> None:
+    from app.api.v1.admin import _mask_proxy_url
+
+    assert _mask_proxy_url("http://host:8080") == "http://host:8080"
+    assert _mask_proxy_url(None) is None
+    assert _mask_proxy_url("") == ""
+
+
+# ── POST /proxies/bulk ───────────────────────────────────────────────────────
+
+
+def test_bulk_upsert_counts_inserted_and_updated(client: TestClient, db: MagicMock) -> None:
+    # первая строка вставлена (xmax=0 → True), вторая обновлена (False)
+    db.execute.side_effect = [_scalar_result(True), _scalar_result(False)]
+    r = client.post(
+        "/api/v1/admin/proxies/bulk",
+        json={
+            "proxies": [
+                {"url": "http://u:p@host1:8080", "provider_affinity": "avito"},
+                {"url": "http://u:p@host2:8080", "provider_affinity": "cian", "kind": "socks5"},
+            ]
+        },
+    )
+    assert r.status_code == 200, r.text
+    assert r.json() == {"inserted": 1, "updated": 1}
+    assert db.commit.called
+
+
+def test_bulk_invalid_affinity_422(client: TestClient) -> None:
+    r = client.post(
+        "/api/v1/admin/proxies/bulk",
+        json={"proxies": [{"url": "http://host:8080", "provider_affinity": "garbage"}]},
+    )
+    assert r.status_code == 422
+
+
+def test_bulk_invalid_kind_422(client: TestClient) -> None:
+    r = client.post(
+        "/api/v1/admin/proxies/bulk",
+        json={"proxies": [{"url": "http://host:8080", "kind": "ftp"}]},
+    )
+    assert r.status_code == 422
+
+
+def test_bulk_empty_list_422(client: TestClient) -> None:
+    r = client.post("/api/v1/admin/proxies/bulk", json={"proxies": []})
+    assert r.status_code == 422
+
+
+def test_bulk_defaults_affinity_any(client: TestClient, db: MagicMock) -> None:
+    db.execute.side_effect = [_scalar_result(True)]
+    r = client.post(
+        "/api/v1/admin/proxies/bulk",
+        json={"proxies": [{"url": "http://host:8080"}]},
+    )
+    assert r.status_code == 200
+    # affinity по умолчанию 'any' попал в параметры INSERT
+    _, kwargs = db.execute.call_args
+    assert kwargs is not None
+    params = db.execute.call_args.args[1]
+    assert params["aff"] == "any"
+    assert params["kind"] == "http"
+
+
+# ── GET /proxies ─────────────────────────────────────────────────────────────
+
+
+def _proxy_db_row(**over: Any) -> dict[str, Any]:
+    base: dict[str, Any] = {
+        "id": 1,
+        "label": "test",
+        "url": "socks5://user:secret@host:1080",
+        "kind": "socks5",
+        "provider_affinity": "avito",
+        "rotate_url": "http://user:secret@rot:9000/changeip",
+        "enabled": True,
+        "consecutive_fails": 0,
+        "exit_ip": "1.2.3.4",
+        "latency_ms": 120,
+        "last_check_at": datetime(2026, 7, 1, tzinfo=UTC),
+        "last_ok_at": datetime(2026, 7, 1, tzinfo=UTC),
+        "leased_by": None,
+        "leased_at": None,
+        "geo": "RU",
+        "operator": "beeline",
+        "expires_at": None,
+        "created_at": datetime(2026, 6, 1, tzinfo=UTC),
+        "updated_at": datetime(2026, 7, 1, tzinfo=UTC),
+    }
+    base.update(over)
+    return base
+
+
+def test_list_masks_password(client: TestClient, db: MagicMock) -> None:
+    result = MagicMock()
+    result.mappings.return_value.all.return_value = [_proxy_db_row()]
+    db.execute.return_value = result
+
+    r = client.get("/api/v1/admin/proxies")
+    assert r.status_code == 200, r.text
+    rows = r.json()
+    assert len(rows) == 1
+    assert rows[0]["url"] == "socks5://user:***@host:1080"
+    assert rows[0]["rotate_url"] == "http://user:***@rot:9000/changeip"
+    assert "secret" not in r.text
+
+
+def test_list_passes_filters(client: TestClient, db: MagicMock) -> None:
+    result = MagicMock()
+    result.mappings.return_value.all.return_value = []
+    db.execute.return_value = result
+
+    r = client.get("/api/v1/admin/proxies", params={"provider": "cian", "enabled": "false"})
+    assert r.status_code == 200
+    params = db.execute.call_args.args[1]
+    assert params["provider"] == "cian"
+    assert params["enabled"] is False
+
+
+# ── PATCH /proxies/{id} ──────────────────────────────────────────────────────
+
+
+def test_patch_disable(client: TestClient, db: MagicMock) -> None:
+    result = MagicMock()
+    result.mappings.return_value.fetchone.return_value = _proxy_db_row(enabled=False)
+    db.execute.return_value = result
+
+    r = client.patch("/api/v1/admin/proxies/1", json={"enabled": False})
+    assert r.status_code == 200, r.text
+    assert r.json()["enabled"] is False
+    assert db.commit.called
+
+
+def test_patch_not_found_404(client: TestClient, db: MagicMock) -> None:
+    result = MagicMock()
+    result.mappings.return_value.fetchone.return_value = None
+    db.execute.return_value = result
+
+    r = client.patch("/api/v1/admin/proxies/999", json={"enabled": True})
+    assert r.status_code == 404