""" Детектор шпионов - анализ активности пользователей """ import json import logging from datetime import datetime, timedelta from typing import List, Dict, Tuple from sqlalchemy import select, func from sqlalchemy.ext.asyncio import AsyncSession from database.models import User, Message, SpyLog from config import ADMIN_USER_ID logger = logging.getLogger(__name__) class SpyDetector: """Детектор подозрительной активности""" def __init__(self, session: AsyncSession): self.session = session self.patterns = self._load_patterns() def _load_patterns(self) -> Dict: """Загрузка паттернов из JSON""" try: with open('data/spies_patterns.json', 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: logger.error(f'Не удалось загрузить паттерны: {e}') return self._get_default_patterns() def _get_default_patterns(self) -> Dict: """Паттерны по умолчанию""" return { 'patterns': { 'new_account_days': 7, 'no_avatar_score': 10, 'generic_username_score': 15, 'unverified_hours': 24, 'unverified_score': 25, 'provocation_score': 30, 'office_hours_score': 15, 'multicount_score': 50, }, 'thresholds': { 'normal': 30, 'suspicious': 60, }, 'office_hours': { 'start': 9, 'end': 18, 'weekdays': [0, 1, 2, 3, 4], } } async def analyze_user(self, user: User) -> Tuple[int, List[str]]: """ Анализ пользователя на подозрительность (v2.0) """ score = 0 flags = [] patterns = self.patterns['patterns'] # 1. Анонимность (нет аватарки или странный ник) if not user.username or self._is_generic_username(user.username): score += patterns.get('no_avatar_score', 10) flags.append('anon_user') # 2. Не прошёл верификацию вовремя if not user.verified: join_date = user.created_at if hasattr(user, 'created_at') else datetime.utcnow() hours_since_join = (datetime.utcnow() - join_date).total_seconds() / 3600 if hours_since_join > patterns.get('unverified_hours', 24): score += patterns.get('unverified_score', 25) flags.append('unverified_timeout') # 3. Активность только в рабочие часы (офисный планктон УК) is_office = await self._check_office_hours_activity(user.user_id) if is_office: score += patterns.get('office_hours_score', 20) flags.append('office_hours_warrior') # 4. Провокационные сообщения has_provocation = await self._check_provocations(user.user_id) if has_provocation: score += patterns.get('provocation_score', 30) flags.append('uk_provocator') # 5. Токсичность и агрессия is_toxic = await self._check_toxicity(user.user_id) if is_toxic: score += patterns.get('toxicity_score', 15) flags.append('toxic_behavior') # Ограничиваем score до 100 score = min(score, 100) return score, flags async def _check_toxicity(self, user_id: int) -> bool: """Проверка на токсичность: КАПС, знаки, мат-триггеры""" stmt = select(Message.text).where(Message.user_id == user_id).limit(20) result = await self.session.execute(stmt) messages = result.scalars().all() if not messages: return False toxic_count = 0 for text in messages: # 1. Проверка на КАПС (если > 50% букв заглавные) letters = [c for c in text if c.isalpha()] if letters and sum(1 for c in letters if c.isupper()) / len(letters) > 0.5: toxic_count += 1 # 2. Избыточные знаки !!! ??? if "!!!" in text or "???" in text: toxic_count += 1 # 3. Ключевые слова токсичности for kw in self.patterns.get('keywords_toxicity', []): if kw in text.lower(): toxic_count += 1 break return toxic_count >= 3 # Срабатывает если 3+ признака в последних 20 сообщениях async def _check_office_hours_activity(self, user_id: int) -> bool: """Проверка: активность в часы 9-18 пн-пт""" stmt = select(Message.created_at).where(Message.user_id == user_id).limit(50) result = await self.session.execute(stmt) dates = result.scalars().all() if len(dates) < 10: return False office_config = self.patterns['office_hours'] office_messages = 0 for dt in dates: if dt.weekday() in office_config['weekdays'] and office_config['start'] <= dt.hour < office_config['end']: office_messages += 1 return (office_messages / len(dates)) > 0.8 # 80%+ в рабочее время async def _check_provocations(self, user_id: int) -> bool: """ Проверка на провокационные сообщения в пользу УК """ keywords = self.patterns.get('keywords_provocation', [ 'управляющая компания права', 'ук говорит что', 'надо платить больше', ]) # Ищем сообщения с ключевыми словами for keyword in keywords: stmt = ( select(func.count()) .select_from(Message) .where(Message.user_id == user_id) .where(Message.text.ilike(f'%{keyword}%')) ) result = await self.session.execute(stmt) count = result.scalar() or 0 if count > 0: return True return False def _is_generic_username(self, username: str) -> bool: """Проверка на стандартный никнейм""" if not username: return True import re patterns = [ r'^user\d+$', r'^user_\w+$', r'^id\d+$', ] return any(re.match(p, username.lower()) for p in patterns) async def update_spy_score(self, user: User) -> bool: """ Обновить рейтинг подозрительности пользователя Возвращает True, если score изменился значительно """ old_score = user.spy_score new_score, flags = await self.analyze_user(user) user.spy_score = new_score user.spy_flags = json.dumps(flags) # Логгируем изменение if abs(new_score - old_score) >= 20: await self._log_spy_detection(user, new_score, flags) return True return False async def _log_spy_detection(self, user: User, score: int, flags: List[str]): """Логирование детекта""" spy_log = SpyLog( user_id=user.user_id, spy_score=score, flags=json.dumps(flags), ) self.session.add(spy_log) await self.session.commit() async def get_suspicious_users(self, min_score: int = 30) -> List[User]: """Получить список подозрительных пользователей""" stmt = ( select(User) .where(User.spy_score >= min_score) .order_by(User.spy_score.desc()) ) result = await self.session.execute(stmt) return list(result.scalars().all()) async def get_spies(self) -> List[User]: """Получить пользователей со статусом «шпион» (score > 60)""" return await self.get_suspicious_users(min_score=61)