589 lines
20 KiB
Python
589 lines
20 KiB
Python
"""Покупка и управление WireGuard-конфигами 3x-ui (лимит 10 при активной подписке)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import secrets
|
||
import string
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
from remnawave import RemnawaveSDK
|
||
|
||
from services.db import Database
|
||
from services.purchase import get_my_subscription_info
|
||
from services.remnawave_users import find_panel_user
|
||
from services.traffic import (
|
||
BASE_TRAFFIC_GB,
|
||
ensure_base_limit_bytes,
|
||
format_gb,
|
||
gb_to_bytes,
|
||
)
|
||
from services.xui_client import XuiClient, XuiError, build_subscription_link
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
MAX_WG_CONFIGS = 10
|
||
SETTING_ENABLED = "xui_wg_purchase_enabled"
|
||
SETTING_PRICE = "xui_wg_config_price"
|
||
DEFAULT_PRICE = 0.0
|
||
|
||
|
||
def _truthy(value: str) -> bool:
|
||
return (value or "").strip().lower() in ("1", "true", "yes", "on", "да")
|
||
|
||
|
||
def _as_utc(dt: datetime) -> datetime:
|
||
if dt.tzinfo is None:
|
||
return dt.replace(tzinfo=timezone.utc)
|
||
return dt.astimezone(timezone.utc)
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class WgPurchaseSettings:
|
||
enabled: bool = False
|
||
price: float = DEFAULT_PRICE
|
||
max_configs: int = MAX_WG_CONFIGS
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"enabled": self.enabled,
|
||
"price": self.price,
|
||
"price_label": _format_price(self.price),
|
||
"max_configs": self.max_configs,
|
||
}
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class WgActionResult:
|
||
ok: bool
|
||
message: str
|
||
balance: float = 0.0
|
||
config: dict[str, Any] | None = None
|
||
configs: list[dict[str, Any]] | None = None
|
||
|
||
|
||
def _format_price(price: float) -> str:
|
||
if price <= 0:
|
||
return "бесплатно"
|
||
if price == int(price):
|
||
return f"{int(price)} ₽"
|
||
return f"{price:.2f} ₽"
|
||
|
||
|
||
def load_wg_purchase_settings(db: Database) -> WgPurchaseSettings:
|
||
raw = db.get_settings_map([SETTING_ENABLED, SETTING_PRICE])
|
||
price_raw = (raw.get(SETTING_PRICE) or "").strip()
|
||
try:
|
||
price = float(price_raw) if price_raw else DEFAULT_PRICE
|
||
except ValueError:
|
||
price = DEFAULT_PRICE
|
||
return WgPurchaseSettings(
|
||
enabled=_truthy(raw.get(SETTING_ENABLED, "")),
|
||
price=max(0.0, price),
|
||
max_configs=MAX_WG_CONFIGS,
|
||
)
|
||
|
||
|
||
def save_wg_purchase_settings(
|
||
db: Database,
|
||
*,
|
||
enabled: bool | None = None,
|
||
price: float | None = None,
|
||
) -> WgPurchaseSettings:
|
||
data: dict[str, str] = {}
|
||
if enabled is not None:
|
||
data[SETTING_ENABLED] = "1" if enabled else "0"
|
||
if price is not None:
|
||
data[SETTING_PRICE] = str(max(0.0, float(price)))
|
||
if data:
|
||
db.set_settings_map(data)
|
||
return load_wg_purchase_settings(db)
|
||
|
||
|
||
def _client_from_server(row: dict[str, Any]) -> XuiClient:
|
||
return XuiClient(
|
||
str(row["api_url"]),
|
||
username=str(row.get("username") or ""),
|
||
password=str(row.get("password") or ""),
|
||
api_key=str(row.get("api_key") or ""),
|
||
verify_ssl=bool(row.get("verify_ssl", True)),
|
||
)
|
||
|
||
|
||
def _new_email(telegram_id: int) -> str:
|
||
suffix = "".join(secrets.choice(string.ascii_lowercase + string.digits) for _ in range(6))
|
||
return f"wg_{telegram_id}_{suffix}"
|
||
|
||
|
||
def serialize_wg_config(row: dict[str, Any], server: dict[str, Any] | None = None) -> dict[str, Any]:
|
||
expire = row.get("expire_at")
|
||
traffic_bytes = int(row.get("traffic_bytes") or 0)
|
||
if traffic_bytes <= 0:
|
||
traffic_bytes = int(gb_to_bytes(BASE_TRAFFIC_GB))
|
||
return {
|
||
"id": int(row["id"]),
|
||
"server_id": int(row["server_id"]),
|
||
"server_name": (server or {}).get("name") or row.get("server_name") or "",
|
||
"email": row.get("email") or "",
|
||
"sub_id": row.get("sub_id") or "",
|
||
"subscription_url": row.get("subscription_url") or "",
|
||
"remark": row.get("remark") or "",
|
||
"enabled": bool(row.get("enabled", True)),
|
||
"expire_at": expire.isoformat() if isinstance(expire, datetime) else expire,
|
||
"traffic_bytes": traffic_bytes,
|
||
"traffic_label": format_gb(traffic_bytes / (1024**3), digits=0),
|
||
"qr_url": f"/api/wg/configs/{int(row['id'])}/qr",
|
||
"created_at": row["created_at"].isoformat()
|
||
if isinstance(row.get("created_at"), datetime)
|
||
else row.get("created_at"),
|
||
}
|
||
|
||
|
||
async def _subscription_limits(
|
||
sdk: RemnawaveSDK,
|
||
telegram_id: int,
|
||
*,
|
||
login: str | None = None,
|
||
db: Database | None = None,
|
||
) -> tuple[datetime, int]:
|
||
"""Срок и лимит трафика как у подписки в кабинете (база ≥ 1000 ГБ)."""
|
||
info = await get_my_subscription_info(sdk, telegram_id, login=login, db=db)
|
||
if not info.expire_at:
|
||
raise RuntimeError("Нет срока подписки")
|
||
expire = _as_utc(info.expire_at)
|
||
panel = await find_panel_user(sdk, telegram_id, login=login)
|
||
if panel is not None:
|
||
traffic_bytes = int(ensure_base_limit_bytes(getattr(panel, "traffic_limit_bytes", 0)))
|
||
elif info.traffic:
|
||
traffic_bytes = int(ensure_base_limit_bytes(info.traffic.limit_bytes))
|
||
else:
|
||
traffic_bytes = int(gb_to_bytes(BASE_TRAFFIC_GB))
|
||
return expire, traffic_bytes
|
||
|
||
|
||
async def subscription_is_active(
|
||
sdk: RemnawaveSDK,
|
||
telegram_id: int,
|
||
*,
|
||
login: str | None = None,
|
||
db: Database | None = None,
|
||
) -> tuple[bool, str]:
|
||
info = await get_my_subscription_info(sdk, telegram_id, login=login, db=db)
|
||
if not info.has_subscription or not info.expire_at:
|
||
return False, "Нет активной подписки. Сначала купите тариф."
|
||
now = datetime.now(timezone.utc)
|
||
if _as_utc(info.expire_at) <= now:
|
||
return (
|
||
False,
|
||
"Подписка истекла. Продлите или купите тариф — после этого снова можно включать конфиги.",
|
||
)
|
||
if info.devices and info.devices.over_limit:
|
||
return False, "Подписка отключена из‑за превышения устройств."
|
||
return True, ""
|
||
|
||
|
||
def list_available_servers(db: Database) -> list[dict[str, Any]]:
|
||
rows = db.list_xui_servers(enabled_only=True)
|
||
out: list[dict[str, Any]] = []
|
||
for r in rows:
|
||
if r.get("inbound_id") is None:
|
||
continue
|
||
if not str(r.get("subscription_url") or "").strip():
|
||
continue
|
||
if not (str(r.get("api_key") or "").strip() or str(r.get("password") or "").strip()):
|
||
continue
|
||
out.append(
|
||
{
|
||
"id": int(r["id"]),
|
||
"name": r.get("name") or f"Server #{r['id']}",
|
||
"inbound_id": r.get("inbound_id"),
|
||
"inbound_remark": r.get("inbound_remark") or "",
|
||
}
|
||
)
|
||
return out
|
||
|
||
|
||
async def purchase_wg_config(
|
||
*,
|
||
db: Database,
|
||
sdk: RemnawaveSDK,
|
||
telegram_id: int,
|
||
server_id: int,
|
||
remark: str = "",
|
||
login: str | None = None,
|
||
) -> WgActionResult:
|
||
settings = load_wg_purchase_settings(db)
|
||
bot_user = db.ensure_user(telegram_id)
|
||
balance = bot_user.balance
|
||
login = login or bot_user.login
|
||
|
||
if not settings.enabled:
|
||
return WgActionResult(
|
||
ok=False,
|
||
message="Покупка WireGuard-конфигов временно отключена.",
|
||
balance=balance,
|
||
)
|
||
|
||
active, reason = await subscription_is_active(
|
||
sdk, telegram_id, login=login, db=db
|
||
)
|
||
if not active:
|
||
return WgActionResult(ok=False, message=reason, balance=balance)
|
||
|
||
count = db.count_xui_wg_configs(telegram_id)
|
||
if count >= settings.max_configs:
|
||
return WgActionResult(
|
||
ok=False,
|
||
message=f"Лимит: максимум {settings.max_configs} конфигов на аккаунт.",
|
||
balance=balance,
|
||
)
|
||
|
||
server = db.get_xui_server(server_id)
|
||
if not server or not server.get("enabled"):
|
||
return WgActionResult(ok=False, message="Сервер недоступен.", balance=balance)
|
||
inbound_id = server.get("inbound_id")
|
||
if inbound_id is None:
|
||
return WgActionResult(
|
||
ok=False,
|
||
message="У сервера не выбран WireGuard inbound.",
|
||
balance=balance,
|
||
)
|
||
sub_base = str(server.get("subscription_url") or "").strip()
|
||
if not sub_base:
|
||
return WgActionResult(
|
||
ok=False,
|
||
message="У сервера не задан URL подписки.",
|
||
balance=balance,
|
||
)
|
||
|
||
price = float(settings.price)
|
||
if price > 0:
|
||
ok, balance = db.try_charge(
|
||
telegram_id,
|
||
price,
|
||
kind="wg_config",
|
||
meta=f"wg:{server_id}",
|
||
)
|
||
if not ok:
|
||
return WgActionResult(
|
||
ok=False,
|
||
message=(
|
||
f"Недостаточно средств.\n"
|
||
f"Нужно: <b>{_format_price(price)}</b>\n"
|
||
f"Баланс: <b>{balance:.0f} ₽</b>"
|
||
),
|
||
balance=balance,
|
||
)
|
||
|
||
email = _new_email(telegram_id)
|
||
remark = (remark or "").strip()[:80] or f"WG {bot_user.display_name}"
|
||
try:
|
||
expire_at, traffic_bytes = await _subscription_limits(
|
||
sdk, telegram_id, login=login, db=db
|
||
)
|
||
expiry_ms = int(expire_at.timestamp() * 1000)
|
||
async with _client_from_server(server) as client:
|
||
created = await client.add_wireguard_client(
|
||
int(inbound_id),
|
||
email=email,
|
||
tg_id=telegram_id if telegram_id > 0 else 0,
|
||
comment=remark,
|
||
enable=True,
|
||
expiry_ms=expiry_ms,
|
||
total_gb=traffic_bytes, # в 3x-ui clients API totalGB — байты
|
||
)
|
||
sub_id = str(created["sub_id"])
|
||
sub_url = build_subscription_link(sub_base, sub_id)
|
||
row = db.create_xui_wg_config(
|
||
telegram_id=telegram_id,
|
||
server_id=server_id,
|
||
email=email,
|
||
sub_id=sub_id,
|
||
subscription_url=sub_url,
|
||
remark=remark,
|
||
enabled=True,
|
||
expire_at=expire_at,
|
||
traffic_bytes=traffic_bytes,
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.exception("WG config create failed")
|
||
if price > 0:
|
||
balance = db.add_balance(
|
||
telegram_id,
|
||
price,
|
||
kind="refund",
|
||
meta=f"wg_refund:{server_id}",
|
||
)
|
||
return WgActionResult(
|
||
ok=False,
|
||
message=f"Не удалось создать конфиг: {exc}",
|
||
balance=balance,
|
||
)
|
||
|
||
cfg = serialize_wg_config(row, server)
|
||
expire_str = expire_at.strftime("%d.%m.%Y %H:%M UTC")
|
||
return WgActionResult(
|
||
ok=True,
|
||
message=(
|
||
f"WireGuard-конфиг создан"
|
||
+ (f" (−{_format_price(price)})" if price > 0 else "")
|
||
+ f".\nДо: <b>{expire_str}</b>"
|
||
+ f"\nТрафик: <b>{format_gb(traffic_bytes / (1024**3), digits=0)}</b>"
|
||
+ f".\n🔗 <code>{sub_url}</code>"
|
||
),
|
||
balance=balance if price <= 0 else balance,
|
||
config=cfg,
|
||
)
|
||
|
||
|
||
async def set_wg_config_enabled(
|
||
*,
|
||
db: Database,
|
||
telegram_id: int,
|
||
config_id: int,
|
||
enable: bool,
|
||
sdk: RemnawaveSDK | None = None,
|
||
login: str | None = None,
|
||
) -> WgActionResult:
|
||
bot_user = db.ensure_user(telegram_id)
|
||
login = login or bot_user.login
|
||
row = db.get_xui_wg_config(config_id, telegram_id=telegram_id)
|
||
if not row:
|
||
return WgActionResult(ok=False, message="Конфиг не найден.", balance=bot_user.balance)
|
||
|
||
if enable:
|
||
if sdk is None:
|
||
return WgActionResult(
|
||
ok=False,
|
||
message="Не удалось проверить подписку.",
|
||
balance=bot_user.balance,
|
||
)
|
||
active, reason = await subscription_is_active(
|
||
sdk, telegram_id, login=login, db=db
|
||
)
|
||
if not active:
|
||
return WgActionResult(ok=False, message=reason, balance=bot_user.balance)
|
||
|
||
server = db.get_xui_server(int(row["server_id"]))
|
||
if not server or server.get("inbound_id") is None:
|
||
return WgActionResult(
|
||
ok=False,
|
||
message="Сервер конфига недоступен.",
|
||
balance=bot_user.balance,
|
||
)
|
||
|
||
try:
|
||
async with _client_from_server(server) as client:
|
||
await client.set_client_enabled(
|
||
int(server["inbound_id"]),
|
||
email=str(row["email"]),
|
||
enable=enable,
|
||
)
|
||
updated = db.set_xui_wg_config_enabled(config_id, telegram_id, enable)
|
||
except XuiError as exc:
|
||
return WgActionResult(
|
||
ok=False,
|
||
message=f"Панель: {exc}",
|
||
balance=bot_user.balance,
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.exception("WG toggle failed")
|
||
return WgActionResult(
|
||
ok=False,
|
||
message=str(exc),
|
||
balance=bot_user.balance,
|
||
)
|
||
|
||
return WgActionResult(
|
||
ok=True,
|
||
message="Конфиг включён." if enable else "Конфиг выключен.",
|
||
balance=bot_user.balance,
|
||
config=serialize_wg_config(updated or row, server) if updated or row else None,
|
||
)
|
||
|
||
|
||
async def _disable_wg_row(db: Database, row: dict[str, Any]) -> bool:
|
||
"""Выключает один WG-конфиг на панели и в БД. True если изменили."""
|
||
if not row.get("enabled"):
|
||
return False
|
||
server = db.get_xui_server(int(row["server_id"]))
|
||
if not server or server.get("inbound_id") is None:
|
||
db.set_xui_wg_config_enabled(int(row["id"]), int(row["telegram_id"]), False)
|
||
return True
|
||
try:
|
||
async with _client_from_server(server) as client:
|
||
await client.set_client_enabled(
|
||
int(server["inbound_id"]),
|
||
email=str(row["email"]),
|
||
enable=False,
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.warning(
|
||
"WG auto-disable panel failed config=%s: %s",
|
||
row.get("id"),
|
||
exc,
|
||
)
|
||
db.set_xui_wg_config_enabled(int(row["id"]), int(row["telegram_id"]), False)
|
||
return True
|
||
|
||
|
||
async def _enable_wg_row(db: Database, row: dict[str, Any]) -> bool:
|
||
if row.get("enabled"):
|
||
return False
|
||
server = db.get_xui_server(int(row["server_id"]))
|
||
if not server or server.get("inbound_id") is None:
|
||
return False
|
||
try:
|
||
async with _client_from_server(server) as client:
|
||
await client.set_client_enabled(
|
||
int(server["inbound_id"]),
|
||
email=str(row["email"]),
|
||
enable=True,
|
||
)
|
||
db.set_xui_wg_config_enabled(int(row["id"]), int(row["telegram_id"]), True)
|
||
return True
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.warning(
|
||
"WG auto-enable panel failed config=%s: %s",
|
||
row.get("id"),
|
||
exc,
|
||
)
|
||
return False
|
||
|
||
|
||
async def enforce_wg_configs_subscription(
|
||
*,
|
||
db: Database,
|
||
sdk: RemnawaveSDK,
|
||
telegram_id: int,
|
||
login: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""В реальном времени: нет подписки → выключить все WG."""
|
||
bot_user = db.ensure_user(telegram_id)
|
||
login = login or bot_user.login
|
||
active, reason = await subscription_is_active(
|
||
sdk, telegram_id, login=login, db=db
|
||
)
|
||
disabled = 0
|
||
if not active:
|
||
for row in db.list_xui_wg_configs(telegram_id):
|
||
if await _disable_wg_row(db, row):
|
||
disabled += 1
|
||
return {
|
||
"subscription_active": active,
|
||
"reason": "" if active else reason,
|
||
"disabled": disabled,
|
||
}
|
||
|
||
|
||
async def restore_wg_configs_after_renew(
|
||
*,
|
||
db: Database,
|
||
telegram_id: int,
|
||
) -> int:
|
||
"""После продления снова включает ранее отключённые WG-конфиги."""
|
||
rows = db.list_xui_wg_configs(telegram_id)
|
||
restored = 0
|
||
for row in rows:
|
||
if row.get("enabled"):
|
||
continue
|
||
if await _enable_wg_row(db, row):
|
||
restored += 1
|
||
return restored
|
||
|
||
|
||
async def sync_wg_configs_limits(
|
||
*,
|
||
db: Database,
|
||
telegram_id: int,
|
||
expire_at: datetime | None = None,
|
||
traffic_bytes: int | None = None,
|
||
restore_enabled: bool = False,
|
||
) -> int:
|
||
"""Обновляет срок/трафик всех WG-конфигов пользователя на панелях 3x-ui."""
|
||
if expire_at is None and traffic_bytes is None and not restore_enabled:
|
||
return 0
|
||
rows = db.list_xui_wg_configs(telegram_id)
|
||
updated = 0
|
||
expiry_ms = int(_as_utc(expire_at).timestamp() * 1000) if expire_at else None
|
||
for row in rows:
|
||
server = db.get_xui_server(int(row["server_id"]))
|
||
if not server or server.get("inbound_id") is None:
|
||
continue
|
||
try:
|
||
async with _client_from_server(server) as client:
|
||
current = await client.get_client(
|
||
int(server["inbound_id"]),
|
||
email=str(row["email"]),
|
||
)
|
||
if not current:
|
||
continue
|
||
patch = dict(current)
|
||
patch["email"] = str(row["email"])
|
||
if expiry_ms is not None:
|
||
patch["expiryTime"] = expiry_ms
|
||
if traffic_bytes is not None:
|
||
patch["totalGB"] = int(traffic_bytes)
|
||
if restore_enabled:
|
||
patch["enable"] = True
|
||
await client.update_client(
|
||
int(server["inbound_id"]),
|
||
str(row["email"]),
|
||
patch,
|
||
)
|
||
db.update_xui_wg_config_limits(
|
||
int(row["id"]),
|
||
expire_at=expire_at,
|
||
traffic_bytes=traffic_bytes,
|
||
)
|
||
if restore_enabled and not row.get("enabled"):
|
||
db.set_xui_wg_config_enabled(
|
||
int(row["id"]), telegram_id, True
|
||
)
|
||
updated += 1
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.warning(
|
||
"WG sync limits failed config=%s: %s",
|
||
row.get("id"),
|
||
exc,
|
||
)
|
||
return updated
|
||
|
||
|
||
def list_user_wg_configs(db: Database, telegram_id: int) -> list[dict[str, Any]]:
|
||
rows = db.list_xui_wg_configs(telegram_id)
|
||
servers = {int(s["id"]): s for s in db.list_xui_servers()}
|
||
return [serialize_wg_config(r, servers.get(int(r["server_id"]))) for r in rows]
|
||
|
||
|
||
async def delete_wg_config(
|
||
*,
|
||
db: Database,
|
||
telegram_id: int,
|
||
config_id: int,
|
||
) -> WgActionResult:
|
||
bot_user = db.ensure_user(telegram_id)
|
||
row = db.get_xui_wg_config(config_id, telegram_id=telegram_id)
|
||
if not row:
|
||
return WgActionResult(ok=False, message="Конфиг не найден.", balance=bot_user.balance)
|
||
|
||
server = db.get_xui_server(int(row["server_id"]))
|
||
if server and server.get("inbound_id") is not None:
|
||
try:
|
||
async with _client_from_server(server) as client:
|
||
await client.delete_client_by_email(
|
||
int(server["inbound_id"]),
|
||
str(row["email"]),
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.warning("WG delete on panel failed (will drop locally): %s", exc)
|
||
|
||
db.delete_xui_wg_config(config_id, telegram_id)
|
||
return WgActionResult(
|
||
ok=True,
|
||
message="Конфиг удалён. Слот освобождён.",
|
||
balance=bot_user.balance,
|
||
)
|