ba4676b0af
✅ Регистрация и верификация жильцов ✅ Рейтинг активности (уровни) ✅ Детект шпионов (Score 0-100) ✅ Анти-мат система (3 предупреждения → бан) ✅ Админ-панель с рассылкой ✅ Учёт квартир (несколько жильцов) ✅ Телефоны экстренных служб и мастеров ✅ Еженедельный экспорт JSON ✅ Прокси (socks5) для обхода РКН Бекап: backups/versions/v1.0_mvp_2026-02-26/ Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
354 lines
14 KiB
Python
354 lines
14 KiB
Python
"""
|
||
Анти-мат система
|
||
Удаляет сообщения с матом, выдаёт предупреждения
|
||
"""
|
||
import logging
|
||
import json
|
||
import re
|
||
from datetime import datetime
|
||
from aiogram import Router, F
|
||
from aiogram.types import Message
|
||
from aiogram.filters import Command
|
||
from aiogram.exceptions import TelegramBadRequest
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy import select
|
||
from database.db import AsyncSessionLocal
|
||
from database.models import User
|
||
from utils.decorators import admin_only
|
||
import config
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = Router()
|
||
|
||
# Глобальный словарь предупреждений: user_id -> count
|
||
warnings = {}
|
||
|
||
|
||
def load_bad_words() -> list:
|
||
"""Загрузка чёрного списка слов"""
|
||
try:
|
||
with open('data/bad_words.json', 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
return data.get('words', [])
|
||
except FileNotFoundError:
|
||
# Возвращаем базовый список
|
||
return [
|
||
'хуй', 'хуи', 'хуя', 'хуем', 'хую',
|
||
'пизд', 'пизде', 'пизду',
|
||
'ебал', 'ебать', 'ебуч', 'ебан',
|
||
'бляд', 'блять',
|
||
'сука', 'суки',
|
||
'мудак', 'мудаки',
|
||
'уебок', 'уебки',
|
||
'говно', 'говна',
|
||
'залуп', 'залупа',
|
||
'манда', 'манды',
|
||
'трах', 'трахал',
|
||
'еблан', 'ебланы',
|
||
'пидор', 'пидоры',
|
||
'долбоёб', 'долбоёбы',
|
||
'о хуе', 'на хуй', 'по хуй',
|
||
'за хуй', 'под хуй',
|
||
]
|
||
|
||
|
||
def load_exclamations() -> list:
|
||
"""Загрузка списка восклицаний (типа @user, !команда)"""
|
||
try:
|
||
with open('data/bad_words.json', 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
return data.get('exclamations', [])
|
||
except:
|
||
return ['@', '!', '/', 'http', 'www', 't.me']
|
||
|
||
|
||
def contains_bad_word(text: str) -> tuple:
|
||
"""
|
||
Проверка текста на мат
|
||
Возвращает: (есть_ли_мат, найденное_слово)
|
||
"""
|
||
if not text:
|
||
return False, None
|
||
|
||
text_lower = text.lower()
|
||
bad_words = load_bad_words()
|
||
exclamations = load_exclamations()
|
||
|
||
# Проверяем каждое плохое слово
|
||
for bad_word in bad_words:
|
||
# Простая проверка (слово содержится в тексте)
|
||
if bad_word.lower() in text_lower:
|
||
# Проверяем что это не часть другого слова (не ссылка, не упоминание)
|
||
skip = False
|
||
for excl in exclamations:
|
||
if excl in text_lower:
|
||
# Это может быть ссылка или упоминание - пропускаем
|
||
skip = True
|
||
break
|
||
|
||
if skip:
|
||
continue
|
||
|
||
# Проверяем что это отдельное слово (окружено пробелами или знаками препинания)
|
||
pattern = r'(^|\s|\.|,|!|\?|:|;)' + re.escape(bad_word) + r'($|\s|\.|,|!|\?|:|;)'
|
||
if re.search(pattern, text_lower, re.IGNORECASE):
|
||
return True, bad_word
|
||
|
||
return False, None
|
||
|
||
|
||
async def get_user_warnings(user_id: int) -> int:
|
||
"""Получить количество предупреждений пользователя"""
|
||
return warnings.get(user_id, 0)
|
||
|
||
|
||
async def add_warning(user_id: int) -> int:
|
||
"""Добавить предупреждение пользователю"""
|
||
current = warnings.get(user_id, 0)
|
||
warnings[user_id] = current + 1
|
||
return warnings[user_id]
|
||
|
||
|
||
async def clear_warnings(user_id: int):
|
||
"""Сбросить предупреждения пользователя"""
|
||
if user_id in warnings:
|
||
del warnings[user_id]
|
||
|
||
|
||
@router.message()
|
||
async def check_profanity(message: Message):
|
||
"""
|
||
Проверка всех сообщений на мат
|
||
"""
|
||
# Пропускаем сообщения от ботов
|
||
if message.from_user.is_bot:
|
||
return
|
||
|
||
# Пропускаем админов (им можно всё)
|
||
if config.is_admin(message.from_user.id):
|
||
return
|
||
|
||
# Пропускаем если нет текста
|
||
if not message.text:
|
||
return
|
||
|
||
# Проверяем на мат
|
||
has_bad_word, bad_word = contains_bad_word(message.text)
|
||
|
||
if has_bad_word:
|
||
logger.warning(f'Обнаружен мат от user_id={message.from_user.id}, слово: {bad_word}')
|
||
|
||
# Удаляем сообщение
|
||
try:
|
||
await message.delete()
|
||
logger.info(f'Сообщение удалено: message_id={message.message_id}')
|
||
except TelegramBadRequest as e:
|
||
logger.error(f'Не удалось удалить сообщение: {e}')
|
||
# Если не удалось удалить (нет прав), продолжаем но с предупреждением
|
||
|
||
# Добавляем предупреждение
|
||
warning_count = await add_warning(message.from_user.id)
|
||
|
||
# Получаем пользователя из БД
|
||
async with AsyncSessionLocal() as session:
|
||
stmt = select(User).where(User.user_id == message.from_user.id)
|
||
result = await session.execute(stmt)
|
||
user = result.scalar_one_or_none()
|
||
|
||
user_name = user.full_name if user else f'user{message.from_user.id}'
|
||
|
||
# Пишем нарушителю в личку
|
||
try:
|
||
if warning_count == 1:
|
||
# Первое предупреждение
|
||
await message.bot.send_message(
|
||
message.from_user.id,
|
||
f'⚠️ <b>Предупреждение #1</b>\n\n'
|
||
f'{user_name}, в вашем сообщении обнаружена нецензурная лексика.\n\n'
|
||
f'❗ <b>Правила чата:</b>\n'
|
||
f'Запрещено использовать мат и оскорбления.\n\n'
|
||
f'📊 <b>Ваши нарушения:</b> 1 из 3\n\n'
|
||
f'При следующем нарушении будет публичное предупреждение.\n'
|
||
f'При третьем — бан.',
|
||
parse_mode='HTML'
|
||
)
|
||
|
||
# Тихое уведомление админу
|
||
try:
|
||
await message.bot.send_message(
|
||
config.ADMIN_USER_ID,
|
||
f'⚠️ <b>Предупреждение выдано</b>\n\n'
|
||
f'Нарушитель: {user_name}\n'
|
||
f'ID: <code>{message.from_user.id}</code>\n'
|
||
f'Слово: {bad_word}\n\n'
|
||
f'Это первое предупреждение.',
|
||
parse_mode='HTML'
|
||
)
|
||
except:
|
||
pass
|
||
|
||
elif warning_count == 2:
|
||
# Второе предупреждение
|
||
await message.bot.send_message(
|
||
message.from_user.id,
|
||
f'🚨 <b>Предупреждение #2</b>\n\n'
|
||
f'{user_name}, вы нарушили правила чата повторно!\n\n'
|
||
f'❗ <b>Ваше сообщение удалено за:</b>\n'
|
||
f'Использование нецензурной лексики.\n\n'
|
||
f'📊 <b>Ваши нарушения:</b> 2 из 3\n\n'
|
||
f'⚠️ <b>Внимание!</b>\n'
|
||
f'При следующем нарушении вы будете забанены.',
|
||
parse_mode='HTML'
|
||
)
|
||
|
||
# Публичное уведомление в чат
|
||
try:
|
||
await message.bot.send_message(
|
||
config.ADMIN_CHAT_ID,
|
||
f'⚠️ <b>Внимание!</b>\n\n'
|
||
f'Пользователь {user_name} получил второе предупреждение за мат.\n\n'
|
||
f'Пожалуйста, соблюдайте правила чата!',
|
||
parse_mode='HTML'
|
||
)
|
||
except:
|
||
pass
|
||
|
||
elif warning_count >= 3:
|
||
# Третье предупреждение — бан
|
||
await message.bot.send_message(
|
||
message.from_user.id,
|
||
f'🚫 <b>ВЫ ЗАБАНЕНЫ</b>\n\n'
|
||
f'{user_name}, вы получили третье предупреждение.\n\n'
|
||
f'Причина: Систематическое нарушение правил чата (мат).\n\n'
|
||
f'Вы больше не можете писать в чат.',
|
||
parse_mode='HTML'
|
||
)
|
||
|
||
# Бан в БД
|
||
async with AsyncSessionLocal() as session:
|
||
stmt = select(User).where(User.user_id == message.from_user.id)
|
||
result = await session.execute(stmt)
|
||
user = result.scalar_one_or_none()
|
||
|
||
if user:
|
||
user.is_banned = True
|
||
await session.commit()
|
||
|
||
# Удаляем из чата
|
||
try:
|
||
await message.chat.ban(message.from_user.id)
|
||
except:
|
||
pass
|
||
|
||
# Уведомление админу
|
||
try:
|
||
await message.bot.send_message(
|
||
config.ADMIN_USER_ID,
|
||
f'🚫 <b>ПОЛЬЗОВАТЕЛЬ ЗАБАНЕН</b>\n\n'
|
||
f'Нарушитель: {user_name}\n'
|
||
f'ID: <code>{message.from_user.id}</code>\n'
|
||
f'Причина: 3 предупреждения за мат\n\n'
|
||
f'Пользователь удалён из чата.',
|
||
parse_mode='HTML'
|
||
)
|
||
except:
|
||
pass
|
||
except TelegramBadRequest as e:
|
||
logger.error(f'Не удалось отправить предупреждение: {e}')
|
||
|
||
# Логируем
|
||
logger.warning(f'Предупреждение #{warning_count} выдано user_id={message.from_user.id}')
|
||
|
||
|
||
@router.message(Command('warnings'))
|
||
@admin_only
|
||
async def cmd_warnings_list(message: Message):
|
||
"""Список предупреждений"""
|
||
if not warnings:
|
||
await message.answer('✅ Нарушений нет')
|
||
return
|
||
|
||
text = '⚠️ <b>Предупреждения</b>\n\n'
|
||
for user_id, count in warnings.items():
|
||
text += f'• ID <code>{user_id}</code> — {count} предупр.\n'
|
||
|
||
text += '\n💡 Для сброса: <code>/clear_warnings <user_id></code>'
|
||
|
||
await message.answer(text, parse_mode='HTML')
|
||
|
||
|
||
@router.message(Command('clear_warnings'))
|
||
@admin_only
|
||
async def cmd_clear_warnings(message: Message):
|
||
"""Сбросить предупреждения пользователя"""
|
||
args = message.text.split()
|
||
|
||
if len(args) < 2:
|
||
await message.answer('Использование: <code>/clear_warnings <user_id></code>', parse_mode='HTML')
|
||
return
|
||
|
||
try:
|
||
user_id = int(args[1])
|
||
except ValueError:
|
||
await message.answer('❌ Неверный ID')
|
||
return
|
||
|
||
await clear_warnings(user_id)
|
||
|
||
await message.answer(f'✅ Предупреждения пользователя {user_id} сброшены')
|
||
|
||
|
||
@router.message(Command('add_bad_word'))
|
||
@admin_only
|
||
async def cmd_add_bad_word(message: Message):
|
||
"""Добавить слово в чёрный список"""
|
||
args = message.text.split(maxsplit=1)
|
||
|
||
if len(args) < 2:
|
||
await message.answer('Использование: <code>/add_bad_word слово</code>', parse_mode='HTML')
|
||
return
|
||
|
||
word = args[1].strip().lower()
|
||
|
||
try:
|
||
with open('data/bad_words.json', 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
except:
|
||
data = {'words': [], 'exclamations': []}
|
||
|
||
if word not in data['words']:
|
||
data['words'].append(word)
|
||
with open('data/bad_words.json', 'w', encoding='utf-8') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
await message.answer(f'✅ Слово "{word}" добавлено в чёрный список')
|
||
else:
|
||
await message.answer(f'ℹ️ Слово "{word}" уже в списке')
|
||
|
||
|
||
@router.message(Command('remove_bad_word'))
|
||
@admin_only
|
||
async def cmd_remove_bad_word(message: Message):
|
||
"""Удалить слово из чёрного списка"""
|
||
args = message.text.split(maxsplit=1)
|
||
|
||
if len(args) < 2:
|
||
await message.answer('Использование: <code>/remove_bad_word слово</code>', parse_mode='HTML')
|
||
return
|
||
|
||
word = args[1].strip().lower()
|
||
|
||
try:
|
||
with open('data/bad_words.json', 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
except:
|
||
await message.answer('❌ Файл не найден')
|
||
return
|
||
|
||
if word in data['words']:
|
||
data['words'].remove(word)
|
||
with open('data/bad_words.json', 'w', encoding='utf-8') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
await message.answer(f'✅ Слово "{word}" удалено из чёрного списка')
|
||
else:
|
||
await message.answer(f'ℹ️ Слово "{word}" не найдено в списке')
|