ASF-Control-Bot/bot/api/asf.py

import aiohttp
import requests

from bot.config import ASF_PASSWORD, ASF_URL
from bot.logging_utils import get_logger

logger = get_logger(__name__)


def _headers() -> dict[str, str | None]:
    return {"Authentication": ASF_PASSWORD}


def asf_get(path: str) -> requests.Response:
    logger.info("Отправка GET-запроса в ASF: path=%s", path)
    response = requests.get(f"{ASF_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:
    logger.info(
        "Отправка POST-запроса в ASF: path=%s payload_keys=%s",
        path,
        sorted((payload or {}).keys()),
    )

    response = requests.post(f"{ASF_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:
    logger.info("Запрос статуса ASF")
    response = requests.get(f"{ASF_URL}/Api/ASF", headers=_headers())  # type: ignore[arg-type]
    logger.info("Получен статус ASF: status=%s", response.status_code)

    return response


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:
    logger.info(
        "Запрос инвентаря бота: bot=%s appid=%s contextid=%s",
        bot_name,
        appid,
        contextid,
    )

    async with aiohttp.ClientSession() as session:
        async with session.get(
            f"{ASF_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"{ASF_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"{ASF_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:
                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"{ASF_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"{ASF_URL}/Api/Command",
            headers=_headers(),
            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", ""))