funpay_proxy_rotator.py

from __future__ import annotations

import html
import json
import logging
import threading
import time
import uuid as uuid_lib
from pathlib import Path
from typing import TYPE_CHECKING, Any
from urllib.parse import quote, urlparse

import requests
import telebot
from telebot.types import InlineKeyboardButton as B
from telebot.types import InlineKeyboardMarkup as K

from tg_bot import CBT

try:
    from Utils import cardinal_tools
except Exception:  # pragma: no cover
    cardinal_tools = None

if TYPE_CHECKING:
    from cardinal import Cardinal


NAME = "FunPay Proxy Rotator"
VERSION = "0.1.1"
DESCRIPTION = "Автоматическая ротация прокси с проверкой валидности через PX6 и кнопочным управлением."
CREDITS = "@takouq, @llzzvvww"
UUID = "7deeb978-2e33-4d50-b087-4d68d8804d89"
SETTINGS_PAGE = True

BIND_TO_DELETE = []

logger = logging.getLogger("FPC.funpay_proxy_rotator")

S_DIR = Path(f"storage/plugins/{UUID}")
SETTINGS_FILE = S_DIR / "settings.json"
STATE_FILE = S_DIR / "state.json"

CBT_OPEN = f"{CBT.PLUGIN_SETTINGS}:{UUID}"


def _cb(action: str) -> str:
    return f"proxyrot:{action}"


CBT_BACK = _cb("back")
CBT_ADD = _cb("add")
CBT_ADD_BULK = _cb("add_bulk")
CBT_ADD_FORCE = _cb("add_force")
CBT_ADD_CANCEL = _cb("add_cancel")
CBT_LIST = _cb("list")
CBT_SELECT = _cb("select")
CBT_DELETE = _cb("delete")
CBT_ROTATE_NOW = _cb("rotate_now")
CBT_CHECK_ALL = _cb("check_all")
CBT_TOGGLE = _cb("toggle")
CBT_SET_KEY = _cb("set_key")
CBT_INTERVAL_10 = _cb("int.600")
CBT_INTERVAL_15 = _cb("int.900")
CBT_INTERVAL_30 = _cb("int.1800")
CBT_INTERVAL_60 = _cb("int.3600")

INPUT_ADD_PROXY = "add_proxy"
INPUT_ADD_PROXY_BULK = "add_proxy_bulk"
INPUT_SET_KEY = "set_key"
INPUT_SELECT_PROXY = "select_proxy"
INPUT_DELETE_PROXY = "delete_proxy"

DEFAULT_SETTINGS = {
    "enabled": False,
    "rotation_interval_sec": 900,
    "check_interval_sec": 15,
    "validate_before_rotate": True,
    "px6_api_key": "",
    "notify_chat_id": 0,
    "proxies": [],
}

DEFAULT_STATE = {
    "current_proxy_id": "",
    "last_rotate_ts": 0.0,
    "next_rotate_ts": 0.0,
    "last_check_ts": 0.0,
    "last_error": "",
}


class RT:
    def __init__(self) -> None:
        self.lock = threading.RLock()
        self.settings: dict[str, Any] = {}
        self.state: dict[str, Any] = {}
        self.loop_started = False
        self.tg_registered = False
        self.pending_inputs: dict[tuple[int, int], dict[str, Any]] = {}
        self.pending_invalid: dict[tuple[int, int], dict[str, Any]] = {}


R = RT()


def _ensure_paths() -> None:
    S_DIR.mkdir(parents=True, exist_ok=True)


def _jcopy(v: Any) -> Any:
    return json.loads(json.dumps(v))


def _settings() -> dict[str, Any]:
    return R.settings


def _state() -> dict[str, Any]:
    return R.state


def _as_bool(v: Any, default: bool = False) -> bool:
    if isinstance(v, bool):
        return v
    if isinstance(v, (int, float)):
        return bool(int(v))
    s = str(v or "").strip().lower()
    if s in {"1", "true", "yes", "on", "y", "да"}:
        return True
    if s in {"0", "false", "no", "off", "n", "нет"}:
        return False
    return default


def _as_int(v: Any, default: int = 0) -> int:
    try:
        return int(float(str(v).strip()))
    except Exception:
        return default


def _load_settings(force: bool = False) -> None:
    with R.lock:
        if R.settings and not force:
            return
        _ensure_paths()
        data = _jcopy(DEFAULT_SETTINGS)
        if SETTINGS_FILE.exists():
            try:
                raw = json.loads(SETTINGS_FILE.read_text(encoding="utf-8"))
                if isinstance(raw, dict):
                    data.update(raw)
            except Exception:
                logger.exception("proxy_rotator: settings load failed")
        if not isinstance(data.get("proxies"), list):
            data["proxies"] = []
        R.settings = data
        _save_settings()


