diff --git a/main.py b/main.py
index 4e5498c..0bbd4b3 100644
--- a/main.py
+++ b/main.py
@@ -1,13 +1,15 @@
import asyncio, aiohttp, json, html, requests, re, os
from datetime import datetime, timezone
+from typing import cast
from aiogram import Bot, Dispatcher
from aiogram.filters import Command, BaseFilter
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton, CallbackQuery, Message
+from aiogram.client.default import DefaultBotProperties
from dotenv import load_dotenv
load_dotenv()
API_TOKEN = os.getenv("API_TOKEN")
-ADMIN_ID = int(os.getenv("ADMIN_ID"))
+ADMIN_ID = int(os.getenv("ADMIN_ID")) # type: ignore
ASF_URL = os.getenv("ASF_URL")
ASF_PASSWORD = os.getenv("ASF_PASSWORD")
@@ -18,24 +20,22 @@ GAMES = {
570: "Dota 2"}
appid = 753
contextid = 6
-console_users = set()
-redeem_users = set()
-bot_redeem_users = {}
-bot_redeem_messages = {}
-redeem_messages = {}
-twofa_tasks = {}
-inventory_cache = {}
+console_users, redeem_users = set(), set()
+twofa_tasks, inventory_cache, redeem_messages, bot_redeem_users, bot_redeem_messages = {}, {}, {}, {}, {}
-bot = Bot(token=API_TOKEN)
+bot = Bot(token=API_TOKEN, default=DefaultBotProperties(parse_mode="HTML")) # type: ignore
dp = Dispatcher()
class AdminFilter(BaseFilter):
async def __call__(self, message: Message) -> bool:
- return message.from_user.id == ADMIN_ID
+ return message.from_user.id == ADMIN_ID # type: ignore
class AdminCallbackFilter(BaseFilter):
async def __call__(self, callback: CallbackQuery) -> bool:
return callback.from_user.id == ADMIN_ID
+
+def asf_get(path: str):
+ return requests.get(f"http://127.0.0.1:1242{path}", headers={"Authentication": ASF_PASSWORD})
def get_currency_name(currency_id: int) -> str:
if currency_id == 1:
@@ -118,7 +118,7 @@ def get_bot_status_icon(bot: dict) -> str:
if farming:
return "🎴"
if idle_games and online:
- return "🎮"
+ return "⌚"
if online:
return "🟢"
return "🔴"
@@ -127,9 +127,9 @@ def format_bot_ui(bot: dict) -> str:
online = get_bot_status_icon(bot)
now_farming = bot.get("CardsFarmer", {}).get("NowFarming")
if now_farming:
- status_line = "Идет фарм карточек"
+ status_line = "🃏 Идет фарм карточек"
else:
- status_line = "Карточки не фармится"
+ status_line = "🃏 Карточки не фармится"
time = bot.get("CardsFarmer", {}).get("TimeRemaining", "00:00:00")
games_to_farm = len(bot.get("CardsFarmer", {}).get("GamesToFarm", []))
twofa = "есть" if bot.get("HasMobileAuthenticator") else "нет"
@@ -137,6 +137,8 @@ def format_bot_ui(bot: dict) -> str:
currency = get_currency_name(bot.get("WalletCurrency", 0))
redeem = bot.get("GamesToRedeemInBackgroundCount", 0)
steam_id = bot.get("SteamID")
+ steam_level = bot.get("SteamLevel")
+ games_count = bot.get("GamesOwned")
nickname = html.escape(str(bot.get("Nickname") or ""))
profile_url = f"https://steamcommunity.com/profiles/{steam_id}"
text = (
@@ -148,16 +150,16 @@ def format_bot_ui(bot: dict) -> str:
if idle_games:
names = [GAMES.get(g, str(g)) for g in idle_games]
if bot.get("IsConnectedAndLoggedOn"):
- idle_text = f"Idle: включен ({', '.join(names)})"
+ idle_text = f"⌚ Накрутка часов: включена ({', '.join(names)})"
else:
- idle_text = f"Idle: включен ({', '.join(names)}) применится после запуска"
- text += f"{idle_text}\n"
+ idle_text = f"⌚ Накрутка часов: включена ({', '.join(names)}) применится после запуска"
+ text += f"{idle_text}\n"
else:
- text += "Idle: выключен\n"
+ text += "⌚ Накрутка часов: выключена\n"
text += (
- f"Steam id: {steam_id}\n"
- f"2FA: {twofa}\n\n"
- f"Баланс: {balance:.2f} {currency}\n")
+ f"🆔 : {steam_id}\n"
+ f"🔐 2FA: {twofa}\n\n"
+ f"💰 Баланс: {balance:.2f} {currency}\n")
if redeem > 0:
text += f"\nКлючей в очереди: {redeem}\n"
if now_farming:
@@ -167,27 +169,14 @@ def format_bot_ui(bot: dict) -> str:
return text
def get_asf_status_text():
- response = requests.get(ASF_URL, headers={"Authentication": ASF_PASSWORD})
+ response = requests.get(ASF_URL, headers={"Authentication": ASF_PASSWORD}) # type: ignore
if response.status_code != 200:
return f"Ошибка API: {response.status_code}"
data = response.json()
if not data.get("Success"):
return "ASF вернул ошибку"
status_text = format_asf(data)
- return f"Панель управления ASF\n\n{status_text}"
-
-async def get_2fa_code(bot_name: str) -> str:
- async with aiohttp.ClientSession() as session:
- async with session.get(f"http://127.0.0.1:1242/Api/Bot/{bot_name}/TwoFactorAuthentication/Token", headers={"Authentication": ASF_PASSWORD}) 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:
- return "Не удалось получить код"
+ return f"💻 Панель управления ASF\n\n{status_text}"
def get_farm_summary(bots: dict) -> str:
total_games = 0
@@ -209,9 +198,78 @@ def get_farm_summary(bots: dict) -> str:
f"Осталось времени: {hours}ч {minutes}м\n"
f"Карт осталось: ~{total_games * 3}")
+def format_bot(bot_data: dict) -> str:
+ return json.dumps(bot_data, indent=2, ensure_ascii=False)
+
+def main_keyboard():
+ return InlineKeyboardMarkup(inline_keyboard=[
+ [InlineKeyboardButton(text="🆙 Обновить", callback_data="refresh")],
+ [InlineKeyboardButton(text="🤖 Боты", callback_data="bots")],
+ [InlineKeyboardButton(text="🔑 Активация ключей на всех аккаунтах", callback_data="redeem_keys")],
+ [InlineKeyboardButton(text="📁 Плагины", callback_data="plugins")],
+ [InlineKeyboardButton(text="💻 Консоль", callback_data="console")],
+ [InlineKeyboardButton(text="♻ Перезапустить ASF", callback_data="restart_asf_confirm")],
+ ])
+
+def back_keyboard():
+ return InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="🔙 Назад", callback_data="back")]])
+
+def games_keyboard(bot_name, is_enabled):
+ if is_enabled:
+ text = "Остановить накрутку CS2"
+ else:
+ text = "⌚ Накрутить часы CS2"
+ return InlineKeyboardMarkup(inline_keyboard=[
+ [InlineKeyboardButton(text=text, callback_data=f"farm|{bot_name}|730")],
+ [InlineKeyboardButton(text="🔙 Назад", callback_data=f"bot_{bot_name}")]])
+
+def bots_keyboard(bots: dict):
+ keyboard = []
+ for name, bot in bots.items():
+ status = get_bot_status_icon(bot)
+ loaded = (bool(bot.get("BotName")) and bool(bot.get("SteamID")))
+ if not loaded:
+ callback_data = "bot_loading"
+ else:
+ callback_data = f"bot_{name}"
+ keyboard.append([InlineKeyboardButton(text=f"{status} {name}", callback_data=callback_data)])
+ keyboard.append([InlineKeyboardButton(text="🔙 Назад", callback_data="back")])
+ return InlineKeyboardMarkup(inline_keyboard=keyboard)
+
+def is_bot_playing(bot_name: str) -> bool:
+ response = asf_get("/Api/Bot/ASF")
+ if response.status_code != 200:
+ return False
+ data = response.json()
+ bot = data.get("Result", {}).get(bot_name, {})
+ # правильный способ
+ return bool(bot.get("PlayingBlocked", False)) or bool(bot.get("CurrentGamesPlayed", []))
+
+def is_idle_enabled(bot_name: str, app_id: int) -> bool:
+ response = requests.get(f"http://127.0.0.1:1242/Api/Bot/{bot_name}", headers={"Authentication": ASF_PASSWORD})
+ if response.status_code != 200:
+ return False
+ data = response.json()
+ bot = data.get("Result", {}).get(bot_name, {})
+ config = bot.get("BotConfig", {})
+ return app_id in config.get("GamesPlayedWhileIdle", [])
+
+async def get_2fa_code(bot_name: str) -> str:
+ async with aiohttp.ClientSession() as session:
+ async with session.get(f"http://127.0.0.1:1242/Api/Bot/{bot_name}/TwoFactorAuthentication/Token", headers={"Authentication": ASF_PASSWORD}) as response: # type: ignore
+ 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:
+ return "Не удалось получить код"
+
async def get_confirmations(bot_name: str):
async with aiohttp.ClientSession() as session:
- async with session.get(f"http://127.0.0.1:1242/Api/Bot/{bot_name}/TwoFactorAuthentication/Confirmations", headers={"Authentication": ASF_PASSWORD}) as response:
+ async with session.get(f"http://127.0.0.1:1242/Api/Bot/{bot_name}/TwoFactorAuthentication/Confirmations", headers={"Authentication": ASF_PASSWORD}) as response: # type: ignore
if response.status != 200:
return []
data = await response.json()
@@ -225,7 +283,7 @@ async def get_confirmations(bot_name: str):
async def get_inventory(bot_name: str, appid: int, contextid: int = 2):
url = (f"http://127.0.0.1:1242/Api/Bot/{bot_name}/Inventory/{appid}/{contextid}")
async with aiohttp.ClientSession() as session:
- async with session.get(url, headers={"Authentication": ASF_PASSWORD}) as response:
+ async with session.get(url, headers={"Authentication": ASF_PASSWORD}) as response: # type: ignore
if response.status != 200:
return None
data = await response.json()
@@ -237,16 +295,16 @@ async def get_inventory(bot_name: str, appid: int, contextid: int = 2):
return None
async def render_inventory_page(callback: CallbackQuery, bot_name: str, inventory_type: str, appid: int, contextid: int, page: int):
- await callback.message.edit_text("Загрузка inventory...")
+ await callback.message.edit_text("Загрузка inventory...") # type: ignore
inventory = await get_inventory(bot_name, appid, contextid)
if not inventory:
- await callback.message.edit_text("Не удалось загрузить inventory")
+ await callback.message.edit_text("Не удалось загрузить inventory") # type: ignore
return
bot_inventory = inventory.get(bot_name, {})
assets = bot_inventory.get("Assets", [])
descriptions = bot_inventory.get("Descriptions", [])
if not assets:
- await callback.message.edit_text("Инвентарь пуст", reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="Назад", callback_data=f"inventory_{bot_name}")]]))
+ await callback.message.edit_text("Инвентарь пуст", reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="🔙 Назад", callback_data=f"inventory_{bot_name}")]])) # type: ignore
return
desc_map = {}
for desc in descriptions:
@@ -308,46 +366,33 @@ async def render_inventory_page(callback: CallbackQuery, bot_name: str, inventor
keyboard = []
if nav_buttons:
keyboard.append(nav_buttons)
- keyboard.append([InlineKeyboardButton(text="Назад", callback_data=f"inventory_{bot_name}")])
- await callback.message.edit_text(text[:4000], reply_markup=InlineKeyboardMarkup(inline_keyboard=keyboard))
+ keyboard.append([InlineKeyboardButton(text="🔙 Назад", callback_data=f"inventory_{bot_name}")])
+ await callback.message.edit_text(text[:4000], reply_markup=InlineKeyboardMarkup(inline_keyboard=keyboard)) # type: ignore
-def format_bot(bot_data: dict) -> str:
- return json.dumps(bot_data, indent=2, ensure_ascii=False)
+async def reload_bot(bot_name: str):
+ try:
+ requests.post("http://127.0.0.1:1242/Api/Command", headers={"Authentication": ASF_PASSWORD}, json={"Command": f"!reload {bot_name}"})
+ except:
+ pass
-def main_keyboard():
- return InlineKeyboardMarkup(inline_keyboard=[
- [InlineKeyboardButton(text="Обновить", callback_data="refresh")],
- [InlineKeyboardButton(text="Боты", callback_data="bots")],
- [InlineKeyboardButton(text="Активация ключей на всех аккаунтах", callback_data="redeem_keys")],
- [InlineKeyboardButton(text="Плагины", callback_data="plugins")],
- [InlineKeyboardButton(text="Консоль", callback_data="console")],
- [InlineKeyboardButton(text="♻ Restart ASF", callback_data="restart_asf_confirm")],
- ])
+async def stop_twofa_task(message_id: int):
+ task = twofa_tasks.pop(message_id, None)
+ if task:
+ task.cancel()
+ try:
+ await task
+ except:
+ pass
-def back_keyboard():
- return InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="Назад", callback_data="back")]])
-
-def games_keyboard(bot_name, is_enabled):
- if is_enabled:
- text = "Остановить накрутку CS2"
- else:
- text = "Накрутить часы CS2"
- return InlineKeyboardMarkup(inline_keyboard=[
- [InlineKeyboardButton(text=text, callback_data=f"farm|{bot_name}|730")],
- [InlineKeyboardButton(text="Назад", callback_data=f"bot_{bot_name}")]])
-
-def bots_keyboard(bots: dict):
- keyboard = []
- for name, bot in bots.items():
- status = get_bot_status_icon(bot)
- loaded = (bool(bot.get("BotName")) and bool(bot.get("SteamID")))
- if not loaded:
- callback_data = "bot_loading"
- else:
- callback_data = f"bot_{name}"
- keyboard.append([InlineKeyboardButton(text=f"{status} {name}", callback_data=callback_data)])
- keyboard.append([InlineKeyboardButton(text="Назад", callback_data="back")])
- return InlineKeyboardMarkup(inline_keyboard=keyboard)
+async def redeem_key(bot_name: str, key: str):
+ async with aiohttp.ClientSession() as session:
+ async with session.post("http://127.0.0.1:1242/Api/Command", headers={"Authentication": ASF_PASSWORD}, json={"Command": f"!redeem {bot_name} {key}"}) as response: # type: ignore
+ 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", ""))
@dp.callback_query(lambda c: c.data == "bot_loading", AdminCallbackFilter())
async def bot_loading_handler(callback: CallbackQuery):
@@ -364,12 +409,12 @@ async def auto_update_2fa(message, bot_name):
keyboard = []
if confirmations:
keyboard.append([InlineKeyboardButton(text="Подтверждения", callback_data=f"confirm_list_{bot_name}")])
- keyboard.append([InlineKeyboardButton(text="Назад", callback_data=f"bot_{bot_name}")])
+ keyboard.append([InlineKeyboardButton(text="🔙 Назад", callback_data=f"bot_{bot_name}")])
text = (
f"🔐 {bot_name}\n\n"
f"{code}")
try:
- await message.edit_text(text, parse_mode="HTML", reply_markup=InlineKeyboardMarkup(inline_keyboard=keyboard))
+ await message.edit_text(text, reply_markup=InlineKeyboardMarkup(inline_keyboard=keyboard))
last_code = code
except:
break
@@ -381,15 +426,6 @@ async def auto_update_2fa(message, bot_name):
finally:
twofa_tasks.pop(message_id, None)
-async def stop_twofa_task(message_id: int):
- task = twofa_tasks.pop(message_id, None)
- if task:
- task.cancel()
- try:
- await task
- except:
- pass
-
@dp.message(Command("start"), AdminFilter())
async def start_handler(message: Message):
try:
@@ -406,26 +442,26 @@ async def start_handler(message: Message):
async def console_enter(callback: CallbackQuery):
await callback.answer()
console_users.add(callback.from_user.id)
- await callback.message.edit_text("Консоль ASF\n\nОтправь команду (без !)", reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="Назад", callback_data="console_exit")]]))
+ await callback.message.edit_text("💻 Консоль ASF\n\nОтправь команду (без !)", reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="🔙 Назад", callback_data="console_exit")]])) # type: ignore
@dp.callback_query(lambda c: c.data == "console_exit", AdminCallbackFilter())
async def console_exit(callback: CallbackQuery):
await callback.answer()
console_users.discard(callback.from_user.id)
text = get_asf_status_text()
- await callback.message.edit_text(text, reply_markup=main_keyboard())
+ await callback.message.edit_text(text, reply_markup=main_keyboard()) # type: ignore
@dp.callback_query(lambda c: c.data == "delete_msg", AdminCallbackFilter())
async def delete_message(callback: CallbackQuery):
await callback.answer()
try:
- await callback.message.delete()
+ await callback.message.delete() # type: ignore
except:
pass
@dp.callback_query(lambda c: c.data == "bots", AdminCallbackFilter())
async def bots_handler(callback: CallbackQuery):
- await stop_twofa_task(callback.message.message_id)
+ await stop_twofa_task(callback.message.message_id) # type: ignore
await callback.answer()
try:
response = requests.get(
@@ -433,33 +469,31 @@ async def bots_handler(callback: CallbackQuery):
headers={"Authentication": ASF_PASSWORD}
)
if response.status_code != 200:
- await callback.message.edit_text("Ошибка API")
+ await callback.message.edit_text("Ошибка API") # type: ignore
return
data = response.json()
if not data.get("Success"):
- await callback.message.edit_text("ASF вернул ошибку")
+ await callback.message.edit_text("ASF вернул ошибку") # type: ignore
return
bots = data.get("Result", {})
count = len(bots)
summary = get_farm_summary(bots)
text = (
- f"Ботов всего: {count}\n\n"
- f"{summary}\n\n"
- f"Выбери бота:"
- )
- await callback.message.edit_text(text, reply_markup=bots_keyboard(bots))
+ f"🤖 Ботов всего: {count}\n\n"
+ f"{summary}")
+ await callback.message.edit_text(text, reply_markup=bots_keyboard(bots)) # type: ignore
except Exception as e:
- await callback.message.edit_text(f"Ошибка: {e}")
+ await callback.message.edit_text(f"Ошибка: {e}") # type: ignore
@dp.callback_query(lambda c: c.data and c.data.startswith("bot_"), AdminCallbackFilter())
async def bot_selected(callback: CallbackQuery):
- await stop_twofa_task(callback.message.message_id)
+ await stop_twofa_task(callback.message.message_id) # type: ignore
await callback.answer()
if not callback.data:
return
bot_name = callback.data.replace("bot_", "")
try:
- response = requests.get("http://127.0.0.1:1242/Api/Bot/ASF", headers={"Authentication": ASF_PASSWORD})
+ response = asf_get("/Api/Bot/ASF")
data = response.json()
bots = data.get("Result", {})
bot = bots.get(bot_name)
@@ -467,20 +501,20 @@ async def bot_selected(callback: CallbackQuery):
await callback.answer("Аккаунт ещё загружается. Попробуйте через несколько секунд.", show_alert=True)
return
if not bot:
- await callback.message.edit_text("Бот не найден")
+ await callback.message.edit_text("Бот не найден") # type: ignore
return
text = format_bot_ui(bot)
keyboard = []
if bot.get("HasMobileAuthenticator"):
- keyboard.append([InlineKeyboardButton(text="2FA", callback_data=f"2fa_{bot_name}")])
- keyboard.append([InlineKeyboardButton(text="Игры", callback_data=f"games_{bot_name}")])
- keyboard.append([InlineKeyboardButton(text="Активировать ключ", callback_data=f"redeem_bot_{bot_name}")])
- keyboard.append([InlineKeyboardButton(text="Инвентарь", callback_data=f"inventory_{bot_name}")])
- keyboard.append([InlineKeyboardButton(text="Назад", callback_data="bots")])
+ keyboard.append([InlineKeyboardButton(text="🔐 2FA", callback_data=f"2fa_{bot_name}")])
+ keyboard.append([InlineKeyboardButton(text="⌚ Накрутка часов", callback_data=f"games_{bot_name}")])
+ keyboard.append([InlineKeyboardButton(text="🔑 Активировать ключ", callback_data=f"redeem_bot_{bot_name}")])
+ keyboard.append([InlineKeyboardButton(text="📦 Инвентарь", callback_data=f"inventory_{bot_name}")])
+ keyboard.append([InlineKeyboardButton(text="🔙 Назад", callback_data="bots")])
- await callback.message.edit_text(text, parse_mode="HTML", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=keyboard))
+ await callback.message.edit_text(text, disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup(inline_keyboard=keyboard)) # type: ignore
except Exception as e:
- await callback.message.edit_text(f"Ошибка: {e}")
+ await callback.message.edit_text(f"Ошибка: {e}") # type: ignore
@dp.callback_query(lambda c: c.data and c.data.startswith("games_"), AdminCallbackFilter())
async def games(callback: CallbackQuery):
@@ -489,7 +523,7 @@ async def games(callback: CallbackQuery):
return
bot_name = callback.data.replace("games_", "")
idle_enabled = is_idle_enabled(bot_name, 730)
- await callback.message.edit_text("Выберите игру для накрутки часов", reply_markup=games_keyboard(bot_name, idle_enabled))
+ await callback.message.edit_text("⌚ Выберите игру для накрутки часов", reply_markup=games_keyboard(bot_name, idle_enabled)) # type: ignore
@dp.callback_query(lambda c: c.data == "back", AdminCallbackFilter())
async def back_button(callback: CallbackQuery):
@@ -498,27 +532,27 @@ async def back_button(callback: CallbackQuery):
console_users.discard(callback.from_user.id)
bot_redeem_users.pop(callback.from_user.id, None)
bot_redeem_messages.pop(callback.from_user.id, None)
- await stop_twofa_task(callback.message.message_id)
+ await stop_twofa_task(callback.message.message_id) # type: ignore
await callback.answer()
try:
text = get_asf_status_text()
- await callback.message.edit_text(text, reply_markup=main_keyboard())
+ await callback.message.edit_text(text, reply_markup=main_keyboard()) # type: ignore
except Exception as e:
- await callback.message.edit_text(f"Ошибка: {e}")
+ await callback.message.edit_text(f"Ошибка: {e}") # type: ignore
@dp.callback_query(lambda c: c.data == "refresh", AdminCallbackFilter())
async def refresh_handler(callback: CallbackQuery):
await callback.answer()
try:
text = get_asf_status_text()
- await callback.message.edit_text(text, reply_markup=main_keyboard())
+ await callback.message.edit_text(text, reply_markup=main_keyboard()) # type: ignore
except Exception as e:
- await callback.message.edit_text(f"Ошибка: {e}")
+ await callback.message.edit_text(f"Ошибка: {e}") # type: ignore
@dp.callback_query(lambda c: c.data == "restart_asf_confirm", AdminCallbackFilter())
async def restart_asf_confirm(callback: CallbackQuery):
await callback.answer()
- await callback.message.edit_text(
+ await callback.message.edit_text( # type: ignore
"♻ Точно перезапустить ASF?\n\n"
"Во время перезапуска IPC будет недоступен несколько секунд.",
reply_markup=InlineKeyboardMarkup(
@@ -530,16 +564,16 @@ async def restart_asf_confirm(callback: CallbackQuery):
async def restart_asf(callback: CallbackQuery):
await callback.answer()
try:
- await callback.message.edit_text("♻ ASF перезапускается...")
+ await callback.message.edit_text("♻ ASF перезапускается...") # type: ignore
response = requests.post("http://127.0.0.1:1242/Api/Command", headers={"Authentication": ASF_PASSWORD}, json={"Command": "!restart"})
if response.status_code != 200:
- await callback.message.edit_text(f"Ошибка HTTP {response.status_code}", reply_markup=main_keyboard())
+ await callback.message.edit_text(f"Ошибка HTTP {response.status_code}", reply_markup=main_keyboard()) # type: ignore
return
await asyncio.sleep(6)
text = get_asf_status_text()
- await callback.message.edit_text(f"ASF успешно перезапущен\n\n{text}", reply_markup=main_keyboard())
+ await callback.message.edit_text(f"ASF успешно перезапущен\n\n{text}", reply_markup=main_keyboard()) # type: ignore
except Exception as e:
- await callback.message.edit_text(f"Ошибка: {e}", reply_markup=main_keyboard())
+ await callback.message.edit_text(f"Ошибка: {e}", reply_markup=main_keyboard()) # type: ignore
@dp.callback_query(lambda c: c.data and c.data.startswith("2fa_") and not c.data.startswith("2fa_refresh_"), AdminCallbackFilter())
async def twofa_handler(callback: CallbackQuery):
@@ -547,7 +581,7 @@ async def twofa_handler(callback: CallbackQuery):
if not callback.data:
return
bot_name = callback.data.replace("2fa_", "")
- message_id = callback.message.message_id
+ message_id = callback.message.message_id # type: ignore
# убиваем старую task
old_task = twofa_tasks.pop(message_id, None)
if old_task:
@@ -557,9 +591,9 @@ async def twofa_handler(callback: CallbackQuery):
except:
pass
# мгновенный экран загрузки
- await callback.message.edit_text(
+ await callback.message.edit_text( # type: ignore
f"🔐 {bot_name}\n\nЗагрузка 2FA...",
- reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="Назад", callback_data=f"bot_{bot_name}")]]))
+ reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="🔙 Назад", callback_data=f"bot_{bot_name}")]]))
# запускаем updater
task = asyncio.create_task(auto_update_2fa(callback.message, bot_name))
twofa_tasks[message_id] = task
@@ -567,7 +601,7 @@ async def twofa_handler(callback: CallbackQuery):
@dp.callback_query(lambda c: c.data and c.data.startswith("confirm_") and not c.data.startswith("confirm_list_") and not c.data.startswith("confirm_all_"), AdminCallbackFilter())
async def confirm_trades(callback: CallbackQuery):
await callback.answer()
- bot_name = callback.data.replace("confirm_", "")
+ bot_name = callback.data.replace("confirm_", "") # type: ignore
try:
response = requests.post(f"http://127.0.0.1:1242/Api/Bot/{bot_name}/TwoFactorAuthentication/Confirmations/Accept", headers={"Authentication": ASF_PASSWORD})
if response.status_code != 200:
@@ -577,39 +611,37 @@ async def confirm_trades(callback: CallbackQuery):
# обновляем экран
await twofa_handler(callback)
except Exception as e:
- await callback.message.edit_text(f"Ошибка: {e}")
+ await callback.message.edit_text(f"Ошибка: {e}") # type: ignore
@dp.callback_query(lambda c: c.data == "plugins", AdminCallbackFilter())
async def plugins_handler(callback: CallbackQuery):
- await stop_twofa_task(callback.message.message_id)
+ await stop_twofa_task(callback.message.message_id) # type: ignore
await callback.answer()
try:
response = requests.get("http://127.0.0.1:1242/Api/Plugins", headers={"Authentication": ASF_PASSWORD})
if response.status_code != 200:
- await callback.message.edit_text("Ошибка API")
+ await callback.message.edit_text("Ошибка API") # type: ignore
return
data = response.json()
if not data.get("Success"):
- await callback.message.edit_text("ASF вернул ошибку")
+ await callback.message.edit_text("ASF вернул ошибку") # type: ignore
return
plugins = data.get("Result", [])
if not plugins:
text = "Плагины не установлены"
else:
- text = "Плагины ASF:\n\n"
+ text = "📁 Плагины ASF:\n\n"
for plugin in plugins:
name = plugin.get("Name", "???")
version = plugin.get("Version", "???")
text += f"{name} ({version})\n"
- await callback.message.edit_text(
- text.strip(),
- reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="Назад", callback_data="back")]]))
+ await callback.message.edit_text(text.strip(), reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="🔙 Назад", callback_data="back")]])) # type: ignore
except Exception as e:
- await callback.message.edit_text(f"Ошибка: {e}")
+ await callback.message.edit_text(f"Ошибка: {e}") # type: ignore
@dp.callback_query(lambda c: c.data and c.data.startswith("confirm_list_"), AdminCallbackFilter())
async def confirm_list(callback: CallbackQuery):
- await stop_twofa_task(callback.message.message_id)
+ await stop_twofa_task(callback.message.message_id) # type: ignore
await callback.answer()
if not callback.data:
return
@@ -625,16 +657,16 @@ async def confirm_list(callback: CallbackQuery):
f"- {conf['type_name']}\n"
f"ID: {conf['id']}\n\n"
)
- await callback.message.edit_text(text, parse_mode="HTML",
+ await callback.message.edit_text(text, # type: ignore
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="Подтвердить всё", callback_data=f"confirm_all_{bot_name}")],
- [InlineKeyboardButton(text="Назад", callback_data=f"2fa_{bot_name}")]]))
+ [InlineKeyboardButton(text="🔙 Назад", callback_data=f"2fa_{bot_name}")]]))
except Exception as e:
- await callback.message.edit_text(f"Ошибка: {e}")
+ await callback.message.edit_text(f"Ошибка: {e}") # type: ignore
@dp.callback_query(lambda c: c.data and c.data.startswith("confirm_all_"), AdminCallbackFilter())
async def confirm_all(callback: CallbackQuery):
- await stop_twofa_task(callback.message.message_id)
+ await stop_twofa_task(callback.message.message_id) # type: ignore
await callback.answer()
if not callback.data:
return
@@ -655,39 +687,14 @@ async def confirm_all(callback: CallbackQuery):
f"- {conf['type_name']}\n"
f"ID: {conf['id']}\n\n"
)
- await callback.message.edit_text(
+ await callback.message.edit_text( # type: ignore
text,
- parse_mode="HTML",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="Подтвердить всё", callback_data=f"confirm_all_{bot_name}")],
[InlineKeyboardButton(text="Назад", callback_data=f"2fa_{bot_name}")]]))
await callback.answer(" Все подтверждено", show_alert=True)
except Exception as e:
- await callback.message.edit_text(f"Ошибка: {e}")
-
-def is_bot_playing(bot_name: str) -> bool:
- response = requests.get("http://127.0.0.1:1242/Api/Bot/ASF", headers={"Authentication": ASF_PASSWORD})
- if response.status_code != 200:
- return False
- data = response.json()
- bot = data.get("Result", {}).get(bot_name, {})
- # правильный способ
- return bool(bot.get("PlayingBlocked", False)) or bool(bot.get("CurrentGamesPlayed", []))
-
-def is_idle_enabled(bot_name: str, app_id: int) -> bool:
- response = requests.get(f"http://127.0.0.1:1242/Api/Bot/{bot_name}", headers={"Authentication": ASF_PASSWORD})
- if response.status_code != 200:
- return False
- data = response.json()
- bot = data.get("Result", {}).get(bot_name, {})
- config = bot.get("BotConfig", {})
- return app_id in config.get("GamesPlayedWhileIdle", [])
-
-async def reload_bot(bot_name: str):
- try:
- requests.post("http://127.0.0.1:1242/Api/Command", headers={"Authentication": ASF_PASSWORD}, json={"Command": f"!reload {bot_name}"})
- except:
- pass
+ await callback.message.edit_text(f"Ошибка: {e}") # type: ignore
@dp.callback_query(lambda c: c.data and c.data.startswith("farm|"), AdminCallbackFilter())
async def farm_game(callback: CallbackQuery):
@@ -720,7 +727,7 @@ async def farm_game(callback: CallbackQuery):
asyncio.create_task(reload_bot(bot_name))
await callback.answer(action, show_alert=True)
try:
- await callback.message.edit_reply_markup(reply_markup=games_keyboard(bot_name, app_id in idle_games))
+ await callback.message.edit_reply_markup(reply_markup=games_keyboard(bot_name, app_id in idle_games)) # type: ignore
except:
pass
except Exception as e:
@@ -729,30 +736,30 @@ async def farm_game(callback: CallbackQuery):
@dp.callback_query(lambda c: c.data and c.data.startswith("inventory_"), AdminCallbackFilter())
async def inventory_menu(callback: CallbackQuery):
await callback.answer()
- bot_name = callback.data.replace("inventory_", "")
- await callback.message.edit_text("Проверка inventory...")
+ bot_name = callback.data.replace("inventory_", "") # type: ignore
+ await callback.message.edit_text("Проверка inventory...") # type: ignore
cs2_inventory = await get_inventory(bot_name, 730, 2)
steam_inventory = await get_inventory(bot_name, 753, 6)
cs2_assets = []
steam_assets = []
try:
- cs2_assets = (cs2_inventory.get(bot_name, {}).get("Assets", []))
+ cs2_assets = (cs2_inventory.get(bot_name, {}).get("Assets", [])) # type: ignore
except:
pass
try:
- steam_assets = (steam_inventory.get(bot_name, {}).get("Assets", []))
+ steam_assets = (steam_inventory.get(bot_name, {}).get("Assets", [])) # type: ignore
except:
pass
if not cs2_assets and not steam_assets:
- await callback.message.edit_text(f"📦 Инвентарь {bot_name} пуст", reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="Назад", callback_data=f"bot_{bot_name}")]]))
+ await callback.message.edit_text(f"📦 Инвентарь {bot_name} пуст", reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="🔙 Назад", callback_data=f"bot_{bot_name}")]])) # type: ignore
return
keyboard = []
if cs2_assets:
keyboard.append([InlineKeyboardButton(text=f"CS2 ({len(cs2_assets)})", callback_data=f"inv_cs2_{bot_name}")])
if steam_assets:
keyboard.append([InlineKeyboardButton(text=f"Steam ({len(steam_assets)})", callback_data=f"inv_steam_{bot_name}")])
- keyboard.append([InlineKeyboardButton(text="Назад", callback_data=f"bot_{bot_name}")])
- await callback.message.edit_text(
+ keyboard.append([InlineKeyboardButton(text="🔙 Назад", callback_data=f"bot_{bot_name}")])
+ await callback.message.edit_text( # type: ignore
(
f"📦 Инвентарь {bot_name}\n\n"
"🔄 — можно трейдить\n"
@@ -766,66 +773,54 @@ async def inventory_menu(callback: CallbackQuery):
@dp.callback_query(lambda c: c.data and c.data.startswith("inv_cs2_"), AdminCallbackFilter())
async def cs2_inventory(callback: CallbackQuery):
await callback.answer()
- bot_name = callback.data.replace("inv_cs2_", "")
+ bot_name = callback.data.replace("inv_cs2_", "") # type: ignore
await render_inventory_page(callback, bot_name, "cs2", 730, 2, 0)
@dp.callback_query(lambda c: c.data and c.data.startswith("inv_steam_"), AdminCallbackFilter())
async def steam_inventory(callback: CallbackQuery):
await callback.answer()
- bot_name = callback.data.replace("inv_steam_", "")
+ bot_name = callback.data.replace("inv_steam_", "") # type: ignore
await render_inventory_page(callback, bot_name, "steam", 753, 6, 0)
@dp.callback_query(lambda c: c.data and c.data.startswith("invpage|"), AdminCallbackFilter())
async def inventory_page(callback: CallbackQuery):
await callback.answer()
- _, inv_type, bot_name, page = callback.data.split("|")
+ _, inv_type, bot_name, page = callback.data.split("|") # type: ignore
page = int(page)
if inv_type == "cs2":
await render_inventory_page(callback, bot_name, "cs2", 730, 2, page)
else:
await render_inventory_page(callback, bot_name, "steam", 753, 6, page)
-async def redeem_key(bot_name: str, key: str):
- async with aiohttp.ClientSession() as session:
- async with session.post("http://127.0.0.1:1242/Api/Command", headers={"Authentication": ASF_PASSWORD}, 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", ""))
-
@dp.callback_query(lambda c: c.data == "redeem_keys", AdminCallbackFilter())
async def redeem_keys_menu(callback: CallbackQuery):
await callback.answer()
redeem_users.add(callback.from_user.id)
redeem_messages[callback.from_user.id] = callback.message
- await callback.message.edit_text(
+ await callback.message.edit_text( # type: ignore
"Отправьте ключ Steam для активации\n\n"
"Можно отправить сразу несколько ключей.\n\n"
"Пример:\n"
"AAAAA-BBBBB-CCCCC\nDDDDD-EEEEE-FFFFF",
- parse_mode="HTML",
- reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="Назад", callback_data="back")]]))
+ reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="🔙 Назад", callback_data="back")]]))
@dp.callback_query(lambda c: c.data and c.data.startswith("redeem_bot_"), AdminCallbackFilter())
async def redeem_bot_menu(callback: CallbackQuery):
await callback.answer()
- bot_name = callback.data.replace("redeem_bot_", "")
+ bot_name = callback.data.replace("redeem_bot_", "") # type: ignore
bot_redeem_users[callback.from_user.id] = bot_name
bot_redeem_messages[callback.from_user.id] = callback.message
- await callback.message.edit_text(
+ await callback.message.edit_text( # type: ignore
f"Отправьте ключ Steam для активации на аккаунте:\n"
f"{bot_name}\n\n"
f"Можно отправить сразу несколько ключей.",
- parse_mode="HTML",
- reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="Назад", callback_data=f"bot_{bot_name}")]]))
+ reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="🔙 Назад", callback_data=f"bot_{bot_name}")]]))
@dp.message(AdminFilter())
async def text_router(message: Message):
# console mode
- if message.from_user.id in console_users:
- command = message.text.strip()
+ if message.from_user.id in console_users: # type: ignore
+ command = message.text.strip() # type: ignore
try:
await message.delete()
except:
@@ -844,14 +839,14 @@ async def text_router(message: Message):
f"Команда: {command}\n\n"
f"Ответ:\n{result}"
)
- await message.answer(text, parse_mode="HTML", reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="Удалить", callback_data="delete_msg")]]))
+ await message.answer(text, reply_markup=InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="Удалить", callback_data="delete_msg")]]))
except Exception as e:
await message.answer(f"Ошибка: {e}")
return
# single bot redeem mode
- if message.from_user.id in bot_redeem_users:
- bot_name = bot_redeem_users.pop(message.from_user.id)
- menu_message = bot_redeem_messages.pop(message.from_user.id, None)
+ if message.from_user.id in bot_redeem_users: # type: ignore
+ bot_name = bot_redeem_users.pop(message.from_user.id) # type: ignore
+ menu_message = bot_redeem_messages.pop(message.from_user.id, None) # type: ignore
if menu_message:
try:
response = requests.get("http://127.0.0.1:1242/Api/Bot/ASF",headers={"Authentication": ASF_PASSWORD})
@@ -860,23 +855,22 @@ async def text_router(message: Message):
if bot_data:
await menu_message.edit_text(
format_bot_ui(bot_data),
- parse_mode="HTML",
disable_web_page_preview=True,
reply_markup=InlineKeyboardMarkup(
inline_keyboard=[
- [InlineKeyboardButton(text="2FA", callback_data=f"2fa_{bot_name}")]
+ [InlineKeyboardButton(text="🔐 2FA", callback_data=f"2fa_{bot_name}")]
if bot_data.get("HasMobileAuthenticator") else [],
- [InlineKeyboardButton(text="Игры", callback_data=f"games_{bot_name}")],
- [InlineKeyboardButton(text="Активировать ключ", callback_data=f"redeem_bot_{bot_name}")],
- [InlineKeyboardButton(text="Инвентарь", callback_data=f"inventory_{bot_name}")],
- [InlineKeyboardButton(text="Назад", callback_data="bots")]]))
+ [InlineKeyboardButton(text="⌚ Накрутка часов", callback_data=f"games_{bot_name}")],
+ [InlineKeyboardButton(text="🔑 Активировать ключ", callback_data=f"redeem_bot_{bot_name}")],
+ [InlineKeyboardButton(text="📦 Инвентарь", callback_data=f"inventory_{bot_name}")],
+ [InlineKeyboardButton(text="🔙 Назад", callback_data="bots")]]))
except:
pass
try:
await bot.delete_message(chat_id=message.chat.id, message_id=message.message_id)
except:
pass
- raw_text = message.text.upper()
+ raw_text = message.text.upper() # type: ignore
keys = re.findall(r"[A-Z0-9]{5}(?:-[A-Z0-9]{5}){2}", raw_text)
keys = list(dict.fromkeys(keys))
if not keys:
@@ -885,8 +879,7 @@ async def text_router(message: Message):
checking = await message.answer(
f"Аккаунт: {bot_name}\n"
f"Найдено ключей: {len(keys)}\n"
- f"Начинаю активацию...",
- parse_mode="HTML")
+ f"Начинаю активацию...")
success_count = 0
failed_count = 0
results = []
@@ -926,12 +919,12 @@ async def text_router(message: Message):
+ "\n".join(results[:50]))
if len(text) > 4000:
text = text[:4000] + "\n\n...обрезано"
- await checking.edit_text(text, parse_mode="HTML")
+ await checking.edit_text(text)
return
# redeem mode
- if message.from_user.id in redeem_users:
- redeem_users.discard(message.from_user.id)
- menu_message = redeem_messages.pop(message.from_user.id, None)
+ if message.from_user.id in redeem_users: # type: ignore
+ redeem_users.discard(message.from_user.id) # type: ignore
+ menu_message = redeem_messages.pop(message.from_user.id, None) # type: ignore
if menu_message:
try:
await menu_message.edit_text(get_asf_status_text(), reply_markup=main_keyboard())
@@ -941,7 +934,7 @@ async def text_router(message: Message):
await bot.delete_message(chat_id=message.chat.id, message_id=message.message_id)
except:
pass
- raw_text = message.text.upper()
+ raw_text = message.text.upper() # type: ignore
keys = re.findall(r"[A-Z0-9]{5}(?:-[A-Z0-9]{5}){2}", raw_text)
keys = list(dict.fromkeys(keys))
if not keys:
@@ -949,7 +942,7 @@ async def text_router(message: Message):
return
checking = await message.answer(f"Найдено ключей: {len(keys)}\nНачинаю активацию...")
try:
- response = requests.get("http://127.0.0.1:1242/Api/Bot/ASF", headers={"Authentication": ASF_PASSWORD})
+ response = asf_get("/Api/Bot/ASF")
data = response.json()
bots = list(data.get("Result", {}).keys())
success_count = 0