219 lines
8.5 KiB
Python
219 lines
8.5 KiB
Python
"""
|
||
Детектор шпионов - анализ активности пользователей
|
||
"""
|
||
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)
|