From 412367ef16e7f03871152abe1f93942df06ea757 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 2 Jul 2026 19:30:27 +0300 Subject: [PATCH] feat(proxy-pool): scrape_proxies table + admin bulk-load/list endpoints (#2161) --- tradein-mvp/backend/app/api/v1/admin.py | 281 +++++++++++++++++- .../backend/data/sql/157_scrape_proxies.sql | 64 ++++ .../backend/tests/test_admin_proxies.py | 199 +++++++++++++ 3 files changed, 542 insertions(+), 2 deletions(-) create mode 100644 tradein-mvp/backend/data/sql/157_scrape_proxies.sql create mode 100644 tradein-mvp/backend/tests/test_admin_proxies.py diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index 1720ccf7..8b243d1b 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/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"]), + ) diff --git a/tradein-mvp/backend/data/sql/157_scrape_proxies.sql b/tradein-mvp/backend/data/sql/157_scrape_proxies.sql new file mode 100644 index 00000000..ca3e9d4d --- /dev/null +++ b/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; diff --git a/tradein-mvp/backend/tests/test_admin_proxies.py b/tradein-mvp/backend/tests/test_admin_proxies.py new file mode 100644 index 00000000..277edf0b --- /dev/null +++ b/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