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

[ufame]

DEVV, таки удивился бы если были. Там просто сильно много шарят за вк и его апи, а также пару ребят «держат» свое приложение для прослушивания музыки из вк, как это работает - не знаю)

 

[DEVV]

ufame, да, надеюсь найду решение) спасибо тебе

 

[DEVV]

 

[DEVV]

Возник такой вопрос, есть такой info бот

Спойлер: main.py

import io
import a2s
import asyncio

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

from config import *
from messages import *

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

SERVERS = [
    {'ip': 'example.com', 'port': 27015}
]

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), timeout=5)
        cache[cache_key] = info
        return info
    except Exception as e:
        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=START_MESSAGE,
                           parse_mode='HTML')

@dp.message_handler(commands=['vip'])
async def vip_info(message: types.Message) -> None:
    await message.reply(VIP_INFO, parse_mode='HTML')

def escape_string(str):
    str = str.replace("<","&lt;")
    str = str.replace(">","&gt;")

    return str

async def send_server_info(message, server_info, image_path=None):
    server_name = server_info['info'].server_name
    players = server_info['info'].player_count
    max_players = server_info['info'].max_players
    map_name = server_info['info'].map_name

    try:
        if image_path:
            with open(image_path, "rb") as photo:
                img_bytes = photo.read()

            img_io = io.BytesIO(img_bytes)
        else:
            raise FileNotFoundError
    except FileNotFoundError:
        with open("images/no_image.png", "rb") as no_image:
            no_image_bytes = no_image.read()

        img_io = io.BytesIO(no_image_bytes)

    caption_text = (
        f"💚<b>Сервер:</b> {server_name}\n"
        f"📌<b>IP:</b>  {server_info['ip']}:{server_info['port']}\n"
        f"😎<b>Игроки:</b> {players}/{max_players}\n"
        f"📄<b>Карта:</b> {map_name}\n"
        f"# Ник | Время (М:С)\n"
    )

    Players = await a2s.aplayers((SERVERS[0]['ip'], SERVERS[0]['port']))
    for idx, player in enumerate(Players):
        playerName = escape_string(player.name)
        playerTime = player.duration
        minutes = int(playerTime // 60)
        seconds = round((playerTime % 60))
        caption_text += f'\n{idx + 1}. <code>🥷🏻{playerName} | 🕒{minutes}:{seconds}</code>'

    await message.reply_photo(photo=img_io, caption=caption_text, parse_mode='HTML')

@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:
        for result in results:
            image_path = f"images/{result['info'].map_name}.png"
            await send_server_info(message, result, image_path)
    else:
        await message.reply(text=ERROR)


if __name__ == '__main__':
    from tasks import scheduled

    loop = asyncio.get_event_loop()
    loop.create_task(scheduled([12, 0], "Йоу! Чуваки! сегодня же выходной? Давайте провести время вместе😉\n"
                                     "IP: example.com:27015\n"
                                     "Заходите!", bot))

    executor.start_polling(dispatcher=dp, skip_updates=True)

Бот почему-то не может отправлять некоторые картинки карт... фотки не повреждены

Спойлер: error

Task exception was never retrieved
future: <Task finished name='Task-55' coro=<Dispatcher._process_polling_updates() done, defined at /usr/local/lib/python3.8/dist-packages/aiogram/dispatcher/dispatcher.py:407> exception=BadRequest('Image_process_failed')>
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/dist-packages/aiogram/dispatcher/dispatcher.py", line 415, in _process_polling_updates
    for responses in itertools.chain.from_iterable(await self.process_updates(updates, fast)):
  File "/usr/local/lib/python3.8/dist-packages/aiogram/dispatcher/dispatcher.py", line 235, in process_updates
    return await asyncio.gather(*tasks)
  File "/usr/local/lib/python3.8/dist-packages/aiogram/dispatcher/handler.py", line 117, in notify
    response = await handler_obj.handler(*args, **partial_data)
  File "/usr/local/lib/python3.8/dist-packages/aiogram/dispatcher/dispatcher.py", line 256, in process_update
    return await self.message_handlers.notify(update.message)
  File "/usr/local/lib/python3.8/dist-packages/aiogram/dispatcher/handler.py", line 117, in notify
    response = await handler_obj.handler(*args, **partial_data)
  File "main.py", line 105, in info_serv
    await send_server_info(message, result, image_path)
  File "main.py", line 88, in send_server_info
    await message.reply_photo(photo=img_io, caption=caption_text, parse_mode='HTML')
  File "/usr/local/lib/python3.8/dist-packages/aiogram/types/message.py", line 1760, in reply_photo
    return await self.bot.send_photo(
  File "/usr/local/lib/python3.8/dist-packages/aiogram/bot/bot.py", line 565, in send_photo
    result = await self.request(api.Methods.SEND_PHOTO, payload, files)
  File "/usr/local/lib/python3.8/dist-packages/aiogram/bot/base.py", line 236, in request
    return await api.make_request(await self.get_session(), self.server, self.__token, method, data, files,
  File "/usr/local/lib/python3.8/dist-packages/aiogram/bot/api.py", line 140, in make_request
    return check_result(method, response.content_type, response.status, await response.text())
  File "/usr/local/lib/python3.8/dist-packages/aiogram/bot/api.py", line 115, in check_result
    exceptions.BadRequest.detect(description)
  File "/usr/local/lib/python3.8/dist-packages/aiogram/utils/exceptions.py", line 141, in detect
    raise cls(description)
aiogram.utils.exceptions.BadRequest: Image_process_failed

что не так? Спасибо.

 

[DEVV]

Небольшой пример того, как лучше реализовать функцию /sendall в телеграм ботах. Это можно сказать, небольшой архив :)

1. В config.py указываем свои данные

   1. TOKEN_API = "" # Тут токен вашего бота
      ADMIN_ID = # Тут ваш айди в телеграме (@username_to_id_bot) пример: 12345612321

2. Нам нужно хранить всех пользователей в базе данных, я написал такой код: database.py

   1. import sqlite3 as sql
      
      class DataBase:
          def __init__(self, db_file):
              self.connect = sql.connect(db_file)
              self.cursor = self.connect.cursor()
      
          def create_table(self):
              with self.connect:
                  return self.cursor.execute("""
                  CREATE TABLE IF NOT EXISTS `users` (
                      `user_id` INTEGER PRIMARY KEY,
                      `user_name` TEXT,
                      `active` INTEGER DEFAULT 0
                  )
              """)
      
          def user_exists(self, user_id):
              with self.connect:
                  result = self.cursor.execute("""
                      SELECT * FROM `users` WHERE `user_id` = ?
                  """, (user_id,)).fetchmany(1)
      
              return bool(len(result))
      
          def add_user(self, user_id, user_name):
              with self.connect:
                  return self.cursor.execute("""
                      INSERT INTO `users` (`user_id`, `user_name`) VALUES (?, ?)
                  """, (user_id, user_name,))
          
          def set_active(self, user_id, active):
              with self.connect:
                  return self.cursor.execute("UPDATE `users` SET `active` = ? WHERE `user_id` = ?", (active, user_id,))
      
          def get_all_users(self):
              with self.connect:
                  return self.cursor.execute("SELECT `user_id` FROM `users`").fetchall()

3. Теперь можно приступать к реализации функционала, открываем main.py:

   1. from aiogram import Bot, Dispatcher, types, executor
      
      from aiogram.contrib.fsm_storage.memory import MemoryStorage
      from aiogram.dispatcher import FSMContext
      from aiogram.dispatcher.filters.state import StatesGroup, State
      
      from config import *
      from database import DataBase
      
      storage = MemoryStorage()
      bot = Bot(token=TOKEN_API)
      dp = Dispatcher(bot=bot, storage=storage)
      db = DataBase("database.db")
      
      class States(StatesGroup):
          send_message = State()
      
      @dp.message_handler(commands=["start"])
      async def start_cmd(message: types.Message) -> None:
          user_name = message.from_user.username
          user_id = message.from_user.id
          if not db.user_exists(user_id):
              db.add_user(user_id, user_name)
          await bot.send_message(
              message.from_user.id,
              "Привет, я бот рассылка!\n" \
              "Отправьте команду /sendall"
          )
      
      @dp.message_handler(commands=["sendall"], state="*")
      async def set_send_all(message: types.Message) -> None:
          if message.from_user.id == ADMIN_ID:
              await message.answer("Отправьте ваше объявление")
              await States.send_message.set()
          else:
              await message.answer("Увы, но вы не админ!")
      
      @dp.message_handler(state=States.send_message)
      async def send_message(message: types.Message, state: FSMContext) -> None:
          if message.from_user.id == ADMIN_ID:
              text_to_send = message.text
              users = db.get_all_users()
      
              for user in users:
                  user_id = user[0]
                  try:
                      await bot.send_message(user_id, text_to_send)
                  except Exception as e:
                      print(f"Ошибка при отправке сообщения пользователю {user_id}: {e}")
      
              await state.finish()
              await message.answer(f"Сообщение отправлено всем пользователям:\n{text_to_send}")
          else:
              await message.answer("У вас нет прав для выполнения этой команды.")
      
      if __name__ == "__main__":
          db.create_table()
          executor.start_polling(dispatcher=dp,
                                 skip_updates=True)

И наш бот готов! Я воспользовался состоянием машины (state), думаю это лучшее решение.