def _save_settings() -> None:
    with R.lock:
        _ensure_paths()
        SETTINGS_FILE.write_text(json.dumps(_settings(), ensure_ascii=False, indent=2), encoding="utf-8")


def _load_state() -> None:
    with R.lock:
        _ensure_paths()
        data = _jcopy(DEFAULT_STATE)
        if STATE_FILE.exists():
            try:
                raw = json.loads(STATE_FILE.read_text(encoding="utf-8"))
                if isinstance(raw, dict):
                    data.update(raw)
            except Exception:
                logger.exception("proxy_rotator: state load failed")
        R.state = data
        _save_state()


def _save_state() -> None:
    with R.lock:
        _ensure_paths()
        STATE_FILE.write_text(json.dumps(_state(), ensure_ascii=False, indent=2), encoding="utf-8")


def _key(chat_id: int, user_id: int) -> tuple[int, int]:
    return int(chat_id), int(user_id)


def _set_pending(chat_id: int, user_id: int, mode: str, **extra: Any) -> None:
    with R.lock:
        R.pending_inputs[_key(chat_id, user_id)] = {"mode": mode, **extra}


def _pop_pending(chat_id: int, user_id: int) -> dict[str, Any] | None:
    with R.lock:
        return R.pending_inputs.pop(_key(chat_id, user_id), None)


def _get_pending(chat_id: int, user_id: int) -> dict[str, Any] | None:
    with R.lock:
        return R.pending_inputs.get(_key(chat_id, user_id))


def _set_pending_invalid(chat_id: int, user_id: int, payload: Any) -> None:
    with R.lock:
        R.pending_invalid[_key(chat_id, user_id)] = payload


def _pop_pending_invalid(chat_id: int, user_id: int) -> dict[str, Any] | None:
    with R.lock:
        return R.pending_invalid.pop(_key(chat_id, user_id), None)


def _clear_pending(chat_id: int, user_id: int) -> None:
    with R.lock:
        R.pending_inputs.pop(_key(chat_id, user_id), None)
        R.pending_invalid.pop(_key(chat_id, user_id), None)


def _mask_key(key: str) -> str:
    key = str(key or "").strip()
    if len(key) <= 8:
        return "*" * len(key)
    return key[:4] + "…" + key[-4:]


def _seconds_human(sec: int) -> str:
    sec = max(0, int(sec))
    if sec % 3600 == 0 and sec >= 3600:
        return f"{sec // 3600}h"
    if sec % 60 == 0 and sec >= 60:
        return f"{sec // 60}m"
    return f"{sec}s"


def _proxy_list() -> list[dict[str, Any]]:
    raw = _settings().get("proxies", [])
    return raw if isinstance(raw, list) else []


def _has_proxy(proxy_url: str) -> bool:
    proxy_url = str(proxy_url or "").strip()
    return any(str(item.get("proxy_url") or "").strip() == proxy_url for item in _proxy_list())


def _find_proxy(proxy_id: str) -> dict[str, Any] | None:
    for item in _proxy_list():
        if str(item.get("id") or "") == proxy_id:
            return item
    return None


def _current_proxy() -> dict[str, Any] | None:
    current_id = str(_state().get("current_proxy_id") or "").strip()
    if not current_id:
        return None
    return _find_proxy(current_id)


def _normalize_proxy_input(text: str) -> dict[str, Any]:
    raw = str(text or "").strip()
    if not raw:
        raise ValueError("empty proxy")

    scheme = "http"
    host = ""
    port = ""
    user = ""
    password = ""

    if "://" in raw:
        parsed = urlparse(raw)
        if not parsed.hostname or not parsed.port:
            raise ValueError("bad proxy url")
        scheme = parsed.scheme or "http"
        host = parsed.hostname
        port = str(parsed.port)
        user = parsed.username or ""
        password = parsed.password or ""
    elif "@" in raw:
        auth_part, host_part = raw.rsplit("@", 1)
        auth_parts = auth_part.split(":", 1)
        host_parts = host_part.split(":")
        if len(auth_parts) != 2 or len(host_parts) != 2:
            raise ValueError("expected user:pass@host:port")
        user, password = auth_parts[0].strip(), auth_parts[1].strip()
        host, port = host_parts[0].strip(), host_parts[1].strip()
    else:
        parts = [x.strip() for x in raw.split(":")]
        if len(parts) == 2:
            host, port = parts
        elif len(parts) == 4:
            host, port, user, password = parts
        else:
            raise ValueError("expected ip:port or ip:port:user:pass")

    if not host or not port:
        raise ValueError("host or port is empty")

    port_int = _as_int(port, -1)
    if port_int <= 0 or port_int > 65535:
        raise ValueError("bad port")

    if user or password:
        proxy_raw = f"{host}:{port_int}:{user}:{password}"
        proxy_url = f"{scheme}://{quote(user)}:{quote(password)}@{host}:{port_int}"
    else:
        proxy_raw = f"{host}:{port_int}"
        proxy_url = f"{scheme}://{host}:{port_int}"

    return {
        "id": str(uuid_lib.uuid4()),
        "label": f"{host}:{port_int}",
        "proxy_raw": proxy_raw,
        "proxy_url": proxy_url,
        "enabled": True,
        "last_valid": None,
        "last_check_ts": 0.0,
        "last_error": "",
        "px6_proxy_id": 0,
        "added_force": False,
        "added_ts": time.time(),
    }


