BLOG [Python] - всякие безделушки :)

DEVV · 28 Апр 2023

Привет всем! Было скучно и я решил написать простейшего tg бота для получения информации о сервере)

Python:

from aiogram import Bot, Dispatcher, executor, types
import a2s

bot = Bot(token='YOUR_TOKEN')
dp = Dispatcher(bot=bot)

SERVER_IP = 'SERVER_IP' # your server ip str()
SERVER_PORT = SERVER_PORT # server port int()
ERROR = 'Тех.работа' # error

@dp.message_handler(commands=['start'])
async def cmd_start(message: types.Message) -> None:
    await bot.send_message(chat_id=message.from_user.id, text='''Это бот для получения информации о сервере.
Вводите команду /info''')

@dp.message_handler(commands=['info'])
async def info_serv(message: types.Message):
    try:
        info = a2s.info((SERVER_IP, SERVER_PORT))
        max_players = info.max_players
        player = info.player_count
        map_name = info.map_name
        server_name = info.server_name

        ttext_info = f'''<b>Сервер:</b> {server_name}
<b>IP:</b> {SERVER_IP}:{SERVER_PORT}
<b>Игроки:</b> {player}/{max_players}
<b>Карта:</b> {map_name}'''
await message.reply(text=text_info, parse_mode='HTML')

    except Exception:
        await message.reply(text=ERROR)


if __name__ == '__main__':
    executor.start_polling(dispatcher=dp,
                           skip_updates=True)

для работы бота требуется установить библиотеки:
pip install aiogram
pip3 install python-a2s

Документация по использованию библиотеки a2s если кому-то нужно :)

Denzer · 28 Апр 2023

Вы используете асинхронный тг фреймворк, но a2s функции для получения информации синхронные. При больших нагрузках (rps) на бота, он будет тормозить.

В этой же либе есть асинк функции для получения инфы a2s.

DEVV · 28 Апр 2023

Denzer, спасибо за наводки, учту, исправлю. Просто я с этой библиотекой первый раз работаю (10 минут учил). Попробую как вы сказали

Maga_610 · 29 Апр 2023

DEVV, а возможно подключить несколько серверов?

DEVV · 29 Апр 2023

Maga_610, сделаю

DEVV · 7 Июл 2023

В итоге получился вот такой код:

Python:

from aiogram import Bot, Dispatcher, executor, types
import a2s
import asyncio

bot = Bot(token='YOUR_TOKEN')
dp = Dispatcher(bot=bot)

SERVERS = [
    {'ip': '123.456.123.45', 'port': 27015},  # Пример сервера 1
    {'ip': '123.456.123.45', 'port': 27016},  # Пример сервера 2
    # Добавьте другие серверы в список
]

ERROR = 'Тех.работа'  # Ошибка

async def get_server_info(ip, port):
    try:
        loop = asyncio.get_event_loop()
        info = await loop.run_in_executor(None, lambda: a2s.info((ip, port)))
        return info
    except Exception:
        return None

@dp.message_handler(commands=['start'])
async def cmd_start(message: types.Message) -> None:
    await bot.send_message(chat_id=message.from_user.id, text='''Это бот для получения информации о сервере.
Вводите команду /info''')

@dp.message_handler(commands=['info'])
async def info_serv(message: types.Message):
    results = []
    for server in SERVERS:
        info = await get_server_info(server['ip'], server['port'])
        if info:
            results.append({
                'ip': server['ip'],
                'port': server['port'],
                'info': info
            })

    if results:
        text_info = ""
        for result in results:
            server_name = result['info'].server_name
            players = result['info'].player_count
            max_players = result['info'].max_players
            map_name = result['info'].map_name

            text_info += f'''<b>Сервер:</b> {server_name}
<b>IP:</b>  {result['ip']}:{result['port']}
<b>Игроки:</b> {players}/{max_players}
<b>Карта:</b> {map_name}
\n'''

        await message.reply(text=text_info, parse_mode='HTML')
    else:
        await message.reply(text=ERROR)

if __name__ == '__main__':
    executor.start_polling(dispatcher=dp, skip_updates=True)

Denzer · 7 Июл 2023

Код:

info = await loop.run_in_executor(None, lambda: a2s.info((ip, port)))

Вы через пул потоков вызываете синхронную функцию, это не ок

В этой либе есть асинхронная функция a2s.ainfo

и в вашем коде будет что-то такое

Код:

