forked from FOSS/ASF-Control-Bot
119 lines
3.8 KiB
Python
119 lines
3.8 KiB
Python
import aiohttp
|
|
import requests
|
|
|
|
from bot.config import ASF_PASSWORD, ASF_URL
|
|
|
|
IPC_BASE_URL = "http://127.0.0.1:1242"
|
|
|
|
|
|
def _headers() -> dict[str, str | None]:
|
|
return {"Authentication": ASF_PASSWORD}
|
|
|
|
|
|
def asf_get(path: str) -> requests.Response:
|
|
return requests.get(f"{IPC_BASE_URL}{path}", headers=_headers())
|
|
|
|
|
|
def asf_post(path: str, payload: dict | None = None) -> requests.Response:
|
|
return requests.post(f"{IPC_BASE_URL}{path}", headers=_headers(), json=payload)
|
|
|
|
|
|
def get_asf_status() -> requests.Response:
|
|
return requests.get(ASF_URL, headers=_headers()) # type: ignore[arg-type]
|
|
|
|
|
|
def get_all_bots() -> requests.Response:
|
|
return asf_get("/Api/Bot/ASF")
|
|
|
|
|
|
def get_bot(bot_name: str) -> requests.Response:
|
|
return asf_get(f"/Api/Bot/{bot_name}")
|
|
|
|
|
|
def get_plugins() -> requests.Response:
|
|
return asf_get("/Api/Plugins")
|
|
|
|
|
|
def send_command(command: str) -> requests.Response:
|
|
return asf_post("/Api/Command", {"Command": command})
|
|
|
|
|
|
def restart_asf() -> requests.Response:
|
|
return send_command("!restart")
|
|
|
|
|
|
async def get_bot_inventory(bot_name: str, appid: int, contextid: int) -> dict | None:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
f"{IPC_BASE_URL}/Api/Bot/{bot_name}/Inventory/{appid}/{contextid}",
|
|
headers=_headers(),
|
|
) as response:
|
|
if response.status != 200:
|
|
return None
|
|
data = await response.json()
|
|
if not data.get("Success"):
|
|
return None
|
|
return data.get("Result")
|
|
|
|
|
|
async def get_2fa_token(bot_name: str) -> str:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
f"{IPC_BASE_URL}/Api/Bot/{bot_name}/TwoFactorAuthentication/Token",
|
|
headers=_headers(),
|
|
) as response:
|
|
if response.status != 200:
|
|
return f"Ошибка HTTP {response.status}"
|
|
data = await response.json()
|
|
if not data.get("Success"):
|
|
return "Ошибка ASF"
|
|
try:
|
|
return data["Result"][bot_name]["Result"]
|
|
except Exception:
|
|
return "Не удалось получить код"
|
|
|
|
|
|
async def get_confirmations(bot_name: str) -> list:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
f"{IPC_BASE_URL}/Api/Bot/{bot_name}/TwoFactorAuthentication/Confirmations",
|
|
headers=_headers(),
|
|
) as response:
|
|
if response.status != 200:
|
|
return []
|
|
data = await response.json()
|
|
if not data.get("Success"):
|
|
return []
|
|
try:
|
|
return data["Result"][bot_name]["Result"]
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
async def accept_confirmations(bot_name: str) -> int:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
f"{IPC_BASE_URL}/Api/Bot/{bot_name}/TwoFactorAuthentication/Confirmations/Accept",
|
|
headers=_headers(),
|
|
) as response:
|
|
return response.status
|
|
|
|
|
|
def save_bot_config(bot_name: str, config: dict) -> requests.Response:
|
|
return asf_post(f"/Api/Bot/{bot_name}", {"BotConfig": config})
|
|
|
|
|
|
async def redeem_key(bot_name: str, key: str) -> tuple[bool, str]:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
f"{IPC_BASE_URL}/Api/Command",
|
|
headers=_headers(),
|
|
json={"Command": f"!redeem {bot_name} {key}"},
|
|
) as response:
|
|
if response.status != 200:
|
|
return False, f"HTTP_ERROR_{response.status}"
|
|
data = await response.json()
|
|
if not data.get("Success"):
|
|
return False, "ASF_ERROR"
|
|
return True, str(data.get("Result", ""))
|