def _px6_check(entry: dict[str, Any]) -> tuple[bool, str]:
    api_key = str(_settings().get("px6_api_key") or "").strip()
    if not api_key:
        return False, "PX6 API key is not set"

    url = f"https://px6.link/api/{quote(api_key, safe='')}/check"
    try:
        resp = requests.get(url, params={"proxy": str(entry.get("proxy_raw") or "")}, timeout=20)
        resp.raise_for_status()
        data = resp.json()
    except Exception as exc:
        entry["last_valid"] = False
        entry["last_check_ts"] = time.time()
        entry["last_error"] = str(exc)
        return False, str(exc)

    ok = str(data.get("status") or "").lower() == "yes" and bool(data.get("proxy_status")) is True
    entry["last_valid"] = bool(ok)
    entry["last_check_ts"] = time.time()
    entry["last_error"] = "" if ok else str(data)
    entry["px6_proxy_id"] = _as_int(data.get("proxy_id"), 0)
    return bool(ok), entry["last_error"] or "ok"


def _proxy_dict(proxy_url: str) -> dict[str, str]:
    return {"http": proxy_url, "https": proxy_url}


def _apply_proxy(cardinal: Any, entry: dict[str, Any]) -> tuple[bool, str]:
    proxy_url = str(entry.get("proxy_url") or "").strip()
    if not proxy_url:
        return False, "empty proxy url"

    proxy_dict = _proxy_dict(proxy_url)

    try:
        cardinal.proxy = dict(proxy_dict)
    except Exception:
        pass

    try:
        acc = getattr(cardinal, "account", None)
        if acc is not None:
            try:
                setattr(acc, "proxy", dict(proxy_dict))
            except Exception:
                pass
            for attr in ("session", "requests_session", "http_session"):
                obj = getattr(acc, attr, None)
                if isinstance(obj, requests.Session):
                    obj.proxies.clear()
                    obj.proxies.update(proxy_dict)
    except Exception:
        logger.debug("proxy_rotator: account proxy apply failed", exc_info=True)

    try:
        cfg = getattr(cardinal, "MAIN_CFG", None)
        if cfg is not None:
            if not cfg.has_section("Proxy"):
                cfg.add_section("Proxy")
            cfg["Proxy"]["enable"] = "1"
            cfg["Proxy"]["proxy"] = proxy_url
            cfg["Proxy"]["check"] = "0"
    except Exception:
        logger.debug("proxy_rotator: MAIN_CFG proxy update failed", exc_info=True)

    try:
        if cardinal_tools is not None and hasattr(cardinal, "proxy_dict"):
            existing = getattr(cardinal, "proxy_dict", {}) or {}
            if proxy_url not in existing.values():
                next_id = max(existing.keys(), default=-1) + 1
                existing[next_id] = proxy_url
                setattr(cardinal, "proxy_dict", existing)
                cardinal_tools.cache_proxy_dict(existing)
    except Exception:
        logger.debug("proxy_rotator: proxy dict cache update failed", exc_info=True)

    _state()["current_proxy_id"] = str(entry.get("id") or "")
    _state()["last_rotate_ts"] = time.time()
    _state()["next_rotate_ts"] = _state()["last_rotate_ts"] + max(60, _as_int(_settings().get("rotation_interval_sec"), 900))
    _state()["last_error"] = ""
    _save_state()
    return True, "ok"


def _render_panel() -> str:
    _load_settings(False)
    _load_state()
    proxies = _proxy_list()
    valid_count = sum(1 for x in proxies if x.get("last_valid") is True)
    current = _current_proxy()
    next_ts = float(_state().get("next_rotate_ts") or 0.0)
    next_in = max(0, int(next_ts - time.time())) if next_ts > 0 else 0
    lines = [
        "<b>Proxy Rotator</b>",
        "",
        f"Разработчики: <code>{html.escape(CREDITS)}</code>",
        f"Авто-ротация: <b>{'включена' if _as_bool(_settings().get('enabled'), False) else 'выключена'}</b>",
        f"Интервал: <code>{_seconds_human(_as_int(_settings().get('rotation_interval_sec'), 900))}</code>",
        f"PX6 API key: <code>{html.escape(_mask_key(str(_settings().get('px6_api_key') or '')) or '-')}</code>",
        f"Всего прокси: <code>{len(proxies)}</code>",
        f"Валидных: <code>{valid_count}</code>",
        f"Текущий: <code>{html.escape(str(current.get('label') if current else '-'))}</code>",
        f"Следующая ротация: <code>{_seconds_human(next_in) if next_in else '-'}</code>",
    ]
    last_error = str(_state().get("last_error") or "").strip()
    if last_error:
        lines.extend(["", f"Последняя ошибка: <code>{html.escape(last_error[:180])}</code>"])
    lines.extend(
        [
            "",
            "Важно: встроенную прокси-проверку Cardinal на старте лучше отключить,",
            "иначе битый стартовый прокси может уронить бот до загрузки плагина.",
        ]
    )
    return "\n".join(lines)


