"""Ротация exit-IP прокси ASocks по требованию, со счётчиком и громким отказом (#2600 п.5).

АДДИТИВНО. НЕ трогает app.services.proxy_pool (pick/lease/health — параллельный
PR #2609, конфликт исключён: вся новая логика тут, в новом модуле).

Контекст (эмпирика, issue #2600 п.5 — проверено владельцем аккаунта/пробой):
  - Документированный публичный API ASocks (GET /v2/proxy/refresh/{portId}?apiKey=)
    для безлимитных портов НЕ работает.
  - Ротация сменой session-суффикса логина (-session-N) НЕ работает — exit-IP
    не меняется (три варианта дали один и тот же IP).
  - Единственный рабочий путь — ручка веб-кабинета:
        POST https://api.asocks.com/unlimited-proxy/{portId}/refresh-ip
        Authorization: Bearer <токен>
    Без заголовка провайдер отдаёт 401 {"success": false, "message": "Unauthenticated"}.
    scrape_proxies.rotate_url уже несёт этот URL (миграция 199) — токен НЕ в URL,
    он только в ASOCKS_API_TOKEN (env, app.core.config.settings.asocks_api_token).
  - Лимит провайдера: 3 ротации в сутки на порт.
  - Токен — сессионный, однажды протухнет (осознанное решение владельца аккаунта).
    Когда это случится, провайдер ответит 401 — это ГРОМКИЙ отказ ниже
    (logger.error + Sentry/GlitchTip capture_message), а не молчаливая остановка.

Суточный лимит и таблица истории (scrape_proxy_rotations, миграция 198):
  Против лимита 3/сутки считаются ТОЛЬКО попытки, реально дошедшие до провайдера
  и обработанные им — т.е. любой HTTP-ответ провайдера, КРОМЕ 401. Обоснование:
  401 — это буквально описание провайдера "Unauthenticated": запрос отсеян на
  уровне аутентификации ДО обращения к самой логике ротации порта, провайдер не
  мог засчитать использование ротации тому, кого даже не подтвердил. Сетевые
  ошибки (таймаут / разрыв соединения — ответа вообще нет) по той же логике не
  считаются: нет подтверждения, что запрос вообще дошёл до провайдера. Локальные
  отказы (нет rotate_url / нет токена / лимит уже исчерпан) до HTTP-вызова не
  доходят вовсе — в таблицу не пишутся и лимит не трогают.

    quota-consuming := http_status IS NOT NULL AND http_status != 401
    (успех 200 И любой не-401 ответ провайдера, включая его собственные 4xx/5xx —
    если провайдер прошёл auth и ответил бизнес-ошибкой, запрос точно дошёл до
    реальной rotate-логики и мог быть учтён в лимите на его стороне).

⛔ Токен никогда не должен появиться в возвращаемом клиенту reason, в тексте
исключения, ни в одной записи scrape_proxy_rotations. Прецедент утечки через
str(exc) — тот же паттерн, что закрывал (до удаления #2616 шаг 3) changeip-путь
admin.rotate_proxy_ip: httpx-исключения несут полный request URL/детали,
поэтому наружу — только нейтральный reason, полные детали — в лог с exc_info=True.

⛔ Хост-пиннинг (security review PR #2611): scrape_proxies.rotate_url колонка
НЕОДНОРОДНА — часть строк пула (id 3/4/5 на проде) несёт mobileproxy changeip-
ссылки (`https://changeip.mobileproxy.space/?proxy_key=<секрет mobileproxy>`,
тот же формат, что читал удалённый #2616 шаг 2/3 admin.rotate_proxy_ip /
Settings.avito_proxy_rotate_url), не ASocks.
Без явной проверки хоста наш `Authorization: Bearer <ASOCKS_API_TOKEN>` ушёл бы
на ЧУЖОЙ провайдер (mobileproxy) — плюс сам GET/POST по их changeip, вероятно,
реально ротирует ИХ IP и тратит ИХ суточный лимит, а мы бы записали это как
успех ASocks. rotate_proxy ПЕРЕД любым HTTP-вызовом проверяет
urlparse(rotate_url).hostname == ALLOWED_ROTATE_HOST (https-only) — несовпадение
это ОТКАЗ (ok=False, нейтральный reason), а НЕ попытка безголового запроса без
Authorization: смысл ручной ротации — конкретный провайдер (ASocks), молчаливый
вызов чужой ручки без авторизации — это сюрприз оператору (он думает "ASocks
ротировал", а фактически задел mobileproxy), которого проще не допустить, чем
потом объяснять админу расхождение счётчиков.

psycopg v3 / SQLAlchemy text(): все параметры через CAST(:x AS type), НЕ :x::type.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Any
from urllib.parse import urlparse

import httpx
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.config import settings
from app.services.proxy_pool import clear_source_bans

logger = logging.getLogger(__name__)

__all__ = [
    "ALLOWED_ROTATE_HOST",
    "DAILY_ROTATION_LIMIT",
    "RotationResult",
    "rotate_proxy",
]

# Лимит провайдера (ASocks, безлимитные порты): 3 ротации в сутки на порт (эмпирика).
DAILY_ROTATION_LIMIT = 3

# Таймаут POST refresh-ip. Пункт задачи требует "~30с".
_ROTATE_TIMEOUT_S = 30.0

# Единственный хост, на который разрешено уходить с ASOCKS_API_TOKEN в заголовке
# (см. "⛔ Хост-пиннинг" в docstring модуля). scrape_proxies.rotate_url может
# нести ЧУЖИЕ changeip-ссылки (mobileproxy и т.п.) — сравнение ДО HTTP-вызова.
ALLOWED_ROTATE_HOST = "api.asocks.com"


def _is_allowed_rotate_url(url: str) -> bool:
    """https-only + hostname точно ALLOWED_ROTATE_HOST (регистронезависимо —
    urlparse().hostname уже лоуеркейзит). Не бросает исключений на кривом url."""
    try:
        parsed = urlparse(url)
    except ValueError:
        return False
    return parsed.scheme == "https" and parsed.hostname == ALLOWED_ROTATE_HOST


@dataclass
class RotationResult:
    """Результат попытки ротации exit-IP одного прокси. reason — ВСЕГДА нейтральный
    (безопасен для HTTP-ответа клиенту), никогда не несёт токен/секреты."""

    ok: bool
    reason: str | None
    new_ip: str | None = None
    # Сколько quota-consuming попыток остаётся сегодня ПОСЛЕ этой попытки (см. модуль
    # docstring за определением quota-consuming). Для локально отклонённых попыток
    # (no rotate_url/no token) не относится к текущему прокси — просто текущий остаток.
    rotations_remaining_today: int = DAILY_ROTATION_LIMIT


def _quota_used_today(db: Session, proxy_id: int) -> int:
    """Число quota-consuming попыток за последние 24ч (см. docstring модуля).

    http_status IS NOT NULL AND != 401 — успех И любой не-401 ответ провайдера.
    401 (auth-отсев) и сетевые ошибки (http_status IS NULL) не считаются.
    """
    row = (
        db.execute(
            text(
                """
                SELECT count(*) AS n
                FROM scrape_proxy_rotations
                WHERE proxy_id = CAST(:proxy_id AS bigint)
                  AND rotated_at > now() - interval '24 hours'
                  AND http_status IS NOT NULL
                  AND http_status != 401
                """
            ),
            {"proxy_id": proxy_id},
        )
        .mappings()
        .fetchone()
    )
    return int(row["n"]) if row is not None else 0


def _record_attempt(
    db: Session,
    proxy_id: int,
    *,
    success: bool,
    http_status: int | None,
    note: str | None,
) -> None:
    """Записать попытку ротации в аудит-таблицу. Вызывается ТОЛЬКО когда HTTP-запрос
    к провайдеру реально был сделан (локально отклонённые попытки не пишутся —
    см. модуль docstring)."""
    db.execute(
        text(
            """
            INSERT INTO scrape_proxy_rotations (proxy_id, success, http_status, note)
            VALUES (
                CAST(:proxy_id AS bigint),
                CAST(:success AS boolean),
                CAST(:http_status AS integer),
                CAST(:note AS text)
            )
            """
        ),
        {"proxy_id": proxy_id, "success": success, "http_status": http_status, "note": note},
    )
    db.commit()


def _alert_stale_token(proxy_id: int) -> None:
    """Громкий отказ на 401: logger.error + событие в Sentry/GlitchTip (best-effort).

    401 значит, что провайдер отверг Authorization-заголовок — токен протух (issue
    #2600 п.5: "Токен — сессионный, однажды протухнет. Это осознанное решение
    владельца"). Молчаливая остановка ротации недопустима — операторы должны узнать
    об этом сразу, а не когда прокси уже забанены неделю.
    """
    logger.error(
        "proxy_rotation: ASocks REJECTED Authorization (401) for proxy_id=%d — "
        "ASOCKS_API_TOKEN likely EXPIRED, IP rotation is now BLOCKED for this proxy "
        "until the token is refreshed in web-cabinet + env",
        proxy_id,
    )
    try:
        import sentry_sdk

        sentry_sdk.capture_message(
            f"ASocks rotation token rejected (401) for proxy_id={proxy_id} — "
            "ASOCKS_API_TOKEN expired, IP rotation blocked until refreshed",
            level="error",
        )
    except Exception:
        pass  # sentry_sdk not initialised in dev — best-effort only


def _extract_new_ip(resp: httpx.Response) -> str | None:
    """Best-effort вытащить новый exit-IP из ответа провайдера. Формат ответа
    refresh-ip для безлимитных портов ASocks не документирован (issue #2600 п.5) —
    парсинг заведомо defensive, неудача не является ошибкой ротации."""
    try:
        data: Any = resp.json()
    except Exception:
        return None
    if not isinstance(data, dict):
        return None
    for key in ("new_ip", "ip", "exit_ip"):
        val = data.get(key)
        if val:
            return str(val)
    nested = data.get("data")
    if isinstance(nested, dict):
        for key in ("new_ip", "ip", "exit_ip"):
            val = nested.get(key)
            if val:
                return str(val)
    return None


async def rotate_proxy(db: Session, proxy_id: int) -> RotationResult:
    """Сменить exit-IP одного прокси пула через ASocks refresh-ip (#2600 п.5).

    Порядок:
      1. proxy_id не найден в scrape_proxies → ok=False, reason нейтральный.
      2. rotate_url пусто → ok=False, "ротация не поддерживается" (НЕ ошибка).
      3. rotate_url хост != ALLOWED_ROTATE_HOST (https://api.asocks.com) → ok=False
         ДО HTTP-вызова — токен не должен уйти на чужой провайдер (mobileproxy
         changeip и т.п. в этой же колонке пула, см. "⛔ Хост-пиннинг" в модуле).
      4. ASOCKS_API_TOKEN не задан (settings.asocks_api_token) → ok=False,
         внятный отказ, ничего не ломается.
      5. Суточный лимит (см. _quota_used_today) исчерпан → ok=False, отказ БЕЗ
         обращения к API.
      6. POST rotate_url с Authorization: Bearer <token>, timeout ~30с.
         - Сетевая ошибка (нет ответа) → ok=False, аудит-запись http_status=NULL
           (НЕ считается в лимите), нейтральный reason, детали в лог exc_info=True.
         - 401 → громкий отказ (_alert_stale_token) + аудит-запись (НЕ считается
           в лимите), нейтральный reason.
         - Другой 4xx/5xx → аудит-запись (считается в лимите — провайдер прошёл
           auth и ответил своей бизнес-логикой), нейтральный reason.
         - 2xx → аудит-запись success=True (считается в лимите), new_ip best-effort.

    Ни в одном из reason/логов НЕ появляется токен.
    """
    row = (
        db.execute(
            text("SELECT id, rotate_url FROM scrape_proxies WHERE id = CAST(:id AS bigint)"),
            {"id": proxy_id},
        )
        .mappings()
        .fetchone()
    )
    if row is None:
        return RotationResult(ok=False, reason="proxy not found")

    rotate_url = row["rotate_url"]
    if not rotate_url:
        logger.info(
            "proxy_rotation: proxy_id=%d has no rotate_url — rotation not supported", proxy_id
        )
        return RotationResult(
            ok=False, reason="rotation not supported for this proxy (no rotate_url configured)"
        )

    if not _is_allowed_rotate_url(rotate_url):
        # scrape_proxies.rotate_url колонка неоднородна (другие строки пула несут
        # mobileproxy changeip-ссылки с ИХ секретом) — отправлять наш
        # Authorization: Bearer <ASOCKS_API_TOKEN> на непроверенный хост нельзя.
        # Логируем ТОЛЬКО hostname (не полный url — на других провайдерах он
        # несёт их собственный секрет в query-string, тот же класс утечки, что
        # и в rotate_proxy_ip, см. модуль docstring).
        logger.warning(
            "proxy_rotation: proxy_id=%d rotate_url host=%r is not the allowed ASocks host "
            "(%s) — refusing before any HTTP call to avoid leaking the token to it",
            proxy_id,
            urlparse(rotate_url).hostname,
            ALLOWED_ROTATE_HOST,
        )
        return RotationResult(
            ok=False, reason="rotation not supported for this proxy (unexpected rotate host)"
        )

    token = settings.asocks_api_token
    if not token:
        logger.warning(
            "proxy_rotation: ASOCKS_API_TOKEN not configured — proxy_id=%d rotation skipped",
            proxy_id,
        )
        return RotationResult(ok=False, reason="rotation not configured (missing API token)")

    used = _quota_used_today(db, proxy_id)
    if used >= DAILY_ROTATION_LIMIT:
        logger.warning(
            "proxy_rotation: daily limit reached proxy_id=%d used=%d/%d — skipping API call",
            proxy_id,
            used,
            DAILY_ROTATION_LIMIT,
        )
        return RotationResult(
            ok=False,
            reason=f"daily rotation limit reached ({DAILY_ROTATION_LIMIT}/day)",
            rotations_remaining_today=0,
        )

    try:
        async with httpx.AsyncClient(timeout=_ROTATE_TIMEOUT_S) as client:
            resp = await client.post(rotate_url, headers={"Authorization": f"Bearer {token}"})
    except Exception as exc:
        # Ответа не было вообще — не подтверждено, что запрос дошёл до провайдера,
        # значит квота НЕ тратится. str(exc) НИКОГДА не идёт наружу (может нести
        # служебные детали соединения) — только exc_info=True в лог. type(exc).__name__
        # секрета не несёт (это имя класса — ConnectError/ReadTimeout/…) и в note
        # ПОЛЕЗЕН оператору: отличить "не дозвонились" от "дозвонились, зависли".
        logger.warning(
            "proxy_rotation: request failed (no response) proxy_id=%d", proxy_id, exc_info=True
        )
        _record_attempt(
            db,
            proxy_id,
            success=False,
            http_status=None,
            note=f"request failed: {type(exc).__name__}",
        )
        return RotationResult(
            ok=False,
            reason="rotation request failed (network error)",
            rotations_remaining_today=max(0, DAILY_ROTATION_LIMIT - used),
        )

    status = resp.status_code

    if status == 401:
        _alert_stale_token(proxy_id)
        _record_attempt(
            db,
            proxy_id,
            success=False,
            http_status=401,
            note="unauthenticated — token expired/invalid (excluded from daily quota)",
        )
        return RotationResult(
            ok=False,
            reason="rotation service rejected credentials — alerted, contact operator",
            rotations_remaining_today=max(0, DAILY_ROTATION_LIMIT - used),
        )

    if status >= 400:
        logger.warning(
            "proxy_rotation: provider returned error proxy_id=%d status=%d", proxy_id, status
        )
        _record_attempt(
            db, proxy_id, success=False, http_status=status, note="provider returned error"
        )
        return RotationResult(
            ok=False,
            reason=f"rotation request failed (provider status {status})",
            rotations_remaining_today=max(0, DAILY_ROTATION_LIMIT - (used + 1)),
        )

    new_ip = _extract_new_ip(resp)
    logger.info("proxy_rotation: rotated proxy_id=%d status=%d new_ip=%s", proxy_id, status, new_ip)
    _record_attempt(db, proxy_id, success=True, http_status=status, note=None)
    # Площадки банили СТАРЫЙ exit-IP, а строка бана привязана к proxy_id (#2600 п.2) —
    # после смены адреса она держала бы узел вне выдачи уже без причины, вплоть до 72ч
    # при эскалации. Ротация прошла → история банов этого узла недействительна.
    clear_source_bans(db, proxy_id, reason=f"exit ip rotated (status={status})")
    return RotationResult(
        ok=True,
        reason=None,
        new_ip=new_ip,
        rotations_remaining_today=max(0, DAILY_ROTATION_LIMIT - (used + 1)),
    )