116 lines
3.5 KiB
Python
116 lines
3.5 KiB
Python
"""Remnawave: поиск пользователя и синхронизация лимита устройств."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from remnawave import RemnawaveSDK
|
|
from remnawave.enums import UserStatus
|
|
from remnawave.models import UpdateUserRequestDto
|
|
|
|
from services.db import Database
|
|
from services.devices import (
|
|
BASE_DEVICES,
|
|
PANEL_SOFT_CAP,
|
|
DeviceUsage,
|
|
allowed_devices,
|
|
)
|
|
|
|
|
|
def as_utc(dt: datetime) -> datetime:
|
|
if dt.tzinfo is None:
|
|
return dt.replace(tzinfo=timezone.utc)
|
|
return dt.astimezone(timezone.utc)
|
|
|
|
|
|
def username_for(telegram_id: int, *, login: str | None = None) -> str:
|
|
if telegram_id > 0:
|
|
return f"tg_{telegram_id}"
|
|
if login:
|
|
safe = "".join(ch for ch in login if ch.isalnum() or ch == "_")[:24] or "user"
|
|
return f"web_{safe}"
|
|
return f"web_{abs(telegram_id)}"
|
|
|
|
|
|
async def find_panel_user(
|
|
sdk: RemnawaveSDK, telegram_id: int, *, login: str | None = None
|
|
):
|
|
if telegram_id > 0:
|
|
try:
|
|
users = await sdk.users.get_users_by_telegram_id(str(telegram_id))
|
|
if users:
|
|
return users[0]
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
try:
|
|
return await sdk.users.get_user_by_username(
|
|
username_for(telegram_id, login=login)
|
|
)
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
|
|
|
|
def device_limit_for_user(extra_paid: int, *, soft_cap: bool = True) -> int:
|
|
allowed = allowed_devices(extra_paid)
|
|
if soft_cap:
|
|
return max(allowed, min(PANEL_SOFT_CAP, max(allowed * 2, BASE_DEVICES + 20)))
|
|
return allowed
|
|
|
|
|
|
async def count_hwid_devices(sdk: RemnawaveSDK, user_uuid: str) -> int:
|
|
try:
|
|
resp = await sdk.hwid.get_hwid_user(str(user_uuid))
|
|
return int(getattr(resp, "total", None) or len(getattr(resp, "devices", []) or []))
|
|
except Exception: # noqa: BLE001
|
|
return 0
|
|
|
|
|
|
async def sync_device_policy(
|
|
sdk: RemnawaveSDK,
|
|
panel_user,
|
|
*,
|
|
db: Database,
|
|
telegram_id: int,
|
|
extra_paid: int,
|
|
device_blocked: bool,
|
|
) -> tuple[object, DeviceUsage]:
|
|
"""Синхронизация лимита HWID и отключение при превышении оплаченной квоты."""
|
|
used = await count_hwid_devices(sdk, str(panel_user.uuid))
|
|
allowed = allowed_devices(extra_paid)
|
|
over = used > allowed
|
|
soft_limit = device_limit_for_user(extra_paid, soft_cap=True)
|
|
|
|
updates: dict = {
|
|
"uuid": panel_user.uuid,
|
|
"hwid_device_limit": soft_limit,
|
|
}
|
|
|
|
if over:
|
|
updates["status"] = UserStatus.DISABLED
|
|
if not device_blocked:
|
|
db.set_device_blocked(telegram_id, True)
|
|
elif device_blocked:
|
|
expire = as_utc(panel_user.expire_at)
|
|
if expire > datetime.now(timezone.utc):
|
|
updates["status"] = UserStatus.ACTIVE
|
|
db.set_device_blocked(telegram_id, False)
|
|
|
|
try:
|
|
current_limit = int(getattr(panel_user, "hwid_device_limit", 0) or 0)
|
|
current_status = panel_user.status
|
|
need_update = current_limit != soft_limit
|
|
if "status" in updates and updates["status"] != current_status:
|
|
need_update = True
|
|
if need_update:
|
|
panel_user = await sdk.users.update_user(UpdateUserRequestDto(**updates))
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
usage = DeviceUsage(
|
|
used=used,
|
|
allowed=allowed,
|
|
extra_paid=max(0, int(extra_paid)),
|
|
over_limit=over,
|
|
)
|
|
return panel_user, usage
|