def _panel_kb() -> K:
    kb = K(row_width=2)
    kb.row(B("➕ Добавить прокси", callback_data=CBT_ADD), B("📋 Список", callback_data=CBT_LIST))
    kb.row(B("📥 Массовый залив", callback_data=CBT_ADD_BULK))
    kb.row(B("✅ Проверить все", callback_data=CBT_CHECK_ALL), B("🔄 Ротация сейчас", callback_data=CBT_ROTATE_NOW))
    kb.row(
        B("🟢 Авто: ON" if _as_bool(_settings().get("enabled"), False) else "⏸ Авто: OFF", callback_data=CBT_TOGGLE),
        B("🔑 PX6 API", callback_data=CBT_SET_KEY),
    )
    kb.row(B("⏱ 10м", callback_data=CBT_INTERVAL_10), B("⏱ 15м", callback_data=CBT_INTERVAL_15))
    kb.row(B("⏱ 30м", callback_data=CBT_INTERVAL_30), B("⏱ 60м", callback_data=CBT_INTERVAL_60))
    kb.row(B("Прокси со скидкой - 5%", url="https://px6.me/?r=936465"))
    kb.row(B("Промокод на прокси px6me 10%", url="https://px6.me/a/936465"))
    return kb


def _render_list() -> str:
    proxies = _proxy_list()
    current_id = str(_state().get("current_proxy_id") or "")
    lines = ["<b>Список прокси</b>", ""]
    if not proxies:
        lines.append("Список пуст.")
    else:
        for idx, item in enumerate(proxies, start=1):
            is_current = str(item.get("id") or "") == current_id
            valid = item.get("last_valid")
            mark = "🎯" if is_current else "•"
            state_mark = "✅" if valid is True else ("❌" if valid is False else "❔")
            forced = " (force)" if _as_bool(item.get("added_force"), False) else ""
            lines.append(f"{mark} <b>{idx}.</b> <code>{html.escape(str(item.get('label') or '-'))}</code> {state_mark}{forced}")
    lines.extend(
        [
            "",
            "Кнопки ниже:",
            "• выбрать текущий прокси",
            "• удалить прокси",
        ]
    )
    return "\n".join(lines)


def _list_kb() -> K:
    kb = K(row_width=2)
    kb.row(B("🎯 Выбрать", callback_data=CBT_SELECT), B("🗑 Удалить", callback_data=CBT_DELETE))
    kb.row(B("⬅️ Назад", callback_data=CBT_BACK))
    return kb


def _invalid_add_kb() -> K:
    kb = K(row_width=2)
    kb.row(B("✅ Добавить всё равно", callback_data=CBT_ADD_FORCE), B("❌ Не добавлять", callback_data=CBT_ADD_CANCEL))
    kb.row(B("⬅️ Назад", callback_data=CBT_BACK))
    return kb


def _add_entries(entries: list[dict[str, Any]], cardinal: Any | None = None) -> int:
    added = 0
    first_added: dict[str, Any] | None = None
    for entry in entries:
        if _has_proxy(str(entry.get("proxy_url") or "")):
            continue
        _proxy_list().append(entry)
        added += 1
        if first_added is None:
            first_added = entry
    if added:
        _save_settings()
        if cardinal is not None and not _state().get("current_proxy_id") and first_added is not None:
            _apply_proxy(cardinal, first_added)
    return added


def _notify_chat_id(cardinal: Any) -> int:
    chat_id = _as_int(_settings().get("notify_chat_id"), 0)
    if chat_id > 0:
        return chat_id
    tg = getattr(cardinal, "telegram", None)
    auth = getattr(tg, "authorized_users", []) if tg is not None else []
    if isinstance(auth, (list, tuple, set)):
        for item in auth:
            val = _as_int(item, 0)
            if val > 0:
                return val
    return 0


def _tg_send(cardinal: Any, text: str, chat_id: int | None = None, markup: Any | None = None) -> None:
    tg = getattr(cardinal, "telegram", None)
    if tg is None:
        return
    cid = int(chat_id or _notify_chat_id(cardinal) or 0)
    if cid <= 0:
        return
    tg.bot.send_message(cid, text, reply_markup=markup, parse_mode="HTML", disable_web_page_preview=True)


