Compare commits

..

20 Commits

Author SHA1 Message Date
51e1d59b12 fixed bot.send_message 2025-08-14 14:04:20 +03:00
a60c6b2ee9 updated main.py 2025-08-14 14:01:06 +03:00
4cbb60fdf4 added botdata command 2025-08-14 13:08:09 +03:00
f667ac7085 Merge pull request 'added utils.py' (#7) from devel into master
Reviewed-on: #7
2025-07-28 18:24:49 +00:00
853bfcdbb4 added utils.py
1. Добавил message_thread_id для всех команд, убрав костыль, который отвечал за корректную отправку
   сообщений в топик.
2. Функции (определение администратора, удаление сообщений) вынес в utils.py, от куда они будут вызываться в командах. Модули стали более читаемы из-за уменьшения количества строк кода в них.
3. Дописал manual_unban и добавил error в config.py
4. Оптимизация
2025-07-28 21:03:28 +03:00
d73c0079f0 deleted the to-do list in README.md 2025-07-27 11:32:57 +03:00
3f23b4c708 mute and ban commands with photos 2025-07-25 20:19:07 +03:00
a714e0d05c fix: updated help in all commands 2025-07-21 14:16:17 +03:00
40b60baa93 added sending logs to the admin chat 2025-07-16 10:28:42 +03:00
fc3ef20145 chat definition 2025-07-15 20:55:56 +03:00
bc909adf4e added and updaded help 2025-07-14 20:15:48 +03:00
37c596c5c7 removed indentation 2025-07-14 20:14:11 +03:00
de079a131f fixed the time for deleting messages with a manual 2025-07-14 19:56:27 +03:00
d5c207e1bc finalize chat_events.py 2025-07-13 14:03:58 +03:00
40f75f7ad8 updated the log in the mute command 2025-07-13 13:34:20 +03:00
2648d6f4f1 added the unmute command 2025-07-13 13:28:17 +03:00
c05703c1b1 finalized the unban command 2025-07-13 12:57:43 +03:00
a7f9ab26ad updated logging 2025-07-12 13:19:28 +03:00
5f38f8c603 Merge pull request 'finalized the mute command' (#6) from updade/mute-command into master
Reviewed-on: #6
2025-07-12 09:44:00 +00:00
ed4bbacaf1 finalized the mute command 2025-07-12 12:42:47 +03:00
16 changed files with 1444 additions and 582 deletions

View File

@@ -1 +1,4 @@
BOT_TOKEN = "..." # Токен бота получать у @BotFather BOT_TOKEN = "..." # Токен бота получать у @BotFather
ADMIN_CHAT_ID = -1001111111111 # ID админ-чата получать у @username_to_id_bot
LOG_THREAD_ID = 2 # ID топика, брать из ссылки сообщения
ADMIN_IDS = "11111,22222" # ID администраторов получать у @username_to_id_bot

View File

@@ -1,21 +1,9 @@
<div align="center"> <div align="center">
<h1 align="center">LGBot</h1> <h1 align="center">LGBot</h1>
<p align="center">Бот-модератор для @linux_gaming_ru </p> <p align="center">Бот-администратор для @linux_gaming_ru </p>
</div> </div>
## Список дел ### Установка зависимостей (pyenv)
- [X] Команда /start
- [ ] Команда /help
- [ ] Команда /mute
- [ ] Команда /unmute
- [X] Команда /ban
- [X] Команда /unban
- [ ] Фильтрация сообщений
- [ ] Удаление сообщений с матом
- [ ] Удаление рекламы
### Установка зависимостей (через pyenv)
```sh ```sh
pyenv install 3.11.0 pyenv install 3.11.0
@@ -30,7 +18,7 @@ pip install -r requirements.txt
### Настройка ### Настройка
Создатите файл **.env** и внесите в него ваш токен, который вы получили у @BotFather. Создайте файл **.env** и внесите в него токен бота, ID админ-чата и топика.
### Запуск ### Запуск

87
src/action_reporter.py Normal file
View File

@@ -0,0 +1,87 @@
from telebot.async_telebot import AsyncTeleBot
from telebot.types import Message
import logging
import os
from database import db
# Инициализация
class ActionReporter:
def __init__(self, bot: AsyncTeleBot, log_chat_id: int, log_thread_id: int):
self.bot = bot
self.log_chat_id = log_chat_id
self.log_thread_id = log_thread_id
# Получает информацию о пользователе из базы данных
async def _get_user_info(self, user_id: int) -> str:
user_info = db.get_user(user_id)
if user_info:
# Вытаскиваем данные с кортежа
_, nickname, tag = user_info
# Формируем справку о пользователе
text = "👤 <b>Пользователь:</b>\n"
if nickname:
text += f"• Name: <code>{nickname}</code>\n"
if tag:
text += f"• Tag: <code>@{tag}</code>\n"
text += f"• ID: <code>{user_id}</code>"
return text
# Получает информацию об администраторе
async def _get_admin_info(self, admin_id: int) -> str:
admin_info = db.get_user(admin_id)
if admin_info:
# Вытаскиваем данные с кортежа
_, nickname, tag = admin_info
# Формируем справку об администраторе
text = "🛡 <b>Администратор:</b>\n"
if nickname:
text += f"• Name: <code>{nickname}</code>\n"
if tag:
text += f"• Tag: <code>@{tag}</code>\n"
text += f"• ID: <code>{admin_id}</code>"
return text
# Отправляет лог действия в админ-чат
async def log_action(self, action: str, user_id: int, admin_id: int, reason: str, duration: str, photo_path: str):
try:
# Получаем информацию о пользователе и администраторе
user_info = await self._get_user_info(user_id)
admin_info = await self._get_admin_info(admin_id)
# Формируем сообщение
msg = f"⚙️ <b>Действие:</b> {action}\n"
if duration:
msg += f"⏱ <b>Длительность:</b> {duration}\n"
if reason:
msg += f"📝 <b>Причина:</b> <i>{reason}</i>\n"
msg += f"\n{user_info}\n\n{admin_info}"
# Отправляем лог с изображением
if photo_path and os.path.exists(photo_path):
with open(photo_path, 'rb') as photo_file:
await self.bot.send_photo(self.log_chat_id, photo_file, caption=msg, message_thread_id=self.log_thread_id)
# Отправляем лог без изображения
else:
await self.bot.send_message(self.log_chat_id, msg, message_thread_id=self.log_thread_id)
except Exception as e:
logging.error(f"Ошибка отправки лога: {str(e)}")
# Создаем глобальный экземпляр для импорта
action_reporter = None
# Инициализирует логгер модерации
def init_action_reporter(bot: AsyncTeleBot, log_chat_id: int, log_thread_id: int):
global action_reporter
action_reporter = ActionReporter(bot, log_chat_id, log_thread_id)

View File

@@ -4,27 +4,78 @@ MODULES_DIR = 'modules'
# Название файла db sqlite # Название файла db sqlite
DATABASE_NAME = 'users.db' DATABASE_NAME = 'users.db'
# Название файла для логов
LOG_FILE_NAME = 'bot.log'
# Сообщения команд # Сообщения команд
COMMAND_MESSAGES = { COMMAND_MESSAGES = {
'start': 'Бот-модератор для чата @linux_gaming_ru', 'start': 'Бот-администратор для чата @linux_gaming_ru',
'help': 'пусто', 'help': (
"<b>📚 Справочник команд администратора</b>\n\n"
"<u>Основные команды:</u>\n"
"• <code>/start</code> - Начало работы\n"
"• <code>/help</code> - Этот справочник\n\n"
"<u>🛠 Команды модерации:</u>\n"
"• <code>/mute help</code> - Инструкция по муту\n"
"• <code>/unmute help</code> - Снятие мута\n"
"• <code>/ban help</code> - Инструкция по бану\n"
"• <code>/unban help</code> - Снятие бана\n\n"
"<i> Для подробностей по конкретной команде используйте: /команда help</i>"
),
'manual_mute': ( 'manual_mute': (
' Использование мута:\n' "<b>🔇 Команда /mute</b>\n\n"
'1⃣ Ответьте на сообщение: <code>/mute время</code>\n' "<i>Ограничивает права пользователя на указанное время</i>\n\n"
'2⃣ Укажите тэг: <code>/mute @username время</code>\n' "<u>🕒 Форматы времени:</u>\n"
'3⃣ Укажите ID: <code>/mute 123456789 время</code>\n\n' "• Минуты: <code>10м</code>, <code>30м</code>\n"
"• Часы: <code>1ч</code>, <code>3ч</code>\n"
"• Дни: <code>1д</code>, <code>7д</code>\n\n"
"<u>🎯 Способы использования:</u>\n"
"1. Ответ на сообщение:\n"
" <code>/mute 30м причина</code>\n"
"2. По тегу пользователя:\n"
" <code>/mute @username 1ч спам</code>\n"
"3. По ID пользователя:\n"
" <code>/mute 123456789 1д нарушение правил</code>\n\n"
"<b>⚠️ Максимальный срок: 30 дней</b>\n"
"<i> Причину стараться указывать</i>"
),
'manual_unmute': (
"<b>🔊 Команда /unmute</b>\n\n"
"<i>Снимает ограничения с пользователя</i>\n\n"
"<u>🎯 Способы использования:</u>\n"
"1. Ответ на сообщение:\n"
" <code>/unmute</code>\n"
"2. По тегу пользователя:\n"
" <code>/unmute @username</code>\n"
"3. По ID пользователя:\n"
" <code>/unmute 123456789</code>\n\n"
"<i> Работает только для временно замученных пользователей</i>"
), ),
'manual_ban': ( 'manual_ban': (
' Использование бана:\n' "<b>🚫 Команда /ban</b>\n\n"
'1⃣ Ответьте на сообщение: <code>/ban</code>\n' "<i>Навсегда исключает пользователя из чата</i>\n\n"
'2⃣ Укажите тэг: <code>/ban @username</code>\n' "<u>🎯 Способы использования:</u>\n"
'3⃣ Укажите ID: <code>/ban 123456789</code>' "1. Ответ на сообщение:\n"
" <code>/ban причина</code>\n"
"2. По тегу пользователя:\n"
" <code>/ban @username спам</code>\n"
"3. По ID пользователя:\n"
" <code>/ban 123456789 нарушение правил</code>\n\n"
"<b>⚠️ Добавляет в ЧС</b>\n"
"<i> Для разбана используйте /unban</i>"
), ),
'manual_unban': ( 'manual_unban': (
' Использование разбана:\n' "<b>✅ Команда /unban</b>\n\n"
'1⃣ Ответьте на сообщение: <code>/unban</code>\n' "<i>Снимает бан с пользователя</i>\n\n"
'2⃣ Укажите тэг: <code>/unban @username</code>\n' "<u>🎯 Способы использования:</u>\n"
'3⃣ Укажите ID: <code>/unban 123456789</code>' "1. Ответ на сообщение:\n"
" <code>/unban</code>\n"
"2. По тегу пользователя:\n"
" <code>/unban @username</code>\n"
"3. По ID пользователя:\n"
" <code>/unban 123456789</code>\n\n"
"<b>⚠️ Работает только для забаненных через /ban</b>\n"
"<i> Пользователь сможет снова присоединиться</i>"
), ),
'no_admin_rights': '❌ Только администраторы могут использовать эту команду.', 'no_admin_rights': '❌ Только администраторы могут использовать эту команду.',
'no_restrict_rights': 'У вас недостаточно прав.', 'no_restrict_rights': 'У вас недостаточно прав.',
@@ -35,8 +86,10 @@ COMMAND_MESSAGES = {
'cant_mute_admin': '❌ Невозможно замутить администратора.', 'cant_mute_admin': '❌ Невозможно замутить администратора.',
'cant_ban_admin': '❌ Невозможно забанить администратора.', 'cant_ban_admin': '❌ Невозможно забанить администратора.',
'muted': '✅ Пользователь замучен на {time_display}.', 'muted': '✅ Пользователь замучен на {time_display}.',
'unmuted': '✅ Пользователь размучен.',
'banned': '✅ Пользователь успешно забанен.', 'banned': '✅ Пользователь успешно забанен.',
'unbanned': '✅ Пользователь успешно разбанен.', 'unbanned': '✅ Пользователь успешно разбанен.',
'error': '⚠️ Ошибка: {e}',
'general_error': '⚠️ Произошла непредвиденная ошибка.' 'general_error': '⚠️ Произошла непредвиденная ошибка.'
} }

View File

@@ -2,6 +2,8 @@ import logging
import time import time
import os import os
from config import LOG_FILE_NAME
class ColoredFormatter(logging.Formatter): # Цветные логи (для терминала) class ColoredFormatter(logging.Formatter): # Цветные логи (для терминала)
LEVEL_COLORS = { LEVEL_COLORS = {
logging.INFO: '\033[92m', logging.INFO: '\033[92m',
@@ -39,6 +41,13 @@ class UncoloredFormatter(logging.Formatter): # Бесцветные логи (д
def setup_logging(): # Инициализирует систему логирования def setup_logging(): # Инициализирует систему логирования
# Добавляем разделитель для нового сеанса
if os.path.exists(LOG_FILE_NAME):
with open(LOG_FILE_NAME, "a", encoding="utf-8") as f:
f.write("\n" + "=" * 60 + "\n")
f.write(f"{'ЗАПУЩЕН НОВЫЙ СЕАНС':^60}\n")
f.write("=" * 60 + "\n\n")
# Создаем корневой логгер # Создаем корневой логгер
logger = logging.getLogger() logger = logging.getLogger()
logger.setLevel(logging.INFO) logger.setLevel(logging.INFO)
@@ -49,7 +58,7 @@ def setup_logging(): # Инициализирует систему логиро
console_handler.setFormatter(ColoredFormatter()) console_handler.setFormatter(ColoredFormatter())
# Сохраняем логи в файл # Сохраняем логи в файл
file_handler = logging.FileHandler("bot.log", encoding='utf-8') file_handler = logging.FileHandler(LOG_FILE_NAME, encoding='utf-8')
file_handler.setFormatter(UncoloredFormatter()) file_handler.setFormatter(UncoloredFormatter())
logger.addHandler(console_handler) logger.addHandler(console_handler)

View File

@@ -12,13 +12,21 @@ from logger import setup_logging
from database import db from database import db
from action_reporter import init_action_reporter
from config import MODULES_DIR from config import MODULES_DIR
load_dotenv() # Загружаем токен бота из .env # Загружаем токен бота из .env
load_dotenv()
bot = AsyncTeleBot(os.getenv("BOT_TOKEN"), parse_mode="html") bot = AsyncTeleBot(os.getenv("BOT_TOKEN"), parse_mode="html")
logger = logging.getLogger(__name__) # Получаем логгер для текущего модуля # Загружаем ID админ-чата из .env и инициализируемся для логов в чат
init_action_reporter(bot, os.getenv("ADMIN_CHAT_ID"), os.getenv("LOG_THREAD_ID"))
# Получаем логгер для текущего модуля
logger = logging.getLogger(__name__)
# Middleware для автоматического обновления информации о пользователях в базе данных
class UserUpdateMiddleware(BaseMiddleware): class UserUpdateMiddleware(BaseMiddleware):
def __init__(self, db): def __init__(self, db):
super().__init__() super().__init__()
@@ -27,6 +35,7 @@ class UserUpdateMiddleware(BaseMiddleware):
self.update_types = ['message', 'chat_member'] self.update_types = ['message', 'chat_member']
self.db = db self.db = db
# Обработчик, вызываемый ДО обработки сообщения основными хэндлерами
async def pre_process(self, message, data): async def pre_process(self, message, data):
# Обработка пользователей, отправившие сообщение # Обработка пользователей, отправившие сообщение
@@ -47,43 +56,78 @@ class UserUpdateMiddleware(BaseMiddleware):
) )
return data return data
# Обработчик, вызываемый ПОСЛЕ обработки сообщения основными хэндлерами
async def post_process(self, message, data, exception): async def post_process(self, message, data, exception):
pass pass
# Регистрируем middleware # Регистрируем middleware
bot.setup_middleware(UserUpdateMiddleware(db)) bot.setup_middleware(UserUpdateMiddleware(db))
async def load_modules(): # Загружает все модули из директории /modules # Загружает все модули из директории /modules
async def load_modules():
setup_logging() # Инициализация логирования # Инициализация логирования
setup_logging()
loaded_count = 0 # Переменная для подсчёта модулей # Переменная для подсчёта модулей
modules_path = os.path.join(os.path.dirname(__file__), MODULES_DIR) # Импортируем относительный путь проекта loaded_count = 0
# Импортируем относительный путь проекта
modules_path = os.path.join(os.path.dirname(__file__), MODULES_DIR)
for filename in os.listdir(modules_path): for filename in os.listdir(modules_path):
# Если файл содержит в конце .py (кроме __init__.py)
if filename.endswith(".py") and filename != "__init__.py": if filename.endswith(".py") and filename != "__init__.py":
module_name = filename[:-3] # Убираем расширение .py
# Убираем расширение .py
module_name = filename[:-3]
try: try:
# Импортируем модуль (modules.start)
module = importlib.import_module(f"{MODULES_DIR}.{module_name}") module = importlib.import_module(f"{MODULES_DIR}.{module_name}")
# Если присутствует register_handlers
if hasattr(module, "register_handlers"): if hasattr(module, "register_handlers"):
module.register_handlers(bot) module.register_handlers(bot)
loaded_count += 1 loaded_count += 1
logger.info(f"Модуль {module_name} успешно загружен") logger.info(f"Модуль {module_name} успешно загружен.")
# Если нет register_handlers
else: else:
logger.warning(f"Модуль {module_name} не содержит функцию register_handlers")
# Записываем действие в логи
logger.warning(f"Модуль {module_name} не содержит функцию register_handlers.")
except Exception as e: except Exception as e:
# Записываем ошибку в логи
logger.error(f"Ошибка при загрузке модуля {module_name}: {str(e)}") logger.error(f"Ошибка при загрузке модуля {module_name}: {str(e)}")
# Записываем отчет о модулях в логи
logger.info(f"Загружено модулей: {loaded_count} шт. Бот запущен.") logger.info(f"Загружено модулей: {loaded_count} шт. Бот запущен.")
async def main(): async def main():
os.system('clear') # Очищаем терминал
# Очищаем терминал
os.system('clear')
try: try:
await load_modules() # Проверяем и загружаем модули
await bot.infinity_polling() # Запускаем бота # Проверяем и загружаем модули
except (KeyboardInterrupt, asyncio.CancelledError): await load_modules()
logger.info("Бот остановлен.")
# Запускаем бота
await bot.infinity_polling()
except Exception as e: except Exception as e:
# Записываем критическую ошибку в логи
logger.critical(f"Критическая ошибка: {str(e)}") logger.critical(f"Критическая ошибка: {str(e)}")
# Завершаем скрипт с критической ошибкой
sys.exit(1) sys.exit(1)
if __name__ == "__main__": if __name__ == "__main__":

View File

@@ -4,34 +4,27 @@ import asyncio
import logging import logging
from database import db from database import db
from action_reporter import action_reporter
from utils import (
delete_messages,
check_admin_status,
check_target_status,
)
from config import COMMAND_MESSAGES from config import COMMAND_MESSAGES
logger = logging.getLogger(__name__) # Получаем логгер для текущего модуля logger = logging.getLogger(__name__) # Получаем логгер для текущего модуля
# Возвращает причину бана # Регистрирует все обработчики команд
def extract_reason(words: str) -> str: def register_handlers(bot: AsyncTeleBot):
if words == []: # Обработчик команды /ban
reason = 'отсутствует' @bot.message_handler(commands=['ban'])
else: async def _ban_command_wrapper(message: Message):
reason = ' '.join(words) await ban_command(bot, message)
return reason # Основная функция команды /ban
async def ban_command(bot: AsyncTeleBot, message: Message, photo_path: str = None):
# Удаляет два последних сообщения
async def delete_messages(bot: AsyncTeleBot, message: Message, time_sleep: int):
await asyncio.sleep(time_sleep)
await bot.delete_message(message.chat.id, message.message_id)
await bot.delete_message(message.chat.id, message.message_id+1)
def register_handlers(bot: AsyncTeleBot): # Регистрирует все обработчики команд
@bot.message_handler(commands=['ban']) # Обработчик команды /ban
async def ban_command(message: Message):
# Отправка сообщения в тему или обычный чат
send_message = bot.reply_to if message.is_topic_message else bot.send_message
chat_id = message if message.is_topic_message else message.chat.id
# Определяем целевого пользователя # Определяем целевого пользователя
target_user = None target_user = None
@@ -42,91 +35,81 @@ def register_handlers(bot: AsyncTeleBot): # Регистрирует все об
# Разбиваем текст сообщения на части # Разбиваем текст сообщения на части
parts_msg = message.text.split() parts_msg = message.text.split()
try: # Команда /ban help
# Проверяем, является ли отправитель администратором if len(parts_msg) == 2 and parts_msg[1].strip() in ('help', 'помощь'):
try:
admin_status = await bot.get_chat_member(message.chat.id, message.from_user.id)
# Проверяем статус администратора (создателя) # Отправляем инструкцию
if admin_status.status not in ['administrator', 'creator']: await bot.send_message(
await send_message(chat_id, COMMAND_MESSAGES['no_admin_rights']) chat_id=message.chat.id,
text=COMMAND_MESSAGES['manual_ban'],
# Удаляем сообщения через 5 секунд message_thread_id=message.message_thread_id,
await delete_messages(bot, message, 5) )
return
# Проверяем право администратора на бан
if admin_status.status == 'administrator' and not admin_status.can_restrict_members:
await send_message(chat_id, COMMAND_MESSAGES['no_restrict_rights'])
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)
return
except Exception as e:
await send_message(chat_id, f"⚠️ Ошибка: {str(e)}")
logger.error(f"Ошибка при получении статуса администратора: {str(e)}")
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)
return
# Выводим помощь (/ban help)
if len(parts_msg) == 2 and parts_msg[1].strip() in ['help', 'помощь']:
await send_message(chat_id, COMMAND_MESSAGES['manual_ban'])
# Удаляем сообщения через 30 секунд # Удаляем сообщения через 30 секунд
await delete_messages(bot, message, 30) await delete_messages(bot, message, time_sleep=30, number_message=2)
return return
# Если одно слово, то ответом на сообщение try:
# Проверяем, является ли отправитель администратором
if await check_admin_status(bot, message) == 1:
return
# Команда через ответ на сообщение, если одно слово (/ban)
if len(parts_msg) == 1: if len(parts_msg) == 1:
# Если это тема # Если это топик
if message.is_topic_message: if message.is_topic_message:
# Если без ответа на сообщение, ошибка # Если без ответа на сообщение
if message.message_thread_id == message.reply_to_message.message_id: if message.message_thread_id == message.reply_to_message.message_id:
await asyncio.sleep(3)
await bot.delete_message(message.chat.id, message.message_id) # Удаляем сообщение через 3 секунды
await delete_messages(bot, message, time_sleep=3, number_message=1)
return return
# Если с ответом на сообщение # Если с ответом на сообщение
else: else:
target_user = message.reply_to_message.from_user
reason = extract_reason(parts_msg[1:])
# Если это обычный чат # Собираем данные
target_user = message.reply_to_message.from_user
reason = 'отсутствует'
# Если это General (обычный чат)
elif message.reply_to_message and message.is_topic_message is None: elif message.reply_to_message and message.is_topic_message is None:
target_user = message.reply_to_message.from_user
reason = extract_reason(parts_msg[1:])
# Удаляем сообщение, если команда неправильная # Собираем данные
target_user = message.reply_to_message.from_user
reason = 'отсутствует'
message.message_thread_id = None
# Если команда неправильная
else: else:
await asyncio.sleep(3)
await bot.delete_message(message.chat.id, message.message_id) # Удаляем сообщение через 3 секунды
await delete_messages(bot, message, time_sleep=3, number_message=1)
return return
# В сообщении больше одного слова
else: else:
# Если второе слово это тег или ID # Если второе слово это тег или ID
if parts_msg[1].strip().isdigit() or parts_msg[1].startswith('@'): if parts_msg[1].strip().isdigit() or parts_msg[1].startswith('@'):
# Собираем данные
identifier = parts_msg[1].strip() identifier = parts_msg[1].strip()
reason = extract_reason(parts_msg[2:]) reason = ' '.join(parts_msg[2:]) if parts_msg[2:] != [] else 'отсутствует'
# Поиск по ID # Делаем поиск по ID
if identifier.isdigit(): if identifier.isdigit():
# Делаем в int и ищем # Ищем пользователя в базе данных
user_info = db.get_user(int(identifier)) user_info = db.get_user(int(identifier))
# Если нашли пользователя
if user_info: if user_info:
# Создаем объект пользователя из данных базы
# Создаем объект пользователя
target_user = User( target_user = User(
id=user_info[0], id=user_info[0],
first_name=user_info[1], first_name=user_info[1],
@@ -134,14 +117,16 @@ def register_handlers(bot: AsyncTeleBot): # Регистрирует все об
is_bot=False is_bot=False
) )
# Поиск по тэгу # Делаем поиск по тегу
elif identifier.startswith('@'): elif identifier.startswith('@'):
# Убираем @ и ищем # Ищем пользователя в базе данных (убрали @)
user_info = db.get_user_by_username(identifier[1:]) user_info = db.get_user_by_username(identifier[1:])
# Если нашли пользователя
if user_info: if user_info:
# Создаем объект пользователя из данных базы
# Создаем объект пользователя
target_user = User( target_user = User(
id=user_info[0], id=user_info[0],
first_name=user_info[1], first_name=user_info[1],
@@ -149,87 +134,116 @@ def register_handlers(bot: AsyncTeleBot): # Регистрирует все об
is_bot=False is_bot=False
) )
# Команда через ответ на сообщение с указанием причины
else: else:
# Если это тема # Если это топик
if message.is_topic_message: if message.is_topic_message:
# Если без ответа на сообщение, ошибка # Если без ответа на сообщение
if message.message_thread_id == message.reply_to_message.message_id: if message.message_thread_id == message.reply_to_message.message_id:
await asyncio.sleep(3)
await bot.delete_message(message.chat.id, message.message_id) # Удаляем сообщение через 3 секунды
await delete_messages(bot, message, time_sleep=3, number_message=1)
return return
# Если с ответом на сообщение # Если с ответом на сообщение
else: else:
target_user = message.reply_to_message.from_user
reason = extract_reason(parts_msg[1:])
# Если это обычный чат # Собираем данные
target_user = message.reply_to_message.from_user
reason = ' '.join(parts_msg[1:])
# Если это General (обычный чат)
elif message.reply_to_message and message.is_topic_message is None: elif message.reply_to_message and message.is_topic_message is None:
target_user = message.reply_to_message.from_user
reason = extract_reason(parts_msg[1:])
# Удаляем сообщение, если команда неправильная # Собираем данные
target_user = message.reply_to_message.from_user
reason = ' '.join(parts_msg[1:])
message.message_thread_id = None
# Если команда неправильная
else: else:
await asyncio.sleep(3)
await bot.delete_message(message.chat.id, message.message_id) # Удаляем сообщение через 3 секунды
await delete_messages(bot, message, time_sleep=3, number_message=1)
return return
# Если пользователь не найден # Если пользователь не найден в базе данных
if not target_user: if not target_user:
await send_message(chat_id, COMMAND_MESSAGES['user_not_found'])
# Отправляем предупреждение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['user_not_found'],
message_thread_id=message.message_thread_id,
)
# Удаляем сообщения через 5 секунд # Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5) await delete_messages(bot, message, time_sleep=5, number_message=2)
return return
# Проверяем статус целевого пользователя # Проверяем статус целевого пользователя
if await check_target_status(bot, message, target_user) == 1:
return
try: try:
target_status = await bot.get_chat_member(message.chat.id, target_user.id)
# Проверяем, является ли цель администратором или создателем
if target_status.status in ['administrator', 'creator']:
await send_message(chat_id, COMMAND_MESSAGES['cant_ban_admin'])
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)
return
except Exception as e:
await send_message(chat_id, f"⚠️ Ошибка: {str(e)}")
logger.error(f"Ошибка при получении статуса пользователя: {str(e)}")
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)
return
# Выполняем бан # Выполняем бан
try: await bot.ban_chat_member(
await bot.ban_chat_member(message.chat.id, target_user.id) chat_id=message.chat.id,
user_id=target_user.id,
)
# Отправляем лог в админ-чат
await action_reporter.log_action(
action="БАН",
user_id=target_user.id,
admin_id=message.from_user.id,
reason=reason,
duration=None,
photo_path=photo_path,
)
# Отправляем сообщения, что пользователь получил бан # Отправляем сообщения, что пользователь получил бан
await send_message(chat_id, COMMAND_MESSAGES['banned']) await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['banned'],
message_thread_id=message.message_thread_id,
)
# Записываем действие в логи
logger.info(f"Пользователь {target_user.id} забанен администратором {message.from_user.id}.") logger.info(f"Пользователь {target_user.id} забанен администратором {message.from_user.id}.")
# Удаляем сообщения через 5 секунд # Удаляем сообщения через 5 секунд
await asyncio.sleep(5) await delete_messages(bot, message, time_sleep=5, number_message=2)
await bot.delete_message(message.chat.id, message.message_id)
await bot.delete_message(message.chat.id, message.message_id+2)
except Exception as e: except Exception as e:
await send_message(chat_id, f"⚠️ Ошибка: {str(e)}")
# Отправляем ошибку
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['error'].format(e=str(e)),
message_thread_id=message.message_thread_id,
)
# Записываем ошибку в логи
logger.error(f"Ошибка бана: {str(e)}") logger.error(f"Ошибка бана: {str(e)}")
# Удаляем сообщения через 5 секунд # Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5) await delete_messages(bot, message, time_sleep=5, number_message=2)
except Exception as e: except Exception as e:
await send_message(chat_id, COMMAND_MESSAGES['general_error'])
# Отправляем ошибку
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['general_error'],
message_thread_id=message.message_thread_id,
)
# Записываем ошибку в логи
logger.error(f"Общая ошибка в ban_command: {str(e)}") logger.error(f"Общая ошибка в ban_command: {str(e)}")
# Удаляем сообщения через 5 секунд # Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5) await delete_messages(bot, message, time_sleep=5, number_message=2)

