0.6.1 Официальный шаблон, который позволяет управлять всем, что необходимо
This commit is contained in:
0
Test/commands/__init__.py
Normal file
0
Test/commands/__init__.py
Normal file
63
Test/commands/ban_cmd.py
Normal file
63
Test/commands/ban_cmd.py
Normal file
@@ -0,0 +1,63 @@
|
||||
# BotCode/routers/commands/admin_cmd/ban_cmd.py
|
||||
# Работа с админ-командой /ban, для блокировки пользователей (в разработке)
|
||||
# Проверка на наличие блокировки пользователя в боте
|
||||
|
||||
|
||||
from aiogram import Router, types
|
||||
from aiogram.filters import Command
|
||||
from BotLibrary import *
|
||||
|
||||
# Создание роутера и настройка экспорта
|
||||
__all__ = ("router", "banned_user", "ban_user_by_username",)
|
||||
router = Router(name="ban_router")
|
||||
command_text = "BAN"
|
||||
|
||||
|
||||
# Функция проверки блокировки пользователя в боте
|
||||
@router.message(lambda message: message.from_user.id in ListId.ban_list_id)
|
||||
async def banned_user(message: types_msg.Message):
|
||||
try:
|
||||
# Вывод сообщения пользователю
|
||||
chat_id = await find_chat_id(message)
|
||||
text = (f"{TextDecorator.RED}Получено сообщение от забанненго пользователя"
|
||||
f" из ({chat_id}) : {message.text}{TextDecorator.RESET_DECORATOR}")
|
||||
await message.answer(f"Вы были забаннены в боте @{BotInfo.username}!")
|
||||
|
||||
# Активация логгера
|
||||
await cmd_logginger(message, command_text, text)
|
||||
|
||||
return text
|
||||
|
||||
# Проверка на ошибку и ее логирование
|
||||
except Exception as e:
|
||||
text_error = await error_cmd_logginger(message, command_text, e)
|
||||
return text_error
|
||||
|
||||
|
||||
# Обработчик команды /ban
|
||||
@router.message(Command("ban", "ифт", "бан", ",fy", prefix=BotEdit.prefixs, ignore_case=True))
|
||||
async def ban_user_by_username(message: types_msg.Message):
|
||||
try:
|
||||
text = f"использовал(а) команду /{command_text.lower()}"
|
||||
|
||||
# Получаем аргументы команды
|
||||
args = message.get_args() # Вернет все, что идет после /ban
|
||||
|
||||
# Проверка на наличие аргумента
|
||||
if not args:
|
||||
text = f"Пожалуйста, укажите ID или имя пользователя для бана. Пример: /ban 123456"
|
||||
await message.reply(text)
|
||||
return text
|
||||
|
||||
# Вывод сообщения пользователю
|
||||
await message.reply(text=f"Вы попытались забанить, обидно да?")
|
||||
|
||||
# Активация логгера
|
||||
await cmd_logginger(message, command_text, text)
|
||||
|
||||
return text
|
||||
|
||||
# Проверка на ошибку и ее логирование
|
||||
except Exception as e:
|
||||
text_error = await error_cmd_logginger(message, command_text, e)
|
||||
return text_error
|
||||
63
Test/commands/download_channel_avatar.py
Normal file
63
Test/commands/download_channel_avatar.py
Normal file
@@ -0,0 +1,63 @@
|
||||
# BotCode/routers/downloads/download_channel_avatar.py
|
||||
# Закачка аватарок канала (в разработке + сделать логирование!!!)
|
||||
|
||||
import requests
|
||||
from aiogram import Router
|
||||
from aiogram.types import Message
|
||||
from BotLibrary import *
|
||||
|
||||
# Создание роутера и настройка экспорта
|
||||
__all__ = ("router", "download_channel_avatar",)
|
||||
router = Router(name="avatar_channel_router")
|
||||
type_messages = "AvatarChannel"
|
||||
|
||||
|
||||
# Функция для скачивания аватарки канала
|
||||
@router.message(lambda message: message.chat.type == 'channel')
|
||||
async def download_channel_avatar(message: Message):
|
||||
try:
|
||||
# Логирование для получения сообщения из канала
|
||||
logger.bind(custom_variable=type_messages, user_var=f"{message.chat.id}").info(f"Получено сообщение из канала ID {message.chat.id}: {message.text}")
|
||||
|
||||
# Получаем информацию о чате канала
|
||||
chat_info = await bot.get_chat(message.chat.id) # message.chat.id для получения ID канала
|
||||
|
||||
# Логирование полученной информации
|
||||
logger.bind(custom_variable=type_messages, user_var=f"{chat_info.id}").info(f"Информация о канале: {chat_info.title}")
|
||||
|
||||
# Проверка наличия аватара
|
||||
if not chat_info.photo:
|
||||
text_error = f"Канал с ID {message.chat.id} не имеет аватара."
|
||||
logger.error(text_error)
|
||||
return text_error
|
||||
|
||||
# Получаем file_id для фото (используем big_file_id для лучшего качества)
|
||||
file_info = await bot.get_file(chat_info.photo.big_file_id)
|
||||
|
||||
# Строим URL для скачивания файла
|
||||
file_url = f"https://api.telegram.org/file/bot{bot.token}/{file_info.file_path}"
|
||||
|
||||
# Формируем имя и путь для сохранения фото
|
||||
save_dir = f"{ImportantPath.channel_avatar}/{message.chat.id}"
|
||||
os.makedirs(save_dir, exist_ok=True)
|
||||
file_name = file_info.file_path.split("/")[-1] # Имя файла (например, "abc.jpg")
|
||||
save_path = os.path.join(save_dir, file_name)
|
||||
|
||||
# Скачиваем аватар
|
||||
response = requests.get(file_url)
|
||||
if response.status_code == 200:
|
||||
with open(save_path, "wb") as file:
|
||||
file.write(response.content)
|
||||
text = f"Фото аватара канала с ID {message.chat.id} успешно скачано и сохранено как {save_path}"
|
||||
logger.info(text)
|
||||
return text
|
||||
|
||||
else:
|
||||
text_error = f"Не удалось скачать аватар канала с ID {message.chat.id}. Статус: {response.status_code}"
|
||||
logger.bind(custom_variable=type_messages, user_var=f"{message.chat.id}").error(text_error)
|
||||
return text_error
|
||||
|
||||
except Exception as e:
|
||||
text_error = f"Ошибка при скачивании фото аватара канала с ID {message.chat.id}: {e}"
|
||||
logger.bind(custom_variable=type_messages, user_var=f"{message.chat.id}").error(text_error)
|
||||
return text_error
|
||||
25
Test/commands/find_username.py
Normal file
25
Test/commands/find_username.py
Normal file
@@ -0,0 +1,25 @@
|
||||
# BotLibrary/analitics/find_username.py
|
||||
# Нахождение юзернейма пользователя по id (в разработке)
|
||||
|
||||
from loguru import logger
|
||||
from BotLibrary.library.bots import bot
|
||||
|
||||
# Настройка экспорта
|
||||
__all__ = ("get_user_id_by_username",)
|
||||
type_messages = "ID_USERNAME"
|
||||
|
||||
|
||||
# Получение ID пользователя по юзернейму (в разработке)
|
||||
async def get_user_id_by_username(chat_id, username):
|
||||
try:
|
||||
user = await bot.get_chat_member_by_username(chat_id, username)
|
||||
if user:
|
||||
return user.user.id
|
||||
else:
|
||||
return None
|
||||
|
||||
# Проверка на ошибку и ее логирование (в разработке)
|
||||
except Exception as e:
|
||||
text_error = f"Ошибка при получении ID пользователя: {e}"
|
||||
logger.bind(custom_variable="IDS", user_var=type_messages).error(text_error)
|
||||
return text_error
|
||||
65
Test/commands/send_to_user.py
Normal file
65
Test/commands/send_to_user.py
Normal file
@@ -0,0 +1,65 @@
|
||||
# BotCode/routers/commands/admin_cmd/send_to_user.py
|
||||
# Работа с админ-командой /send, для отправки конкретного сообщения пользователю (в разработке)
|
||||
|
||||
from aiogram import Router, types, F
|
||||
from aiogram.filters import Command
|
||||
from BotLibrary import *
|
||||
|
||||
# Создание роутера и настройка экспорта
|
||||
__all__ = ("router", "send_message",)
|
||||
router = Router(name="send_router")
|
||||
command_text = "Send"
|
||||
|
||||
|
||||
# Обработчик команды /send для отправки сообщения определенному пользователю (в разработке)
|
||||
@router.message(F.from_user.id.in_(ListId.important),
|
||||
Command("send", "отправить", "отправ", "s", "ыутв", "jnghfdbnm", "jnghfd",
|
||||
prefix=BotEdit.prefixs, ignore_case=True))
|
||||
async def send_message(message: types_msg.Message):
|
||||
try:
|
||||
if message.chat.id in ListId.adm_list_id:
|
||||
text = f"использовал(а) команду /{command_text.lower()}"
|
||||
|
||||
# Разбиваем сообщение на аргументы
|
||||
args = message.text.split()
|
||||
|
||||
# Проверка на правильность команды /send
|
||||
if len(args) < 3:
|
||||
texts = "Некорректный формат команды. Используйте: /send <user_id> <текст>"
|
||||
await message.reply(texts)
|
||||
return texts
|
||||
|
||||
# Получаем user_id и текст сообщения
|
||||
user_id = int(args[1])
|
||||
text_send = ' '.join(args[2:])
|
||||
|
||||
# Отправляем сообщение пользователю
|
||||
await bot.send_message(chat_id=user_id, text=text_send)
|
||||
|
||||
# Логирование
|
||||
user_id = find_imp_id(user_id)
|
||||
await cmd_logginger(message, command_text, text)
|
||||
|
||||
# Логирование и отчет об отправке
|
||||
await message.reply(f"Сообщение успешно отправлено пользователю с ID {user_id}")
|
||||
return text
|
||||
|
||||
# Проверка на ошибку и ее логирование
|
||||
except Exception as e:
|
||||
text_error = await error_cmd_logginger(message, command_text, e)
|
||||
return text_error
|
||||
|
||||
# Проверка заблокирован ли бот для пользователя
|
||||
# except exceptions.BotBlocked:
|
||||
# await message.answer("Пользователь заблокировал бота")
|
||||
# except aiogram.utils.exceptions.ChatNotFound:
|
||||
# await message.answer("Чат с пользователем не найден")
|
||||
# except exceptions.RetryAfter as e:
|
||||
# await asyncio.sleep(e.timeout)
|
||||
# return await send_message(message)
|
||||
# except exceptions.UserDeactivated:
|
||||
# await message.answer("Пользователь деактивирован")
|
||||
# except exceptions.TelegramAPIError:
|
||||
# await message.answer("Произошла ошибка при отправке сообщения")
|
||||
# except:
|
||||
# return
|
||||
Reference in New Issue
Block a user