def _tg_edit(cardinal: Any, call: Any, text: str, markup: Any | None = None) -> None:
    try:
        cardinal.telegram.bot.edit_message_text(
            text,
            call.message.chat.id,
            call.message.id,
            reply_markup=markup,
            parse_mode="HTML",
            disable_web_page_preview=True,
        )
    except Exception:
        _tg_send(cardinal, text, chat_id=call.message.chat.id, markup=markup)


def _open_panel(cardinal: Any, chat_id: int, user_id: int, message_id: int | None = None) -> None:
    _clear_pending(chat_id, user_id)
    text = _render_panel()
    markup = _panel_kb()
    if message_id:
        try:
            cardinal.telegram.bot.edit_message_text(
                text,
                chat_id,
                message_id,
                reply_markup=markup,
                parse_mode="HTML",
                disable_web_page_preview=True,
            )
            return
        except Exception:
            pass
    _tg_send(cardinal, text, chat_id=chat_id, markup=markup)


def _check_entry(entry: dict[str, Any]) -> tuple[bool, str]:
    ok, msg = _px6_check(entry)
    _save_settings()
    return ok, msg


def _rotate_to_next(cardinal: Any, *, chat_id: int | None = None, manual: bool = False) -> tuple[bool, str]:
    proxies = _proxy_list()
    if not proxies:
        _state()["last_error"] = "proxy list is empty"
        _save_state()
        return False, "Список прокси пуст."

    current_id = str(_state().get("current_proxy_id") or "")
    start_index = -1
    for idx, item in enumerate(proxies):
        if str(item.get("id") or "") == current_id:
            start_index = idx
            break

    ordered = proxies[start_index + 1 :] + proxies[: start_index + 1]
    if not ordered:
        ordered = proxies[:]

    for entry in ordered:
        if not _as_bool(entry.get("enabled"), True):
            continue
        if _as_bool(_settings().get("validate_before_rotate"), True):
            ok, msg = _check_entry(entry)
            if not ok:
                logger.warning("proxy_rotator: skip invalid proxy %s (%s)", entry.get("label"), msg)
                continue
        ok_apply, msg_apply = _apply_proxy(cardinal, entry)
        if ok_apply:
            text = f"Прокси переключен на <code>{html.escape(str(entry.get('label') or '-'))}</code>."
            if manual and chat_id:
                _tg_send(cardinal, text, chat_id=chat_id)
            return True, text

    _state()["last_error"] = "no valid proxy available"
    _save_state()
    msg = "Не удалось найти рабочий прокси для переключения."
    if manual and chat_id:
        _tg_send(cardinal, msg, chat_id=chat_id)
    return False, msg


def _loop(cardinal: Any) -> None:
    while True:
        try:
            _load_settings(False)
            _load_state()
            if not _as_bool(_settings().get("enabled"), False):
                time.sleep(max(5, _as_int(_settings().get("check_interval_sec"), 15)))
                continue

            interval = max(60, _as_int(_settings().get("rotation_interval_sec"), 900))
            now_ts = time.time()
            next_rotate_ts = float(_state().get("next_rotate_ts") or 0.0)
            if next_rotate_ts <= 0:
                _state()["next_rotate_ts"] = now_ts + interval
                _save_state()
            elif now_ts >= next_rotate_ts:
                ok, msg = _rotate_to_next(cardinal, chat_id=_notify_chat_id(cardinal), manual=False)
                if not ok:
                    _tg_send(cardinal, f"[proxyrot] {html.escape(msg)}", chat_id=_notify_chat_id(cardinal))
                    _state()["next_rotate_ts"] = time.time() + min(interval, 300)
                    _save_state()
            time.sleep(max(5, _as_int(_settings().get("check_interval_sec"), 15)))
        except Exception:
            logger.exception("proxy_rotator: loop failed")
            time.sleep(10)


def _start_loop(cardinal: Any) -> None:
    with R.lock:
        if R.loop_started:
            return
        R.loop_started = True
    threading.Thread(target=lambda: _loop(cardinal), daemon=True).start()


def _toggle_enabled(cardinal: Any, chat_id: int) -> str:
    _settings()["enabled"] = not _as_bool(_settings().get("enabled"), False)
    _settings()["notify_chat_id"] = int(chat_id)
    if _settings()["enabled"]:
        interval = max(60, _as_int(_settings().get("rotation_interval_sec"), 900))
        _state()["next_rotate_ts"] = time.time() + interval
        _save_state()
    _save_settings()
    return "Авто-ротация включена." if _settings()["enabled"] else "Авто-ротация выключена."


def _set_interval(sec: int) -> None:
    _settings()["rotation_interval_sec"] = int(sec)
    _state()["next_rotate_ts"] = time.time() + int(sec)
    _save_settings()
    _save_state()