67
src/modules/botdata.py Normal file
View File

@@ -0,0 +1,67 @@
from telebot.async_telebot import AsyncTeleBot
from telebot.types import Message
import asyncio
import logging
import os
from utils import delete_messages
from config import COMMAND_MESSAGES, DATABASE_NAME, LOG_FILE_NAME
# Получаем логгер для текущего модуля
logger = logging.getLogger(__name__)
# Загружаем id администраторов из .env
ADMIN_IDS = [int(id_str.strip()) for id_str in os.getenv('ADMIN_IDS').split(',')]
# Регистрирует все обработчики команд
def register_handlers(bot: AsyncTeleBot):
# Обработчик команды /botdata
@bot.message_handler(commands=['botdata'])
async def botdata_command(message: Message):
try:
# Если id администратора совпадает
if message.from_user.id in ADMIN_IDS:
# Отправляем базу данных
await bot.send_document(
chat_id=message.chat.id,
document=open(DATABASE_NAME, 'rb')
)
# Отправляем файл с логами
await bot.send_document(
chat_id=message.chat.id,
document=open(LOG_FILE_NAME, 'rb')
)
# Записываем действие в логи
logger.info(f"Администратор {message.from_user.id} запустил /botdata.")
# Если id администратора не совпадает
else:
# Отправляем предупреждение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['no_admin_rights'],
message_thread_id=message.message_thread_id,
)
# Записываем действие в логи
logger.info(f"Пользователь {message.from_user.id} запустил /botdata.")
except Exception as e:
# Отправляем ошибку
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['general_error'],
message_thread_id=message.message_thread_id,
)
# Записываем ошибку в логи
logger.error(f"Общая ошибка в botdata_command: {str(e)}")

