ba4676b0af
✅ Регистрация и верификация жильцов ✅ Рейтинг активности (уровни) ✅ Детект шпионов (Score 0-100) ✅ Анти-мат система (3 предупреждения → бан) ✅ Админ-панель с рассылкой ✅ Учёт квартир (несколько жильцов) ✅ Телефоны экстренных служб и мастеров ✅ Еженедельный экспорт JSON ✅ Прокси (socks5) для обхода РКН Бекап: backups/versions/v1.0_mvp_2026-02-26/ Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
284 lines
11 KiB
Python
284 lines
11 KiB
Python
"""
|
||
Multicount Detector — детектор мультиаккаунтов
|
||
Анализ паттернов поведения для выявления нескольких аккаунтов от одного человека
|
||
"""
|
||
import logging
|
||
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 MulticountDetector:
|
||
"""Детектор мультиаккаунтов"""
|
||
|
||
def __init__(self, session: AsyncSession):
|
||
self.session = session
|
||
self.config = self._get_default_config()
|
||
|
||
def _get_default_config(self) -> Dict:
|
||
"""Конфигурация по умолчанию"""
|
||
return {
|
||
'time_window_hours': 24, # Окно анализа (часов)
|
||
'activity_similarity_threshold': 0.8, # Порог схожести активности
|
||
'message_pattern_threshold': 0.7, # Порог схожести паттернов сообщений
|
||
'same_device_score': 50, # Баллы за same device
|
||
'same_activity_score': 30, # Баллы за схожую активность
|
||
'same_pattern_score': 20, # Баллы за схожие паттерны
|
||
}
|
||
|
||
async def analyze_all_users(self) -> List[Dict]:
|
||
"""
|
||
Проанализировать всех пользователей на мультиаккаунты
|
||
Returns: список пар/групп подозрительных аккаунтов
|
||
"""
|
||
from database.models import User
|
||
|
||
# Получаем всех активных пользователей
|
||
stmt = select(User).where(User.is_banned == False)
|
||
result = await self.session.execute(stmt)
|
||
users = list(result.scalars().all())
|
||
|
||
if len(users) < 2:
|
||
return []
|
||
|
||
suspicious_pairs = []
|
||
|
||
# Сравниваем каждого с каждым
|
||
for i in range(len(users)):
|
||
for j in range(i + 1, len(users)):
|
||
user1 = users[i]
|
||
user2 = users[j]
|
||
|
||
# Пропускаем если уже в одной квартире
|
||
if user1.apartment and user2.apartment:
|
||
if user1.apartment == user2.apartment:
|
||
continue
|
||
|
||
score, reasons = await self._compare_users(user1, user2)
|
||
|
||
if score >= 50: # Порог подозрительности
|
||
suspicious_pairs.append({
|
||
'user1_id': user1.user_id,
|
||
'user1_name': user1.full_name,
|
||
'user1_apartment': user1.apartment,
|
||
'user2_id': user2.user_id,
|
||
'user2_name': user2.full_name,
|
||
'user2_apartment': user2.apartment,
|
||
'score': score,
|
||
'reasons': reasons,
|
||
})
|
||
|
||
return suspicious_pairs
|
||
|
||
async def _compare_users(self, user1, user2) -> Tuple[int, List[str]]:
|
||
"""
|
||
Сравнить двух пользователей
|
||
Returns: (score, reasons)
|
||
"""
|
||
score = 0
|
||
reasons = []
|
||
|
||
# 1. Проверка активности по времени
|
||
activity_similarity = await self._check_activity_similarity(user1.user_id, user2.user_id)
|
||
if activity_similarity >= self.config['activity_similarity_threshold']:
|
||
score += self.config['same_activity_score']
|
||
reasons.append(f'Схожая активность ({activity_similarity:.0%})')
|
||
|
||
# 2. Проверка паттернов сообщений
|
||
pattern_similarity = await self._check_message_patterns(user1.user_id, user2.user_id)
|
||
if pattern_similarity >= self.config['message_pattern_threshold']:
|
||
score += self.config['same_pattern_score']
|
||
reasons.append(f'Схожие паттерны сообщений ({pattern_similarity:.0%})')
|
||
|
||
# 3. Проверка device ID (если доступно)
|
||
# Telegram не отдаёт device ID через Bot API, но можно проверять по другим признакам
|
||
|
||
# 4. Проверка времени регистрации
|
||
if user1.join_date and user2.join_date:
|
||
delta = abs((user1.join_date - user2.join_date).total_seconds())
|
||
if delta < 3600: # Зарегистрировались в течение часа
|
||
score += 10
|
||
reasons.append('Зарегистрировались в течение часа')
|
||
|
||
# 5. Проверка IP (если бы было доступно)
|
||
# В Telegram ботах IP недоступен
|
||
|
||
return score, reasons
|
||
|
||
async def _check_activity_similarity(self, user1_id: int, user2_id: int) -> float:
|
||
"""
|
||
Проверить схожесть активности по времени
|
||
Returns: коэффициент схожести 0-1
|
||
"""
|
||
from database.models import Message
|
||
|
||
# Получаем временное окно
|
||
time_window = datetime.utcnow() - timedelta(hours=self.config['time_window_hours'])
|
||
|
||
# Активность по часам для user1
|
||
stmt = select(func.strftime('%H', Message.timestamp)).where(
|
||
Message.user_id == user1_id,
|
||
Message.timestamp >= time_window
|
||
)
|
||
result = await self.session.execute(stmt)
|
||
user1_hours = [row[0] for row in result.all()]
|
||
|
||
# Активность по часам для user2
|
||
stmt = select(func.strftime('%H', Message.timestamp)).where(
|
||
Message.user_id == user2_id,
|
||
Message.timestamp >= time_window
|
||
)
|
||
result = await self.session.execute(stmt)
|
||
user2_hours = [row[0] for row in result.all()]
|
||
|
||
if not user1_hours or not user2_hours:
|
||
return 0.0
|
||
|
||
# Считаем активность по часам
|
||
user1_activity = defaultdict(int)
|
||
user2_activity = defaultdict(int)
|
||
|
||
for hour in user1_hours:
|
||
user1_activity[hour] += 1
|
||
|
||
for hour in user2_hours:
|
||
user2_activity[hour] += 1
|
||
|
||
# Сравниваем распределение
|
||
common_hours = set(user1_activity.keys()) & set(user2_activity.keys())
|
||
if not common_hours:
|
||
return 0.0
|
||
|
||
# Коэффициент схожести
|
||
similarity = len(common_hours) / 24 # Доля общих часов активности
|
||
|
||
# Дополнительно проверяем пики активности
|
||
user1_peak = max(user1_activity.keys(), key=lambda h: user1_activity[h]) if user1_activity else None
|
||
user2_peak = max(user2_activity.keys(), key=lambda h: user2_activity[h]) if user2_activity else None
|
||
|
||
if user1_peak and user2_peak and abs(int(user1_peak) - int(user2_peak)) <= 1:
|
||
similarity += 0.2 # Бонус за схожий пик активности
|
||
|
||
return min(similarity, 1.0)
|
||
|
||
async def _check_message_patterns(self, user1_id: int, user2_id: int) -> float:
|
||
"""
|
||
Проверить схожесть паттернов сообщений
|
||
Returns: коэффициент схожести 0-1
|
||
"""
|
||
from database.models import Message
|
||
|
||
# Получаем последние сообщения
|
||
time_window = datetime.utcnow() - timedelta(hours=self.config['time_window_hours'])
|
||
|
||
stmt = select(Message.text).where(
|
||
Message.user_id == user1_id,
|
||
Message.timestamp >= time_window,
|
||
Message.text.isnot(None)
|
||
).limit(50)
|
||
result = await self.session.execute(stmt)
|
||
user1_messages = [row[0] for row in result.all()]
|
||
|
||
stmt = select(Message.text).where(
|
||
Message.user_id == user2_id,
|
||
Message.timestamp >= time_window,
|
||
Message.text.isnot(None)
|
||
).limit(50)
|
||
result = await self.session.execute(stmt)
|
||
user2_messages = [row[0] for row in result.all()]
|
||
|
||
if not user1_messages or not user2_messages:
|
||
return 0.0
|
||
|
||
# Анализируем паттерны
|
||
user1_patterns = self._extract_patterns(user1_messages)
|
||
user2_patterns = self._extract_patterns(user2_messages)
|
||
|
||
# Сравниваем паттерны
|
||
common_patterns = set(user1_patterns.keys()) & set(user2_patterns.keys())
|
||
if not common_patterns:
|
||
return 0.0
|
||
|
||
# Коэффициент схожести паттернов
|
||
similarity = len(common_patterns) / max(len(user1_patterns), len(user2_patterns))
|
||
|
||
return min(similarity, 1.0)
|
||
|
||
def _extract_patterns(self, messages: List[str]) -> Dict:
|
||
"""
|
||
Извлечь паттерны из сообщений
|
||
Returns: dict с паттернами
|
||
"""
|
||
patterns = defaultdict(int)
|
||
|
||
for msg in messages:
|
||
msg = msg.lower().strip()
|
||
|
||
# Длина сообщения
|
||
if len(msg) < 10:
|
||
patterns['short'] += 1
|
||
elif len(msg) < 50:
|
||
patterns['medium'] += 1
|
||
else:
|
||
patterns['long'] += 1
|
||
|
||
# Наличие ссылок
|
||
if 'http' in msg or 'www' in msg or 't.me' in msg:
|
||
patterns['with_links'] += 1
|
||
|
||
# Наличие emoji
|
||
if any(ord(c) > 12700 for c in msg):
|
||
patterns['with_emoji'] += 1
|
||
|
||
# Вопросы
|
||
if '?' in msg:
|
||
patterns['questions'] += 1
|
||
|
||
# Приветствия
|
||
if msg.startswith(('привет', 'здравствуй', 'добрый')):
|
||
patterns['greetings'] += 1
|
||
|
||
# Мат (уже есть в antimat)
|
||
# Можно добавить проверку
|
||
|
||
return patterns
|
||
|
||
async def check_apartment(self, apartment: str) -> List[Dict]:
|
||
"""
|
||
Проверить квартиру на мультиаккаунты
|
||
Returns: список аккаунтов в квартире
|
||
"""
|
||
from database.models import User
|
||
|
||
stmt = select(User).where(
|
||
User.apartment == apartment,
|
||
User.is_banned == False
|
||
)
|
||
result = await self.session.execute(stmt)
|
||
users = list(result.scalars().all())
|
||
|
||
if len(users) <= 1:
|
||
return []
|
||
|
||
# Проверяем пары на подозрительность
|
||
suspicious = []
|
||
for i in range(len(users)):
|
||
for j in range(i + 1, len(users)):
|
||
score, reasons = await self._compare_users(users[i], users[j])
|
||
if score >= 50:
|
||
suspicious.append({
|
||
'user1': {'id': users[i].user_id, 'name': users[i].full_name},
|
||
'user2': {'id': users[j].user_id, 'name': users[j].full_name},
|
||
'score': score,
|
||
'reasons': reasons,
|
||
})
|
||
|
||
return suspicious
|
||
|
||
def get_config(self) -> Dict:
|
||
"""Получить конфигурацию"""
|
||
return self.config
|