def _check_all_worker(cardinal: Any, chat_id: int) -> None:
    proxies = _proxy_list()
    if not proxies:
        _tg_send(cardinal, "Список прокси пуст.", chat_id=chat_id)
        return
    good = 0
    bad = 0
    for entry in proxies:
        ok, _ = _check_entry(entry)
        if ok:
            good += 1
        else:
            bad += 1
    _save_settings()
    _state()["last_check_ts"] = time.time()
    _save_state()
    _tg_send(cardinal, f"Проверка завершена.\nРабочих: <code>{good}</code>\nНерабочих: <code>{bad}</code>", chat_id=chat_id)


def _authorized(cardinal: Any, user_id: int) -> bool:
    tg = getattr(cardinal, "telegram", None)
    auth = getattr(tg, "authorized_users", []) if tg is not None else []
    return True if not auth else int(user_id) in set(auth)


def _handle_callback(cardinal: Any, call: Any) -> None:
    if not _authorized(cardinal, call.from_user.id):
        return
    try:
        cardinal.telegram.bot.answer_callback_query(call.id)
    except Exception:
        pass

    chat_id = int(call.message.chat.id)
    user_id = int(call.from_user.id)
    data = str(call.data or "")

    if data.startswith(CBT_OPEN) or data == CBT_BACK:
        _open_panel(cardinal, chat_id, user_id, call.message.id)
        return

    if data == CBT_LIST:
        _clear_pending(chat_id, user_id)
        _tg_edit(cardinal, call, _render_list(), _list_kb())
        return

    if data == CBT_ADD:
        _clear_pending(chat_id, user_id)
        if not str(_settings().get("px6_api_key") or "").strip():
            _tg_send(cardinal, "Сначала укажи PX6 API key через кнопку <b>🔑 PX6 API</b>.", chat_id=chat_id)
            _open_panel(cardinal, chat_id, user_id, call.message.id)
            return
        _set_pending(chat_id, user_id, INPUT_ADD_PROXY)
        _tg_send(cardinal, "Отправь прокси в формате <code>ip:port:user:pass</code> или <code>ip:port</code>.", chat_id=chat_id)
        return

    if data == CBT_ADD_BULK:
        _clear_pending(chat_id, user_id)
        if not str(_settings().get("px6_api_key") or "").strip():
            _tg_send(cardinal, "Сначала укажи PX6 API key через кнопку <b>🔑 PX6 API</b>.", chat_id=chat_id)
            _open_panel(cardinal, chat_id, user_id, call.message.id)
            return
        _set_pending(chat_id, user_id, INPUT_ADD_PROXY_BULK)
        _tg_send(
            cardinal,
            (
                "Отправь список прокси одним сообщением.\n"
                "Формат: <code>1 строка = 1 прокси</code>\n\n"
                "Поддерживается:\n"
                "• <code>ip:port:user:pass</code>\n"
                "• <code>ip:port</code>"
            ),
            chat_id=chat_id,
        )
        return

    if data == CBT_SET_KEY:
        _clear_pending(chat_id, user_id)
        _set_pending(chat_id, user_id, INPUT_SET_KEY)
        _tg_send(cardinal, "Отправь PX6 API key одним сообщением.", chat_id=chat_id)
        return

    if data == CBT_SELECT:
        _clear_pending(chat_id, user_id)
        _set_pending(chat_id, user_id, INPUT_SELECT_PROXY)
        _tg_send(cardinal, "Отправь номер прокси из списка, который нужно сделать текущим.", chat_id=chat_id)
        return

    if data == CBT_DELETE:
        _clear_pending(chat_id, user_id)
        _set_pending(chat_id, user_id, INPUT_DELETE_PROXY)
        _tg_send(cardinal, "Отправь номер прокси из списка, который нужно удалить.", chat_id=chat_id)
        return

    if data == CBT_TOGGLE:
        msg = _toggle_enabled(cardinal, chat_id)
        _tg_edit(cardinal, call, _render_panel(), _panel_kb())
        _tg_send(cardinal, msg, chat_id=chat_id)
        return

    if data in {CBT_INTERVAL_10, CBT_INTERVAL_15, CBT_INTERVAL_30, CBT_INTERVAL_60}:
        mapping = {
            CBT_INTERVAL_10: 600,
            CBT_INTERVAL_15: 900,
            CBT_INTERVAL_30: 1800,
            CBT_INTERVAL_60: 3600,
        }
        _set_interval(mapping[data])
        _tg_edit(cardinal, call, _render_panel(), _panel_kb())
        return

    if data == CBT_ROTATE_NOW:
        threading.Thread(target=lambda: _rotate_to_next(cardinal, chat_id=chat_id, manual=True), daemon=True).start()
        _tg_edit(cardinal, call, _render_panel(), _panel_kb())
        return

    if data == CBT_CHECK_ALL:
        threading.Thread(target=lambda: _check_all_worker(cardinal, chat_id), daemon=True).start()
        _tg_send(cardinal, "Запустил проверку всех прокси.", chat_id=chat_id)
        _tg_edit(cardinal, call, _render_panel(), _panel_kb())
        return

    if data == CBT_ADD_FORCE:
        payload = _pop_pending_invalid(chat_id, user_id)
        if payload is None:
            _open_panel(cardinal, chat_id, user_id, call.message.id)
            return
        if isinstance(payload, list):
            for entry in payload:
                entry["added_force"] = True
            added = _add_entries(payload, cardinal)
            _tg_send(cardinal, f"Принудительно добавлено прокси: <code>{added}</code>.", chat_id=chat_id)
        else:
            entry = payload
            entry["added_force"] = True
            _add_entries([entry], cardinal)
            _tg_send(cardinal, f"Прокси <code>{html.escape(str(entry.get('label') or '-'))}</code> добавлен принудительно.", chat_id=chat_id)
        _open_panel(cardinal, chat_id, user_id, call.message.id)
        return

    if data == CBT_ADD_CANCEL:
        _pop_pending_invalid(chat_id, user_id)
        _tg_send(cardinal, "Добавление отменено.", chat_id=chat_id)
        _open_panel(cardinal, chat_id, user_id, call.message.id)