View File

@@ -10,13 +10,13 @@ def register_handlers(bot: AsyncTeleBot): # Регистрирует все об
@bot.message_handler(content_types=['new_chat_members']) # Обработчик захода @bot.message_handler(content_types=['new_chat_members']) # Обработчик захода
async def handle_new_members(message: Message): async def handle_new_members(message: Message):
await asyncio.sleep(10)
await bot.delete_message(message.chat.id, message.message_id)
for new_member in message.new_chat_members: for new_member in message.new_chat_members:
logger.info(f"Пользователь {new_member.id} зашёл в чат.") logger.info(f"Пользователь {new_member.id} зашёл в чат.")
await asyncio.sleep(10)
await bot.delete_message(message.chat.id, message.message_id)
@bot.message_handler(content_types=['left_chat_member']) # Обработчик выхода @bot.message_handler(content_types=['left_chat_member']) # Обработчик выхода
async def handle_left_members(message: Message): async def handle_left_members(message: Message):
logger.info(f"Пользователь {message.left_chat_member.id} вышел из чата.")
await asyncio.sleep(10) await asyncio.sleep(10)
await bot.delete_message(message.chat.id, message.message_id) await bot.delete_message(message.chat.id, message.message_id)
logger.info(f"Пользователь {message.left_chat_member.id} вышел из чата.")

