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

refactor(asf): use ASF_URL directly, poll restart

Browse Source
This commit is contained in:
Tom Riddle
2026-07-22 22:10:24 +05:00
parent 2fb37d1ee8
commit b1c8f6d9f8
6 changed files with 70 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot/api/asf.py
+17 -12
View File
@@ -4,7 +4,6 @@ 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__)
@@ -14,7 +13,7 @@ def _headers() -> dict[str, str | None]:
def asf_get(path: str) -> requests.Response:
logger.info("Отправка GET-запроса в ASF: path=%s", path)
response = requests.get(f"{IPC_BASE_URL}{path}", headers=_headers())
response = requests.get(f"{ASF_URL}{path}", headers=_headers())
logger.info(
"Получен ответ ASF на GET-запрос: path=%s status=%s",
path,
@@ -29,7 +28,7 @@ def asf_post(path: str, payload: dict | None = None) -> requests.Response:
path,
sorted((payload or {}).keys()),
)
response = requests.post(f"{IPC_BASE_URL}{path}", headers=_headers(), json=payload)
response = requests.post(f"{ASF_URL}{path}", headers=_headers(), json=payload)
logger.info(
"Получен ответ ASF на POST-запрос: path=%s status=%s",
path,
@@ -40,7 +39,7 @@ def asf_post(path: str, payload: dict | None = None) -> requests.Response:
def get_asf_status() -> requests.Response:
logger.info("Запрос статуса ASF")
response = requests.get(ASF_URL, headers=_headers()) # type: ignore[arg-type]
response = requests.get(f"{ASF_URL}/Api/ASF", headers=_headers()) # type: ignore[arg-type]
logger.info("Получен статус ASF: status=%s", response.status_code)
return response
@@ -74,7 +73,7 @@ async def get_bot_inventory(bot_name: str, appid: int, contextid: int) -> dict |
)
async with aiohttp.ClientSession() as session:
async with session.get(
f"{IPC_BASE_URL}/Api/Bot/{bot_name}/Inventory/{appid}/{contextid}",
f"{ASF_URL}/Api/Bot/{bot_name}/Inventory/{appid}/{contextid}",
headers=_headers(),
) as response:
if response.status != 200:
@@ -108,7 +107,7 @@ 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",
f"{ASF_URL}/Api/Bot/{bot_name}/TwoFactorAuthentication/Token",
headers=_headers(),
) as response:
if response.status != 200:
@@ -120,13 +119,17 @@ async def get_2fa_token(bot_name: str) -> str:
return f"Ошибка HTTP {response.status}"
data = await response.json()
if not data.get("Success"):
logger.warning("ASF вернул ошибку при запросе 2FA-кода: bot=%s", bot_name)
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)
logger.exception(
"Не удалось извлечь 2FA-код из ответа ASF: bot=%s", bot_name
)
return "Не удалось получить код"
@@ -134,7 +137,7 @@ 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",
f"{ASF_URL}/Api/Bot/{bot_name}/TwoFactorAuthentication/Confirmations",
headers=_headers(),
) as response:
if response.status != 200:
@@ -171,7 +174,7 @@ 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",
f"{ASF_URL}/Api/Bot/{bot_name}/TwoFactorAuthentication/Confirmations/Accept",
headers=_headers(),
) as response:
logger.info(
@@ -191,7 +194,7 @@ 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",
f"{ASF_URL}/Api/Command",
headers=_headers(),
json={"Command": f"!redeem {bot_name} {key}"},
) as response:
@@ -204,7 +207,9 @@ async def redeem_key(bot_name: str, key: str) -> tuple[bool, str]:
return False, f"HTTP_ERROR_{response.status}"
data = await response.json()
if not data.get("Success"):
logger.warning("ASF вернул ошибку при активации ключа: bot=%s", bot_name)
logger.warning(
"ASF вернул ошибку при активации ключа: bot=%s", bot_name
)
return False, "ASF_ERROR"
logger.info("ASF вернул результат активации ключа: bot=%s", bot_name)
return True, str(data.get("Result", ""))