forked from FOSS/ASF-Control-Bot
143 lines
4.8 KiB
Python
143 lines
4.8 KiB
Python
import asyncio
|
|
|
|
from bot.api.asf import get_all_bots, get_bot, save_bot_config, send_command
|
|
from bot.logging_utils import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
RELOAD_POLL_INTERVAL_SECONDS = 1
|
|
RELOAD_TIMEOUT_SECONDS = 20
|
|
_reloading_bots: set[str] = set()
|
|
|
|
|
|
def is_bot_reloading(bot_name: str) -> bool:
|
|
return bot_name in _reloading_bots
|
|
|
|
|
|
def _is_bot_loaded_and_online(bot_name: str) -> bool:
|
|
response = get_all_bots()
|
|
if response.status_code != 200:
|
|
return False
|
|
|
|
data = response.json()
|
|
bot = data.get("Result", {}).get(bot_name, {})
|
|
return bool(
|
|
bot.get("BotName")
|
|
and bot.get("SteamID")
|
|
and bot.get("IsConnectedAndLoggedOn")
|
|
)
|
|
|
|
|
|
def is_bot_playing(bot_name: str) -> bool:
|
|
logger.info("Проверка, играет ли бот: bot=%s", bot_name)
|
|
response = get_all_bots()
|
|
if response.status_code != 200:
|
|
logger.warning(
|
|
"Не удалось проверить состояние игры бота: bot=%s status=%s",
|
|
bot_name,
|
|
response.status_code,
|
|
)
|
|
return False
|
|
|
|
data = response.json()
|
|
bot = data.get("Result", {}).get(bot_name, {})
|
|
playing = bool(bot.get("PlayingBlocked", False)) or bool(
|
|
bot.get("CurrentGamesPlayed", [])
|
|
)
|
|
logger.info("Состояние игры бота определено: bot=%s playing=%s", bot_name, playing)
|
|
return playing
|
|
|
|
|
|
def is_idle_enabled(bot_name: str, app_id: int) -> bool:
|
|
logger.info("Проверка статуса idle-игры: bot=%s app_id=%s", bot_name, app_id)
|
|
response = get_bot(bot_name)
|
|
|
|
if response.status_code != 200:
|
|
logger.warning(
|
|
"Не удалось получить конфиг бота для проверки idle: bot=%s status=%s",
|
|
bot_name,
|
|
response.status_code,
|
|
)
|
|
return False
|
|
|
|
data = response.json()
|
|
bot = data.get("Result", {}).get(bot_name, {})
|
|
config = bot.get("BotConfig", {})
|
|
enabled = app_id in config.get("GamesPlayedWhileIdle", [])
|
|
logger.info(
|
|
"Статус idle-игры определен: bot=%s app_id=%s enabled=%s",
|
|
bot_name,
|
|
app_id,
|
|
enabled,
|
|
)
|
|
return enabled
|
|
|
|
|
|
def toggle_idle_game(config: dict, app_id: int) -> tuple[dict, bool]:
|
|
idle_games = config.get("GamesPlayedWhileIdle", [])
|
|
if app_id in idle_games:
|
|
idle_games.remove(app_id)
|
|
enabled = False
|
|
else:
|
|
idle_games.append(app_id)
|
|
enabled = True
|
|
config["GamesPlayedWhileIdle"] = idle_games
|
|
return config, enabled
|
|
|
|
|
|
async def reload_bot(bot_name: str) -> None:
|
|
_reloading_bots.add(bot_name)
|
|
try:
|
|
logger.info("Отправка команды перезагрузки бота: bot=%s", bot_name)
|
|
send_command(f"!reload {bot_name}")
|
|
for _ in range(RELOAD_TIMEOUT_SECONDS):
|
|
await asyncio.sleep(RELOAD_POLL_INTERVAL_SECONDS)
|
|
if _is_bot_loaded_and_online(bot_name):
|
|
logger.info("Перезагрузка бота завершена: bot=%s", bot_name)
|
|
return
|
|
logger.warning("Истекло время ожидания перезагрузки бота: bot=%s", bot_name)
|
|
except Exception:
|
|
logger.exception(
|
|
"Не удалось отправить команду перезагрузки бота: bot=%s", bot_name
|
|
)
|
|
finally:
|
|
_reloading_bots.discard(bot_name)
|
|
|
|
|
|
async def update_idle_game(bot_name: str, app_id: int) -> tuple[bool, str, bool]:
|
|
logger.info("Изменение настройки idle-игры: bot=%s app_id=%s", bot_name, app_id)
|
|
|
|
response = get_bot(bot_name)
|
|
if response.status_code != 200:
|
|
logger.warning(
|
|
"Не удалось загрузить конфиг бота для idle: bot=%s status=%s",
|
|
bot_name,
|
|
response.status_code,
|
|
)
|
|
|
|
return False, "Ошибка API", False
|
|
|
|
data = response.json()
|
|
bot_data = data.get("Result", {}).get(bot_name, {})
|
|
config = bot_data.get("BotConfig", {})
|
|
config, enabled = toggle_idle_game(config, app_id)
|
|
save = save_bot_config(bot_name, config)
|
|
if save.status_code != 200:
|
|
logger.warning(
|
|
"Не удалось сохранить конфиг idle-игры: bot=%s app_id=%s status=%s",
|
|
bot_name,
|
|
app_id,
|
|
save.status_code,
|
|
)
|
|
return False, "Ошибка сохранения", enabled
|
|
|
|
_reloading_bots.add(bot_name)
|
|
asyncio.create_task(reload_bot(bot_name))
|
|
action = "Накрутка часов включена" if enabled else "Накрутка часов выключена"
|
|
logger.info(
|
|
"Настройка idle-игры изменена: bot=%s app_id=%s enabled=%s",
|
|
bot_name,
|
|
app_id,
|
|
enabled,
|
|
)
|
|
return True, action, enabled
|