Lolz/ASF-Control-Bot
1
0
Fork 0
forked from FOSS/ASF-Control-Bot
Code Pull Requests Activity

refactor(logging): logging and rename token env

Browse Source
This commit is contained in:
Tom Riddle
2026-07-22 21:36:33 +05:00
parent e4de501472
commit 2fb37d1ee8
21 changed files with 741 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot/api/asf.py
+96 -4
View File
@@ -2,8 +2,10 @@ import aiohttp
import requests
from bot.config import ASF_PASSWORD, ASF_URL
from bot.logging_utils import get_logger
IPC_BASE_URL = "http://127.0.0.1:1242"
logger = get_logger(__name__)
def _headers() -> dict[str, str | None]:
@@ -11,15 +13,36 @@ def _headers() -> dict[str, str | None]:
def asf_get(path: str) -> requests.Response:
return requests.get(f"{IPC_BASE_URL}{path}", headers=_headers())
logger.info("Отправка GET-запроса в ASF: path=%s", path)
response = requests.get(f"{IPC_BASE_URL}{path}", headers=_headers())
logger.info(
"Получен ответ ASF на GET-запрос: path=%s status=%s",
path,
response.status_code,
)
return response
def asf_post(path: str, payload: dict | None = None) -> requests.Response:
return requests.post(f"{IPC_BASE_URL}{path}", headers=_headers(), json=payload)
logger.info(
"Отправка POST-запроса в ASF: path=%s payload_keys=%s",
path,
sorted((payload or {}).keys()),
)
response = requests.post(f"{IPC_BASE_URL}{path}", headers=_headers(), json=payload)
logger.info(
"Получен ответ ASF на POST-запрос: path=%s status=%s",
path,
response.status_code,
)
return response
def get_asf_status() -> requests.Response:
return requests.get(ASF_URL, headers=_headers()) # type: ignore[arg-type]
logger.info("Запрос статуса ASF")
response = requests.get(ASF_URL, headers=_headers()) # type: ignore[arg-type]
logger.info("Получен статус ASF: status=%s", response.status_code)
return response
def get_all_bots() -> requests.Response:
@@ -43,67 +66,129 @@ def restart_asf() -> requests.Response:
async def get_bot_inventory(bot_name: str, appid: int, contextid: int) -> dict | None:
logger.info(
"Запрос инвентаря бота: bot=%s appid=%s contextid=%s",
bot_name,
appid,
contextid,
)
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:
logger.warning(
"Не удалось получить инвентарь бота: bot=%s appid=%s contextid=%s status=%s",
bot_name,
appid,
contextid,
response.status,
)
return None
data = await response.json()
if not data.get("Success"):
logger.warning(
"ASF вернул ошибку при получении инвентаря: bot=%s appid=%s contextid=%s",
bot_name,
appid,
contextid,
)
return None
logger.info(
"Инвентарь бота получен: bot=%s appid=%s contextid=%s",
bot_name,
appid,
contextid,
)
return data.get("Result")
async def get_2fa_token(bot_name: str) -> str:
logger.info("Запрос 2FA-кода: bot=%s", bot_name)
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:
logger.warning(
"Не удалось получить 2FA-код: bot=%s status=%s",
bot_name,
response.status,
)
return f"Ошибка HTTP {response.status}"
data = await response.json()
if not data.get("Success"):
logger.warning("ASF вернул ошибку при запросе 2FA-кода: bot=%s", bot_name)
return "Ошибка ASF"
try:
logger.info("2FA-код успешно получен: bot=%s", bot_name)
return data["Result"][bot_name]["Result"]
except Exception:
logger.exception("Не удалось извлечь 2FA-код из ответа ASF: bot=%s", bot_name)
return "Не удалось получить код"
async def get_confirmations(bot_name: str) -> list:
logger.info("Запрос подтверждений 2FA: bot=%s", bot_name)
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:
logger.warning(
"Не удалось получить подтверждения 2FA: bot=%s status=%s",
bot_name,
response.status,
)
return []
data = await response.json()
if not data.get("Success"):
logger.warning(
"ASF вернул ошибку при запросе подтверждений 2FA: bot=%s",
bot_name,
)
return []
try:
return data["Result"][bot_name]["Result"]
result = data["Result"][bot_name]["Result"]
logger.info(
"Подтверждения 2FA получены: bot=%s count=%s",
bot_name,
len(result),
)
return result
except Exception:
logger.exception(
"Не удалось извлечь подтверждения 2FA из ответа ASF: bot=%s",
bot_name,
)
return []
async def accept_confirmations(bot_name: str) -> int:
logger.info("Отправка запроса на подтверждение всех 2FA-операций: bot=%s", bot_name)
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:
logger.info(
"Получен ответ на подтверждение 2FA-операций: bot=%s status=%s",
bot_name,
response.status,
)
return response.status
def save_bot_config(bot_name: str, config: dict) -> requests.Response:
logger.info("Сохранение конфигурации бота: bot=%s", bot_name)
return asf_post(f"/Api/Bot/{bot_name}", {"BotConfig": config})
async def redeem_key(bot_name: str, key: str) -> tuple[bool, str]:
logger.info("Запрос активации ключа: bot=%s", bot_name)
async with aiohttp.ClientSession() as session:
async with session.post(
f"{IPC_BASE_URL}/Api/Command",
@@ -111,8 +196,15 @@ async def redeem_key(bot_name: str, key: str) -> tuple[bool, str]:
json={"Command": f"!redeem {bot_name} {key}"},
) as response:
if response.status != 200:
logger.warning(
"Не удалось активировать ключ: bot=%s status=%s",
bot_name,
response.status,
)
return False, f"HTTP_ERROR_{response.status}"
data = await response.json()
if not data.get("Success"):
logger.warning("ASF вернул ошибку при активации ключа: bot=%s", bot_name)
return False, "ASF_ERROR"
logger.info("ASF вернул результат активации ключа: bot=%s", bot_name)
return True, str(data.get("Result", ""))