89 lines
2.9 KiB
Python
89 lines
2.9 KiB
Python
"""
|
|
Проверка, что Remnawave Panel API принимает запросы.
|
|
|
|
Запуск (Docker Compose):
|
|
docker compose run --rm bot python check_api.py
|
|
docker compose run --rm bot python check_api.py --verbose
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
import traceback
|
|
|
|
import httpx
|
|
|
|
from config import get_settings
|
|
from remnawave_client import create_sdk_ready, probe_users_write_access
|
|
|
|
|
|
async def probe_http(base_url: str) -> tuple[bool, str]:
|
|
"""Быстрая проверка доступности панели без SDK."""
|
|
url = f"{base_url.rstrip('/')}/api/system/health"
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
|
|
resp = await client.get(url)
|
|
if resp.status_code in (200, 401, 403):
|
|
return True, f"HTTP {resp.status_code} от {url}"
|
|
return False, f"Неожиданный HTTP {resp.status_code} от {url}: {resp.text[:200]}"
|
|
except httpx.RequestError as exc:
|
|
return False, f"Сеть: не удалось достучаться до {url}: {exc}"
|
|
|
|
|
|
async def check_with_sdk(sdk) -> None:
|
|
print("→ system.get_health() ...")
|
|
health = await sdk.system.get_health()
|
|
print(f" OK health: {health}")
|
|
|
|
print("→ users.get_all_users(size=1) ...")
|
|
users = await sdk.users.get_all_users(size=1)
|
|
print(f" OK users read: total={getattr(users, 'total', '?')}")
|
|
|
|
print("→ users.create_user (write probe) ...")
|
|
ok, msg = await probe_users_write_access(sdk)
|
|
if ok:
|
|
print(f" OK write: {msg}")
|
|
else:
|
|
print(f" FAIL write:\n{msg}")
|
|
raise SystemExit(3)
|
|
|
|
|
|
async def main() -> int:
|
|
settings = get_settings()
|
|
print(f"Panel: {settings.remnawave_base_url}")
|
|
if settings.remnawave_token:
|
|
tok = settings.remnawave_token
|
|
print(f"Token: {tok[:8]}…{tok[-4:]}")
|
|
if settings.remnawave_username:
|
|
print(f"Login: {settings.remnawave_username}")
|
|
print()
|
|
|
|
ok, msg = await probe_http(settings.remnawave_base_url)
|
|
print(f"1) HTTP probe: {msg}")
|
|
if not ok:
|
|
print("API недоступен. Проверьте URL / DNS / SSL / firewall.")
|
|
return 1
|
|
|
|
print("2) SDK auth check:")
|
|
try:
|
|
sdk = await create_sdk_ready(settings)
|
|
await check_with_sdk(sdk)
|
|
except SystemExit as exc:
|
|
return int(exc.code) if isinstance(exc.code, int) else 3
|
|
except Exception as exc: # noqa: BLE001
|
|
print()
|
|
print("ОШИБКА: API не принял запрос с этим токеном.")
|
|
print(f" {type(exc).__name__}: {exc}")
|
|
if "--verbose" in sys.argv:
|
|
traceback.print_exc()
|
|
return 2
|
|
|
|
print()
|
|
print("✓ Remnawave API: чтение и запись OK. Можно продавать тарифы.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(asyncio.run(main()))
|