address = (ip, port)
info = await a2s.ainfo(address)

И кэшировать надо это дело, хотя бы на 1 минуту.

DEVV · 7 Июл 2023

Denzer написал(а):
В этой либе есть асинхронная функция a2s.ainfo

Спасибо, сделал.

Denzer написал(а):
И кэшировать надо это дело, хотя бы на 1 минуту.

Код:

cache = {}

async def get_server_info(ip, port):
    cache_key = f"{ip}:{port}"
    if cache_key in cache:
        return cache[cache_key]

вот так?

Denzer · 7 Июл 2023

DEVV написал(а):
Код:

cache = {} async def get_server_info(ip, port): cache_key = f"{ip}:{port}" if cache_key in cache: return cache[cache_key]

вот так?

Вам нужен TTL кэш, можете погуглить либы либо сами реализовать это.

thesuper · 11 Июл 2023

Traceback (most recent call last):
File "main.py", line 2, in <module>
import a2s
ModuleNotFoundError: No module named 'a2s'

KeyboardInterrupt
ошибка,чтоделать?

DEVV · 11 Июл 2023

thesuper
Установить библиотеку a2s. в первом посте читайте.

Python:

from aiogram import Bot, Dispatcher, executor, types
import a2s

bot = Bot(token='TOKEN_API')
dp = Dispatcher(bot=bot)

SERVERS = [
    {'ip': '123.456.12.34', 'port': 27015},  # Пример сервера 1
    {'ip': '123.456.12.34', 'port': 27016},  # Пример сервера 2
    # Добавьте другие серверы в список
]

ERROR = 'Тех.работа'  # Ошибка

async def get_server_info(ip, port):
    try:
        info = await a2s.ainfo((ip, port))
        return info
    except Exception:
        return None

@dp.message_handler(commands=['start'])
async def cmd_start(message: types.Message) -> None:
    await bot.send_message(chat_id=message.from_user.id, text='''Это бот для получения информации о сервере.
Вводите команду /info''')

@dp.message_handler(commands=['info'])
async def info_serv(message: types.Message):
    results = []
    for server in SERVERS:
        info = await get_server_info(server['ip'], server['port'])
        if info:
            results.append({
                'ip': server['ip'],
                'port': server['port'],
                'info': info
            })

    if results:
        text_info = ""
        for result in results:
            server_name = result['info'].server_name
            players = result['info'].player_count
            max_players = result['info'].max_players
            map_name = result['info'].map_name

            text_info += f'''<b>Сервер:</b> {server_name}
<b>IP:</b>  {result['ip']}:{result['port']}
<b>Игроки:</b> {players}/{max_players}
<b>Карта:</b> {map_name}
\n'''

        await message.reply(text=text_info, parse_mode='HTML')
    else:
        await message.reply(text=ERROR)

if __name__ == '__main__':
    executor.start_polling(dispatcher=dp, skip_updates=True)

а еще использовать слегка исправленный код)

DEVV · 11 Июл 2023

Denzer
Можете проверить, как кэширование?

Python:

from aiogram import Bot, Dispatcher, executor, types
import a2s
from cachetools import TTLCache

bot = Bot(token='YOUR_BOT_TOKEN')
dp = Dispatcher(bot=bot)

SERVERS = [
    {'ip': '127.0.0.1', 'port': 27015},  # Пример сервера 1
    {'ip': '127.0.0.1', 'port': 27016},  # Пример сервера 2
    # Добавьте другие серверы в список
]

ERROR = 'Тех.работа'  # Ошибка

cache = TTLCache(maxsize=100, ttl=60) 

async def get_server_info(ip, port):
    cache_key = f"{ip}:{port}"
    cached_result = cache.get(cache_key)
    if cached_result:
        return cached_result

    try:
        info = await a2s.ainfo((ip, port))
        cache[cache_key] = info
        return info
    except Exception:
        return None

@dp.message_handler(commands=['start'])
async def cmd_start(message: types.Message) -> None:
    await bot.send_message(chat_id=message.from_user.id, text='''Это бот для получения информации о сервере.
Вводите команду /info''')