View File

@@ -0,0 +1,70 @@
from telebot.async_telebot import AsyncTeleBot
from telebot.types import Message
import asyncio
import os
import logging
from database import db
# Импортируем обработчики команд
from modules.mute import mute_command
from modules.ban import ban_command
logger = logging.getLogger(__name__) # Получаем логгер для текущего модуля
def register_handlers(bot: AsyncTeleBot): # Регистрирует все обработчики событий
# Обработчик изображений
@bot.message_handler(content_types=['photo'])
async def message_photo(message: Message):
# Определяем путь к изображению
photo_path = None
try:
# Проверяем, есть ли подпись
if not message.caption:
return
# Разделяем подпись на части
parts = message.caption.split()
# Определяем команду (первое слово в подписи)
command = parts[0].lower()
# Поддерживаемые команды
supported_commands = {
'/mute': mute_command,
'/ban': ban_command
}
# Проверяем, является ли первое слово командой
if command not in supported_commands:
return
# Скачиваем изображение
file_info = await bot.get_file(message.photo[-1].file_id)
os.makedirs("tmp", exist_ok=True)
photo_path = f"tmp/{file_info.file_id}.jpg"
downloaded_file = await bot.download_file(file_info.file_path)
with open(photo_path, 'wb') as new_file:
new_file.write(downloaded_file)
# Переносим caption в text
message.text = message.caption
# Вызываем обработчик команды
handler = supported_commands[command]
await handler(bot, message, photo_path=photo_path)
except Exception as e:
logger.error(f"Ошибка обработки команды с изображением: {str(e)}")
finally:
if photo_path and os.path.exists(photo_path):
try:
os.remove(photo_path)
except Exception as e:
logger.error(f"Ошибка удаления временного изображения: {str(e)}")

