""" Toxicity Detector — детектор токсичности сообщений Анализ лексики на оскорбления, агрессию, токсичность """ import logging import re from datetime import datetime, timedelta from typing import Dict, List, Tuple, Optional from collections import defaultdict from sqlalchemy import select, func from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger(__name__) class ToxicityDetector: """Детектор токсичности""" def __init__(self, session: AsyncSession): self.session = session self.toxic_words = self._load_toxic_words() self.aggression_patterns = self._load_aggression_patterns() def _load_toxic_words(self) -> Dict: """Загрузить токсичные слова по категориям""" return { 'insults': [ # Оскорбления 'дурак', 'идиот', 'кретин', 'придурок', 'тупой', 'глупый', 'неадекват', 'псих', 'сумасшедший', 'ненормальный', 'хам', 'грубиян', 'скотина', 'свинья', ], 'aggression': [ # Агрессия 'заткнись', 'молчи', 'убью', 'убить', 'ударю', 'ударить', 'ненавижу', 'ненавижу тебя', 'достал', 'задолбал', 'пошел на хуй', 'пошла на хуй', 'нахуй', ], 'threats': [ # Угрозы 'угрожаю', 'угроза', 'пожалеешь', 'пожалеете', 'найду', 'встретимся', 'разберусь', 'поимеешь', ], 'discrimination': [ # Дискриминация 'национальн', 'расов', 'религиозн', ], } def _load_aggression_patterns(self) -> List: """Загрузить паттерны агрессивного поведения""" return [ r'!{2,}', # Множественные восклицательные знаки r'\?{2,}', # Множественные вопросительные знаки r'[А-Я]{3,}', # CAPS LOCK (3+ заглавных подряд) r'(.)\1{2,}', # Повторяющиеся символы (аааа) ] async def analyze_message(self, text: str) -> Tuple[float, List[str]]: """ Проанализировать сообщение на токсичность Returns: (toxicity_score 0-100, reasons) """ if not text: return 0.0, [] text_lower = text.lower() score = 0 reasons = [] # 1. Проверка токсичных слов for category, words in self.toxic_words.items(): for word in words: if word in text_lower: score += 15 reasons.append(f'{category}: "{word}"') # 2. Проверка паттернов агрессии for pattern in self.aggression_patterns: if re.search(pattern, text): score += 10 reasons.append(f'паттерн: {pattern}') # 3. Проверка длины (слишком длинные сообщения могут быть агрессивными) if len(text) > 500: score += 5 reasons.append('длинное сообщение') # 4. Проверка на капс (более 50% заглавных) uppercase_count = sum(1 for c in text if c.isupper()) total_letters = sum(1 for c in text if c.isalpha()) if total_letters > 0 and uppercase_count / total_letters > 0.5: score += 15 reasons.append('CAPS LOCK') # Ограничиваем score до 100 score = min(score, 100) return score, reasons async def calculate_user_toxicity(self, user_id: int, days: int = 7) -> Dict: """ Рассчитать токсичность пользователя за период Returns: dict со статистикой """ from database.models import Message time_window = datetime.utcnow() - timedelta(days=days) # Получаем сообщения пользователя stmt = ( select(Message.text) .where(Message.user_id == user_id) .where(Message.timestamp >= time_window) .where(Message.text.isnot(None)) ) result = await self.session.execute(stmt) messages = [row[0] for row in result.all()] if not messages: return { 'total_messages': 0, 'toxic_messages': 0, 'toxicity_percent': 0, 'avg_toxicity': 0, 'max_toxicity': 0, 'categories': {} } # Анализируем каждое сообщение toxic_count = 0 total_toxicity = 0 max_toxicity = 0 category_counts = defaultdict(int) for text in messages: score, reasons = await self.analyze_message(text) if score > 0: toxic_count += 1 total_toxicity += score max_toxicity = max(max_toxicity, score) # Считаем категории for reason in reasons: if ':' in reason: category = reason.split(':')[0] category_counts[category] += 1 return { 'total_messages': len(messages), 'toxic_messages': toxic_count, 'toxicity_percent': round(toxic_count / len(messages) * 100, 1), 'avg_toxicity': round(total_toxicity / len(messages), 1), 'max_toxicity': max_toxicity, 'categories': dict(category_counts) } async def get_toxic_users(self, days: int = 7, limit: int = 10) -> List[Dict]: """ Получить топ токсичных пользователей Returns: список пользователей с токсичностью """ from database.models import User, Message # Получаем всех активных пользователей stmt = select(User).where(User.is_banned == False).limit(100) result = await self.session.execute(stmt) users = list(result.scalars().all()) toxic_users = [] for user in users: stats = await self.calculate_user_toxicity(user.user_id, days) if stats['toxic_messages'] > 0: toxic_users.append({ 'user_id': user.user_id, 'username': user.username, 'full_name': user.full_name, 'apartment': user.apartment, 'stats': stats }) # Сортируем по токсичности toxic_users.sort(key=lambda x: x['stats']['avg_toxicity'], reverse=True) return toxic_users[:limit] async def get_toxicity_trend(self, user_id: int, days: int = 14) -> Dict: """ Получить тренд токсичности по дням Returns: dict с датами и токсичностью """ from database.models import Message time_window = datetime.utcnow() - timedelta(days=days) stmt = ( select(Message.text, func.date(Message.timestamp).label('msg_date')) .where(Message.user_id == user_id) .where(Message.timestamp >= time_window) .where(Message.text.isnot(None)) ) result = await self.session.execute(stmt) messages = [(row[0], row[1]) for row in result.all()] daily_toxicity = defaultdict(list) for text, date in messages: score, _ = await self.analyze_message(text) daily_toxicity[date].append(score) # Считаем среднюю токсичность по дням result = {} for date, scores in sorted(daily_toxicity.items()): result[date] = round(sum(scores) / len(scores), 1) return { 'user_id': user_id, 'days': days, 'daily_toxicity': result } async def get_complaints_stats(self, user_id: int) -> Dict: """ Получить статистику жалоб на пользователя Returns: dict со статистикой жалоб """ from database.models import User user = await self.session.get(User, user_id) if not user: return {'complaints': 0, 'last_complaint': None} # Пока просто возвращаем заглушку # В будущем можно добавить таблицу complaints return { 'complaints': 0, 'last_complaint': None } def get_toxicity_level(self, score: float) -> Tuple[str, str]: """ Получить уровень токсичности Returns: (level_name, emoji) """ if score < 10: return '😊 Нормальный', '😊' elif score < 30: return '😐 Нейтральный', '😐' elif score < 50: return '⚠️ Подозрительный', '⚠️' elif score < 70: return '😠 Токсичный', '😠' else: return '🚫 Очень токсичный', '🚫'