def _handle_message(cardinal: Any, m: Any) -> None:
    if not _authorized(cardinal, m.from_user.id):
        return
    pending = _get_pending(m.chat.id, m.from_user.id)
    if pending is None:
        return

    mode = str(pending.get("mode") or "")
    text = str(m.text or "").strip()
    if not text:
        return

    if mode == INPUT_SET_KEY:
        _settings()["px6_api_key"] = text
        _settings()["notify_chat_id"] = int(m.chat.id)
        _save_settings()
        _pop_pending(m.chat.id, m.from_user.id)
        _tg_send(cardinal, "PX6 API key сохранён.", chat_id=m.chat.id)
        return

    if mode == INPUT_ADD_PROXY:
        try:
            entry = _normalize_proxy_input(text)
        except Exception as exc:
            _tg_send(cardinal, f"Не удалось разобрать прокси: <code>{html.escape(str(exc))}</code>", chat_id=m.chat.id)
            return
        if _has_proxy(str(entry.get("proxy_url") or "")):
            _pop_pending(m.chat.id, m.from_user.id)
            _tg_send(cardinal, f"Прокси <code>{html.escape(entry['label'])}</code> уже есть в списке.", chat_id=m.chat.id)
            return
        ok, msg = _check_entry(entry)
        _pop_pending(m.chat.id, m.from_user.id)
        if ok:
            _settings()["notify_chat_id"] = int(m.chat.id)
            _add_entries([entry], cardinal)
            _tg_send(cardinal, f"Прокси <code>{html.escape(entry['label'])}</code> добавлен.", chat_id=m.chat.id)
        else:
            _set_pending_invalid(m.chat.id, m.from_user.id, entry)
            _tg_send(
                cardinal,
                (
                    f"PX6 считает прокси <code>{html.escape(entry['label'])}</code> невалидным.\n"
                    f"Причина: <code>{html.escape(msg[:180])}</code>\n\n"
                    "Добавить его всё равно?"
                ),
                chat_id=m.chat.id,
                markup=_invalid_add_kb(),
            )
        return

    if mode == INPUT_ADD_PROXY_BULK:
        lines = [line.strip() for line in text.splitlines() if line.strip()]
        if not lines:
            _tg_send(cardinal, "Список пуст. Отправь хотя бы один прокси.", chat_id=m.chat.id)
            return

        parsed: list[dict[str, Any]] = []
        parse_errors: list[str] = []
        for idx, line in enumerate(lines, start=1):
            try:
                entry = _normalize_proxy_input(line)
                if _has_proxy(str(entry.get("proxy_url") or "")) or any(
                    str(x.get("proxy_url") or "") == str(entry.get("proxy_url") or "") for x in parsed
                ):
                    parse_errors.append(f"{idx}. дубль <code>{html.escape(str(entry.get('label') or '-'))}</code>")
                    continue
                parsed.append(entry)
            except Exception as exc:
                parse_errors.append(f"{idx}. <code>{html.escape(str(exc))}</code>")

        valid_entries: list[dict[str, Any]] = []
        invalid_entries: list[dict[str, Any]] = []
        for entry in parsed:
            ok, _msg = _check_entry(entry)
            if ok:
                valid_entries.append(entry)
            else:
                invalid_entries.append(entry)

        _pop_pending(m.chat.id, m.from_user.id)
        _settings()["notify_chat_id"] = int(m.chat.id)
        added = _add_entries(valid_entries, cardinal)

        summary = [
            "<b>Массовый залив завершён</b>",
            "",
            f"Строк получено: <code>{len(lines)}</code>",
            f"Добавлено: <code>{added}</code>",
            f"Невалидных: <code>{len(invalid_entries)}</code>",
            f"Пропущено: <code>{len(parse_errors)}</code>",
        ]
        if parse_errors:
            summary.extend(["", "<b>Что пропущено:</b>"])
            summary.extend(parse_errors[:12])

        if invalid_entries:
            _set_pending_invalid(m.chat.id, m.from_user.id, invalid_entries)
            labels = ", ".join(html.escape(str(x.get("label") or "-")) for x in invalid_entries[:6])
            if len(invalid_entries) > 6:
                labels += ", ..."
            summary.extend(
                [
                    "",
                    f"PX6 считает невалидными: <code>{labels}</code>",
                    "Добавить их всё равно?",
                ]
            )
            _tg_send(cardinal, "\n".join(summary), chat_id=m.chat.id, markup=_invalid_add_kb())
        else:
            _tg_send(cardinal, "\n".join(summary), chat_id=m.chat.id)
        return

    if mode == INPUT_SELECT_PROXY:
        idx = _as_int(text, 0)
        proxies = _proxy_list()
        if idx <= 0 or idx > len(proxies):
            _tg_send(cardinal, "Нет прокси с таким номером.", chat_id=m.chat.id)
            return
        entry = proxies[idx - 1]
        ok, msg = _check_entry(entry)
        if not ok:
            _tg_send(cardinal, f"Прокси не прошёл проверку PX6: <code>{html.escape(msg[:180])}</code>", chat_id=m.chat.id)
            return
        ok_apply, msg_apply = _apply_proxy(cardinal, entry)
        _pop_pending(m.chat.id, m.from_user.id)
        if ok_apply:
            _tg_send(cardinal, f"Текущий прокси: <code>{html.escape(entry['label'])}</code>", chat_id=m.chat.id)
        else:
            _tg_send(cardinal, f"Не удалось применить прокси: <code>{html.escape(msg_apply)}</code>", chat_id=m.chat.id)
        return

    if mode == INPUT_DELETE_PROXY:
        idx = _as_int(text, 0)
        proxies = _proxy_list()
        if idx <= 0 or idx > len(proxies):
            _tg_send(cardinal, "Нет прокси с таким номером.", chat_id=m.chat.id)
            return
        entry = proxies.pop(idx - 1)
        if str(_state().get("current_proxy_id") or "") == str(entry.get("id") or ""):
            _state()["current_proxy_id"] = ""
            _save_state()
        _save_settings()
        _pop_pending(m.chat.id, m.from_user.id)
        _tg_send(cardinal, f"Прокси <code>{html.escape(str(entry.get('label') or '-'))}</code> удалён.", chat_id=m.chat.id)