48
src/modules/help.py Normal file
View File

@@ -0,0 +1,48 @@
from telebot.async_telebot import AsyncTeleBot
from telebot.types import Message
import asyncio
import logging
from utils import delete_messages
from config import COMMAND_MESSAGES
# Получаем логгер для текущего модуля
logger = logging.getLogger(__name__)
# Регистрирует все обработчики команд
def register_handlers(bot: AsyncTeleBot):
# Обработчик команды /help
@bot.message_handler(commands=['help'])
async def help_command(message: Message):
try:
# Отправляем сообщение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['help'],
message_thread_id=message.message_thread_id,
)
# Записываем действие в логи
logger.info(f"Пользователь {message.from_user.id} запустил /help.")
# Если пользователь писал в чат
if message.chat.id != message.from_user.id:
# Удаляем сообщения через 30 секунд
await delete_messages(bot, message, time_sleep=30, number_message=2)
except Exception as e:
# Отправляем ошибку
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['general_error'],
message_thread_id=message.message_thread_id,
)
# Записываем ошибку в логи
logger.error(f"Общая ошибка в help_command: {str(e)}")

View File

@@ -5,106 +5,30 @@ import logging
import time import time
from database import db from database import db
from action_reporter import action_reporter
from utils import (
delete_messages,
check_admin_status,
check_target_status,
parse_mute_time,
format_mute_time,
)
from config import COMMAND_MESSAGES from config import COMMAND_MESSAGES
logger = logging.getLogger(__name__) # Получаем логгер для текущего модуля # Получаем логгер для текущего модуля
logger = logging.getLogger(__name__)
# Возвращает количество секунд # Регистрирует все обработчики команд
def parse_mute_time(time_str: str) -> int: def register_handlers(bot: AsyncTeleBot):
# Парсим строку времени # Обработчик команды /mute
time_str = time_str.strip().lower() @bot.message_handler(commands=['mute'])
async def _mute_command_wrapper(message: Message):
await mute_command(bot, message)
# Минуты # Основная функция команды /mute
if time_str.endswith('m') or time_str.endswith('м'): async def mute_command(bot: AsyncTeleBot, message: Message, photo_path: str = None):
try:
minutes = int(time_str[:-1])
return abs(minutes) * 60
except:
return None
# Часы
elif time_str.endswith('h') or time_str.endswith('ч'):
try:
hours = int(time_str[:-1])
return abs(hours) * 3600
except:
return None
# Дни
elif time_str.endswith('d') or time_str.endswith('д'):
try:
days = int(time_str[:-1])
return abs(days) * 86400
except:
return None
# Число без указания единицы (по умолчанию минуты)
elif time_str.isdigit():
try:
minutes = int(time_str)
return abs(minutes) * 60
except:
return None
return None
# Возвращает причину мута
def extract_reason(words: str) -> str:
if words == []:
reason = 'отсутствует'
else:
reason = ' '.join(words)
return reason
# Форматирует время в нормальный вид
def format_time(seconds: int) -> str:
# Для минут
minutes = seconds // 60
if minutes < 60:
if minutes % 10 == 1 and minutes % 100 != 11:
return f"{minutes} минута"
elif 2 <= minutes % 10 <= 4 and minutes % 100 not in (12, 13, 14):
return f"{minutes} минуты"
else:
return f"{minutes} минут"
# Для часов
hours = minutes // 60
if hours < 24:
if hours % 10 == 1 and hours % 100 != 11:
return f"{hours} час"
elif 2 <= hours % 10 <= 4 and hours % 100 not in (12, 13, 14):
return f"{hours} часа"
else:
return f"{hours} часов"
# Для дней
days = hours // 24
if days % 10 == 1 and days % 100 != 11:
return f"{days} день"
elif 2 <= days % 10 <= 4 and days % 100 not in (12, 13, 14):
return f"{days} дня"
else:
return f"{days} дней"
# Удаляет два последних сообщения
async def delete_messages(bot: AsyncTeleBot, message: Message, time_sleep: int):
await asyncio.sleep(time_sleep)
await bot.delete_message(message.chat.id, message.message_id)
await bot.delete_message(message.chat.id, message.message_id+1)
def register_handlers(bot: AsyncTeleBot): # Регистрирует все обработчики команд
@bot.message_handler(commands=['mute']) # Обработчик команды /mute
async def mute_command(message: Message):
# Отправка сообщения в тему или обычный чат
send_message = bot.reply_to if message.is_topic_message else bot.send_message
chat_id = message if message.is_topic_message else message.chat.id
# Определяем целевого пользователя # Определяем целевого пользователя
target_user = None target_user = None
@@ -118,182 +42,215 @@ def register_handlers(bot: AsyncTeleBot): # Регистрирует все об
# Разбиваем текст сообщения на части # Разбиваем текст сообщения на части
parts_msg = message.text.split() parts_msg = message.text.split()
try: # Команда /mute help
# Проверяем, является ли отправитель администратором if len(parts_msg) == 2 and parts_msg[1].strip() in ('help', 'помощь'):
try:
admin_status = await bot.get_chat_member(message.chat.id, message.from_user.id)
# Проверяем статус администратора (создателя) # Отправляем инструкцию
if admin_status.status not in ['administrator', 'creator']: await bot.send_message(
await send_message(chat_id, COMMAND_MESSAGES['no_admin_rights']) chat_id=message.chat.id,
text=COMMAND_MESSAGES['manual_mute'],
# Удаляем сообщения через 5 секунд message_thread_id=message.message_thread_id,
await delete_messages(bot, message, 5) )
return
# Проверяем право администратора на мут
if admin_status.status == 'administrator' and not admin_status.can_restrict_members:
await send_message(chat_id, COMMAND_MESSAGES['no_restrict_rights'])
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)
return
except Exception as e:
await send_message(chat_id, f"⚠️ Ошибка: {str(e)}")
logger.error(f"Ошибка при получении статуса администратора: {str(e)}")
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)
return
# Выводим помощь
if len(parts_msg) == 2 and parts_msg[1].strip() in ['help', 'помощь']:
await send_message(chat_id, COMMAND_MESSAGES['manual_mute'])
# Удаляем сообщения через 30 секунд # Удаляем сообщения через 30 секунд
await delete_messages(bot, message, 30) await delete_messages(bot, message, time_sleep=30, number_message=2)
return return
# Случай №1 - Команда используется в ответ на сообщение try:
if len(parts_msg) >= 2:
# Если мутят в теме # Проверяем, является ли отправитель администратором
if await check_admin_status(bot, message) == 1:
return
# Если одно слово (/mute)
if len(parts_msg) == 1:
# Удаляем сообщение через 3 секунды
await delete_messages(bot, message, time_sleep=3, number_message=1)
return
# Команда через ответ на сообщение, если два слова (/mute 2m)
elif len(parts_msg) == 2:
# Если это топик
if message.is_topic_message: if message.is_topic_message:
# Если без ответа на сообщение # Если без ответа на сообщение
if message.message_thread_id == message.reply_to_message.message_id: if message.message_thread_id == message.reply_to_message.message_id:
await asyncio.sleep(3)
await bot.delete_message(message.chat.id, message.message_id) # Удаляем сообщение через 3 секунды
await delete_messages(bot, message, time_sleep=3, number_message=1)
return return
# Если с ответом на сообщение # Если с ответом на сообщение
else: else:
# Собираем данные
target_user = message.reply_to_message.from_user target_user = message.reply_to_message.from_user
time_arg = parts_msg[1] time_arg = parts_msg[1]
reason = extract_reason(parts_msg[2:]) reason = 'отсутствует'
# Если мутят в обычном чате # Если это General (обычный чат)
elif message.reply_to_message and message.is_topic_message is None: elif message.reply_to_message and message.is_topic_message is None:
# Собираем данные
target_user = message.reply_to_message.from_user target_user = message.reply_to_message.from_user
time_arg = parts_msg[1] time_arg = parts_msg[1]
reason = extract_reason(parts_msg[2:]) reason = 'отсутствует'
message.message_thread_id = None
# Не выводим сообщение, что команда неправильная
else:
await asyncio.sleep(3)
await bot.delete_message(message.chat.id, message.message_id)
return
# Случай №2 - Команда через тег или ID
elif len(parts_msg) >= 3:
identifier = parts_msg[1].strip()
time_arg = parts_msg[2]
reason = extract_reason(parts_msg[3:])
# Поиск по ID
if identifier.isdigit():
# Делаем в int и ищем
user_info = db.get_user(int(identifier))
if user_info:
# Создаем объект пользователя из данных базы
target_user = User(
id=user_info[0],
first_name=user_info[1],
username=user_info[2],
is_bot=False
)
# Поиск по тегу
elif identifier.startswith('@'):
# Убираем @ и ищем
user_info = db.get_user_by_username(identifier[1:])
if user_info:
# Создаем объект пользователя из данных базы
target_user = User(
id=user_info[0],
first_name=user_info[1],
username=user_info[2],
is_bot=False
)
# Если команда неправильная # Если команда неправильная
else: else:
await asyncio.sleep(3)
await bot.delete_message(message.chat.id, message.message_id) # Удаляем сообщение через 3 секунды
await delete_messages(bot, message, time_sleep=3, number_message=1)
return return
print(reason) # В сообщении больше двух слов
else:
# Если пользователь не найден # Если второе слово это тег или ID
if parts_msg[1].strip().isdigit() or parts_msg[1].startswith('@'):
# Собираем данные
identifier = parts_msg[1].strip()
time_arg = parts_msg[2]
reason = ' '.join(parts_msg[3:]) if parts_msg[3:] != [] else 'отсутствует'
# Делаем поиск по ID
if identifier.isdigit():
# Ищем пользователя в базе данных
user_info = db.get_user(int(identifier))
# Если нашли пользователя
if user_info:
# Создаем объект пользователя
target_user = User(
id=user_info[0],
first_name=user_info[1],
username=user_info[2],
is_bot=False
)
# Делаем поиск по тегу
elif identifier.startswith('@'):
# Ищем пользователя в базе данных (убрали @)
user_info = db.get_user_by_username(identifier[1:])
# Если нашли пользователя
if user_info:
# Создаем объект пользователя
target_user = User(
id=user_info[0],
first_name=user_info[1],
username=user_info[2],
is_bot=False
)
# Команда через ответ на сообщение с указанием причины
else:
# Если это топик
if message.is_topic_message:
# Если без ответа на сообщение
if message.message_thread_id == message.reply_to_message.message_id:
# Удаляем сообщение через 3 секунды
await delete_messages(bot, message, time_sleep=3, number_message=1)
return
# Если с ответом на сообщение
else:
# Собираем данные
target_user = message.reply_to_message.from_user
time_arg = parts_msg[1]
reason = ' '.join(parts_msg[2:])
# Если это General (обычный чат)
elif message.reply_to_message and message.is_topic_message is None:
# Собираем данные
target_user = message.reply_to_message.from_user
time_arg = parts_msg[1]
reason = ' '.join(parts_msg[2:])
message.message_thread_id = None
# Если команда неправильная
else:
# Удаляем сообщение через 3 секунды
await delete_messages(bot, message, time_sleep=3, number_message=1)
return
# Если пользователь не найден в базе данных
if not target_user: if not target_user:
await send_message(chat_id, COMMAND_MESSAGES['user_not_found'])
# Отправляем предупреждение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['user_not_found'],
message_thread_id=message.message_thread_id,
)
# Удаляем сообщения через 5 секунд # Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5) await delete_messages(bot, message, time_sleep=5, number_message=2)
return return
# Парсинг времени мута # Парсинг времени мута
mute_seconds = parse_mute_time(time_arg) mute_seconds = parse_mute_time(time_arg)
# Если не указали время
if mute_seconds is None: if mute_seconds is None:
await send_message(chat_id, COMMAND_MESSAGES['incorrect_time_format'])
# Отправляем предупреждение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['incorrect_time_format'],
message_thread_id=message.message_thread_id,
)
# Удаляем сообщения через 5 секунд # Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5) await delete_messages(bot, message, time_sleep=5, number_message=2)
return return
# Минимальный мут 1 минута (60 секунд) # Минимальное время мута - 1 минута (60 секунд)
if mute_seconds < 60: if mute_seconds < 60:
await send_message(chat_id, COMMAND_MESSAGES['min_mute'])
# Отправляем предупреждение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['min_mute'],
message_thread_id=message.message_thread_id,
)
# Удаляем сообщения через 5 секунд # Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5) await delete_messages(bot, message, time_sleep=5, number_message=2)
return return
# Максимальный мут 30 дней (2592000 секунд) # Максимальное время мута - 30 дней (2592000 секунд)
if mute_seconds > 2592000: if mute_seconds > 2592000:
await send_message(chat_id, COMMAND_MESSAGES['max_mute'])
# Отправляем предупреждение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['max_mute'],
message_thread_id=message.message_thread_id,
)
# Удаляем сообщения через 5 секунд # Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5) await delete_messages(bot, message, time_sleep=5, number_message=2)
return return
# Проверяем статус целевого пользователя # Проверяем статус целевого пользователя
try: if await check_target_status(bot, message, target_user) == 1:
target_status = await bot.get_chat_member(message.chat.id, target_user.id)
# Проверяем, является ли цель администратором или создателем
if target_status.status in ['administrator', 'creator']:
await send_message(chat_id, COMMAND_MESSAGES['cant_mute_admin'])
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)
return return
except Exception as e:
await send_message(chat_id, f"⚠️ Ошибка: {str(e)}")
logger.error(f"Ошибка при получении статуса пользователя: {str(e)}")
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)
return
# Выполняем мут
try: try:
# Вычисляем время окончания мута # Вычисляем время окончания мута
@@ -311,6 +268,7 @@ def register_handlers(bot: AsyncTeleBot): # Регистрирует все об
can_pin_messages=False, can_pin_messages=False,
) )
# Выполняем мут
await bot.restrict_chat_member( await bot.restrict_chat_member(
chat_id=message.chat.id, chat_id=message.chat.id,
user_id=target_user.id, user_id=target_user.id,
@@ -318,26 +276,58 @@ def register_handlers(bot: AsyncTeleBot): # Регистрирует все об
until_date=until_date until_date=until_date
) )
# Форматирование времени # Форматируем время в удобный формат
time_display = format_time(mute_seconds) time_display = format_mute_time(mute_seconds)
# Отправляем сообщения, что пользователь получил мут # Отправляем сообщение-лог в админ-чат
await send_message(chat_id, COMMAND_MESSAGES['muted'].format(time_display=time_display)) await action_reporter.log_action(
logger.info(f"Пользователь {target_user.id} замучен на {time_display} администратором {message.from_user.id}.") action="МУТ",
user_id=target_user.id,
admin_id=message.from_user.id,
reason=reason,
duration=time_display,
photo_path=photo_path,
)
# Отправляем сообщение, что пользователь получил мут
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['muted'].format(time_display=time_display),
message_thread_id=message.message_thread_id,
)
# Записываем действие в логи
logger.info(f"Пользователь {target_user.id} получил мут на {time_display} от администратора {message.from_user.id}.")
# Удаляем сообщения через 5 секунд # Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5) await delete_messages(bot, message, time_sleep=5, number_message=2)
except Exception as e: except Exception as e:
await send_message(chat_id, f"⚠️ Ошибка: {str(e)}")
# Отправляем ошибку
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['error'].format(e=str(e)),
message_thread_id=message.message_thread_id,
)
# Записываем ошибку в логи
logger.error(f"Ошибка мута: {str(e)}") logger.error(f"Ошибка мута: {str(e)}")
# Удаляем сообщения через 5 секунд # Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5) await delete_messages(bot, message, time_sleep=5, number_message=2)
except Exception as e: except Exception as e:
await send_message(chat_id, COMMAND_MESSAGES['general_error'])
# Отправляем ошибку
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['general_error'],
message_thread_id=message.message_thread_id,
)
# Записываем ошибку в логи
logger.error(f"Общая ошибка в mute_command: {str(e)}") logger.error(f"Общая ошибка в mute_command: {str(e)}")
# Удаляем сообщения через 5 секунд # Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5) await delete_messages(bot, message, time_sleep=5, number_message=2)