@dp.message_handler(commands=['info'])
async def info_serv(message: types.Message):
    results = []
    for server in SERVERS:
        info = await get_server_info(server['ip'], server['port'])
        if info:
            results.append({
                'ip': server['ip'],
                'port': server['port'],
                'info': info
            })

    if results:
        text_info = ""
        for result in results:
            server_name = result['info'].server_name
            players = result['info'].player_count
            max_players = result['info'].max_players
            map_name = result['info'].map_name

            text_info += f'''<b>Сервер:</b> {server_name}
<b>IP:</b>  {result['ip']}:{result['port']}
<b>Игроки:</b> {players}/{max_players}
<b>Карта:</b> {map_name}
\n'''

        await message.reply(text=text_info, parse_mode='HTML')
    else:
        await message.reply(text=ERROR)

if __name__ == '__main__':
    executor.start_polling(dispatcher=dp, skip_updates=True)

thesuper · 11 Июл 2023

момент

11 Июл 2023

DEVV, да с пк запускаю, все гуд, на хосте такая лажа.... ,Дай телеграм или отпиши @thesuperlativee

DEVV · 11 Июл 2023

thesuper, позже покажу как установить Python и т.д, на ubuntu

thesuper · 11 Июл 2023

DEVV, дай контакты

DEVV · 13 Июл 2023

thesuper, попробуйте вот так

thesuper · 18 Июл 2023

DEVV Переделал твой код, сделал чуть проще. Твой выдавал ошибки + не ставился на хост. Тут все работает. Дарю

pip install telebot
pip install python-a2s

/

import a2s
import telebot

TOKEN__BOT = 'TOKEN'

bot = telebot.TeleBot(TOKEN__BOT)

@bot.message_handler(commands=['info'])
def start(message):
server_info = a2s.info(('91.211.118.48', 27019))
msg = f"Адрес: {server_info.address}\n"
msg += f"Название сервера: {server_info.server_name}\n"
msg += f"Карта: {server_info.map_name}\n"
msg += f"Количество игроков: {server_info.player_count}/{server_info.max_players}\n"

bot.send_message(message.chat.id,
text=msg.format(
message.from_user), parse_mode='html')

def telegram_polling():
try:
bot.polling()
except Exception as e:
print(e)
telegram_polling()

if name == 'main':
telegram_polling()

DEVV · 18 Июл 2023

thesuper написал(а):
Твой выдавал ошибки + не ставился на хост.

Возможно стоило показать ошибки в моей теме. У меня на ubuntu 18.04 всё работает (могу ссылку на своего бота дать для наглядности). О твоём коде ничего говорить не буду, каждому своё. Лайк за старания :)

DEVV · 18 Июл 2023

Вопрос к опытным программистам, у меня есть такой код:

Python:

from aiogram import Bot, Dispatcher, executor, types
import logging

from config import TOKEN_API
from keyboard import kb
from functions import ControlServer

logging.basicConfig(level=logging.INFO)
bot = Bot(token=TOKEN_API)
dp = Dispatcher(bot=bot)
cs = ControlServer()

@dp.message_handler(commands=['start'])
async def cmd_start(message: types.Message) -> None:
    await bot.send_message(chat_id=message.from_user.id,
                           text=f'''Привет, через меня ты можешь запустить/перезапустить сервера.''',
                           reply_markup=kb)

@dp.message_handler(text='🟢Запуск')
async def start_server(message: types.Message):
    try:
        cs.start_server()
        await message.reply(f'''Сервер успешно запущен!''',
                            reply_markup=kb)
    except Exception as e:
        await message.reply(f'Ошибка: {e}')
    
if __name__ == '__main__':
    executor.start_polling(dp, skip_updates=True)

Python:

import subprocess
from config import cd, cmd_start

class ControlServer:
    def start_server(self):
        subprocess.call(cd, shell=True)
        subprocess.call(cmd_start, shell=True)

Python:

TOKEN_API = '...' # Токен бота
cd = 'cd /home/servers/3454/' # Путь к файлу start.sh
cmd_start = 'screen -S 3454 ./start.sh' # команда запуска

Python:

from aiogram.types import ReplyKeyboardMarkup, KeyboardButton, ReplyKeyboardRemove

kb = ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=True)
kb1 = KeyboardButton(text='🟢Запуск')
kb.add(kb1)

ну а сам start.sh:

CSS:

./hlds_run -game cstrike +ip 0.0.0.0 -port 3454 +maxplayers 2 +map de_dust2 +exec server.cfg -fps_max 1001 -sys_ticrate 1001

Почему-то сервер не запускается, директории и тд правильно указал. В чем может быть проблема?

Denzer · 18 Июл 2023

DEVV, попробуйте объединить эти две команды через &&, т.е. subprocess.call("cd path && screen...")