def _cmd_open(cardinal: Any, m: Any) -> None:
    if not _authorized(cardinal, m.from_user.id):
        return
    _settings()["notify_chat_id"] = int(m.chat.id)
    _save_settings()
    _open_panel(cardinal, m.chat.id, m.from_user.id)


def _register_tg_handlers(cardinal: Any) -> None:
    tg = getattr(cardinal, "telegram", None)
    if tg is None or R.tg_registered:
        return

    tg.cbq_handler(lambda c: _handle_callback(cardinal, c), lambda c: str(c.data or "").startswith(CBT_OPEN))
    tg.cbq_handler(lambda c: _handle_callback(cardinal, c), lambda c: str(c.data or "").startswith("proxyrot:"))
    tg.msg_handler(lambda m: _cmd_open(cardinal, m), commands=["proxyrot"])
    tg.msg_handler(lambda m: _handle_message(cardinal, m), func=lambda m: _get_pending(m.chat.id, m.from_user.id) is not None)
    try:
        cardinal.add_telegram_commands(UUID, [("proxyrot", "Proxy rotator", True)])
    except Exception:
        logger.debug("proxy_rotator: add_telegram_commands failed", exc_info=True)
    R.tg_registered = True


def init_plugin(cardinal: Any, *_args: Any, **_kwargs: Any) -> None:
    _load_settings()
    _load_state()
    _register_tg_handlers(cardinal)


def post_init(cardinal: Any, *_args: Any, **_kwargs: Any) -> None:
    _load_settings()
    _load_state()
    _register_tg_handlers(cardinal)
    _start_loop(cardinal)

    current = _current_proxy()
    if current is not None:
        try:
            _apply_proxy(cardinal, current)
        except Exception:
            logger.debug("proxy_rotator: initial proxy apply failed", exc_info=True)


BIND_TO_PRE_INIT = [init_plugin]
BIND_TO_POST_INIT = [post_init]
BIND_TO_NEW_MESSAGE = []
BIND_TO_NEW_ORDER = []
BIND_TO_ORDER_STATUS_CHANGED = []