Compare commits

..

20 Commits

Author SHA1 Message Date
61956d9808 Добавление возможности менять внешний вид бота 2026-02-25 17:50:37 +07:00
d646c1eb50 Модуль изменения описания бота 2026-02-25 17:50:23 +07:00
54125b82ac Добавление работы с конфликтными частями и исправление вайтлиста 2026-02-25 17:50:11 +07:00
6a4e56c367 Улучшение элиасов команд 2026-02-25 17:49:28 +07:00
9e38397e85 Исправление не соответствий в ядре бота 2026-02-25 17:49:15 +07:00
8b5d567536 Модуль установки настроек бота 2026-02-25 17:48:50 +07:00
82d40ad6e8 Модуль установки виджета бота 2026-02-25 17:48:41 +07:00
66889721c2 Модуль установки имени бота 2026-02-25 17:48:35 +07:00
9b56d5a45a Модуль установки аватара (В разработке!) 2026-02-25 17:48:18 +07:00
b23fc81eac Тесты запуска бота (В разработке!) 2026-02-23 14:59:37 +07:00
922ee0d986 Уведомления о спаме 2026-02-23 14:39:27 +07:00
cd7d6512dd Фильтр для првоерки содержимого сообщения 2026-02-23 14:39:18 +07:00
c74732cbd4 Утилиты работы с командами 2026-02-23 14:38:52 +07:00
4d1eb3e231 Фильтр проверки активных режимов анти-конфликт и "молчанки" 2026-02-23 14:38:37 +07:00
b79446b0ed Модуль ИИ (В РАЗРАБОТКЕ!) 2026-02-23 14:38:19 +07:00
fee19ff1aa Автоматическое удаление сообщений 2026-02-23 14:38:02 +07:00
8170d7a588 Фильтр обработки callback-запросов 2026-02-23 14:37:52 +07:00
5a52f62afd Хранилище инлайн-клавиатур под сообщением 2026-02-23 14:37:34 +07:00
4f382e4197 Обработчик команды /pin для закрепов (В бета!) 2026-02-23 14:37:16 +07:00
36e721fd3d Обработчик команды /start 2026-02-23 14:37:05 +07:00
32 changed files with 3536 additions and 352 deletions

32
.gitea/workflows/ci.yml Normal file
View File

@@ -0,0 +1,32 @@
name: Bot CI
on:
push:
branches: [ main ]
pull_request:
jobs:
test-bot:
runs-on: ubuntu-latest
env:
BOT_TOKEN: test_token
DATABASE_URL: sqlite+aiosqlite:///./test.db
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Install Python
uses: actions/setup-python@v5
with:
python-version: "3.13"
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
- name: Run bot (startup test)
run: |
timeout 10s python main.py

View File