View File

@@ -1,24 +1,48 @@
from telebot.async_telebot import AsyncTeleBot from telebot.async_telebot import AsyncTeleBot
from telebot.types import Message from telebot.types import Message
import asyncio
import logging import logging
from utils import delete_messages
from config import COMMAND_MESSAGES from config import COMMAND_MESSAGES
logger = logging.getLogger(__name__) # Получаем логгер для текущего модуля # Получаем логгер для текущего модуля
logger = logging.getLogger(__name__)
def register_handlers(bot: AsyncTeleBot): # Регистрирует все обработчики команд # Регистрирует все обработчики команд
def register_handlers(bot: AsyncTeleBot):
@bot.message_handler(commands=['start']) # Обработчик команды /start # Обработчик команды /start
@bot.message_handler(commands=['start'])
async def start_command(message: Message): async def start_command(message: Message):
# Отправка сообщения в тему или обычный чат
send_message = bot.reply_to if message.is_topic_message else bot.send_message
chat_id = message if message.is_topic_message else message.chat.id
try: try:
await send_message(chat_id, COMMAND_MESSAGES['start']) # Отправляем сообщение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['start'],
message_thread_id=message.message_thread_id,
)
# Записываем действие в логи
logger.info(f"Пользователь {message.from_user.id} запустил /start.") logger.info(f"Пользователь {message.from_user.id} запустил /start.")
# Если пользователь писал в чат
if message.chat.id != message.from_user.id:
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)
except Exception as e: except Exception as e:
logger.error(f"Пользователь {message.from_user.id} запустил /start: {str(e)}")
# Отправляем ошибку
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['general_error'],
message_thread_id=message.message_thread_id,
)
# Записываем ошибку в логи
logger.error(f"Общая ошибка в start_command: {str(e)}")

