ba4676b0af
✅ Регистрация и верификация жильцов ✅ Рейтинг активности (уровни) ✅ Детект шпионов (Score 0-100) ✅ Анти-мат система (3 предупреждения → бан) ✅ Админ-панель с рассылкой ✅ Учёт квартир (несколько жильцов) ✅ Телефоны экстренных служб и мастеров ✅ Еженедельный экспорт JSON ✅ Прокси (socks5) для обхода РКН Бекап: backups/versions/v1.0_mvp_2026-02-26/ Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
254 lines
9.7 KiB
Python
254 lines
9.7 KiB
Python
"""
|
||
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 '🚫 Очень токсичный', '🚫'
|