@@ -65,7 +65,7 @@ class BotInfo:
last_name: str = None
username: str = None
description: str = None
short_description: str = None
widget: str = None
is_premium: bool = False
# Возможности бота
@@ -159,11 +159,15 @@ class BotInfo:
logger.info("Получение информации о боте", log_type='BOT')
bot_info: User = await bots.get_me()
description_obj: BotDescription = await bots.get_my_description()
short_obj: BotShortDescription = await bots.get_my_short_description()
cls.id = bot_info.id
cls.url = f'tg://user?id={cls.id}'
cls.first_name = bot_info.first_name
cls.last_name = bot_info.last_name
cls.description = description_obj.description if description_obj else None
cls.widget = short_obj.short_description if short_obj else None
cls.username = bot_info.username
cls.can_join_groups = getattr(bot_info, 'can_join_groups', False)
cls.can_read_all_group_messages = getattr(bot_info, 'can_read_all_group_messages', False)
@@ -183,6 +187,8 @@ class BotInfo:
'username': cls.username,
'prefix': cls.prefix,
'is_premium': cls.is_premium,
'description': cls.description,
'short_description': cls.widget,
'can_join_groups': cls.can_join_groups,
'can_read_all_group_messages': cls.can_read_all_group_messages,
'supports_inline_queries': cls.supports_inline_queries,
@@ -310,6 +316,8 @@ class BotInfo:
f"║ • Имя: {cls.first_name} {cls.last_name or ''}".ljust(60) + "",
f"║ • Username: @{cls.username}".ljust(60) + "",
f"║ • ID: {cls.id}".ljust(60) + "",
f"║ • Description: {cls.description}".ljust(60) + "",
f"║ • Widget: {cls.widget}".ljust(60) + "",
f"",
f"║ ⚙️ ВОЗМОЖНОСТИ БОТА:",
f"║ • Вступать в группы: {'' if cls.can_join_groups else ''}".ljust(60) + "",
@@ -371,6 +379,7 @@ class BotInfo:
setup_webhook: Устанавливать ли webhook (по умолчанию True)
"""
perm = perm if perm is not None else settings.BOT_EDIT
await BotInfo.info()
logger.info("🚀 Процесс запуска бота!", log_type='START')

253
bot/filters/callback.py Normal file
View File

@@ -0,0 +1,253 @@
"""
Фильтры для обработки callback-запросов
"""
import re
from typing import Union
from aiogram.filters import BaseFilter
from aiogram.types import CallbackQuery
from middleware.loggers import logger
__all__ = (
'CallbackStartsWith',
'CallbackEndsWith',
'CallbackContains',
'CallbackMatches',
'CallbackIn'
)
class CallbackStartsWith(BaseFilter):
"""
Проверяет, начинается ли callback_data с указанного префикса.
Attributes:
№ prefix: Префикс для проверки (строка или список строк)
ignore_case: Игнорировать регистр
Example:
```python
# Один префикс
@router.callback_query(CallbackStartsWith("menu:"))
async def menu_handler(callback: CallbackQuery):
await callback.answer("Меню")
# Несколько префиксов
@router.callback_query(CallbackStartsWith(["admin:", "mod:"]))
async def admin_handler(callback: CallbackQuery):
await callback.answer("Админ панель")
```
"""
def __init__(self, prefix: Union[str, list[str]], ignore_case: bool = True):
"""
Args:
prefix: Префикс или список префиксов
ignore_case: Игнорировать регистр букв
"""
self.prefixes = [prefix] if isinstance(prefix, str) else prefix
self.ignore_case = ignore_case
if self.ignore_case:
self.prefixes = [p.lower() for p in self.prefixes]
async def __call__(self, callback: CallbackQuery) -> Union[bool, dict]:
if not callback.data:
return False
data = callback.data.lower() if self.ignore_case else callback.data
for prefix in self.prefixes:
if data.startswith(prefix):
# Извлекаем данные после префикса
value = callback.data[len(prefix):]
logger.debug(
f"Callback с префиксом '{prefix}': {callback.data}",
log_type='CALLBACK'
)
return {
'matched': True,
'prefix': prefix,
'value': value,
'full_data': callback.data
}
return False
class CallbackEndsWith(BaseFilter):
"""
Проверяет, заканчивается ли callback_data на указанный суффикс.
Example:
```python
@router.callback_query(CallbackEndsWith(":confirm"))
async def confirm_handler(callback: CallbackQuery, matched: dict):
action = matched['value']
await callback.answer(f"Подтверждение: {action}")
```
"""
def __init__(self, suffix: Union[str, list[str]], ignore_case: bool = True):
"""
Args:
suffix: Суффикс или список суффиксов
ignore_case: Игнорировать регистр букв
"""
self.suffixes = [suffix] if isinstance(suffix, str) else suffix
self.ignore_case = ignore_case
if self.ignore_case:
self.suffixes = [s.lower() for s in self.suffixes]
async def __call__(self, callback: CallbackQuery) -> Union[bool, dict]:
if not callback.data:
return False
data = callback.data.lower() if self.ignore_case else callback.data
for suffix in self.suffixes:
if data.endswith(suffix):
# Извлекаем данные до суффикса
value = callback.data[:-len(suffix)]
return {
'matched': True,
'suffix': suffix,
'value': value,
'full_data': callback.data
}
return False
class CallbackContains(BaseFilter):
"""
Проверяет, содержит ли callback_data указанную подстроку.
Example:
```python
@router.callback_query(CallbackContains("delete"))
async def delete_handler(callback: CallbackQuery):
await callback.answer("Удаление...")
```
"""
def __init__(self, substring: Union[str, list[str]], ignore_case: bool = True):
"""
Args:
substring: Подстрока или список подстрок
ignore_case: Игнорировать регистр букв
"""
self.substrings = [substring] if isinstance(substring, str) else substring
self.ignore_case = ignore_case
if self.ignore_case:
self.substrings = [s.lower() for s in self.substrings]
async def __call__(self, callback: CallbackQuery) -> Union[bool, dict]:
if not callback.data:
return False
data = callback.data.lower() if self.ignore_case else callback.data
for substring in self.substrings:
if substring in data:
return {
'matched': True,
'substring': substring,
'full_data': callback.data
}
return False
class CallbackMatches(BaseFilter):
"""
Проверяет callback_data по regex паттерну.
Example:
```python
# Паттерн: user_123, user_456 и т.д.
@router.callback_query(CallbackMatches(r'^user_(\\d+)$'))
async def user_handler(callback: CallbackQuery, matched: dict):
user_id = matched['groups']
await callback.answer(f"Пользователь {user_id}")
```
"""
def __init__(self, pattern: Union[str, re.Pattern], flags: int = 0):
"""
Args:
pattern: Regex паттерн (строка или скомпилированный Pattern)
flags: Флаги для regex (например, re.IGNORECASE)
"""
if isinstance(pattern, str):
self.pattern = re.compile(pattern, flags)
else:
self.pattern = pattern
async def __call__(self, callback: CallbackQuery) -> Union[bool, dict]:
if not callback.data:
return False
match = self.pattern.match(callback.data)
if match:
logger.debug(
f"Callback соответствует паттерну {self.pattern.pattern}: {callback.data}",
log_type='CALLBACK'
)
return {
'matched': True,
'pattern': self.pattern.pattern,
'groups': match.groups(),
'groupdict': match.groupdict(),
'full_data': callback.data
}
return False
class CallbackIn(BaseFilter):
"""
Проверяет, находится ли callback_data в списке разрешенных значений.
Example:
```python
@router.callback_query(CallbackIn(["yes", "no", "cancel"]))
async def choice_handler(callback: CallbackQuery):
choice = callback.data
await callback.answer(f"Выбрано: {choice}")
```
"""
def __init__(self, values: list[str], ignore_case: bool = True):
"""
Args:
values: Список разрешенных значений
ignore_case: Игнорировать регистр букв
"""
self.values = values
self.ignore_case = ignore_case
if self.ignore_case:
self.values = [v.lower() for v in values]
async def __call__(self, callback: CallbackQuery) -> Union[bool, dict]:
if not callback.data:
return False
data = callback.data.lower() if self.ignore_case else callback.data
if data in self.values:
return {
'matched': True,
'value': callback.data
}
return False

184
bot/filters/modes.py Normal file
View File

@@ -0,0 +1,184 @@
"""
Фильтры для проверки активных режимов бота (silence, conflict)
"""
from datetime import datetime
from typing import Optional
from aiogram.filters import BaseFilter
from aiogram.types import Message
from middleware.loggers import logger
__all__ = ('IsSilenceActive', 'IsConflictModeActive')
class IsSilenceActive(BaseFilter):
"""
Проверяет, активен ли режим тишины (silence mode).
В режиме тишины удаляются ВСЕ сообщения (кроме админов).
Attributes:
silence_until: Время до которого активен режим (None = неактивен)
Example:
```python
# В handler-файле
silence_filter = IsSilenceActive()
@router.message(silence_filter)
async def silence_mode_active(message: Message):
# Удаляем все сообщения в режиме тишины
await message.delete()
```
"""
def __init__(self, silence_until: Optional[datetime] = None):
"""
Args:
silence_until: Datetime до которого активен режим
"""
self.silence_until = silence_until
def update_silence_until(self, new_datetime: Optional[datetime]) -> None:
"""
Обновляет время окончания режима тишины.
Args:
new_datetime: Новое время окончания или None для отключения
"""
self.silence_until = new_datetime
if new_datetime:
logger.info(
f"Режим тишины активирован до {new_datetime.strftime('%H:%M:%S')}",
log_type='SILENCE'
)
else:
logger.info("Режим тишины отключен", log_type='SILENCE')
def is_active(self) -> bool:
"""
Проверяет, активен ли режим сейчас.
Returns:
bool: True если режим активен
"""
if self.silence_until is None:
return False
# Проверка истечения времени
if datetime.now() >= self.silence_until:
logger.info("Режим тишины автоматически завершен", log_type='SILENCE')
self.silence_until = None
return False
return True
async def __call__(self, event: Message) -> Optional[dict]:
"""
Проверка активности режима тишины.
Returns:
dict или None: Информация о режиме если активен, иначе None
"""
if self.is_active():
remaining = (self.silence_until - datetime.now()).total_seconds()
logger.debug(
f"Режим тишины активен (осталось {remaining:.0f}с)",
log_type='SILENCE',
message=event
)
return {
'is_active': True,
'until': self.silence_until,
'remaining_seconds': remaining
}
return None
class IsConflictModeActive(BaseFilter):
"""
Проверяет, активен ли режим антиконфликта (conflict mode).
В режиме антиконфликта удаляются сообщения с конфликтными словами.
Attributes:
conflict_until: Время до которого активен режим (None = неактивен)
Example:
```python
conflict_filter = IsConflictModeActive()
@router.message(conflict_filter)
async def conflict_mode_active(message: Message):
# Проверяем на конфликтные слова и удаляем
if has_conflict_words(message.text):
await message.delete()
```
"""
def __init__(self, conflict_until: Optional[datetime] = None):
"""
Args:
conflict_until: Datetime до которого активен режим
"""
self.conflict_until = conflict_until
def update_conflict_until(self, new_datetime: Optional[datetime]) -> None:
"""
Обновляет время окончания режима антиконфликта.
Args:
new_datetime: Новое время окончания или None для отключения
"""
self.conflict_until = new_datetime
if new_datetime:
logger.info(
f"Режим антиконфликта активирован до {new_datetime.strftime('%H:%M:%S')}",
log_type='CONFLICT'
)
else:
logger.info("Режим антиконфликта отключен", log_type='CONFLICT')
def is_active(self) -> bool:
"""
Проверяет, активен ли режим сейчас.
Returns:
bool: True если режим активен
"""
if self.conflict_until is None:
return False
# Проверка истечения времени
if datetime.now() >= self.conflict_until:
logger.info("Режим антиконфликта автоматически завершен", log_type='CONFLICT')
self.conflict_until = None
return False
return True
async def __call__(self, event: Message) -> Optional[dict]:
"""
Проверка активности режима антиконфликта.
Returns:
dict или None: Информация о режиме если активен, иначе None
"""
if self.is_active():
remaining = (self.conflict_until - datetime.now()).total_seconds()
logger.debug(
f"Режим антиконфликта активен (осталось {remaining:.0f}с)",
log_type='CONFLICT',
message=event
)
return {
'is_active': True,
'until': self.conflict_until,
'remaining_seconds': remaining
}
return None

395
bot/filters/msg_content.py Normal file
View File

@@ -0,0 +1,395 @@
"""
Фильтры для проверки содержимого сообщений
"""
import re
from typing import Optional, Union
from aiogram.filters import BaseFilter
from aiogram.types import Message, ContentType
from middleware.loggers import logger
__all__ = (
'IsReply',
'IsForwarded',
'HasMedia',
'ContainsURL',
'HasText',
'HasCaption',
'HasEntities',
'MediaType'
)
class IsReply(BaseFilter):
"""
Проверяет, является ли сообщение ответом на другое сообщение.
Example:
```python
@router.message(IsReply())
async def handle_reply(message: Message):
original = message.reply_to_message
await message.answer(f"Это ответ на: {original.text}")
```
"""
async def __call__(self, message: Message) -> Union[bool, dict]:
is_reply = message.reply_to_message is not None
if is_reply:
return {
'is_reply': True,
'reply_to_message': message.reply_to_message,
'reply_to_user_id': message.reply_to_message.from_user.id if message.reply_to_message.from_user else None
}
return False
class IsForwarded(BaseFilter):
"""
Проверяет, является ли сообщение пересланным.
Поддерживает:
- Пересылку от пользователей (forward_from)
- Пересылку из каналов/групп (forward_from_chat)
- Скрытую пересылку (forward_sender_name)
Example:
```python
@router.message(IsForwarded())
async def handle_forwarded(message: Message, forward_info: dict):
await message.answer(f"Переслано из: {forward_info['origin']}")
```
"""
async def __call__(self, message: Message) -> Union[bool, dict]:
# Проверка различных типов пересылки
is_forwarded = (
message.forward_origin is not None or # Новый API (aiogram 3.x)
message.forward_from is not None or
message.forward_from_chat is not None or
message.forward_sender_name is not None
)
if is_forwarded:
origin = "неизвестно"
if message.forward_from:
origin = f"пользователь @{message.forward_from.username or message.forward_from.id}"
elif message.forward_from_chat:
origin = f"чат {message.forward_from_chat.title or message.forward_from_chat.id}"
elif message.forward_sender_name:
origin = f"скрытый пользователь ({message.forward_sender_name})"
logger.debug(
f"Обнаружено пересланное сообщение из: {origin}",
log_type='FORWARD',
message=message
)
return {
'is_forwarded': True,
'origin': origin,
'forward_date': message.forward_date
}
return False
class HasMedia(BaseFilter):
"""
Проверяет, содержит ли сообщение медиа-контент.
Attributes:
media_types: Список типов медиа для проверки (если None, проверяются все)
Example:
```python
# Любое медиа
@router.message(HasMedia())
async def handle_media(message: Message):
await message.answer("Получено медиа!")
# Только фото и видео
@router.message(HasMedia(['photo', 'video']))
async def handle_visual(message: Message):
await message.answer("Фото или видео!")
```
"""
def __init__(self, media_types: Optional[list[str]] = None):
"""
Args:
media_types: Список типов медиа ('photo', 'video', 'document', и т.д.)
Если None, проверяются все типы
"""
self.media_types = media_types
async def __call__(self, message: Message) -> Union[bool, dict]:
# Все возможные типы медиа
media_checks = {
'photo': message.photo,
'video': message.video,
'document': message.document,
'audio': message.audio,
'voice': message.voice,
'video_note': message.video_note,
'sticker': message.sticker,
'animation': message.animation,
}
# Если указаны конкретные типы, проверяем только их
if self.media_types:
has_media = any(
media_checks[media_type]
for media_type in self.media_types
if media_type in media_checks
)
detected_type = next(
(media_type for media_type in self.media_types if media_checks.get(media_type)),
None
)
else:
# Проверяем все типы
has_media = any(media_checks.values())
detected_type = next(
(media_type for media_type, value in media_checks.items() if value),
None
)
if has_media:
return {
'has_media': True,
'media_type': detected_type,
'content': media_checks[detected_type]
}
return False
class ContainsURL(BaseFilter):
"""
Проверяет, содержит ли сообщение ссылки.
Поддерживает:
- HTTP/HTTPS ссылки
- Telegram ссылки (t.me, tg://)
- Проверку через entities (более точная)
Attributes:
strict: Использовать строгую проверку через entities
Example:
```python
@router.message(ContainsURL())
async def handle_url(message: Message, url_info: dict):
urls = url_info['urls']
await message.answer(f"Обнаружено {len(urls)} ссылок")
```
"""
def __init__(self, strict: bool = False):
"""
Args:
strict: Если True, проверяет через entities (игнорирует текст в коде/pre)
"""
self.strict = strict
# Паттерн для поиска URL
self.url_pattern = re.compile(
r'https?://[^\s]+|' # http(s)://
r't\.me/[^\s]+|' # t.me/
r'tg://[^\s]+', # tg://
re.IGNORECASE
)
async def __call__(self, message: Message) -> Union[bool, dict]:
if not message.text and not message.caption:
return False
text = message.text or message.caption
if self.strict and message.entities:
# Строгая проверка через entities
url_entities = [
entity for entity in message.entities
if entity.type in ('url', 'text_link')
]
if url_entities:
urls = []
for entity in url_entities:
if entity.type == 'url':
url = text[entity.offset:entity.offset + entity.length]
urls.append(url)
elif entity.type == 'text_link':
urls.append(entity.url)
return {
'contains_url': True,
'urls': urls,
'url_count': len(urls)
}
else:
# Простая проверка через regex
urls = self.url_pattern.findall(text)
if urls:
return {
'contains_url': True,
'urls': urls,
'url_count': len(urls)
}
return False
class HasText(BaseFilter):
"""
Проверяет, содержит ли сообщение текст.
Attributes:
min_length: Минимальная длина текста (по умолчанию 1)
max_length: Максимальная длина текста (по умолчанию None)
Example:
```python
# Любой текст
@router.message(HasText())
async def handle_text(message: Message):
await message.answer("Получен текст!")
# Текст от 10 до 100 символов
@router.message(HasText(min_length=10, max_length=100))
async def handle_medium_text(message: Message):
await message.answer("Текст подходящей длины!")
```
"""
def __init__(self, min_length: int = 1, max_length: Optional[int] = None):
self.min_length = min_length
self.max_length = max_length
async def __call__(self, message: Message) -> Union[bool, dict]:
if not message.text:
return False
text_length = len(message.text)
# Проверка длины
if text_length < self.min_length:
return False
if self.max_length and text_length > self.max_length:
return False
return {
'has_text': True,
'text_length': text_length,
'text': message.text
}
class HasCaption(BaseFilter):
"""
Проверяет, есть ли у медиа подпись.
Example:
```python
@router.message(HasCaption())
async def handle_caption(message: Message):
await message.answer(f"Подпись: {message.caption}")
```
"""
async def __call__(self, message: Message) -> Union[bool, dict]:
if message.caption:
return {
'has_caption': True,
'caption': message.caption,
'caption_length': len(message.caption)
}
return False
class HasEntities(BaseFilter):
"""
Проверяет наличие entities (упоминания, хештеги, команды и т.д.).
Attributes:
entity_types: Список типов entities для проверки
Example:
```python
# Любые entities
@router.message(HasEntities())
async def handle_entities(message: Message):
pass
# Только упоминания и хештеги
@router.message(HasEntities(['mention', 'hashtag']))
async def handle_mentions(message: Message):
pass
```
"""
def __init__(self, entity_types: Optional[list[str]] = None):
"""
Args:
entity_types: Список типов ('mention', 'hashtag', 'bot_command', и т.д.)
"""
self.entity_types = entity_types
async def __call__(self, message: Message) -> Union[bool, dict]:
if not message.entities:
return False
if self.entity_types:
# Фильтруем по типам
matching_entities = [
entity for entity in message.entities
if entity.type in self.entity_types
]
if matching_entities:
return {
'has_entities': True,
'entities': matching_entities,
'entity_count': len(matching_entities)
}
else:
# Любые entities
return {
'has_entities': True,
'entities': message.entities,
'entity_count': len(message.entities)
}
return False
class MediaType(BaseFilter):
"""
Проверяет точный тип контента сообщения.
Attributes:
content_type: Тип контента из ContentType enum
Example:
```python
@router.message(MediaType(ContentType.PHOTO))
async def handle_photo(message: Message):
await message.answer("Это фото!")
```
"""
def __init__(self, content_type: Union[ContentType, str]):
"""
Args:
content_type: Тип контента (ContentType enum или строка)
"""
self.content_type = content_type if isinstance(content_type, str) else content_type.value
async def __call__(self, message: Message) -> bool:
return message.content_type == self.content_type

View File

@@ -0,0 +1,77 @@
from asyncio import create_task
from aiogram import Router, F
from aiogram.filters import Command
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery
from bot.core.bots import BotInfo, bot
from bot.filters import IsOwner
from bot.templates import msg
from bot.utils import status_clear
from bot.utils.auto_delete import auto_delete_message
from configs import COMMANDS
__all__ = ("router",)
CMD: str = "pin".lower()
router: Router = Router(name=f"{CMD}_cmd_router")
@router.message(Command(*COMMANDS[CMD], prefix=BotInfo.prefix, ignore_case=True), IsOwner())
async def pin_cmd(message: Message, state: FSMContext) -> None:
"""
Обработчик команды /pin для закрепления последнего сообщения или ответа.
"""
# Если есть reply → закрепляем его, иначе закрепляем предыдущее сообщение
if message.reply_to_message:
target_message_id = message.reply_to_message.message_id
else:
# Закрепляем предыдущее сообщение (команда - 1)
target_message_id = message.message_id - 1
try:
await bot.pin_chat_message(
chat_id=message.chat.id,
message_id=target_message_id,
disable_notification=False
)
# Автоудаление через 7 суток (удаляем закрепленное сообщение)
create_task(
auto_delete_message(
chat_id=message.chat.id,
message_id=target_message_id,
delay=604800
)
)
await msg(update=message, text="✅ Сообщение успешно закреплено", state=state)
except Exception as e:
await msg(update=message, text=f"❌ Ошибка закрепления: {e}", state=state)
@router.callback_query(F.data.casefold().isin(COMMANDS[CMD]), IsOwner())
async def pin_callback(callback: CallbackQuery, state: FSMContext) -> None:
"""
Обработчик кнопки с callback_data="pin".
"""
await status_clear(update=callback.message, state=state)
try:
await bot.pin_chat_message(
chat_id=callback.message.chat.id,
message_id=callback.message.message_id,
disable_notification=False
)
create_task(
auto_delete_message(
chat_id=callback.message.chat.id,
message_id=callback.message.message_id,
delay=604800
)
)
await callback.answer("✅ Сообщение закреплено")
except Exception as e:
await callback.answer(f"❌ Ошибка: {e}", show_alert=True)

View File

@@ -0,0 +1,223 @@
"""
Модуль смены аватарки бота.
Совместим с aiogram 3.22.0 и Bot API 9.4+
Использует:
- SetMyProfilePhoto
- InputProfilePhotoStatic
"""
from __future__ import annotations
import os
from typing import Union
from aiogram import Router, Bot, F
from aiogram.filters import Command
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import StatesGroup, State
from aiogram.types import (
Message,
CallbackQuery,
FSInputFile,
InputProfilePhotoStatic,
)
from aiogram.methods.set_my_profile_photo import SetMyProfilePhoto
from aiogram.exceptions import TelegramAPIError, TelegramRetryAfter
from aiogram.utils.i18n import gettext as _
from bot.filters import IsSuperAdmin
from bot.templates import msg
from bot.utils import format_retry_time, status_clear
from bot.core.bots import BotInfo
from bot.handlers.commands.settings.settings_cmd import settings_keyboard
from configs import COMMANDS
from middleware.loggers import logger
__all__ = ("router",)
CMD: str = "set_avatar".lower()
router: Router = Router(name=f"{CMD}_router")
# ================= FSM =================
class SetBotAvatarForm(StatesGroup):
"""
FSM состояния для смены аватарки.
"""
waiting_for_photo: State = State()
# ================= CORE =================
async def handle_set_avatar(
update: Union[Message, CallbackQuery],
state: FSMContext,
bot: Bot
) -> None:
"""
Устанавливает новую аватарку бота.
Args:
update: Message или CallbackQuery
state: FSM контекст
bot: Экземпляр бота
"""
message: Message = update.message if isinstance(update, CallbackQuery) else update
if not message.photo:
return
largest_photo = message.photo[-1]
# Получаем файл от Telegram
tg_file = await bot.get_file(largest_photo.file_id)
temp_path: str = f"/tmp/{largest_photo.file_unique_id}.jpg"
await bot.download_file(tg_file.file_path, destination=temp_path)
try:
method = SetMyProfilePhoto(
photo=InputProfilePhotoStatic(
photo=FSInputFile(temp_path)
)
)
result: bool = await bot(method)
if result:
logger.info("Аватарка бота успешно обновлена", log_type="BOT_SETUP")
await state.clear()
await msg(
update=update,
text=_("✅ Аватарка бота успешно обновлена."),
markup=settings_keyboard(),
state=state
)
except TelegramRetryAfter as e:
retry_text: str = format_retry_time(e.retry_after)
logger.warning(
f"Rate limit при смене аватарки. Повтор через {retry_text}",
log_type="BOT_SETUP"
)
await msg(
update=update,
text=_(
"⚠️ Слишком частая смена аватарки.\n"
"Попробуйте снова через: <b>{retry}</b>"
).format(retry=retry_text),
markup=settings_keyboard(),
state=state
)
except TelegramAPIError as e:
logger.error(
f"Ошибка Telegram API при смене аватарки: {e}",
log_type="BOT_SETUP"
)
await msg(
update=update,
text=_(
"❌ Ошибка Telegram API:\n"
"<pre>{error}</pre>"
).format(error=str(e)),
markup=settings_keyboard(),
state=state
)
except Exception as e:
logger.error(
f"Непредвиденная ошибка при смене аватарки: {e}",
log_type="BOT_SETUP"
)
await msg(
update=update,
text=_(
"❌ Непредвиденная ошибка:\n"
"<pre>{error}</pre>"
).format(error=str(e)),
markup=settings_keyboard(),
state=state
)
finally:
if os.path.exists(temp_path):
os.remove(temp_path)
# ================= COMMAND =================
@router.callback_query(F.data.lower() == CMD, IsSuperAdmin())
@router.message(
Command(*COMMANDS.get(CMD, [CMD]), prefix=BotInfo.prefix, ignore_case=True),
IsSuperAdmin()
)
async def set_avatar_cmd(
message: Message | CallbackQuery,
state: FSMContext,
bot: Bot
) -> None:
"""
Команда /set_avatar
Поддерживает:
- Фото вместе с командой
- Ответ на фото
- FSM режим
"""
await status_clear(update=message, state=state)
msg_obj: Message = message.message if isinstance(message, CallbackQuery) else message
if msg_obj.photo:
await handle_set_avatar(message, state, bot)
return
if msg_obj.reply_to_message and msg_obj.reply_to_message.photo:
await handle_set_avatar(msg_obj.reply_to_message, state, bot)
return
await msg(
update=message,
text=_(
"🖼 <b>Смена аватарки бота</b>\n\n"
"Отправьте фотографию."
),
markup=settings_keyboard(),
state=state
)
await state.set_state(SetBotAvatarForm.waiting_for_photo)
# ================= FSM =================
@router.message(SetBotAvatarForm.waiting_for_photo, IsSuperAdmin(), F.photo)
async def process_avatar_photo(
message: Message,
state: FSMContext,
bot: Bot
) -> None:
"""
Обработка фото через FSM.
"""
await handle_set_avatar(message, state, bot)
@router.message(SetBotAvatarForm.waiting_for_photo, IsSuperAdmin())
async def invalid_input(message: Message) -> None:
"""
Обработка некорректного ввода.
"""
await message.answer(_("❌ Пожалуйста, отправьте фотографию."))

View File

@@ -6,8 +6,8 @@ from aiogram.fsm.state import StatesGroup, State
from aiogram.types import Message, CallbackQuery
from aiogram.utils.i18n import gettext as _
from bot.filters import IsSuperAdmin
from bot.core.bots import BotInfo
from bot.filters import IsOwner
from bot.handlers.commands.settings.settings_cmd import settings_keyboard
from bot.templates import msg
from bot.utils import format_retry_time, status_clear
@@ -60,7 +60,7 @@ async def handle_set_bot_description(
await bot.set_my_short_description(short_description=description)
# Сохраняем текущее значение в BotInfo
BotInfo.short_description = description
BotInfo.widget = description
# Сбрасываем состояние FSM
await state.clear()
@@ -108,9 +108,9 @@ async def handle_set_bot_description(
)
@router.callback_query(F.data.lower() == CMD, IsOwner())
@router.message(Command(*COMMANDS[CMD], prefix=BotInfo.prefix, ignore_case=True), IsOwner())
async def settings_cmd(
@router.callback_query(F.data.lower() == CMD, IsSuperAdmin())
@router.message(Command(*COMMANDS[CMD], prefix=BotInfo.prefix, ignore_case=True), IsSuperAdmin())
async def set_description_cmd(
message: Message | CallbackQuery,
state: FSMContext,
bot: Bot,
@@ -155,7 +155,7 @@ async def settings_cmd(
await state.set_state(SetBotDescriptionForm.new_description)
@router.message(SetBotDescriptionForm.new_description, IsOwner())
@router.message(SetBotDescriptionForm.new_description, IsSuperAdmin())
async def process_new_bot_description(
message: Message,
state: FSMContext,
@@ -171,3 +171,4 @@ async def process_new_bot_description(
return
await handle_set_bot_description(description, message, state, bot)
BotInfo.description = description

View File

@@ -6,8 +6,8 @@ from aiogram.fsm.state import StatesGroup, State
from aiogram.types import Message, CallbackQuery
from aiogram.utils.i18n import gettext as _
from bot.filters import IsSuperAdmin
from bot.core.bots import BotInfo
from bot.filters import IsOwner
from bot.handlers.commands.settings.settings_cmd import settings_keyboard
from bot.templates import msg
from configs import COMMANDS
@@ -98,9 +98,9 @@ async def handle_set_name(
)
@router.callback_query(F.data.lower() == CMD, IsOwner())
@router.message(Command(*COMMANDS[CMD], prefix=BotInfo.prefix, ignore_case=True), IsOwner())
async def settings_cmd(
@router.callback_query(F.data.lower() == CMD, IsSuperAdmin())
@router.message(Command(*COMMANDS[CMD], prefix=BotInfo.prefix, ignore_case=True), IsSuperAdmin())
async def set_name_cmd(
message: Message | CallbackQuery,
state: FSMContext,
bot: Bot,
@@ -143,7 +143,7 @@ async def settings_cmd(
await state.set_state(SetNameForm.new_name)
@router.message(SetNameForm.new_name, IsOwner())
@router.message(SetNameForm.new_name, IsSuperAdmin())
async def process_new_name(message: Message, state: FSMContext, bot: Bot):
"""
Обработка ввода нового имени через FSM
@@ -155,3 +155,4 @@ async def process_new_name(message: Message, state: FSMContext, bot: Bot):
return
await handle_set_name(new_name, message, state, bot)
BotInfo.first_name = new_name

View File

@@ -7,7 +7,7 @@ from aiogram.types import Message, CallbackQuery
from aiogram.utils.i18n import gettext as _
from bot.core.bots import BotInfo
from bot.filters import IsOwner
from bot.filters import IsSuperAdmin
from bot.handlers.commands.settings.settings_cmd import settings_keyboard
from bot.templates import msg
from bot.utils import format_retry_time, status_clear
@@ -107,9 +107,9 @@ async def handle_set_widget(
)
@router.callback_query(F.data.lower() == CMD, IsOwner())
@router.message(Command(*COMMANDS[CMD], prefix=BotInfo.prefix, ignore_case=True), IsOwner())
async def settings_cmd(
@router.callback_query(F.data.lower() == CMD, IsSuperAdmin())
@router.message(Command(*COMMANDS[CMD], prefix=BotInfo.prefix, ignore_case=True), IsSuperAdmin())
async def set_widget_cmd(
message: Message | CallbackQuery,
state: FSMContext,
bot: Bot,
@@ -124,7 +124,7 @@ async def settings_cmd(
3. FSM ввод.
"""
# Получаем текущее значение виджета
current_widget: str = BotInfo.short_description
current_widget: str = BotInfo.widget
# Вариант 1: пользователь ввёл аргумент сразу (/set_widget TEXT)
if command and command.args:
@@ -155,7 +155,7 @@ async def settings_cmd(
await state.set_state(SetWidgetForm.new_widget)
@router.message(SetWidgetForm.new_widget, IsOwner())
@router.message(SetWidgetForm.new_widget, IsSuperAdmin())
async def process_new_widget(
message: Message,
state: FSMContext,
@@ -172,3 +172,4 @@ async def process_new_widget(
return
await handle_set_widget(new_widget, message, state, bot)
BotInfo.widget = new_widget

View File

@@ -6,7 +6,7 @@ from aiogram.utils.i18n import gettext as _
from aiogram.utils.keyboard import InlineKeyboardBuilder
from bot.core.bots import BotInfo
from bot.filters import IsOwner
from bot.filters import IsSuperAdmin
from bot.templates import msg
from bot.utils import status_clear
from configs import COMMANDS
@@ -20,12 +20,12 @@ router: Router = Router(name=f"{CMD}_cmd_router")
def settings_keyboard() -> InlineKeyboardBuilder:
"""Клавиатура настроек"""
ikb: InlineKeyboardBuilder = InlineKeyboardBuilder()
ikb.row(InlineKeyboardButton(text="🔙 Вернуться", callback_data="settings"))
ikb.row(InlineKeyboardButton(text="🔙 Вернуться", callback_data=CMD))
return ikb
@router.callback_query(F.data.lower() == CMD, IsOwner())
@router.message(Command(*COMMANDS[CMD], prefix=BotInfo.prefix, ignore_case=True), IsOwner())
@router.callback_query(F.data.lower() == CMD, IsSuperAdmin())
@router.message(Command(*COMMANDS[CMD], prefix=BotInfo.prefix, ignore_case=True), IsSuperAdmin())
async def settings_cmd(message: Message | CallbackQuery, state: FSMContext) -> None:
"""Обработчик команды /settings"""
await status_clear(update=message, state=state)
@@ -35,7 +35,9 @@ async def settings_cmd(message: Message | CallbackQuery, state: FSMContext) -> N
ikb.row(InlineKeyboardButton(text="Имя бота⚜️", callback_data='set_name'))
ikb.row(InlineKeyboardButton(text="Описание бота📝", callback_data='set_description'))
ikb.row(InlineKeyboardButton(text="Виджет🧩", callback_data='set_widget'))
ikb.row(InlineKeyboardButton(text="Назад◀️", callback_data='menu'))
ikb.row(InlineKeyboardButton(text="Аватарка🖼", callback_data='set_avatar'))
ikb.row(InlineKeyboardButton(text="Назад◀️", callback_data='settings'))
ikb.adjust(2)
# Формируем приветственное сообщение
text: str = _("""
@@ -46,3 +48,19 @@ async def settings_cmd(message: Message | CallbackQuery, state: FSMContext) -> N
# Отправляем сообщение
await msg(update=message, text=text, markup=ikb, state=state)
@router.callback_query(F.data.lower() == "set_avatar", IsSuperAdmin())
async def avatar_zaglushka(
callback: CallbackQuery,
state: FSMContext
) -> None:
"""
Аватары нельзя менять в aiogram хнык
"""
await status_clear(update=callback, state=state)
await callback.answer(
text="Ну бля, я не виноват что тг говно и не даст поменять",
show_alert=True
)

View File

@@ -11,7 +11,6 @@ from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.exceptions import TelegramBadRequest
from middleware.loggers import logger
from bot.filters.admin import IsAdmin
from database import get_manager
@@ -48,6 +47,7 @@ def create_settings_menu() -> InlineKeyboardBuilder:
ikb.button(text="📊 Чат репортов", callback_data="settings:report_chat")
ikb.button(text="🧵 Топик репортов", callback_data="settings:report_thread")
ikb.button(text="🔄 Обновить", callback_data="settings:refresh")
ikb.button(text="⚙️ Настройка бота", callback_data="botsettings")
ikb.button(text="❌ Закрыть", callback_data="settings:close")
ikb.adjust(2)
return ikb
@@ -62,6 +62,7 @@ def cancel_keyboard():
# MAIN HANDLER
# ======================================================================
@router.callback_query(F.data.lower() == "settings", IsAdmin())
@router.message(Command("settings"), IsAdmin())
async def settings_cmd(message: Message, state: FSMContext) -> None:
"""Главная команда /settings"""

View File

@@ -99,7 +99,7 @@ async def add_conflict_word_cmd(message: Message) -> None:
try:
added = await manager.add_banword(
word=word,
word_type=BanWordType.CONFLICT_SUBSTRING,
word_type=BanWordType.CONFLICT_WORD,
added_by=message.from_user.id,
reason="Конфликтное слово"
)
@@ -192,7 +192,7 @@ async def remove_conflict_word_cmd(message: Message) -> None:
try:
removed = await manager.remove_banword(
word=word,
word_type=BanWordType.CONFLICT_SUBSTRING
word_type=BanWordType.CONFLICT_WORD
)
if removed:
@@ -433,3 +433,113 @@ async def conflict_status_cmd(message: Message) -> None:
except Exception as e:
logger.error(f"Ошибка получения статуса режима: {e}", log_type="CONFLICT")
await message.answer("❌ <b>Ошибка получения статуса</b>", parse_mode="HTML")
@router.message(
Command(*COMMANDS.get("addconflictpart", ["addconflictpart"]),
prefix=settings.PREFIX,
ignore_case=True),
IsAdmin()
)
@log_action(action_name="ADD_CONFLICT_PART", log_args=True)
async def add_conflict_part_cmd(message: Message) -> None:
"""
Добавляет конфликтную часть.
Использование: /addconflictpart <комбинация>
"""
success, result = parse_conflict_args(
message.text,
"addconflictpart",
need_minutes=False
)
if not success:
await message.answer(result, parse_mode="HTML")
return
word = result[0].lower().strip()
manager = get_manager()
try:
added = await manager.add_banword(
word=word,
word_type=BanWordType.CONFLICT_PART,
added_by=message.from_user.id,
reason="Конфликтная часть"
)
if added:
text = (
f"✅ <b>Конфликтная часть добавлена</b>\n\n"
f"🧩 Часть: <code>{word}</code>\n"
f"🔍 Тип: поиск без пробелов\n\n"
f"⚔️ <i>Работает только в режиме антиконфликта</i>\n"
f"Активируйте: <code>/stopconflict [минуты]</code>"
)
else:
text = f"⚠️ Конфликтная часть <code>{word}</code> уже существует"
await message.answer(text, parse_mode="HTML")
except Exception as e:
logger.error(
f"Ошибка добавления конфликтной части: {e}",
log_type="CONFLICT"
)
await message.answer(
"❌ <b>Ошибка добавления</b>\n\nПопробуйте позже",
parse_mode="HTML"
)
@router.message(
Command(*COMMANDS.get("remconflictpart", ["remconflictpart"]),
prefix=settings.PREFIX,
ignore_case=True),
IsAdmin()
)
@log_action(action_name="REMOVE_CONFLICT_PART", log_args=True)
async def remove_conflict_part_cmd(message: Message) -> None:
"""
Удаляет конфликтную часть.
Использование: /remconflictpart <комбинация>
"""
success, result = parse_conflict_args(
message.text,
"remconflictpart",
need_minutes=False
)
if not success:
await message.answer(result, parse_mode="HTML")
return
word = result[0].lower().strip()
manager = get_manager()
try:
removed = await manager.remove_banword(
word=word,
word_type=BanWordType.CONFLICT_PART
)
if removed:
text = (
f"🗑 <b>Конфликтная часть удалена</b>\n\n"
f"🧩 Часть: <code>{word}</code>"
)
else:
text = f"⚠️ Конфликтная часть <code>{word}</code> не найдена"
await message.answer(text, parse_mode="HTML")
except Exception as e:
logger.error(
f"Ошибка удаления конфликтной части: {e}",
log_type="CONFLICT"
)
await message.answer(
"❌ <b>Ошибка удаления</b>\n\nПопробуйте позже",
parse_mode="HTML"
)

View File

@@ -7,7 +7,6 @@ from aiogram.types import Message, CallbackQuery
from aiogram.utils.keyboard import InlineKeyboardBuilder
from aiogram.exceptions import TelegramBadRequest
from bot.filters.admin import IsAdmin
from configs import settings, COMMANDS
from database import get_manager
from middleware.loggers import logger
@@ -27,7 +26,7 @@ def get_refresh_kb(page: int = 0):
return ikb.as_markup()
async def format_banwords_list(page: int = 0) -> str:
async def format_banwords_list() -> str:
"""
Форматирует список всех банвордов с разбивкой по типам.
@@ -46,13 +45,14 @@ async def format_banwords_list(page: int = 0) -> str:
stats = await manager.get_stats()
# Извлекаем данные из словаря
permanent_words = list(data.get('substring', set()))
permanent_words = list(data.get('word', set()))
permanent_lemmas = list(data.get('lemma', set()))
permanent_parts = list(data.get('part', set()))
temp_words = list(data.get('temp_substring', set()))
temp_words = list(data.get('temp_word', set()))
temp_lemmas = list(data.get('temp_lemma', set()))
conflict_words = list(data.get('conflict_substring', set()))
conflict_words = list(data.get('conflict_word', set()))
conflict_lemmas = list(data.get('conflict_lemma', set()))
conflict_parts = list(data.get('conflict_part', set()))
exceptions = list(data.get('whitelist', set()))
except Exception as e:
@@ -67,7 +67,7 @@ async def format_banwords_list(page: int = 0) -> str:
total_count = (
len(permanent_words) + len(permanent_lemmas) + len(permanent_parts) +
len(temp_words) + len(temp_lemmas) +
len(conflict_words) + len(conflict_lemmas)
len(conflict_words) + len(conflict_lemmas) + len(conflict_parts)
)
output += f"📊 <b>Общая статистика:</b>\n"
@@ -81,21 +81,21 @@ async def format_banwords_list(page: int = 0) -> str:
output += "🔴 <b>ПОСТОЯННЫЕ ПРАВИЛА:</b>\n\n"
if permanent_words:
output += f"📝 <b>Подстроки</b> ({len(permanent_words)}):\n"
output += f"📝 <b>Слова</b> ({len(permanent_words)}):\n"
words_str = ', '.join([f"<code>{w}</code>" for w in sorted(permanent_words)[:20]])
if len(permanent_words) > 20:
words_str += f" ... <i>(+{len(permanent_words) - 20} ещё)</i>"
output += f"{words_str}\n\n"
if permanent_lemmas:
output += f"🔤 <b>Леммы</b> ({len(permanent_lemmas)}):\n"
output += f"🔤 <b>Леммы (морф.формы)</b> ({len(permanent_lemmas)}):\n"
lemmas_str = ', '.join([f"<code>{w}</code>" for w in sorted(permanent_lemmas)[:20]])
if len(permanent_lemmas) > 20:
lemmas_str += f" ... <i>(+{len(permanent_lemmas) - 20} ещё)</i>"
output += f"{lemmas_str}\n\n"
if permanent_parts:
output += f"🧩 <b>Части</b> ({len(permanent_parts)}):\n"
output += f"🧩 <b>Части в сообщении</b> ({len(permanent_parts)}):\n"
parts_str = ', '.join([f"<code>{w}</code>" for w in sorted(permanent_parts)[:20]])
if len(permanent_parts) > 20:
parts_str += f" ... <i>(+{len(permanent_parts) - 20} ещё)</i>"
@@ -106,7 +106,7 @@ async def format_banwords_list(page: int = 0) -> str:
output += "⏱ <b>ВРЕМЕННЫЕ ПРАВИЛА:</b>\n\n"
if temp_words:
output += f"📝 <b>Временные подстроки</b> ({len(temp_words)}):\n"
output += f"📝 <b>Временные слова</b> ({len(temp_words)}):\n"
# Для временных слов нужна дополнительная информация о времени истечения
# Пока просто выводим список
words_str = ', '.join([f"<code>{w}</code>" for w in sorted(temp_words)[:15]])
@@ -122,7 +122,7 @@ async def format_banwords_list(page: int = 0) -> str:
output += f"{lemmas_str}\n\n"
# === КОНФЛИКТНЫЕ ПРАВИЛА ===
if conflict_words or conflict_lemmas:
if conflict_words or conflict_lemmas or conflict_parts:
output += "⚔️ <b>КОНФЛИКТНЫЕ ПРАВИЛА:</b>\n"
output += "<i>(работают только в режиме <code>/stopconflict</code> <code>время</code>)</i>\n\n"
@@ -140,10 +140,17 @@ async def format_banwords_list(page: int = 0) -> str:
lemmas_str += f" ... <i>(+{len(conflict_lemmas) - 15} ещё)</i>"
output += f"{lemmas_str}\n\n"
if conflict_parts:
output += f"🧩 <b>Конфликтные части</b> ({len(conflict_parts)}):\n"
parts_str = ', '.join([f"<code>{w}</code>" for w in sorted(conflict_parts)[:15]])
if len(conflict_parts) > 15:
parts_str += f" ... <i>(+{len(conflict_parts) - 15} ещё)</i>"
output += f"{parts_str}\n\n"
# === ИСКЛЮЧЕНИЯ (WHITELIST) ===
if exceptions:
output += f"✅ <b>ИСКЛЮЧЕНИЯ</b> ({len(exceptions)}):\n"
exc_str = ', '.join([f"<code>{exceptions}</code>" for w in sorted(exceptions)[:15]])
exc_str = ', '.join([f"<code>{w}</code>" for w in sorted(exceptions)[:15]])
if len(exceptions) > 15:
exc_str += f" ... <i>(+{len(exceptions) - 15} ещё)</i>"
output += f"{exc_str}\n\n"
@@ -183,7 +190,7 @@ async def format_banwords_list(page: int = 0) -> str:
@router.callback_query(F.data.startswith("listwords:refresh"))
@router.message(Command(*COMMANDS[CMD], prefix=settings.PREFIX, ignore_case=True), IsAdmin())
@router.message(Command(*COMMANDS[CMD], prefix=settings.PREFIX, ignore_case=True))
@log_action(action_name="LISTWORDS_COMMAND")
async def listwords_cmd(update: Message | CallbackQuery) -> None:
"""
@@ -209,7 +216,7 @@ async def listwords_cmd(update: Message | CallbackQuery) -> None:
# Формируем список
try:
text = await format_banwords_list(page)
text = await format_banwords_list()
keyboard = get_refresh_kb(page)
if is_callback:

View File

@@ -0,0 +1,118 @@
"""
Обработчики callback-кнопок уведомлений о спаме
"""
from aiogram import Router, F
from aiogram.types import CallbackQuery
from aiogram.exceptions import TelegramBadRequest
from bot.filters.admin import IsAdmin
from database import get_manager
from middleware.loggers import logger
__all__ = ("router",)
router: Router = Router(name="spam_notifications_router")
# ================= ЗАКРЫТИЕ УВЕДОМЛЕНИЯ =================
@router.callback_query(F.data == "spam_close", IsAdmin())
async def spam_close_callback(callback: CallbackQuery) -> None:
"""
Закрывает (удаляет) уведомление о спаме.
"""
try:
await callback.message.delete()
await callback.answer("✅ Уведомление закрыто")
logger.debug(
f"Уведомление о спаме закрыто админом {callback.from_user.id}",
log_type="SPAM_NOTIFICATION"
)
except TelegramBadRequest as e:
logger.error(f"Ошибка удаления уведомления: {e}", log_type="ERROR")
await callback.answer("Не удалось удалить уведомление", show_alert=True)
# ================= БАН ПОЛЬЗОВАТЕЛЯ =================
@router.callback_query(F.data.startswith("spam_ban:"), IsAdmin())
async def spam_ban_callback(callback: CallbackQuery) -> None:
"""
Банит пользователя прямо из уведомления.
"""
try:
# Парсим данные: spam_ban:user_id:chat_id
parts = callback.data.split(":")
user_id = int(parts[1])
chat_id = int(parts[2])
# Баним пользователя
try:
await callback.bot.ban_chat_member(
chat_id=chat_id,
user_id=user_id
)
# Обновляем сообщение
updated_text = callback.message.text + f"\n\n🔨 <b>Пользователь забанен</b> (@{callback.from_user.username or callback.from_user.id})"
# Убираем кнопки
await callback.message.edit_text(
text=updated_text,
parse_mode="HTML"
)
await callback.answer("✅ Пользователь забанен", show_alert=True)
logger.info(
f"Пользователь {user_id} забанен админом {callback.from_user.id} через уведомление о спаме",
log_type="SPAM_BAN"
)
except TelegramBadRequest as e:
await callback.answer(f"❌ Ошибка бана: {str(e)}", show_alert=True)
except Exception as e:
logger.error(f"Ошибка обработки бана из уведомления: {e}", log_type="ERROR")
await callback.answer("❌ Ошибка выполнения", show_alert=True)
# ================= СТАТИСТИКА ПОЛЬЗОВАТЕЛЯ =================
@router.callback_query(F.data.startswith("spam_stats:"), IsAdmin())
async def spam_stats_callback(callback: CallbackQuery) -> None:
"""
Показывает статистику пользователя.
"""
try:
# Парсим данные: spam_stats:user_id
parts = callback.data.split(":")
user_id = int(parts[1])
manager = get_manager()
# Получаем статистику
spam_count = await manager.get_user_spam_count(user_id)
recent_spam = await manager.get_spam_stats(limit=5, user_id=user_id)
# Формируем текст
text = f"📊 <b>Статистика пользователя</b>\n\n"
text += f"🆔 ID: <code>{user_id}</code>\n"
text += f"🗑 Удалено сообщений: <code>{spam_count}</code>\n\n"
if recent_spam:
text += f"📝 <b>Последние нарушения:</b>\n"
for idx, stat in enumerate(recent_spam, 1):
matched_word = stat.matched_word or "неизвестно"
match_type = stat.match_type or "unknown"
text += f"{idx}. <code>{matched_word}</code> ({match_type})\n"
else:
text += "✅ <i>Нет нарушений</i>"
await callback.answer(text, show_alert=True)
except Exception as e:
logger.error(f"Ошибка получения статистики из уведомления: {e}", log_type="ERROR")
await callback.answer("❌ Ошибка получения статистики", show_alert=True)

View File

@@ -0,0 +1,164 @@
"""
Обработчик команды /start и /help для администраторов.
Показывает список доступных команд для управления банвордами.
"""
from aiogram import Router, F
from aiogram.filters import Command
from aiogram.types import Message, CallbackQuery
from aiogram.utils.keyboard import InlineKeyboardBuilder
from bot.filters.admin import IsAdmin
from configs import settings, COMMANDS
from middleware.loggers import logger
from bot.utils import log_action, tg_emoji
__all__ = ("router",)
CMD: str = "start"
router: Router = Router(name="start_cmd_router")
def kb(text: str = "Создатель⬆️", url: str = "https://t.me/verdise"):
ikb = InlineKeyboardBuilder()
ikb.button(text=text, url=url)
return ikb.as_markup()
@router.callback_query(F.data.casefold() == CMD)
@router.message(Command(*COMMANDS[CMD], prefix=settings.PREFIX, ignore_case=True), IsAdmin())
@log_action(action_name="START_COMMAND", log_args=True)
async def start_cmd(update: Message | CallbackQuery) -> None:
"""
Обработчик команды /start и /help.
Показывает справку по командам бота для администраторов.
Доступно только администраторам (суперадмин или доп. админ из БД).
Args:
update: Message или CallbackQuery
"""
print(123)
# Определяем тип update и извлекаем данные
if isinstance(update, CallbackQuery):
message = update.message
user_id = update.from_user.id
is_callback = True
else:
message = update
user_id = update.from_user.id
is_callback = False
# Проверяем, является ли пользователь суперадмином
is_super_admin = user_id in settings.OWNER_ID
# Формируем текст помощи
help_text = (
f'{tg_emoji("4961073056677103064")} <b>PrimoGuard - Бот-модератор</b>\n\n'
'<blockquote>Автоматическое удаление сообщений с запрещёнными словами.\nПоддержка слов, лемм, временных блокировок и режимов модерации.</blockquote>\n\n'
)
# === Команды просмотра ===
help_text += (
f'{tg_emoji("4961141003059725568")} <b>Просмотр:</b>\n'
'<b>/list</b> — список всех правил и слов\n'
'<b>/stats</b> — статистика по удалениям\n'
'<b>/id</b> — получение айди пользователя\n'
'<b>/chatid</b> — получение айди чата\n\n'
)
# === Постоянные банворды ===
help_text += (
f'{tg_emoji("4961019408240608234")} <b>Добавить банворд (постоянно):</b>\n'
'<code>/word</code> <code>слово</code> — слова (простой поиск)\n'
'<code>/lemma</code> <code>слово</code> — лемма (все формы слова)\n'
'<code>/part</code> <code>комбинация</code> — часть (поиск без пробелов)\n\n'
)
# === Временные банворды ===
help_text += (
f'{tg_emoji("4960719190026618714")} <b>Добавить банворд (временно):</b>\n'
'<code>/tempword</code> <code>слово минуты</code> — временная слова\n'
'<code>/templemma</code> <code>слово минуты</code> — временная лемма\n'
'<i>Пример: /tempword спам 60</i>\n\n'
)
# === Исключения (whitelist) ===
help_text += (
f'{tg_emoji("4963010134172239128")} <b>Исключения (whitelist):</b>\n'
'<code>/addexcept</code> <code>текст</code> — добавить исключение\n'
'<code>/remexcept</code> <code>текст</code> — удалить исключение\n'
'<i>Исключения не проверяются фильтром</i>\n\n'
)
# === Режимы модерации ===
help_text += (
f'{tg_emoji("4960987543878239236")} <b>Режим тишины:</b>\n'
'<code>/silence</code> <code>минуты</code> — удалять ВСЕ сообщения\n'
'<b>/unsilence</b> — отключить режим тишины\n'
'<code>/report</code> — отправить репорт\n\n'
)
help_text += (
f'{tg_emoji("4960986152308835400")} <b>Режим антиконфликта:</b>\n'
'<code>/addconflictword</code> <code>слово</code> — добавить конфликтное слово\n'
'<code>/addconflictlemma</code> <code>слово</code> — добавить конфликтную лемму\n'
'<code>/addconflictpart</code> <code>слово</code> — добавить конфликтную часть\n'
'<code>/stopconflict</code> <code>минуты</code> — активировать режим\n'
'<code>/unstopconflict</code> — отключить режим\n\n'
)
# === Удаление ===
help_text += (
f'{tg_emoji("4961196485447254983")} <b>Удалить:</b>\n'
'<code>/remword</code> <code>слово</code> — удалить слову\n'
'<code>/remlemma</code> <code>слово</code> — удалить лемму\n'
'<code>/rempart</code> <code>комбинация</code> — удалить часть\n'
'<code>/remtempword</code> <code>слово</code> — удалить временную слову\n'
'<code>/remtemplemma</code> <code>слово</code> — удалить временную лемму\n'
'<code>/remconflictword</code> <code>слово</code> — удалить конфликтное слово\n'
'<code>/remconflictpart</code> <code>слово</code> — удалить конфликтное часть\n'
'<code>/remconflictlemma</code> <code>слово</code> — удалить конфликтную лемму\n\n'
)
# === Управление админами (только для суперадминов) ===
if is_super_admin:
help_text += (
f'{tg_emoji("4960891456869893259")} <b>Управление админами (только для владельцев):</b>\n'
'<code>/addadmin</code> <i>ID</i> — добавить администратора\n'
'<code>/remadmin</code> <i>ID</i> — удалить администратора\n'
'<b>/redactcomment</b> — изменить комментарий под постом\n'
'<b>/listadmins</b> — список всех админов\n\n'
)
# === Типы проверок ===
help_text += (
f'{tg_emoji("4961021096162755737")} <b>Типы проверок:</b>\n'
'• <b>Слово</b> — простой поиск в тексте\n'
'• <b>Лемма</b> — все формы слова (купить→куплю, купил, купишь...)\n'
'• <b>Часть</b> — поиск без пробелов (обходит \"к у п и т ь\")\n'
'• <b>Временные</b> — автоматически удаляются через N минут\n'
'• <b>Конфликтные</b> — работают только в режиме /stopconflict\n\n'
)
# Отправляем ответ
try:
if is_callback:
await message.edit_text(
text=help_text,
parse_mode="HTML",
reply_markup=kb()
)
await update.answer()
else:
await message.answer(
text=help_text,
parse_mode="HTML",
reply_markup=kb()
)
except Exception as e:
logger.error(
f"Ошибка отправки help сообщения: {e}",
log_type="ERROR"
)
if is_callback:
await update.answer(f'{tg_emoji("4963277744994518278")} Ошибка отображения справки', show_alert=True)

View File

@@ -153,7 +153,13 @@ async def stats_cmd(update: Message | CallbackQuery) -> None:
conflict_words_count = len(data.get('conflict_substring', set()))
conflict_lemmas_count = len(data.get('conflict_lemma', set()))
total_conflict = conflict_words_count + conflict_lemmas_count
conflict_parts_count = len(data.get('conflict_part', set()))
total_conflict = (
conflict_words_count +
conflict_lemmas_count +
conflict_parts_count
)
output += f"⚔️ <b>Режим антиконфликта</b>\n"
output += f"├─ ⏱ Осталось: <code>{format_time_remaining(time_left_minutes)}</code>\n"

View File

@@ -108,7 +108,7 @@ async def add_word_cmd(message: Message) -> None:
try:
added = await manager.add_banword(
word=word,
word_type=BanWordType.SUBSTRING,
word_type=BanWordType.WORD,
added_by=message.from_user.id,
reason=f"Добавлено через команду"
)
@@ -126,7 +126,7 @@ async def add_word_cmd(message: Message) -> None:
await message.answer(text, parse_mode="HTML")
except Exception as e:
logger.error(f"Ошибка добавления банворда: {e}", log_type="CMD", exc_info=True)
logger.error(f"Ошибка добавления банворда: {e}", log_type="CMD")
await message.answer("❌ <b>Ошибка добавления</b>\n\nПопробуйте позже", parse_mode="HTML")
@@ -168,7 +168,7 @@ async def add_lemma_cmd(message: Message) -> None:
await message.answer(text, parse_mode="HTML")
except Exception as e:
logger.error(f"Ошибка добавления леммы: {e}", log_type="CMD", exc_info=True)
logger.error(f"Ошибка добавления леммы: {e}", log_type="CMD")
await message.answer("❌ <b>Ошибка добавления</b>\n\nПопробуйте позже", parse_mode="HTML")
@@ -210,7 +210,7 @@ async def add_part_cmd(message: Message) -> None:
await message.answer(text, parse_mode="HTML")
except Exception as e:
logger.error(f"Ошибка добавления части: {e}", log_type="CMD", exc_info=True)
logger.error(f"Ошибка добавления части: {e}", log_type="CMD")
await message.answer("❌ <b>Ошибка добавления</b>\n\nПопробуйте позже", parse_mode="HTML")
@@ -246,7 +246,7 @@ async def add_temp_word_cmd(message: Message) -> None:
try:
added = await manager.add_temp_banword(
word=word,
word_type=BanWordType.SUBSTRING,
word_type=BanWordType.WORD,
minutes=minutes,
added_by=message.from_user.id
)
@@ -265,7 +265,7 @@ async def add_temp_word_cmd(message: Message) -> None:
await message.answer(text, parse_mode="HTML")
except Exception as e:
logger.error(f"Ошибка добавления временного банворда: {e}", log_type="CMD", exc_info=True)
logger.error(f"Ошибка добавления временного банворда: {e}", log_type="CMD")
await message.answer("❌ <b>Ошибка добавления</b>\n\nПопробуйте позже", parse_mode="HTML")
@@ -319,7 +319,7 @@ async def add_temp_lemma_cmd(message: Message) -> None:
await message.answer(text, parse_mode="HTML")
except Exception as e:
logger.error(f"Ошибка добавления временной леммы: {e}", log_type="CMD", exc_info=True)
logger.error(f"Ошибка добавления временной леммы: {e}", log_type="CMD")
await message.answer("❌ <b>Ошибка добавления</b>\n\nПопробуйте позже", parse_mode="HTML")
@@ -360,7 +360,7 @@ async def add_exception_cmd(message: Message) -> None:
await message.answer(text, parse_mode="HTML")
except Exception as e:
logger.error(f"Ошибка добавления исключения: {e}", log_type="CMD", exc_info=True)
logger.error(f"Ошибка добавления исключения: {e}", log_type="CMD")
await message.answer("❌ <b>Ошибка добавления</b>\n\nПопробуйте позже", parse_mode="HTML")
@@ -384,7 +384,7 @@ async def remove_word_cmd(message: Message) -> None:
manager = get_manager()
try:
removed = await manager.remove_banword(word=word, word_type=BanWordType.SUBSTRING)
removed = await manager.remove_banword(word=word, word_type=BanWordType.WORD)
if removed:
text = format_success_message("удалена", word, "подстрока")
@@ -394,7 +394,7 @@ async def remove_word_cmd(message: Message) -> None:
await message.answer(text, parse_mode="HTML")
except Exception as e:
logger.error(f"Ошибка удаления банворда: {e}", log_type="CMD", exc_info=True)
logger.error(f"Ошибка удаления банворда: {e}", log_type="CMD")
await message.answer("❌ <b>Ошибка удаления</b>\n\nПопробуйте позже", parse_mode="HTML")
@@ -422,7 +422,7 @@ async def remove_lemma_cmd(message: Message) -> None:
await message.answer(text, parse_mode="HTML")
except Exception as e:
logger.error(f"Ошибка удаления леммы: {e}", log_type="CMD", exc_info=True)
logger.error(f"Ошибка удаления леммы: {e}", log_type="CMD")
await message.answer("❌ <b>Ошибка удаления</b>\n\nПопробуйте позже", parse_mode="HTML")
@@ -450,7 +450,7 @@ async def remove_part_cmd(message: Message) -> None:
await message.answer(text, parse_mode="HTML")
except Exception as e:
logger.error(f"Ошибка удаления части: {e}", log_type="CMD", exc_info=True)
logger.error(f"Ошибка удаления части: {e}", log_type="CMD")
await message.answer("❌ <b>Ошибка удаления</b>\n\nПопробуйте позже", parse_mode="HTML")
@@ -469,7 +469,7 @@ async def remove_temp_word_cmd(message: Message) -> None:
manager = get_manager()
try:
removed = await manager.remove_temp_banword(word=word, word_type=BanWordType.SUBSTRING)
removed = await manager.remove_temp_banword(word=word, word_type=BanWordType.WORD)
if removed:
text = format_success_message("удалена", word, "временная подстрока")
@@ -479,7 +479,7 @@ async def remove_temp_word_cmd(message: Message) -> None:
await message.answer(text, parse_mode="HTML")
except Exception as e:
logger.error(f"Ошибка удаления временного банворда: {e}", log_type="CMD", exc_info=True)
logger.error(f"Ошибка удаления временного банворда: {e}", log_type="CMD")
await message.answer("❌ <b>Ошибка удаления</b>\n\nПопробуйте позже", parse_mode="HTML")
@@ -508,7 +508,7 @@ async def remove_temp_lemma_cmd(message: Message) -> None:
await message.answer(text, parse_mode="HTML")
except Exception as e:
logger.error(f"Ошибка удаления временной леммы: {e}", log_type="CMD", exc_info=True)
logger.error(f"Ошибка удаления временной леммы: {e}", log_type="CMD")
await message.answer("❌ <b>Ошибка удаления</b>\n\nПопробуйте позже", parse_mode="HTML")
@@ -536,5 +536,5 @@ async def remove_exception_cmd(message: Message) -> None:
await message.answer(text, parse_mode="HTML")
except Exception as e:
logger.error(f"Ошибка удаления исключения: {e}", log_type="CMD", exc_info=True)
logger.error(f"Ошибка удаления исключения: {e}", log_type="CMD")
await message.answer("❌ <b>Ошибка удаления</b>\n\nПопробуйте позже", parse_mode="HTML")

View File

@@ -1,15 +1,9 @@
from aiogram import Router
from .default_msg import router as default_message_router
from .ping_test import router as ping_test_message_router
# Настройка экспорта и роутера
router: Router = Router(name=__name__)
# Подготовка роутера команд
# router.include_routers(
# ping_test_message_router,
# )
# Подключение стандартного роутера
router.include_router(default_message_router)

View File

@@ -0,0 +1,242 @@
"""
Триггер-хэндлер: реагирует на обращения к Лайле с именем персонажа.
Формат: "Лайла [что угодно] [имя или псевдоним]"
"""
from typing import Dict, Optional
import random
from aiogram import Router
from aiogram.types import Message
__all__ = ("router",)
router: Router = Router(name="triggers_router")
CHARACTERS: Dict[str, Dict] = {
"эвелин": {
"aliases": ["эвелин", "эва", "эви"],
"answers": [
"Эвелин умеет молчать так, что хочется говорить.",
"Эва всегда знает больше, чем говорит. Это немного пугает.",
"С ней рядом становится спокойно. Не знаю почему.",
"Интересно, о чём она думает в тишине...",
"Эвелин тихая снаружи. Но внутри — целый ураган, я уверена.",
],
},
"лео": {
"aliases": ["лео", "лёва", "лёня"],
"answers": [
"Лео громкий, яркий и немного безрассудный. Мне нравится!",
"Он смеётся первым и уходит последним. Настоящий.",
"Лео всегда найдёт повод для праздника, даже если его нет.",
"Кажется, он боится тишины. Поэтому и заполняет её собой.",
"За его смехом прячется что-то очень серьёзное...",
],
},
"маркус": {
"aliases": ["маркус", "марк"],
"answers": [
"Маркус говорит мало, но каждое слово весит.",
"Он из тех, кто держит слово даже когда это неудобно.",
"С Маркусом не поспоришь. Не потому что нельзя — просто незачем.",
"Он смотрит так, будто видит тебя насквозь. Немного жутковато.",
"Маркус — тот, на кого можно положиться в самый плохой день.",
],
},
"мари": {
"aliases": ["мари", "маришка", "мариша"],
"answers": [
"Мари — это как утренний свет. Мягко и неожиданно тепло.",
"Она помнит мелочи, которые другие не замечают. Это её суперсила.",
"С Мари любой разговор становится важным.",
"Она улыбается даже когда грустно. Не притворяется — просто верит.",
"Мари умеет прощать. Это редкость.",
],
},
"либе": {
"aliases": ["либе", "либ"],
"answers": [
"Либе... имя звучит как песня на незнакомом языке.",
"Она всегда чуть в стороне, но именно к ней тянутся люди.",
"Либе видит красоту там, где другие видят хаос.",
"Она не объясняет себя. И не должна.",
"С Либе можно молчать — и это не будет неловко.",
],
},
"мотциэль": {
"aliases": ["мотциэль", "мотц", "моц"],
"answers": [
"Мотциэль... даже имя звучит как заклинание.",
"Он существует между мирами. Буквально.",
"Спрашивать его о прошлом — плохая идея. Очень плохая.",
"Мотциэль помнит вещи, которых не было. Или были?",
"Его глаза смотрят в разные эпохи одновременно.",
],
},
"виктор": {
"aliases": ["виктор", "вик", "витя"],
"answers": [
"Виктор всегда побеждает. Это в имени.",
"Он не злой. Просто у него другая шкала ценностей.",
"Виктор говорит правду даже когда это больно. Особенно когда больно.",
"Не стоит играть с ним в слова — проиграешь.",
"За его холодностью — старая-старая усталость.",
],
},
"кситти": {
"aliases": ["кситти", "кси", "ксит"],
"answers": [
"Кситти — маленький хаос в красивой упаковке.",
"Она никогда не делает то, что от неё ожидают. Никогда.",
"С Кситти скучно не бывает. Опасно — бывает. Скучно — нет.",
"Она собирает странные вещи и странных людей.",
"Кситти смеётся над правилами. Потому что сама их придумывает.",
],
},
"кадфаль": {
"aliases": ["кадфаль", "кад", "кадф"],
"answers": [
"Кадфаль несёт что-то древнее в каждом шаге.",
"Он не торопится. У него другое ощущение времени.",
"Кадфаль знает цену словам — поэтому тратит их редко.",
"В его присутствии хочется стоять прямо.",
"Он видел многое. Слишком многое для одной жизни.",
],
},
"вайш": {
"aliases": ["вайш", "вай"],
"answers": [
"Вайш появляется неожиданно и исчезает так же.",
"Её след — это вопросы без ответов.",
"Вайш знает что-то, что тебе лучше не знать.",
"Она не объясняет своих решений. Просто делает.",
"С Вайш никогда не знаешь, друг она или нет.",
],
},
"скаф": {
"aliases": ["скаф"],
"answers": [
"Скаф — имя, которое не забывается.",
"Он работает в тени. Не потому что боится света — просто так удобнее.",
"Скаф знает цену всему. Буквально всему.",
"Его нельзя купить. Его можно только нанять. Это разница.",
"Те, кто встречал Скафа, редко рассказывают об этом дважды.",
],
},
"куарти": {
"aliases": ["куарти", "куар"],
"answers": [
"Куарти — четыре буквы и миллион вопросов.",
"Он улыбается, когда другие нервничают. Это не успокаивает.",
"Куарти коллекционирует долги. Чужие.",
"Говорят, он никогда не проигрывает. Говорят.",
"Куарти появляется именно тогда, когда тебе нужна помощь. И это не случайно.",
],
},
"саэрин": {
"aliases": ["саэрин", "саэ", "сэрин"],
"answers": [
"Саэрин — как туман. Красиво и немного опасно.",
"Она говорит загадками не потому что хочет запутать — просто иначе не умеет.",
"Саэрин помнит всё, что ей говорят. Всё.",
"Её спокойствие пугает больше, чем чужой гнев.",
"Саэрин выбирает слова как оружие — точно и без лишнего.",
],
},
"котики": {
"aliases": ["котики", "котик", "кот", "кошка"],
"answers": [
"Котики — это лучшее, что есть в этом мире. Без обсуждений.",
"Котик сел на тебя — ты избран.",
"Кот смотрит на тебя и думает что-то важное. Наверное.",
"Котики всегда правы. Это научный факт.",
"Маленький тёплый комочек счастья. Что ещё нужно?",
],
},
"нотик": {
"aliases": ["нотик", "нота", "нотка"],
"answers": [
"Нотик! Звучит как маленькая музыкальная нота! 🎵",
"Нотик — тот, кто приносит мелодию туда, где её не хватает.",
"Маленький, но важный. Как все хорошие вещи.",
"Нотик — это и ласково, и загадочно одновременно.",
"Из таких маленьких нотиков складываются большие истории.",
],
},
"илья": {
"aliases": ["илья", "илюха", "илюша"],
"answers": [
"Илья — имя с характером. Твёрдое и живое.",
"Илья всегда знает что делать. Или уверенно делает вид.",
"С Ильёй легко — он не усложняет лишнего.",
"Он из тех, кто сделает, а потом расскажет. Не наоборот.",
"Илья редко жалуется. Чаще просто решает.",
],
},
"ина": {
"aliases": ["ина", "инка", "инуля"],
"answers": [
"Ина — короткое имя, за которым много всего.",
"Она тихая, но запоминается.",
"Ина умеет слушать так, что хочется говорить.",
"В ней есть что-то очень своё, неповторимое.",
"Ина — как маленький огонёк. Незаметный, но греет.",
],
},
"абсцисс": {
"aliases": ["абсцисс", "абс"],
"answers": [
"Абсцисс! Это математика или имя? Хи-хи!",
"Ось абсцисс — горизонталь жизни. Всё движется по ней.",
"Абсцисс звучит как заклинание из учебника.",
"Кто-то мечтает о приключениях, а кто-то — об осях координат!",
"Абсцисс и ордината. Звучит как имена двух загадочных персонажей!",
],
},
}
# Имена, на которые Лайла откликается
LAYLA_NAMES = ["лайла", "лайл", "лая"]
def find_character_answer(text: str) -> Optional[str]:
"""
Проверяет:
1. Есть ли в тексте обращение к Лайле
2. Есть ли имя персонажа
Возвращает случайный ответ или None.
"""
text_lower = text.lower()
# Проверяем обращение к Лайле
if not any(name in text_lower for name in LAYLA_NAMES):
return None
# Ищем персонажа
for character, data in CHARACTERS.items():
for alias in data["aliases"]:
if alias in text_lower:
return random.choice(data["answers"])
return None
@router.message()
async def handle_triggers(message: Message) -> None:
"""
Реагирует только если:
- Сообщение от живого человека
- В тексте есть обращение к Лайле
- В тексте есть имя персонажа
На всё остальное — молчит.
"""
#if not message.text or not message.from_user or message.from_user.is_bot:
#return
#response = find_character_answer(message.text)
#if response:
#await message.reply(response)
return

0
bot/keyboards/inline.py Normal file
View File

View File

@@ -47,8 +47,8 @@ def setup_middlewares(
bot: Bot,
admin_ids: list[int] = settings.ADMIN_ID+settings.OWNER_ID,
channel_ids: list[int | str] | None = None,
enable_spam_check: bool = False,
enable_subscription_check: bool = False,
enable_spam_check: bool = settings.enable_spam_check,
enable_subscription_check: bool = settings.enable_subscription_check,
) -> dict:
"""
Регистрирует все middleware в диспетчере.
@@ -138,5 +138,3 @@ def setup_middlewares(
)
return instances

View File

@@ -1,13 +1,5 @@
"""
Middleware для проверки сообщений на запрещённые слова (банворды).
✅ ИСПРАВЛЕНО:
- Полная нормализация текста с использованием UNICODE_MAP
- Удаление повторов символов (леееейн → лейн)
- Игнорирование разделителей (л.е.й.н → лейн)
- Поддержка всех типов проверок (SUBSTRING, LEMMA, PART, CONFLICT)
- Белый список и режимы тишины/конфликта
- Нет уведомлений в режиме тишины
"""
from typing import Callable, Dict, Any, Awaitable, Optional
@@ -15,7 +7,7 @@ import re
import unicodedata
from aiogram import BaseMiddleware
from aiogram.types import Message, InlineKeyboardMarkup, InlineKeyboardButton
from aiogram.types import Message
from aiogram.exceptions import TelegramBadRequest
from configs import settings, UNICODE_MAP, LATIN_TO_CYRILLIC, CYRILLIC_NORMALIZE
@@ -23,97 +15,86 @@ from database import get_manager, BanWordType
from bot.special import extract_words, get_lemma
from middleware.loggers import logger
__all__ = ("BanWordsMiddleware",)
__all__ = ("BanWordsMiddleware",)
URL_PATTERN = re.compile(
r'(https?://\S+|www\.\S+)',
re.IGNORECASE
)
class TextNormalizer:
"""
Класс для многоступенчатой нормализации текста.
Приводит различные юникод-символы к базовым буквам,
удаляет повторы, убирает разделители.
"""
# Объединяем все словари замен в один
FULL_MAP = {}
FULL_MAP: Dict[str, str] = {}
FULL_MAP.update(LATIN_TO_CYRILLIC)
FULL_MAP.update(CYRILLIC_NORMALIZE)
FULL_MAP.update(UNICODE_MAP)
# Символы-разделители, которые могут быть вставлены между буквами
SEPARATORS = re.compile(r'[\s.\-_,;:|]+', re.UNICODE)
# Паттерн для поиска повторяющихся букв (3+ раза)
REPEAT_PATTERN = re.compile(r'([а-яёa-z])\1{2,}', re.IGNORECASE)
@classmethod
def normalize_characters(cls, text: str) -> str:
"""
Заменяет все символы из FULL_MAP на их базовые эквиваленты.
Проходит по строке посимвольно для максимальной замены.
"""
result = []
result: list[str] = []
for ch in text:
# Сначала пробуем заменить по карте
if ch in cls.FULL_MAP:
result.append(cls.FULL_MAP[ch])
else:
result.append(ch)
# Приводим к нижнему регистру после замен (чтобы избежать потери регистра в карте)
result.append(cls.FULL_MAP.get(ch, ch))
return ''.join(result).lower()
@classmethod
def remove_separators(cls, text: str) -> str:
"""Удаляет разделители между буквами (пробелы, точки и т.д.)"""
return cls.SEPARATORS.sub('', text)
@classmethod
def collapse_repeats(cls, text: str, max_repeat: int = 2) -> str:
def repl(m):
ch = m.group(1)
return ch # вместо ch * 2 — теперь схлопываем до одного символа
def collapse_repeats(cls, text: str) -> str:
def repl(match: re.Match[str]) -> str:
return match.group(1)
return cls.REPEAT_PATTERN.sub(repl, text)
@classmethod
def normalize_full(cls, text: str, remove_sep: bool = True, collapse: bool = True) -> str:
"""
Полная нормализация:
1. Unicode нормализация (NFKC) для разложения составных символов
2. Замена по карте
3. Приведение к нижнему регистру
4. Удаление разделителей (опционально)
5. Схлопывание повторов (опционально)
"""
# NFKC разлагает символы типа "ё" в "е" + умляут, но нам лучше оставить как есть,
# т.к. у нас есть прямые замены. Однако для совместимости применим.
def normalize_full(
cls,
text: str,
remove_sep: bool = True,
collapse: bool = True
) -> str:
text = unicodedata.normalize('NFKC', text)
# Замена символов
text = cls.normalize_characters(text)
# Удаление разделителей
if remove_sep:
text = cls.remove_separators(text)
# Схлопывание повторов
if collapse:
text = cls.collapse_repeats(text)
return text
@classmethod
def normalize_for_part(cls, text: str) -> str:
def normalize_for_part_token(cls, text: str) -> str:
"""
Нормализация для типа PART:
- Полная нормализация
- Удаление всех не-буквенных символов (кроме пробелов)
- Приведение к нижнему регистру
Нормализация для PART:
- NFKC
- lower()
- удаление zero-width
- схлопывание повторов латиницы
- БЕЗ LATIN_TO_CYRILLIC
"""
text = cls.normalize_full(text, remove_sep=False, collapse=True)
# Оставляем только буквы и пробелы
text = re.sub(r'[^а-яёa-z\s]', '', text, flags=re.IGNORECASE)
text = re.sub(r'\s+', ' ', text).strip()
return text.lower()
text = unicodedata.normalize('NFKC', text)
text = text.lower()
# удаляем zero-width
text = re.sub(r'[\u200B-\u200D\uFEFF]', '', text)
# схлопываем повторы букв (3+ → 1)
text = re.sub(r'([a-z])\1+', r'\1', text)
return text
class BanWordsMiddleware(BaseMiddleware):
def __init__(self):
def __init__(self) -> None:
super().__init__()
self.manager = get_manager()
self.normalizer = TextNormalizer()
@@ -124,219 +105,178 @@ class BanWordsMiddleware(BaseMiddleware):
event: Message,
data: Dict[str, Any]
) -> Any:
# Проверяем наличие текста или подписи
if not event.text and not event.caption:
return await handler(event, data)
message_text = event.text or event.caption
message_text: str = event.text or event.caption
# Игнорируем команды
if message_text.startswith('/'):
return await handler(event, data)
# Проверка на админа
user_id = event.from_user.id
is_super_admin = user_id in settings.OWNER_ID
is_admin = is_super_admin or self.manager.is_admin_cached(user_id)
user_id: int = event.from_user.id
is_super_admin: bool = user_id in settings.OWNER_ID
is_admin: bool = is_super_admin or self.manager.is_admin_cached(user_id)
if is_admin:
return await handler(event, data)
# Проверяем сообщение на спам
spam_result = await self._check_message(message_text)
if spam_result:
await self._handle_spam(event, spam_result)
return None # Сообщение удалено, дальше не обрабатываем
await self._handle_spam(event)
return None
return await handler(event, data)
async def _check_message(self, text: str) -> Optional[Dict[str, str]]:
"""
Многоступенчатая проверка текста.
Возвращает словарь с причиной блокировки или None.
"""
# 1. Повторяющиеся символы (например, "леееейн") — блокируем сразу
# repeat_result = self._check_repeated_chars(text)
# if repeat_result:
# return repeat_result
@staticmethod
def is_allowed_url(url: str, allowed: str) -> bool:
url_lower = url.lower()
allowed_lower = allowed.lower()
if allowed_lower.endswith('/'):
# исключение со слешем: только строгое начало с этим слешем
return url_lower.startswith(allowed_lower)
else:
# исключение без слеша: разрешаем точное совпадение или начало с добавлением слеша
return url_lower == allowed_lower or url_lower.startswith(allowed_lower + '/')
# 2. Получаем кэшированные списки
substring_words = self.manager.get_banwords_cached(BanWordType.SUBSTRING)
async def _check_message(self, text: str) -> Optional[Dict[str, str]]:
whitelist = {
w.lower().strip()
for w in self.manager.get_whitelist_cached()
}
# ================= URL CHECK =================
urls = URL_PATTERN.findall(text)
for url in urls:
url_lower = url.lower()
# если URL начинается с разрешённого исключения — пропускаем
if any(self.is_allowed_url(url_lower, allowed) for allowed in whitelist):
continue
# если нет разрешения — проверяем WORD-правила для URL
for word in self.manager.get_banwords_cached(BanWordType.WORD):
if word in url_lower:
return {"word": word, "type": "word"}
# =============================================
# 2. Убираем URL из текста для word/lemma проверки
text_without_urls = URL_PATTERN.sub(' ', text)
word_words = self.manager.get_banwords_cached(BanWordType.WORD)
lemma_words = self.manager.get_banwords_cached(BanWordType.LEMMA)
part_words = self.manager.get_banwords_cached(BanWordType.PART)
conflict_substring = self.manager.get_banwords_cached(BanWordType.CONFLICT_SUBSTRING)
conflict_word = self.manager.get_banwords_cached(BanWordType.CONFLICT_WORD)
conflict_lemma = self.manager.get_banwords_cached(BanWordType.CONFLICT_LEMMA)
conflict_part = self.manager.get_banwords_cached(BanWordType.CONFLICT_PART)
# 3. Белый список
if self.manager.is_whitelisted(text):
logger.debug(f"⏭️ Пропуск по белому списку: {text[:30]}", log_type="BANWORDS")
return None
# 4. Режим тишины
if await self.manager.is_silence_active():
return {"word": "[режим тишины]", "type": "silence"}
# 5. Режим конфликта (более мягкие правила)
if await self.manager.is_conflict_active():
# Проверка conflict_substring (с нормализацией)
normalized_text = self.normalizer.normalize_full(text, remove_sep=True, collapse=True)
for word in conflict_substring:
norm_word = self.normalizer.normalize_full(word, remove_sep=True, collapse=True)
if norm_word in normalized_text:
return {"word": word, "type": "conflict_substring"}
normalized_text = self.normalizer.normalize_full(text)
for word in conflict_word:
if self.normalizer.normalize_full(word) in normalized_text:
return {"word": word, "type": "conflict_word"}
# conflict_lemma
for word_text in extract_words(text):
lemma = get_lemma(word_text)
if lemma in conflict_lemma:
return {"word": lemma, "type": "conflict_lemma"}
if get_lemma(word_text) in conflict_lemma:
return {"word": word_text, "type": "conflict_lemma"}
# Если в конфликтном режиме ничего не найдено — пропускаем
return None
# 6. Обычный режим: проверка substring (с удалением разделителей и схлопыванием повторов)
normalized_text = self.normalizer.normalize_full(text, remove_sep=True, collapse=True)
for word in substring_words:
norm_word = self.normalizer.normalize_full(word, remove_sep=True, collapse=True)
if norm_word in normalized_text:
logger.info(f"✅ SUBSTRING: '{word}'", log_type="BANWORDS")
return {"word": word, "type": "substring"}
# WORD — строгое совпадение как отдельное слово
for word in word_words:
pattern = r'(?<!\w){}(?!\w)'.format(re.escape(word))
# 7. Проверка part (строгая нормализация, только буквы и пробелы)
part_normalized = self.normalizer.normalize_for_part(text)
for part in part_words:
norm_part = self.normalizer.normalize_for_part(part)
if norm_part in part_normalized:
logger.info(f"✅ PART: '{part}'", log_type="BANWORDS")
return {"word": part, "type": "part"}
for match in re.finditer(pattern, text_without_urls, re.IGNORECASE):
matched = match.group(0).lower()
# 8. Проверка lemma
for word_text in extract_words(text):
# Для леммы тоже применяем нормализацию (удаляем разделители, схлопываем повторы)
normalized_word = self.normalizer.normalize_full(word_text, remove_sep=True, collapse=True)
# если совпавшее слово в whitelist — игнорируем
if matched in whitelist:
continue
# если это начало URL — пропускаем
if text[match.end():match.end() + 3] == '://':
continue
return {"word": word, "type": "word"}
# PART
usernames = re.findall(r'@[\w_]+', text_without_urls)
latin_tokens = re.findall(r'\b[a-zA-Z0-9_]*[a-zA-Z]+[a-zA-Z0-9_]*\b', text_without_urls)
tokens_to_check = usernames + latin_tokens
# PART
for token in tokens_to_check:
token_lower = token.lower()
# если именно этот токен разрешён
normalized_for_whitelist = token_lower.lstrip('@')
if (
token_lower in whitelist or
normalized_for_whitelist in whitelist or
f"@{normalized_for_whitelist}" in whitelist
):
continue
normalized_token = self.normalizer.normalize_for_part_token(token)
for part in part_words:
norm_part = self.normalizer.normalize_for_part_token(part)
if norm_part in normalized_token:
return {"word": part, "type": "part"}
# CONFLICT PART
for token in tokens_to_check:
token_lower = token.lower()
normalized_for_whitelist = token_lower.lstrip('@')
if (
token_lower in whitelist or
normalized_for_whitelist in whitelist or
f"@{normalized_for_whitelist}" in whitelist
):
continue
normalized_token = self.normalizer.normalize_for_part_token(token)
for part in conflict_part:
norm_part = self.normalizer.normalize_for_part_token(part)
if norm_part in normalized_token:
return {"word": part, "type": "conflict_part"}
# LEMMA
for word_text in extract_words(text_without_urls):
word_lower = word_text.lower()
# если слово разрешено — пропускаем
if word_lower in whitelist:
continue
normalized_word = self.normalizer.normalize_full(word_text)
lemma = get_lemma(normalized_word)
if lemma in lemma_words:
logger.info(f"✅ LEMMA: '{lemma}' из '{word_text}'", log_type="BANWORDS")
return {"word": lemma, "type": "lemma"}
return None
def _check_repeated_chars(self, text: str) -> Optional[Dict[str, str]]:
"""
Проверяет на наличие 3+ повторяющихся букв подряд.
Использует сырой текст без нормализации (чтобы поймать "леееейн").
"""
# Ищем повторения букв (только кириллица/латиница)
pattern = re.compile(r'([а-яёa-zA-Z])\1{2,}', re.IGNORECASE)
matches = pattern.finditer(text)
for match in matches:
char = match.group(1)
count = len(match.group(0))
if count >= 3:
logger.info(f"🔥 ПОВТОРЫ: '{match.group(0)}' ({count}x)", log_type="BANWORDS")
return {"word": f"'{match.group(0)}' ({count}x)", "type": "repeated_chars"}
return None
@staticmethod
async def _handle_spam(
message: Message,
) -> None:
async def _handle_spam(self, message: Message, spam_result: Dict[str, str]) -> None:
"""Обрабатывает спам-сообщение: удаляет, логирует, уведомляет (кроме silence)"""
user = message.from_user
matched_word = spam_result["word"]
match_type = spam_result["type"]
message_text = message.text or message.caption or "[нет текста]"
# В режиме тишины удаляем молча
if match_type == "silence":
try:
await message.delete()
logger.info(f"🔇 SILENCE: @{user.username or user.id} удалено молча", log_type="BANWORDS")
except TelegramBadRequest as e:
logger.error(f"Не удалено (silence): {e}", log_type="BANWORDS")
return
# Удаляем сообщение
try:
await message.delete()
logger.info(f"🚫 @{user.username or user.id}: '{matched_word}' ({match_type})", log_type="BANWORDS")
except TelegramBadRequest as e:
logger.error(f"Не удалено: {e}", log_type="BANWORDS")
logger.info(f"Удалено сообщение: {message.text}")
except TelegramBadRequest:
return
# Логируем в БД
await self.manager.log_spam(
user_id=user.id,
username=user.username or f"id{user.id}",
chat_id=message.chat.id,
message_text=message_text,
matched_word=matched_word,
match_type=match_type
)
# Уведомляем админов
await self._notify_admins(message, matched_word, match_type, message_text)
async def _notify_admins(
self,
message: Message,
matched_word: str,
match_type: str,
message_text: str
) -> None:
"""Отправляет уведомление об удалении в админ-чат (берёт ID из БД)"""
user = message.from_user
username = f"@{user.username}" if user.username else f"ID: {user.id}"
spam_count = await self.manager.get_user_spam_count(user.id)
chat_title = message.chat.title or "Без названия"
source_thread_id = message.message_thread_id
notification_text = (
f"🚫 <b>Удалено сообщение</b>\n\n"
f"👤 <b>Пользователь:</b> {username}\n"
f"🆔 <b>ID:</b> <code>{user.id}</code>\n"
f"📊 <b>Нарушений:</b> {spam_count}\n\n"
f"💬 <b>Чат:</b> {self._escape_html(chat_title)}\n"
f"🆔 <b>Chat ID:</b> <code>{message.chat.id}</code>\n"
f"{'📌 <b>Topic ID:</b> <code>' + str(source_thread_id) + '</code>\n' if source_thread_id else ''}"
f"🔗 <b>Message ID:</b> <code>{message.message_id}</code>\n\n"
f"🔍 <b>Триггер:</b> <code>{self._escape_html(matched_word)}</code>\n"
f"📝 <b>Тип:</b> {self._get_type_emoji(match_type)} {self._escape_html(match_type)}\n\n"
f"💬 <b>Текст:</b>\n<code>{self._escape_html(message_text[:500])}</code>"
)
keyboard = InlineKeyboardMarkup(inline_keyboard=[
[
InlineKeyboardButton(text="🔨 Забанить", callback_data=f"spam_ban:{user.id}:{message.chat.id}"),
InlineKeyboardButton(text="✅ Закрыть", callback_data="spam_close")
],
[InlineKeyboardButton(text="📊 Статистика", callback_data=f"spam_stats:{user.id}")]
])
try:
# ✅ Получаем настройки из БД (динамические, установленные через /settings)
admin_chat_id = await self.manager.get_bot_setting("admin_chat_id")
admin_thread_id = await self.manager.get_bot_setting("admin_thread_id")
if admin_chat_id:
await message.bot.send_message(
chat_id=int(admin_chat_id),
text=notification_text,
reply_markup=keyboard,
parse_mode="HTML",
message_thread_id=int(admin_thread_id) if admin_thread_id else None
)
except Exception as e:
logger.error(f"❌ Уведомление админам: {e}", log_type="BANWORDS")
@staticmethod
def _get_type_emoji(match_type: str) -> str:
return {
"substring": "🔤",
"lemma": "📖",
"part": "🧩",
"silence": "🔇",
"conflict_substring": "⚔️",
"conflict_lemma": "⚔️",
"repeated_chars": "🔁"
}.get(match_type, "")
@staticmethod
def _escape_html(text: str) -> str:
return str(text).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")

View File

@@ -105,7 +105,7 @@ class UserSpamStats:
self.total_blocks += 1
self.reputation = max(0.5, self.reputation - 0.3)
def detect_spam_patterns(self, time_window: float = 10.0) -> Dict[str, Any]:
def detect_spam_patterns(self, time_window: float = 2.0) -> Dict[str, Any]:
"""
Умная детекция спама на основе паттернов.
УЛУЧШЕНО: учитывает скорость отправки сообщений.
@@ -120,7 +120,7 @@ class UserSpamStats:
current_time = time()
# 1. КРИТИЧНО: Экстремально быстрая отправка (флуд-бот)
very_recent = [ctx for ctx in recent_contexts if (current_time - ctx.timestamp) < 2.0]
very_recent = [ctx for ctx in recent_contexts if (current_time - ctx.timestamp) < time_window]
if len(very_recent) >= 5:
return {
'is_spam': True,
@@ -133,7 +133,7 @@ class UserSpamStats:
# 2. КРИТИЧНО: 8+ сообщений за 5 секунд => агрессивный флуд
recent_5s = [ctx for ctx in recent_contexts if (current_time - ctx.timestamp) < 5.0]
if len(recent_5s) >= 8:
if len(recent_5s) >= 15:
return {
'is_spam': True,
'reason': 'aggressive_flood',
@@ -145,7 +145,7 @@ class UserSpamStats:
# 3. Медиа-флуд
media_contexts = [ctx for ctx in recent_contexts if ctx.media_type]
if len(media_contexts) >= 7:
if len(media_contexts) >= 15:
media_recent = [ctx for ctx in media_contexts if (current_time - ctx.timestamp) < 5.0]
if len(media_recent) >= 6:
return {
@@ -303,7 +303,8 @@ class AntiSpamMiddleware(BaseMiddleware):
self.enable_reputation = enable_reputation
self.log_all = log_all
def _extract_context(self, event: TelegramObject) -> MessageContext:
@staticmethod
def _extract_context(event: TelegramObject) -> MessageContext:
"""Извлекает контекст из события"""
context = MessageContext()

View File

@@ -246,23 +246,40 @@ async def msg(
keyboard = markups(markup)
try:
# Попытка редактирования (для callback)
if edit_if_possible and isinstance(update, CallbackQuery):
sent_message = await message.edit_text(
try:
sent_message = await message.edit_text(
text=text,
reply_markup=keyboard,
parse_mode=parse_mode,
disable_web_page_preview=disable_web_page_preview
)
if log:
logger.debug(
f"Сообщение отредактировано: {message.message_id}",
log_type='MESSAGE'
)
except (TelegramBadRequest, TelegramForbiddenError):
sent_message = await message.answer(
text=text,
reply_markup=keyboard,
parse_mode=parse_mode,
disable_web_page_preview=disable_web_page_preview,
disable_notification=disable_notification,
protect_content=protect_content
)
else:
sent_message = await message.answer(
text=text,
reply_markup=keyboard,
parse_mode=parse_mode,
disable_web_page_preview=disable_web_page_preview
disable_web_page_preview=disable_web_page_preview,
disable_notification=disable_notification,
protect_content=protect_content
)
if log:
logger.debug(
f"Сообщение отредактировано: {message.message_id}",
log_type='MESSAGE'
)
else:
raise TelegramBadRequest
except (TelegramBadRequest, TelegramForbiddenError):
# Отправка нового сообщения
try:

688
bot/utils/argument.py Normal file
View File

@@ -0,0 +1,688 @@
"""
Утилиты для работы с командами бота
"""
from typing import Optional, Union, Dict, List, Tuple, Set
from dataclasses import dataclass, field
import re
from aiogram.types import Message
from configs import settings
__all__ = (
'is_command',
'find_argument',
'get_command',
'parse_arguments',
'parse_flags',
'CommandParser',
'ParsedCommand',
'parse_command',
'validate_command',
'get_command_usage',
'extract_mentions',
'extract_user_ids',
'extract_hashtags'
)
@dataclass
class ParsedCommand:
"""
Распарсенная команда.
Attributes:
command: Название команды
prefix: Префикс команды
args: Список аргументов
raw_args: Исходная строка аргументов
flags: Словарь флагов (--flag value)
bot_username: Username бота (если было упоминание)
is_group_command: Команда в группе с упоминанием бота
"""
command: str
prefix: str
args: List[str] = field(default_factory=list)
raw_args: Optional[str] = None
flags: Dict[str, Union[str, bool]] = field(default_factory=dict)
bot_username: Optional[str] = None
is_group_command: bool = False
@property
def has_args(self) -> bool:
"""Есть ли аргументы"""
return len(self.args) > 0
@property
def has_flags(self) -> bool:
"""Есть ли флаги"""
return len(self.flags) > 0
def get_arg(self, index: int, default: Optional[str] = None) -> Optional[str]:
"""Получает аргумент по индексу"""
return self.args[index] if index < len(self.args) else default
def get_flag(self, name: str, default: Optional[Union[str, bool]] = None) -> Union[str, bool, None]:
"""Получает значение флага"""
return self.flags.get(name, default)
def __repr__(self) -> str:
return (
f"ParsedCommand(command='{self.command}', "
f"args={self.args}, flags={self.flags})"
)
class CommandParser:
"""
Парсер команд бота.
Возможности:
- Поддержка нескольких префиксов
- Парсинг аргументов
- Парсинг флагов (--flag value, -f value)
- Поддержка упоминаний бота (@botname)
- Парсинг quoted аргументов ("arg with spaces")
- Валидация команд
Example:
```python
parser = CommandParser()
# Парсинг команды
parsed = parser.parse("/ban @user 7d --reason спам")
print(parsed.command) # "ban"
print(parsed.args) # ["@user", "7d"]
print(parsed.flags) # {"reason": "спам"}
```
"""
def __init__(
self,
prefixes: Optional[List[str]] = None,
bot_username: Optional[str] = None
):
"""
Args:
prefixes: Список префиксов (по умолчанию из settings)
bot_username: Username бота для проверки упоминаний
"""
self.prefixes = prefixes or settings.PREFIX
self.bot_username = bot_username
def is_command(self, text: Optional[str]) -> bool:
"""
Проверяет, является ли текст командой.
Args:
text: Текст для проверки
Returns:
bool: True если это команда
Example:
>> parser.is_command("/start")
True
>> parser.is_command("hello")
False
"""
if not text:
return False
text = text.strip()
# Проверяем все префиксы
return any(text.startswith(prefix) for prefix in self.prefixes)
def get_command(
self,
text: Optional[str],
strip_mention: bool = True
) -> Optional[str]:
"""
Извлекает название команды из текста.
Args:
text: Текст сообщения
strip_mention: Убирать упоминание бота (@botname)
Returns:
Optional[str]: Название команды или None
Example:
>> parser.get_command("/start@mybot arg")
'start'
>> parser.get_command("!help")
'help'
"""
if not self.is_command(text):
return None
text = text.strip()
# Находим префикс
prefix = next(p for p in self.prefixes if text.startswith(p))
# Убираем префикс
without_prefix = text[len(prefix):]
# Берем первое слово
command = without_prefix.split()[0] if without_prefix else ""
# Убираем упоминание бота если есть
if strip_mention and '@' in command:
command = command.split('@')[0]
return command.lower() if command else None
def find_argument(self, text: Optional[str]) -> Optional[str]:
"""
Извлекает аргументы команды (все после команды).
Args:
text: Текст сообщения
Returns:
Optional[str]: Аргументы или None
Example:
>> parser.find_argument("/start referrer")
'referrer'
>> parser.find_argument("/ban @user reason text")
'@user reason text'
"""
if not self.is_command(text):
return None
parts = text.strip().split(maxsplit=1)
return parts[1] if len(parts) > 1 else None
@staticmethod
def parse_arguments(
args_text: Optional[str],
preserve_quotes: bool = False
) -> List[str]:
"""
Парсит аргументы, поддерживает кавычки.
Args:
args_text: Строка аргументов
preserve_quotes: Сохранять кавычки в результате
Returns:
List[str]: Список аргументов
Example:
>> parser.parse_arguments('user 7d "ban reason here"')
['user', '7d', 'ban reason here']
"""
if not args_text:
return []
# Regex для парсинга с кавычками
# Поддерживает: "arg with spaces" 'arg' arg
pattern = r'''(?:[^\s"']+|"[^"]*"|'[^']*')+'''
matches = re.findall(pattern, args_text)
if preserve_quotes:
return matches
# Убираем кавычки
return [m.strip('"').strip("'") for m in matches]
@staticmethod
def parse_flags(
args: List[str]
) -> Tuple[List[str], Dict[str, Union[str, bool]]]:
"""
Парсит флаги из аргументов.
Поддерживает:
- --flag value
- --flag (boolean, True)
- -f value (короткая форма)
Args:
args: Список аргументов
Returns:
Tuple: (аргументы_без_флагов, словарь_флагов)
Example:
>> args = ['user', '--reason', 'spam', '--silent']
>> clean_args, flags = parser.parse_flags(args)
>> print(clean_args) # ['user']
>> print(flags) # {'reason': 'spam', 'silent': True}
"""
clean_args = []
flags = {}
i = 0
while i < len(args):
arg = args[i]
# Длинный флаг --flag
if arg.startswith('--'):
flag_name = arg[2:]
# Проверяем, есть ли значение
if i + 1 < len(args) and not args[i + 1].startswith('-'):
flags[flag_name] = args[i + 1]
i += 2
else:
# Boolean флаг
flags[flag_name] = True
i += 1
# Короткий флаг -f
elif arg.startswith('-') and len(arg) == 2:
flag_name = arg[1]
# Проверяем значение
if i + 1 < len(args) and not args[i + 1].startswith('-'):
flags[flag_name] = args[i + 1]
i += 2
else:
flags[flag_name] = True
i += 1
# Обычный аргумент
else:
clean_args.append(arg)
i += 1
return clean_args, flags
def parse(
self,
text: str,
parse_flags: bool = True
) -> Optional[ParsedCommand]:
"""
Полный парсинг команды.
Args:
text: Текст команды
parse_flags: Парсить флаги
Returns:
Optional[ParsedCommand]: Распарсенная команда или None
Example:
>> parsed = parser.parse('/ban @user 7d --reason "spam bot"')
>> print(parsed.command) # 'ban'
>> print(parsed.args) # ['@user', '7d']
>> print(parsed.flags) # {'reason': 'spam bot'}
"""
if not self.is_command(text):
return None
text = text.strip()
# Находим префикс
prefix = next(p for p in self.prefixes if text.startswith(p))
# Убираем префикс
without_prefix = text[len(prefix):]
# Разделяем на команду и аргументы
parts = without_prefix.split(maxsplit=1)
if not parts:
return None
command_part = parts[0]
raw_args = parts[1] if len(parts) > 1 else None
# Проверяем упоминание бота
bot_username = None
is_group_command = False
if '@' in command_part:
cmd_parts = command_part.split('@')
command_name = cmd_parts[0]
bot_username = cmd_parts[1] if len(cmd_parts) > 1 else None
is_group_command = True
else:
command_name = command_part
# Парсим аргументы
args = self.parse_arguments(raw_args) if raw_args else []
# Парсим флаги
flags = {}
if parse_flags and args:
args, flags = self.parse_flags(args)
return ParsedCommand(
command=command_name.lower(),
prefix=prefix,
args=args,
raw_args=raw_args,
flags=flags,
bot_username=bot_username,
is_group_command=is_group_command
)
def parse_from_message(
self,
message: Message,
parse_flags: bool = True
) -> Optional[ParsedCommand]:
"""
Парсит команду из объекта Message.
Args:
message: Объект сообщения
parse_flags: Парсить флаги
Returns:
Optional[ParsedCommand]: Распарсенная команда
Example:
>> parsed = parser.parse_from_message(message)
>> if parsed:
... print(f"Команда: {parsed.command}")
"""
if not message.text:
return None
return self.parse(message.text, parse_flags=parse_flags)
# Глобальный парсер
_default_parser: Optional[CommandParser] = None
def get_parser() -> CommandParser:
"""Получает глобальный парсер команд"""
global _default_parser
if _default_parser is None:
_default_parser = CommandParser()
return _default_parser
# ================= УДОБНЫЕ ФУНКЦИИ =================
def is_command(text: Optional[str]) -> bool:
"""
Проверяет, является ли текст командой.
Args:
text: Текст для проверки
Returns:
bool: True если это команда
Example:
>> is_command("/start")
True
>> is_command("hello")
False
"""
return get_parser().is_command(text)
def find_argument(text: Optional[str]) -> Optional[str]:
"""
Извлекает аргументы команды.
Args:
text: Текст команды
Returns:
Optional[str]: Аргументы или None
Example:
>> find_argument("/start referrer")
'referrer'
>> find_argument("/ban @user spam")
'@user spam'
"""
return get_parser().find_argument(text)
def get_command(text: Optional[str]) -> Optional[str]:
"""
Извлекает название команды.
Args:
text: Текст сообщения
Returns:
Optional[str]: Название команды или None
Example:
>> get_command("/start@mybot")
'start'
>> get_command("!help")
'help'
"""
return get_parser().get_command(text)
def parse_arguments(args_text: Optional[str]) -> List[str]:
"""
Парсит аргументы команды.
Args:
args_text: Строка аргументов
Returns:
List[str]: Список аргументов
Example:
>> parse_arguments('user 7d "ban reason"')
['user', '7d', 'ban reason']
"""
return get_parser().parse_arguments(args_text)
def parse_flags(args: List[str]) -> Tuple[List[str], Dict[str, Union[str, bool]]]:
"""
Парсит флаги из аргументов.
Args:
args: Список аргументов
Returns:
Tuple: (аргументы, флаги)
Example:
>> args = ['user', '--reason', 'spam', '--silent']
>> clean_args, flags = parse_flags(args)
>> print(flags) # {'reason': 'spam', 'silent': True}
"""
return get_parser().parse_flags(args)
def parse_command(text: str) -> Optional[ParsedCommand]:
"""
Полный парсинг команды.
Args:
text: Текст команды
Returns:
Optional[ParsedCommand]: Распарсенная команда
Example:
>> parsed = parse_command('/ban @user --reason spam')
>> print(parsed.command) # 'ban'
>> print(parsed.args) # ['@user']
>> print(parsed.flags) # {'reason': 'spam'}
"""
return get_parser().parse(text)
# ================= ВАЛИДАЦИЯ КОМАНД =================
def validate_command(
text: str,
expected_command: str,
min_args: int = 0,
max_args: Optional[int] = None,
required_flags: Optional[Set[str]] = None
) -> Tuple[bool, Optional[str]]:
"""
Валидирует команду.
Args:
text: Текст команды
expected_command: Ожидаемая команда
min_args: Минимальное количество аргументов
max_args: Максимальное количество аргументов
required_flags: Обязательные флаги
Returns:
Tuple[bool, Optional[str]]: (валидна, сообщение_об_ошибке)
Example:
>> valid, error = validate_command(
... '/ban user',
... 'ban',
... min_args=1,
... max_args=2
... )
>> if not valid:
... print(error)
"""
parsed = parse_command(text)
if not parsed:
return False, "Невалидная команда"
# Проверка команды
if parsed.command != expected_command:
return False, f"Ожидалась команда '{expected_command}'"
# Проверка количества аргументов
arg_count = len(parsed.args)
if arg_count < min_args:
return False, f"Недостаточно аргументов (минимум {min_args})"
if max_args is not None and arg_count > max_args:
return False, f"Слишком много аргументов (максимум {max_args})"
# Проверка обязательных флагов
if required_flags:
missing_flags = required_flags - set(parsed.flags.keys())
if missing_flags:
return False, f"Отсутствуют обязательные флаги: {', '.join(missing_flags)}"
return True, None
def get_command_usage(
command: str,
args: List[str],
flags: Optional[Dict[str, str]] = None,
description: Optional[str] = None
) -> str:
"""
Формирует строку использования команды.
Args:
command: Название команды
args: Список аргументов
flags: Словарь флагов с описанием
description: Описание команды
Returns:
str: Форматированная строка использования
Example:
>> usage = get_command_usage(
... 'ban',
... ['<user>', '[duration]'],
... {'reason': 'Причина бана', 'silent': 'Тихий бан'},
... 'Банит пользователя'
... )
>> print(usage)
"""
lines = []
# Описание
if description:
lines.append(f"📝 {description}\n")
# Использование
args_str = ' '.join(args)
lines.append(f"<b>Использование:</b>")
lines.append(f"<code>/{command} {args_str}</code>\n")
# Аргументы
if args:
lines.append("<b>Аргументы:</b>")
for arg in args:
# Определяем обязательность
if arg.startswith('<') and arg.endswith('>'):
lines.append(f"{arg} - обязательный")
elif arg.startswith('[') and arg.endswith(']'):
lines.append(f"{arg} - необязательный")
lines.append("")
# Флаги
if flags:
lines.append("<b>Флаги:</b>")
for flag, desc in flags.items():
lines.append(f"• --{flag} - {desc}")
return '\n'.join(lines)
# ================= ИЗВЛЕЧЕНИЕ УПОМИНАНИЙ =================
def extract_mentions(text: str) -> List[str]:
"""
Извлекает все упоминания (@username) из текста.
Args:
text: Текст для анализа
Returns:
List[str]: Список username (без @)
Example:
>> extract_mentions("Бан @user1 и @user2")
['user1', 'user2']
"""
pattern = r'@(\w+)'
return re.findall(pattern, text)
def extract_user_ids(text: str) -> List[int]:
"""
Извлекает все ID пользователей из текста.
Args:
text: Текст для анализа
Returns:
List[int]: Список ID
Example:
>> extract_user_ids("Бан id123456789 и id987654321")
[123456789, 987654321]
"""
pattern = r'id(\d+)'
matches = re.findall(pattern, text)
return [int(m) for m in matches]
def extract_hashtags(text: str) -> List[str]:
"""
Извлекает все хештеги из текста.
Args:
text: Текст для анализа
Returns:
List[str]: Список хештегов (без #)
Example:
>> extract_hashtags("Пост #важное #новости")
['важное', 'новости']
"""
pattern = r'#(\w+)'
return re.findall(pattern, text)

636
bot/utils/auto_delete.py Normal file
View File

@@ -0,0 +1,636 @@
"""
Утилиты для автоматического удаления сообщений
"""
from typing import Optional, Callable, Awaitable, Dict, Any
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from asyncio import sleep, create_task, Task, CancelledError
from aiogram import Bot
from aiogram.types import Message
from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError
from middleware.loggers import logger
from .format_time import format_duration
__all__ = (
'auto_delete_message',
'schedule_delete',
'cancel_delete',
'delete_after',
'auto_delete_manager',
'AutoDeleteManager',
'DeleteTask',
'delete_both_after',
'delete_messages_after',
)
@dataclass
class DeleteTask:
"""
Задача на удаление сообщения.
Attributes:
chat_id: ID чата
message_id: ID сообщения
delete_at: Время удаления
task: Asyncio task
created_at: Время создания задачи
reason: Причина удаления
callback: Callback функция после удаления
"""
chat_id: int
message_id: int
delete_at: datetime
task: Optional[Task] = None
created_at: datetime = field(default_factory=datetime.now)
reason: Optional[str] = None
callback: Optional[Callable[[], Awaitable[None]]] = None
@property
def delay(self) -> int:
"""Задержка до удаления в секундах"""
delta = self.delete_at - datetime.now()
return max(0, int(delta.total_seconds()))
@property
def is_expired(self) -> bool:
"""Истекло ли время удаления"""
return datetime.now() >= self.delete_at
def __repr__(self) -> str:
return (
f"DeleteTask(chat={self.chat_id}, msg={self.message_id}, "
f"delay={self.delay}s, reason={self.reason})"
)
class AutoDeleteManager:
"""
Менеджер автоматического удаления сообщений.
Возможности:
- Планирование удаления с задержкой
- Отмена запланированного удаления
- Массовое удаление
- Callback функции
- История задач
- Автоматическая очистка завершенных задач
Example:
```python
from utils.auto_delete import auto_delete_manager
# Планирование удаления
await auto_delete_manager.schedule(
bot=bot,
chat_id=message.chat.id,
message_id=message.message_id,
delay=60,
reason="Временное сообщение"
)
# Отмена удаления
auto_delete_manager.cancel(message.chat.id, message.message_id)
# Получение статистики
stats = auto_delete_manager.get_stats()
```
"""
def __init__(self):
# Активные задачи: {(chat_id, message_id): DeleteTask}
self.tasks: Dict[tuple[int, int], DeleteTask] = {}
# Завершенные задачи (последние 100)
self.completed: list[DeleteTask] = []
self.max_completed = 100
# Статистика
self.total_scheduled: int = 0
self.total_deleted: int = 0
self.total_failed: int = 0
self.total_cancelled: int = 0
async def schedule(
self,
bot: Bot,
chat_id: int,
message_id: int,
delay: int,
reason: Optional[str] = None,
callback: Optional[Callable[[], Awaitable[None]]] = None,
log: bool = True
) -> DeleteTask:
"""
Планирует удаление сообщения.
Args:
bot: Экземпляр бота
chat_id: ID чата
message_id: ID сообщения
delay: Задержка в секундах
reason: Причина удаления
callback: Callback функция после удаления
log: Логировать планирование
Returns:
DeleteTask: Созданная задача
Example:
>> task = await auto_delete_manager.schedule(
... bot=bot,
... chat_id=message.chat.id,
... message_id=message.message_id,
... delay=60,
... reason="Спам"
... )
"""
# Отменяем предыдущую задачу если есть
key = (chat_id, message_id)
if key in self.tasks:
self.cancel(chat_id, message_id)
# Создаем задачу
delete_at = datetime.now() + timedelta(seconds=delay)
task_data = DeleteTask(
chat_id=chat_id,
message_id=message_id,
delete_at=delete_at,
reason=reason,
callback=callback
)
# Создаем asyncio task
task = create_task(self._delete_task(bot, task_data, log))
task_data.task = task
# Сохраняем
self.tasks[key] = task_data
self.total_scheduled += 1
if log:
delay_str = format_duration(delay)
logger.info(
f"Запланировано удаление сообщения через {delay_str}",
log_type='AUTO_DELETE'
)
return task_data
async def _delete_task(
self,
bot: Bot,
task_data: DeleteTask,
log: bool
) -> None:
"""
Внутренняя функция для выполнения задачи удаления.
Args:
bot: Экземпляр бота
task_data: Данные задачи
log: Логировать выполнение
"""
key = (task_data.chat_id, task_data.message_id)
try:
# Ждем
await sleep(task_data.delay)
# Удаляем сообщение
await bot.delete_message(
chat_id=task_data.chat_id,
message_id=task_data.message_id
)
self.total_deleted += 1
if log:
reason_str = f" (причина: {task_data.reason})" if task_data.reason else ""
logger.info(
f"Сообщение удалено автоматически{reason_str}",
log_type='AUTO_DELETE'
)
# Вызываем callback если есть
if task_data.callback:
try:
await task_data.callback()
except Exception as e:
logger.error(
f"Ошибка в callback автоудаления: {e}",
log_type='AUTO_DELETE'
)
except CancelledError:
# Задача отменена
self.total_cancelled += 1
if log:
logger.debug(
f"Удаление сообщения отменено",
log_type='AUTO_DELETE'
)
raise
except (TelegramBadRequest, TelegramForbiddenError) as e:
# Ошибка удаления
self.total_failed += 1
if log:
logger.warning(
f"Не удалось автоматически удалить сообщение: {e}",
log_type='AUTO_DELETE'
)
finally:
# Удаляем из активных задач
if key in self.tasks:
completed_task = self.tasks.pop(key)
# Сохраняем в завершенные
self.completed.append(completed_task)
if len(self.completed) > self.max_completed:
self.completed.pop(0)
def cancel(
self,
chat_id: int,
message_id: int,
log: bool = True
) -> bool:
"""
Отменяет запланированное удаление.
Args:
chat_id: ID чата
message_id: ID сообщения
log: Логировать отмену
Returns:
bool: True если задача была отменена
Example:
>> cancelled = auto_delete_manager.cancel(
... chat_id=message.chat.id,
... message_id=message.message_id
... )
"""
key = (chat_id, message_id)
if key in self.tasks:
task_data = self.tasks[key]
# Отменяем asyncio task
if task_data.task and not task_data.task.done():
task_data.task.cancel()
# Удаляем из активных
self.tasks.pop(key)
if log:
logger.debug(
f"Автоудаление отменено для сообщения {message_id}",
log_type='AUTO_DELETE'
)
return True
return False
def cancel_all(self, chat_id: Optional[int] = None) -> int:
"""
Отменяет все запланированные удаления.
Args:
chat_id: ID чата (если None, отменяет для всех чатов)
Returns:
int: Количество отмененных задач
Example:
>> # Отменить для всех чатов
>> count = auto_delete_manager.cancel_all()
>> # Отменить для конкретного чата
>> count = auto_delete_manager.cancel_all(chat_id=message.chat.id)
"""
cancelled_count = 0
# Собираем ключи для отмены
keys_to_cancel = []
for key, task_data in self.tasks.items():
if chat_id is None or task_data.chat_id == chat_id:
keys_to_cancel.append(key)
# Отменяем
for key in keys_to_cancel:
if self.cancel(key[0], key[1], log=False):
cancelled_count += 1
if cancelled_count > 0:
logger.info(
f"Отменено {cancelled_count} задач автоудаления",
log_type='AUTO_DELETE'
)
return cancelled_count
def get_task(
self,
chat_id: int,
message_id: int
) -> Optional[DeleteTask]:
"""
Получает задачу по ID чата и сообщения.
Args:
chat_id: ID чата
message_id: ID сообщения
Returns:
Optional[DeleteTask]: Задача или None
"""
key = (chat_id, message_id)
return self.tasks.get(key)
def get_chat_tasks(self, chat_id: int) -> list[DeleteTask]:
"""
Получает все задачи для чата.
Args:
chat_id: ID чата
Returns:
list[DeleteTask]: Список задач
"""
return [
task for task in self.tasks.values()
if task.chat_id == chat_id
]
def get_stats(self) -> Dict[str, Any]:
"""
Возвращает статистику менеджера.
Returns:
Dict: Словарь со статистикой
Example:
>> stats = auto_delete_manager.get_stats()
>> print(f"Активных задач: {stats['active_tasks']}")
"""
return {
'active_tasks': len(self.tasks),
'completed_tasks': len(self.completed),
'total_scheduled': self.total_scheduled,
'total_deleted': self.total_deleted,
'total_failed': self.total_failed,
'total_cancelled': self.total_cancelled,
'success_rate': (
f"{(self.total_deleted / self.total_scheduled * 100):.1f}%"
if self.total_scheduled > 0 else "0%"
)
}
def cleanup_expired(self) -> int:
"""
Удаляет истекшие задачи (которые должны были выполниться, но не выполнились).
Returns:
int: Количество удаленных задач
"""
expired_keys = [
key for key, task in self.tasks.items()
if task.is_expired and (not task.task or task.task.done())
]
for key in expired_keys:
self.tasks.pop(key)
return len(expired_keys)
# Глобальный менеджер
auto_delete_manager = AutoDeleteManager()
# ================= УДОБНЫЕ ФУНКЦИИ =================
async def auto_delete_message(
bot: Bot,
chat_id: int,
message_id: int,
delay: int = 604800,
reason: Optional[str] = None
) -> DeleteTask:
"""
Автоматически удаляет сообщение через указанное время.
Args:
bot: Экземпляр бота
chat_id: ID чата
message_id: ID сообщения
delay: Задержка в секундах (по умолчанию 7 дней)
reason: Причина удаления
Returns:
DeleteTask: Созданная задача
Example:
>> # Удалить через 1 минуту
>> await auto_delete_message(bot, chat_id, message_id, delay=60)
>> # Удалить через 7 дней (по умолчанию)
>> await auto_delete_message(bot, chat_id, message_id)
"""
return await auto_delete_manager.schedule(
bot=bot,
chat_id=chat_id,
message_id=message_id,
delay=delay,
reason=reason
)
async def schedule_delete(
message: Message,
delay: int,
reason: Optional[str] = None
) -> DeleteTask:
"""
Планирует удаление сообщения (упрощенная версия).
Args:
message: Объект сообщения
delay: Задержка в секундах
reason: Причина удаления
Returns:
DeleteTask: Созданная задача
Example:
>> # Планируем удаление через 30 секунд
>> await schedule_delete(message, delay=30, reason="Временное")
"""
return await auto_delete_manager.schedule(
bot=message.bot,
chat_id=message.chat.id,
message_id=message.message_id,
delay=delay,
reason=reason
)
def cancel_delete(message: Message) -> bool:
"""
Отменяет запланированное удаление сообщения.
Args:
message: Объект сообщения
Returns:
bool: True если удаление было отменено
Example:
>> if cancel_delete(message):
... await message.answer("Удаление отменено")
"""
return auto_delete_manager.cancel(
chat_id=message.chat.id,
message_id=message.message_id
)
async def delete_after(
message: Message,
text: str,
delay: int = 10,
**kwargs
) -> Message:
"""
Отправляет сообщение и автоматически удаляет его через указанное время.
Args:
message: Исходное сообщение
text: Текст нового сообщения
delay: Задержка до удаления в секундах
**kwargs: Дополнительные параметры для message.answer()
Returns:
Message: Отправленное сообщение
Example:
>> # Отправить и удалить через 10 секунд
>> await delete_after(message, "Это временное сообщение")
>> # Отправить и удалить через 5 секунд
>> await delete_after(
... message,
... "⚠️ Ошибка!",
... delay=5,
... parse_mode="HTML"
... )
"""
sent_message = await message.answer(text, **kwargs)
await auto_delete_manager.schedule(
bot=message.bot,
chat_id=sent_message.chat.id,
message_id=sent_message.message_id,
delay=delay,
reason="delete_after"
)
return sent_message
async def delete_both_after(
original: Message,
reply_text: str,
delay: int = 10,
**kwargs
) -> Message:
"""
Отправляет ответ и удаляет оба сообщения через указанное время.
Args:
original: Исходное сообщение
reply_text: Текст ответа
delay: Задержка до удаления
**kwargs: Дополнительные параметры
Returns:
Message: Отправленное сообщение
Example:
>> # Удалить и команду, и ответ через 5 секунд
>> await delete_both_after(
... message,
... "✅ Команда выполнена",
... delay=5
... )
"""
# Отправляем ответ
sent = await delete_after(original, reply_text, delay, **kwargs)
# Планируем удаление оригинала
await auto_delete_manager.schedule(
bot=original.bot,
chat_id=original.chat.id,
message_id=original.message_id,
delay=delay,
reason="delete_both"
)
return sent
async def delete_messages_after(
bot: Bot,
chat_id: int,
message_ids: list[int],
delay: int
) -> int:
"""
Планирует удаление нескольких сообщений.
Args:
bot: Экземпляр бота
chat_id: ID чата
message_ids: Список ID сообщений
delay: Задержка до удаления
Returns:
int: Количество запланированных удалений
Example:
>> # Удалить все сообщения через 1 час
>> count = await delete_messages_after(
... bot,
... chat_id,
... [123, 124, 125, 126],
... delay=3600
... )
"""
count = 0
for message_id in message_ids:
await auto_delete_manager.schedule(
bot=bot,
chat_id=chat_id,
message_id=message_id,
delay=delay,
reason="mass_delete",
log=False
)
count += 1
logger.info(
f"Запланировано удаление {count} сообщений через {format_duration(delay)}",
log_type='AUTO_DELETE'
)
return count

View File

@@ -64,7 +64,7 @@ COMMANDS: Final[dict[str, list[str]]] = {
"addtemplemma": [
"addtemplemma", "добавитьвремлемму", # основные
"фввеуьздуььф", "lj,fdbnmdhtvktve", # раскладка
"atl", "addtl", "темплемму", "addtlem", "addtemplem",
"atl", "addtl", "темплемму", "addtlem", "addtemplem", "templemma"
],
# ==================== ДОБАВЛЕНИЕ ИСКЛЮЧЕНИЙ ====================
@@ -78,19 +78,19 @@ COMMANDS: Final[dict[str, list[str]]] = {
"remword": [
"remword", "удалитьслово", # основные
"кутцщкв", "elfkbnmckjdj", # раскладка
"rw", "delword", "dw", "удслово",
"rw", "delword", "dw", "удслово", "rword",
],
"remlemma": [
"remlemma", "удалитьлемму", # основные
"кутдуььф", "elfkbnmktve", # раскладка
"rl", "dellemma", "dl", "удлемму",
"rl", "dellemma", "dl", "удлемму", "rlemma",
],
"rempart": [
"rempart", "удалитьчасть", # основные
"кутзфке", "elfkbnmxfcnm", # раскладка
"rp", "delpart", "dp", "удчасть",
"rp", "delpart", "dp", "удчасть", "rpart",
],
# ==================== УДАЛЕНИЕ ВРЕМЕННЫХ ====================
@@ -110,45 +110,45 @@ COMMANDS: Final[dict[str, list[str]]] = {
"remexcept": [
"remexcept", "удалитьисключение", # основные
"кутучсузе", "elfkbnmbcrkx", # раскладка
"rxc", "remwhite", "удискл",
"rxc", "remwhite", "удискл", "rexcept",
],
# ==================== КОНФЛИКТНЫЕ СЛОВА ====================
"addconflictword": [
"addconflictword", "добавитьконфликт", # основные
"фввсщтакшсецщкв", "lj,fdbnmrjyakbrn", # раскладка
"acw", "addcw", "конфслово", "conflictword",
"acw", "addcw", "конфслово", "conflictword", "cword",
],
"addconflictlemma": [
"addconflictlemma", "добавитьконфлемму", # основные
"фввсщтакшседуььф", "lj,fdbnmrjyaktve", # раскладка
"acl", "addcl", "конфлемму", "conflictlemma",
"acl", "addcl", "конфлемму", "conflictlemma", "clemma", "clema",
],
"remconflictword": [
"remconflictword", "удалитьконфликт", # основные
"кутсщтакшсецщкв", "elfkbnmrjyakbrn", # раскладка
"rcw", "delcw", "удконфликт",
"rcw", "delcw", "удконфликт", "rcword", "rconflictword",
],
"remconflictlemma": [
"remconflictlemma", "удалитьконфлемму", # основные
"кутсщтакшседуььф", "elfkbnmrjyaktve", # раскладка
"rcl", "delcl", "удконфлемму",
"rcl", "delcl", "удконфлемму", "rclemma", "rclema",
],
# ==================== РЕЖИМ АНТИКОНФЛИКТА ====================
"stopconflict": [
"stopconflict", "стопконфликт", # основные
"cnjgsщтакшse", "cnjzrjyakbrn", # раскладка
"sconf", "sc", "стопконф", "stopconf",
"sconf", "sc", "стопконф", "stopconf", "stopc",
],
"unstopconflict": [
"unstopconflict", "отменаконфликта", # основные
"eycnjgsщтакшse", "jnvtyf", # раскладка
"usconf", "usc", "откконф", "unstopconf",
"usconf", "usc", "откконф", "unstopconf", "ustopc",
],
"conflictstatus": [
@@ -161,7 +161,7 @@ COMMANDS: Final[dict[str, list[str]]] = {
"silence": [
"silence", "тишина", # основные
"ышдутсу", "nbibyf", # раскладка
"sl", "sil", "mute", "quiet", "тиш", "ven",
"sl", "sil", "muteall", "quiet", "тиш", "ven",
],
"unsilence": [
@@ -186,31 +186,31 @@ COMMANDS: Final[dict[str, list[str]]] = {
"addadmin": [
"addadmin", "добавитьадмина", # основные
"фввфвьшт", "lj,fdbnmflvbyf", # раскладка
"aa", "addadm", "добадм",
"aa", "addadm", "добадм", "admin",
],
"remadmin": [
"remadmin", "удалитьадмина", # основные
"кутфвьшт", "elfkbnmflvbyf", # раскладка
"ra", "remadm", "deladmin", "удадм",
"ra", "remadm", "deladmin", "удадм", "radmin",
],
"listadmins": [
"listadmins", "списокадминов", # основные
"дшыефвьшты", "cgbcjrflvbyjd", # раскладка
"admins", "adm", "adminlist", "адм", "дшыефвь", "listadm", "la",
"admins", "adm", "adminlist", "адм", "дшыефвь", "listadm", "la", "ladmin"
],
"adminhelp": [
"adminhelp", "помощьадмину", # основные
"фвьштрудз", "gjvjomflvbyt", # раскладка
"admhelp", "ah", "хелпадм",
"admhelp", "ah", "хелпадм", "adminh"
],
"checkadmin": [
"checkadmin", "проверкаадмина", # основные
"сруслфвьшт", "ghjdthrf", # раскладка
"isadmin", "ca", "провадм", "checkadm",
"isadmin", "ca", "провадм", "checkadm", "cadmin"
],
# ==================== ПРОСМОТР ====================
@@ -223,25 +223,25 @@ COMMANDS: Final[dict[str, list[str]]] = {
"listlemmas": [
"listlemmas", "списоклемм", # основные
"дшыедуььфы", "cgbcjrktv", # раскладка
"ll", "lemmas", "леммы",
"ll", "lemmas", "леммы", "llemma", "llema",
],
"listparts": [
"listparts", "списокчастей", # основные
"дшыезфкеы", "cgbcjrxfcntq", # раскладка
"lp", "parts", "части",
"lp", "parts", "части", "lpart"
],
"listexcept": [
"listexcept", "списокисключений", # основные
"дшыеучсузе", "cgbcjrbcrkx", # раскладка
"lxc", "except", "white", "искл",
"lxc", "except", "white", "искл", "lexcept"
],
"listconflict": [
"listconflict", "списокконфликтов", # основные
"дшыесщтакшse", "cgbcjrrjyakbrnjd", # раскладка
"lc", "conflict", "конф",
"lc", "conflict", "конф", "lconflict",
],
# ==================== СТАТИСТИКА ====================
@@ -280,7 +280,7 @@ COMMANDS: Final[dict[str, list[str]]] = {
"report": [
"report", "репорт", "жалоба", # основные
"кузщке", "htgjhn", ";fkj,f", # раскладка
"rep", "r", "жал",
"rep", "r", "жал", "жб"
],
"reporthelp": [
@@ -362,6 +362,15 @@ COMMANDS: Final[dict[str, list[str]]] = {
"redcom", "editcom", "коммент", "rc", # дополнения
],
"set_description": [
"set_description", "description", "set_des",
],
"set_name": [
"set_name",
],
"set_widget": [
"set_widget",
],
"botsettings": [
"botsettings", "bsettings", "botsetting", "bsetting", # основные + сокращения
"bset", "ботнастрйоки", # раскладка

View File

@@ -45,7 +45,6 @@ class _Settings(BaseSettings):
WEBHOOK_URL: Optional[str] = None
WEBAPP_HOST: str = "0.0.0.0"
WEBAPP_PORT: int = 3131
LOG_LEVEL: str = "warning"
ACCES_LOG: bool = False
# API ключи
@@ -137,7 +136,10 @@ class _Settings(BaseSettings):
SHOW_CAPTION_ABOVE_MEDIA: bool = False
# улучшения
ANTI_SPAM: bool = True
ANTI_SPAM: bool = False
enable_spam_check: bool = False
enable_subscription_check: bool = False
BOT_USERNAME: str = "@OvhdLayla2_bot"
# ================= ВАЛИДАТОРЫ =================
@field_validator('PARSE_MODE')

View File

@@ -44,6 +44,7 @@ class BanWordsManager:
"""Инициализирует базу данных и загружает кэш"""
await self.db.init()
await self.init_default_bot_settings() # ← добавлено
await self.init_default_words()
await self.refresh_cache()
logger.info("BanWordsManager инициализирован", log_type="DATABASE")
@@ -452,12 +453,13 @@ class BanWordsManager:
admins = await self.repo.get_admins()
return {
'substring': banwords.get(BanWordType.SUBSTRING, set()),
'word': banwords.get(BanWordType.WORD, set()),
'lemma': banwords.get(BanWordType.LEMMA, set()),
'part': banwords.get(BanWordType.PART, set()),
'conflict_substring': banwords.get(BanWordType.CONFLICT_SUBSTRING, set()),
'conflict_word': banwords.get(BanWordType.CONFLICT_WORD, set()),
'conflict_lemma': banwords.get(BanWordType.CONFLICT_LEMMA, set()),
'temp_substring': temp_banwords.get(BanWordType.SUBSTRING, set()),
'conflict_part': banwords.get(BanWordType.CONFLICT_PART, set()),
'temp_word': temp_banwords.get(BanWordType.WORD, set()),
'temp_lemma': temp_banwords.get(BanWordType.LEMMA, set()),
'whitelist': whitelist,
'admins': admins
@@ -516,6 +518,50 @@ class BanWordsManager:
)
return []
async def init_default_words(self) -> None:
"""
Добавляет базовые банворды и whitelist при первом запуске.
Ничего не перезаписывает, если уже существует.
"""
try:
from configs import settings
# --- Базовые слова ---
default_word = {"http", "t.me/"}
default_part = {"bot"}
default_lemma = {"скам", "мошенник"}
# Проверяем уже существующие
existing = await self.repo.get_all_banwords()
# word
for word in default_word:
if word not in existing.get(BanWordType.WORD, set()):
await self.repo.add_banword(word, BanWordType.WORD)
# PART
for word in default_part:
if word not in existing.get(BanWordType.PART, set()):
await self.repo.add_banword(word, BanWordType.PART)
# LEMMA
for word in default_lemma:
if word not in existing.get(BanWordType.LEMMA, set()):
await self.repo.add_banword(word, BanWordType.LEMMA)
# --- Добавляем username бота в whitelist ---
bot_username = settings.BOT_USERNAME
if bot_username:
whitelist = await self.repo.get_whitelist()
if bot_username.lower() not in whitelist:
await self.repo.add_whitelist(bot_username.lower())
logger.info("Базовые слова и whitelist инициализированы", log_type="DATABASE")
except Exception as e:
logger.error(f"Ошибка инициализации базовых слов: {e}", log_type="DATABASE")
async def cleanup_expired_temp_words(self) -> int:
"""
Удаляет истёкшие временные банворды.

View File

@@ -31,11 +31,12 @@ class Base(DeclarativeBase):
class BanWordType(str, PyEnum):
"""Типы банвордов"""
SUBSTRING = "substring"
WORD = "word"
LEMMA = "lemma"
PART = "part"
CONFLICT_SUBSTRING = "conflict_substring"
CONFLICT_WORD = "conflict_word"
CONFLICT_LEMMA = "conflict_lemma"
CONFLICT_PART = "conflict_part"
class SpamMode(str, PyEnum):
@@ -62,7 +63,11 @@ class BanWord(Base):
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
word: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
type: Mapped[BanWordType] = mapped_column(
Enum(BanWordType, native_enum=False),
Enum(
BanWordType,
native_enum=False,
values_callable=lambda enum: [e.value for e in enum]
),
nullable=False,
index=True
)
@@ -95,8 +100,13 @@ class TempBanWord(Base):
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
word: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
type: Mapped[BanWordType] = mapped_column(
Enum(BanWordType, native_enum=False),
nullable=False
Enum(
BanWordType,
native_enum=False,
values_callable=lambda enum: [e.value for e in enum]
),
nullable=False,
index=True
)
added_by: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
added_at: Mapped[datetime] = mapped_column(

View File

@@ -158,11 +158,12 @@ class BanWordsRepository:
async def get_all_banwords(self) -> dict[BanWordType, Set[str]]:
result = {
BanWordType.SUBSTRING: set(),
BanWordType.WORD: set(),
BanWordType.LEMMA: set(),
BanWordType.PART: set(),
BanWordType.CONFLICT_SUBSTRING: set(),
BanWordType.CONFLICT_WORD: set(),
BanWordType.CONFLICT_LEMMA: set(),
BanWordType.CONFLICT_PART: set(),
}
try:
async with self.db.get_session() as session:
@@ -335,7 +336,7 @@ class BanWordsRepository:
async def get_all_temp_banwords(self) -> dict[BanWordType, Set[str]]:
"""Получает все активные временные банворды по типам"""
result = {
BanWordType.SUBSTRING: set(),
BanWordType.WORD: set(),
BanWordType.LEMMA: set(),
}