domovoy_bot/services/spy_detector.py

"""
Детектор шпионов - анализ активности пользователей
"""
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]]:
        """
        Анализ пользователя на подозрительность
        Возвращает: (score, list_of_flags)
        """
        score = 0
        flags = []
        patterns = self.patterns['patterns']
        
        # 1. Новый аккаунт
        # Примечание: Telegram не отдаёт дату создания аккаунта через Bot API
        # Можно эвристически определять по первым сообщениям
        
        # 2. Нет аватарки (проверяем по наличию username/фото)
        if not user.username or user.username.startswith('user'):
            score += patterns.get('no_avatar_score', 10)
            flags.append('no_avatar')
        
        # 3. Стандартный никнейм
        if self._is_generic_username(user.username):
            score += patterns.get('generic_username_score', 15)
            flags.append('generic_username')
        
        # 4. Не прошёл верификацию
        if not user.verified:
            hours_since_join = (datetime.utcnow() - user.join_date).total_seconds() / 3600
            if hours_since_join > patterns.get('unverified_hours', 24):
                score += patterns.get('unverified_score', 25)
                flags.append('unverified')
        
        # 5. Активность только в рабочие часы
        is_office = await self._check_office_hours_activity(user.user_id)
        if is_office:
            score += patterns.get('office_hours_score', 15)
            flags.append('office_hours')
        
        # 6. Провокационные сообщения
        has_provocation = await self._check_provocations(user.user_id)
        if has_provocation:
            score += patterns.get('provocation_score', 30)
            flags.append('provocation')
        
        # Ограничиваем score до 100
        score = min(score, 100)
        
        return score, flags
    
    async def _check_office_hours_activity(self, user_id: int) -> bool:
        """
        Проверка: активность только в рабочие часы (9-18, пн-пт)
        """
        stmt = (
            select(func.count())
            .select_from(Message)
            .where(Message.user_id == user_id)
        )
        result = await self.session.execute(stmt)
        total = result.scalar() or 0
        
        if total == 0:
            return False
        
        # Считаем сообщения в рабочие часы
        # Примечание: для точной проверки нужно хранить время каждого сообщения
        # и анализировать распределение
        
        # Упрощённая версия: если >80% сообщений в рабочие часы
        # Требуется дополнительная логика анализа временных меток
        
        return False  # Заглушка
    
    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)