View File

@@ -1,88 +1,103 @@
from telebot.async_telebot import AsyncTeleBot from telebot.async_telebot import AsyncTeleBot
from telebot.types import Message, User from telebot.types import Message, User, ChatPermissions
import asyncio import asyncio
import logging import logging
from database import db from database import db
from action_reporter import action_reporter
from utils import (
delete_messages,
check_admin_status,
check_target_status,
)
from config import COMMAND_MESSAGES from config import COMMAND_MESSAGES
logger = logging.getLogger(__name__) # Получаем логгер для текущего модуля # Получаем логгер для текущего модуля
logger = logging.getLogger(__name__)
def register_handlers(bot: AsyncTeleBot): # Регистрирует все обработчики команд # Регистрирует все обработчики команд
def register_handlers(bot: AsyncTeleBot):
@bot.message_handler(commands=['unban']) # Обработчик команды /unban # Обработчик команды /unban
@bot.message_handler(commands=['unban'])
async def unban_command(message: Message): async def unban_command(message: Message):
# Отправка сообщения в тему или обычный чат
send_message = bot.reply_to if message.is_topic_message else bot.send_message
chat_id = message if message.is_topic_message else message.chat.id
# Определяем целевого пользователя # Определяем целевого пользователя
target_user = None target_user = None
# Разбиваем текст сообщения на части # Разбиваем текст сообщения на части
parts_msg = message.text.split() parts_msg = message.text.split()
# Команда /unban help
if len(parts_msg) == 2 and parts_msg[1].strip() in ('help', 'помощь'):
# Отправляем инструкцию
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['manual_unban'],
message_thread_id=message.message_thread_id,
)
# Удаляем сообщения через 30 секунд
await delete_messages(bot, message, time_sleep=30, number_message=2)
return
try: try:
# Проверяем, является ли отправитель администратором # Проверяем, является ли отправитель администратором
try: if await check_admin_status(bot, message) == 1:
admin_status = await bot.get_chat_member(message.chat.id, message.from_user.id)
# Проверяем статус администратора (создателя)
if admin_status.status not in ['administrator', 'creator']:
await send_message(chat_id, COMMAND_MESSAGES['no_admin_rights'])
return return
# Проверяем право администратора на разбан # Команда через ответ на сообщение, если одно слово (/unban)
if admin_status.status == 'administrator' and not admin_status.can_restrict_members:
await send_message(chat_id, COMMAND_MESSAGES['no_restrict_rights'])
return
except Exception as e:
await send_message(chat_id, f"⚠️ Ошибка: {str(e)}")
logger.error(f"Ошибка при получении статуса администратора: {str(e)}")
return
# Случай №1 - Команда используется в ответ на сообщение
if len(parts_msg) == 1: if len(parts_msg) == 1:
# Если банят в теме # Если это топик
if message.is_topic_message: if message.is_topic_message:
# Если без ответа на сообщение # Если без ответа на сообщение
if message.message_thread_id == message.reply_to_message.message_id: if message.message_thread_id == message.reply_to_message.message_id:
# Удаляем сообщение через 3 секунды
await delete_messages(bot, message, time_sleep=3, number_message=1)
return return
# Если команда правильная # Если с ответом на сообщение
else: else:
# Собираем данные
target_user = message.reply_to_message.from_user target_user = message.reply_to_message.from_user
# Если банят в обычном чате # Если это General (обычный чат)
elif message.reply_to_message and message.is_topic_message is None: elif message.reply_to_message and message.is_topic_message is None:
target_user = message.reply_to_message.from_user
# Не выводим сообщение, что команда неправильная # Собираем данные
target_user = message.reply_to_message.from_user
message.message_thread_id = None
# Удаляем сообщение, если команда неправильная
else: else:
# Удаляем сообщение через 3 секунды
await delete_messages(bot, message, time_sleep=3, number_message=1)
return return
# Случай №2 - Команда с аргументом (/unban @username или /unban 12345) # Команда через тег или ID, если два слова
elif len(parts_msg) == 2: elif len(parts_msg) == 2:
# Выводим помощь (/unban help) # Собираем данные
if parts_msg[1].strip() in ['help', 'помощь']:
await send_message(chat_id, COMMAND_MESSAGES['manual_unban'])
return
identifier = parts_msg[1].strip() identifier = parts_msg[1].strip()
# Поиск по ID # Делаем поиск по ID
if identifier.isdigit(): if identifier.isdigit():
# Делаем в int и ищем # Ищем пользователя в базе данных
user_info = db.get_user(int(identifier)) user_info = db.get_user(int(identifier))
# Если нашли пользователя
if user_info: if user_info:
# Создаем объект пользователя из данных базы
# Создаем объект пользователя
target_user = User( target_user = User(
id=user_info[0], id=user_info[0],
first_name=user_info[1], first_name=user_info[1],
@@ -90,14 +105,16 @@ def register_handlers(bot: AsyncTeleBot): # Регистрирует все об
is_bot=False is_bot=False
) )
# Поиск по тэгу # Делаем поиск по тегу
elif identifier.startswith('@'): elif identifier.startswith('@'):
# Убираем @ и ищем # Ищем пользователя в базе данных
user_info = db.get_user_by_username(identifier[1:]) user_info = db.get_user_by_username(identifier[1:])
# Если нашли пользователя
if user_info: if user_info:
# Создаем объект пользователя из данных базы
# Создаем объект пользователя
target_user = User( target_user = User(
id=user_info[0], id=user_info[0],
first_name=user_info[1], first_name=user_info[1],
@@ -105,25 +122,81 @@ def register_handlers(bot: AsyncTeleBot): # Регистрирует все об
is_bot=False is_bot=False
) )
# Если пользователь не найден # Если пользователь не найден в базе данных
if not target_user: if not target_user:
await send_message(chat_id, COMMAND_MESSAGES['user_not_found'])
# Отправляем предупреждение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['user_not_found'],
message_thread_id=message.message_thread_id,
)
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)
return return
# Выполняем разбан # Проверяем статус целевого пользователя
try: if await check_target_status(bot, message, target_user) == 1:
await bot.unban_chat_member(message.chat.id, target_user.id) return
await send_message(chat_id, COMMAND_MESSAGES['unbanned'])
await asyncio.sleep(5) try:
await bot.delete_message(message.chat.id, message.message_id)
await bot.delete_message(message.chat.id, message.message_id+1) # Выполняем разбан
await bot.unban_chat_member(
chat_id=message.chat.id,
user_id=target_user.id,
)
# Отправляем лог в админ-чат
await action_reporter.log_action(
action="РАЗБАН",
user_id=target_user.id,
admin_id=message.from_user.id,
reason=None,
duration=None,
photo_path=None,
)
# Отправляем сообщения, что пользователь получил разбан
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['unbanned'],
message_thread_id=message.message_thread_id,
)
# Записываем действие в логи
logger.info(f"Пользователь {target_user.id} разбанен администратором {message.from_user.id}.") logger.info(f"Пользователь {target_user.id} разбанен администратором {message.from_user.id}.")
except Exception as e: # Удаляем сообщения через 5 секунд
await send_message(chat_id, f"⚠️ Ошибка: {str(e)}") await delete_messages(bot, message, time_sleep=5, number_message=2)
logger.error(f"Ошибка разбана: {str(e)}")
except Exception as e: except Exception as e:
await send_message(chat_id, COMMAND_MESSAGES['general_error'])
# Отправляем ошибку
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['error'].format(e=str(e)),
message_thread_id=message.message_thread_id,
)
# Записываем ошибку в логи
logger.error(f"Ошибка размута: {str(e)}")
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)
except Exception as e:
# Отправляем ошибку
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['general_error'],
message_thread_id=message.message_thread_id,
)
# Записываем ошибку в логи
logger.error(f"Общая ошибка в unban_command: {str(e)}") logger.error(f"Общая ошибка в unban_command: {str(e)}")
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)

215
src/modules/unmute.py Normal file
View File

@@ -0,0 +1,215 @@
from telebot.async_telebot import AsyncTeleBot
from telebot.types import Message, User, ChatPermissions
import asyncio
import logging
from database import db
from action_reporter import action_reporter
from utils import (
delete_messages,
check_admin_status,
check_target_status,
)
from config import COMMAND_MESSAGES
# Получаем логгер для текущего модуля
logger = logging.getLogger(__name__)
# Регистрирует все обработчики команд
def register_handlers(bot: AsyncTeleBot):
# Обработчик команды /unmute
@bot.message_handler(commands=['unmute'])
async def unmute_command(message: Message):
# Определяем целевого пользователя
target_user = None
# Разбиваем текст сообщения на части
parts_msg = message.text.split()
# Команда /unmute help
if len(parts_msg) == 2 and parts_msg[1].strip() in ('help', 'помощь'):
# Отправляем инструкцию
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['manual_unmute'],
message_thread_id=message.message_thread_id,
)
# Удаляем сообщения через 30 секунд
await delete_messages(bot, message, time_sleep=30, number_message=2)
return
try:
# Проверяем, является ли отправитель администратором
if await check_admin_status(bot, message) == 1:
return
# Команда через ответ на сообщение, если одно слово (/unmute)
if len(parts_msg) == 1:
# Если это топик
if message.is_topic_message:
# Если без ответа на сообщение
if message.message_thread_id == message.reply_to_message.message_id:
# Удаляем сообщение через 3 секунды
await delete_messages(bot, message, time_sleep=3, number_message=1)
return
# Если с ответом на сообщение
else:
# Собираем данные
target_user = message.reply_to_message.from_user
# Если это General (обычный чат)
elif message.reply_to_message and message.is_topic_message is None:
# Собираем данные
target_user = message.reply_to_message.from_user
message.message_thread_id = None
# Удаляем сообщение, если команда неправильная
else:
# Удаляем сообщение через 3 секунды
await delete_messages(bot, message, time_sleep=3, number_message=1)
return
# Команда через тег или ID, если два слова
elif len(parts_msg) == 2:
# Собираем данные
identifier = parts_msg[1].strip()
# Делаем поиск по ID
if identifier.isdigit():
# Ищем пользователя в базе данных
user_info = db.get_user(int(identifier))
# Если нашли пользователя
if user_info:
# Создаем объект пользователя
target_user = User(
id=user_info[0],
first_name=user_info[1],
username=user_info[2],
is_bot=False
)
# Делаем поиск по тегу
elif identifier.startswith('@'):
# Ищем пользователя в базе данных
user_info = db.get_user_by_username(identifier[1:])
# Если нашли пользователя
if user_info:
# Создаем объект пользователя
target_user = User(
id=user_info[0],
first_name=user_info[1],
username=user_info[2],
is_bot=False
)
# Если пользователь не найден в базе данных
if not target_user:
# Отправляем предупреждение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['user_not_found'],
message_thread_id=message.message_thread_id,
)
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)
return
# Проверяем статус целевого пользователя
if await check_target_status(bot, message, target_user) == 1:
return
try:
# Убираем ограничения (можно писать в чат)
permissions = ChatPermissions(
can_send_messages=True,
can_send_media_messages=True,
can_send_polls=True,
can_send_other_messages=True,
can_add_web_page_previews=True,
can_change_info=False,
can_invite_users=True,
can_pin_messages=False
)
# Выполняем размут
await bot.restrict_chat_member(
chat_id=message.chat.id,
user_id=target_user.id,
permissions=permissions,
)
# Отправляем лог в админ-чат
await action_reporter.log_action(
action="РАЗМУТ",
user_id=target_user.id,
admin_id=message.from_user.id,
reason=None,
duration=None,
photo_path=None,
)
# Отправляем сообщения, что пользователь получил размут
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['unmuted'],
message_thread_id=message.message_thread_id,
)
# Записываем действие в логи
logger.info(f"Пользователь {target_user.id} получил размут от администратора {message.from_user.id}.")
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)
except Exception as e:
# Отправляем ошибку
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['error'].format(e=str(e)),
message_thread_id=message.message_thread_id,
)
# Записываем ошибку в логи
logger.error(f"Ошибка размута: {str(e)}")
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)
except Exception as e:
# Отправляем ошибку
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['general_error'],
message_thread_id=message.message_thread_id,
)
# Записываем ошибку в логи
logger.error(f"Общая ошибка в unmute_command: {str(e)}")
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)

177
src/utils.py Normal file
View File

@@ -0,0 +1,177 @@
from telebot.async_telebot import AsyncTeleBot
from telebot.types import Message
import asyncio
import logging
# Получаем логгер для текущего модуля
logger = logging.getLogger(__name__)
# Удаляет определённое количество сообщения
async def delete_messages(bot: AsyncTeleBot, message: Message, time_sleep: int, number_message: int):
await asyncio.sleep(time_sleep)
for i in range(number_message):
await bot.delete_message(message.chat.id, message.message_id+i)
# Проверяет, является ли отправитель администратором
async def check_admin_status(bot: AsyncTeleBot, message: Message):
if message.reply_to_message and message.is_topic_message is None:
message.message_thread_id = None
try:
# Получаем статус отправителя
admin_status = await bot.get_chat_member(message.chat.id, message.from_user.id)
# Проверка наличия прав администратора/создателя
if admin_status.status not in ('administrator', 'creator'):
# Отправляем предупреждение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['no_admin_rights'],
message_thread_id=message.message_thread_id,
)
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)
return 1
# Проверка права на ограничение участников
if admin_status.status == 'administrator' and not admin_status.can_restrict_members:
# Отправляем предупреждение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['no_restrict_rights'],
message_thread_id=message.message_thread_id,
)
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)
return 1
except Exception as e:
# Отправляем ошибку
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['error'].format(e=str(e)),
message_thread_id=message.message_thread_id,
)
# Записываем ошибку в логи
logger.error(f"Ошибка при получении статуса администратора: {str(e)}")
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)
return 1
# Проверяет статус целевого пользователя
async def check_target_status(bot: AsyncTeleBot, message: Message, target_user):
if message.reply_to_message and message.is_topic_message is None:
message.message_thread_id = None
try:
# Получаем статус пользователя
target_status = await bot.get_chat_member(
chat_id=message.chat.id,
user_id=target_user.id,
)
# Проверяем, является ли цель администратором или создателем
if target_status.status in ('administrator', 'creator'):
# Отправляем предупреждение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['cant_mute_admin'],
message_thread_id=message.message_thread_id,
)
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)
return
except Exception as e:
# Отправляем ошибку
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['error'].format(e=str(e)),
message_thread_id=message.message_thread_id,
)
# Записываем ошибку в логи
logger.error(f"Ошибка при получении статуса пользователя: {str(e)}")
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)
return
# Возвращает количество секунд
def parse_mute_time(time_str: str) -> int | None:
# Нормализация входной строки
normalized = time_str.strip().lower()
# Словарь для конвертации единиц времени в секунды
time_units = {
'm': 60, 'м': 60, # минуты
'h': 3600, 'ч': 3600, # часы
'd': 86400, 'д': 86400 # дни
}
# Проверям, существует ли строка
if not normalized:
return None
# Проверяем последний символ (единица измерения)
unit = normalized[-1]
if unit not in time_units:
return None
# Парсим числовую часть
try:
value = int(normalized[:-1])
if value <= 0:
return None
return value * time_units[unit]
except (ValueError, TypeError):
return None
# Форматирует время в нормальный вид
def format_mute_time(seconds: int) -> str:
# Обработка минут
minutes = seconds // 60
if minutes < 60:
if minutes % 10 == 1 and minutes % 100 != 11:
return f"{minutes} минуту"
elif 2 <= minutes % 10 <= 4 and (minutes % 100 < 10 or minutes % 100 >= 20):
return f"{minutes} минуты"
else:
return f"{minutes} минут"
# Обработка часов
hours = minutes // 60
if hours < 24:
if hours % 10 == 1 and hours % 100 != 11:
return f"{hours} час"
elif 2 <= hours % 10 <= 4 and (hours % 100 < 10 or hours % 100 >= 20):
return f"{hours} часа"
else:
return f"{hours} часов"
# Обработка дней
days = hours // 24
if days % 10 == 1 and days % 100 != 11:
return f"{days} день"
elif 2 <= days % 10 <= 4 and (days % 100 < 10 or days % 100 >= 20):
return f"{days} дня"
else:
return f"{days} дней"