ba4676b0af
✅ Регистрация и верификация жильцов ✅ Рейтинг активности (уровни) ✅ Детект шпионов (Score 0-100) ✅ Анти-мат система (3 предупреждения → бан) ✅ Админ-панель с рассылкой ✅ Учёт квартир (несколько жильцов) ✅ Телефоны экстренных служб и мастеров ✅ Еженедельный экспорт JSON ✅ Прокси (socks5) для обхода РКН Бекап: backups/versions/v1.0_mvp_2026-02-26/ Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
299 lines
12 KiB
Python
299 lines
12 KiB
Python
"""
|
||
Achievements Service — система достижений пользователей
|
||
"""
|
||
import logging
|
||
import json
|
||
from datetime import datetime
|
||
from typing import Dict, List, Optional, Tuple
|
||
from sqlalchemy import select, func
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class AchievementsService:
|
||
"""Сервис достижений"""
|
||
|
||
def __init__(self, session: AsyncSession):
|
||
self.session = session
|
||
self.achievements = self._get_achievements_config()
|
||
|
||
def _get_achievements_config(self) -> Dict:
|
||
"""Конфигурация достижений"""
|
||
return {
|
||
'newbie': {
|
||
'name': 'Новичок',
|
||
'emoji': '🌱',
|
||
'description': 'Первое сообщение в чате',
|
||
'category': 'activity'
|
||
},
|
||
'active_10': {
|
||
'name': 'Активист 10',
|
||
'emoji': '📝',
|
||
'description': '10 сообщений в чате',
|
||
'category': 'activity'
|
||
},
|
||
'active_100': {
|
||
'name': 'Активист 100',
|
||
'emoji': '✍️',
|
||
'description': '100 сообщений в чате',
|
||
'category': 'activity'
|
||
},
|
||
'active_1000': {
|
||
'name': 'Говорун 1000',
|
||
'emoji': '🗣️',
|
||
'description': '1000 сообщений в чате',
|
||
'category': 'activity'
|
||
},
|
||
'verified': {
|
||
'name': 'Свой человек',
|
||
'emoji': '✅',
|
||
'description': 'Пройдена верификация',
|
||
'category': 'verification'
|
||
},
|
||
'rating_100': {
|
||
'name': 'Сосед',
|
||
'emoji': '🏠',
|
||
'description': 'Рейтинг 100 баллов',
|
||
'category': 'rating'
|
||
},
|
||
'rating_500': {
|
||
'name': 'Активист',
|
||
'emoji': '⭐',
|
||
'description': 'Рейтинг 500 баллов',
|
||
'category': 'rating'
|
||
},
|
||
'rating_1000': {
|
||
'name': 'Лидер',
|
||
'emoji': '👑',
|
||
'description': 'Рейтинг 1000 баллов',
|
||
'category': 'rating'
|
||
},
|
||
'early_bird': {
|
||
'name': 'Жаворонок',
|
||
'emoji': '🐦',
|
||
'description': 'Сообщение в 6-7 утра',
|
||
'category': 'time'
|
||
},
|
||
'night_owl': {
|
||
'name': 'Сова',
|
||
'emoji': '🦉',
|
||
'description': 'Сообщение после 23:00',
|
||
'category': 'time'
|
||
},
|
||
'helper': {
|
||
'name': 'Помощник',
|
||
'emoji': '🤝',
|
||
'description': 'Получил благодарность',
|
||
'category': 'social'
|
||
},
|
||
'spy_hunter': {
|
||
'name': 'Охотник на шпионов',
|
||
'emoji': '🕵️',
|
||
'description': 'Сообщение о подозрительном поведении',
|
||
'category': 'security'
|
||
},
|
||
'week_warrior': {
|
||
'name': 'Недельный воин',
|
||
'emoji': '📅',
|
||
'description': 'Активность 7 дней подряд',
|
||
'category': 'streak'
|
||
},
|
||
'month_master': {
|
||
'name': 'Месячный мастер',
|
||
'emoji': '🗓️',
|
||
'description': 'Активность 30 дней подряд',
|
||
'category': 'streak'
|
||
},
|
||
'poll_participant': {
|
||
'name': 'Участник опроса',
|
||
'emoji': '📊',
|
||
'description': 'Участие в опросе',
|
||
'category': 'polls'
|
||
},
|
||
'ad_publisher': {
|
||
'name': 'Объявитель',
|
||
'emoji': '📢',
|
||
'description': 'Публикация объявления',
|
||
'category': 'ads'
|
||
},
|
||
'helper': {
|
||
'name': 'Помощник',
|
||
'emoji': '🤝',
|
||
'description': 'Получил благодарность',
|
||
'category': 'social'
|
||
},
|
||
}
|
||
|
||
async def check_and_award(self, user_id: int, **kwargs) -> List[str]:
|
||
"""
|
||
Проверить и выдать достижения
|
||
Returns: список полученных достижений
|
||
"""
|
||
from database.models import Achievement, User
|
||
|
||
awarded = []
|
||
user = await self.session.get(User, user_id)
|
||
|
||
if not user:
|
||
return []
|
||
|
||
# Получаем текущие достижения пользователя
|
||
stmt = select(Achievement.achievement_type).where(Achievement.user_id == user_id)
|
||
result = await self.session.execute(stmt)
|
||
existing = set(row[0] for row in result.all())
|
||
|
||
# Проверка по сообщениям
|
||
if 'message_count' in kwargs:
|
||
msg_count = kwargs['message_count']
|
||
|
||
if msg_count >= 1 and 'newbie' not in existing:
|
||
await self._award(user_id, 'newbie')
|
||
awarded.append('newbie')
|
||
|
||
if msg_count >= 10 and 'active_10' not in existing:
|
||
await self._award(user_id, 'active_10')
|
||
awarded.append('active_10')
|
||
|
||
if msg_count >= 100 and 'active_100' not in existing:
|
||
await self._award(user_id, 'active_100')
|
||
awarded.append('active_100')
|
||
|
||
if msg_count >= 1000 and 'active_1000' not in existing:
|
||
await self._award(user_id, 'active_1000')
|
||
awarded.append('active_1000')
|
||
|
||
# Проверка по верификации
|
||
if 'verified' in kwargs and kwargs['verified'] and 'verified' not in existing:
|
||
await self._award(user_id, 'verified')
|
||
awarded.append('verified')
|
||
|
||
# Проверка по рейтингу
|
||
if 'rating' in kwargs:
|
||
rating = kwargs['rating']
|
||
|
||
if rating >= 100 and 'rating_100' not in existing:
|
||
await self._award(user_id, 'rating_100')
|
||
awarded.append('rating_100')
|
||
|
||
if rating >= 500 and 'rating_500' not in existing:
|
||
await self._award(user_id, 'rating_500')
|
||
awarded.append('rating_500')
|
||
|
||
if rating >= 1000 and 'rating_1000' not in existing:
|
||
await self._award(user_id, 'rating_1000')
|
||
awarded.append('rating_1000')
|
||
|
||
# Проверка по времени сообщения
|
||
if 'message_hour' in kwargs:
|
||
hour = kwargs['message_hour']
|
||
|
||
if 6 <= hour < 7 and 'early_bird' not in existing:
|
||
await self._award(user_id, 'early_bird')
|
||
awarded.append('early_bird')
|
||
|
||
if hour >= 23 and 'night_owl' not in existing:
|
||
await self._award(user_id, 'night_owl')
|
||
awarded.append('night_owl')
|
||
|
||
# Проверка по участию в опросе
|
||
if 'poll_vote' in kwargs and kwargs['poll_vote'] and 'poll_participant' not in existing:
|
||
await self._award(user_id, 'poll_participant')
|
||
awarded.append('poll_participant')
|
||
|
||
# Проверка по публикации объявления
|
||
if 'ad_published' in kwargs and kwargs['ad_published'] and 'ad_publisher' not in existing:
|
||
await self._award(user_id, 'ad_publisher')
|
||
awarded.append('ad_publisher')
|
||
|
||
# Проверка по благодарности (помощник)
|
||
if 'helper' in kwargs and kwargs['helper'] and 'helper' not in existing:
|
||
await self._award(user_id, 'helper')
|
||
awarded.append('helper')
|
||
|
||
return awarded
|
||
|
||
async def _award(self, user_id: int, achievement_type: str, metadata: dict = None):
|
||
"""Выдать достижение"""
|
||
from database.models import Achievement
|
||
|
||
achievement = Achievement(
|
||
user_id=user_id,
|
||
achievement_type=achievement_type,
|
||
metadata=json.dumps(metadata) if metadata else None
|
||
)
|
||
self.session.add(achievement)
|
||
await self.session.commit()
|
||
|
||
# Уведомление пользователя
|
||
try:
|
||
from aiogram import Bot
|
||
from config import ADMIN_USER_ID
|
||
|
||
ach_data = self.achievements.get(achievement_type, {})
|
||
|
||
# Бот будет передан извне
|
||
# await bot.send_message(user_id, f"🏆 {ach_data.get('emoji', '🏆')} {ach_data.get('name', 'Достижение')}!\n\n{ach_data.get('description', '')}")
|
||
|
||
logger.info(f"Достижение {achievement_type} выдано пользователю {user_id}")
|
||
except Exception as e:
|
||
logger.error(f"Ошибка уведомления о достижении: {e}")
|
||
|
||
async def get_user_achievements(self, user_id: int) -> List[Dict]:
|
||
"""Получить достижения пользователя"""
|
||
from database.models import Achievement
|
||
|
||
stmt = select(Achievement).where(Achievement.user_id == user_id).order_by(Achievement.earned_at)
|
||
result = await self.session.execute(stmt)
|
||
achievements = list(result.scalars().all())
|
||
|
||
result = []
|
||
for ach in achievements:
|
||
ach_data = self.achievements.get(ach.achievement_type, {})
|
||
result.append({
|
||
'type': ach.achievement_type,
|
||
'name': ach_data.get('name', ach.achievement_type),
|
||
'emoji': ach_data.get('emoji', '🏆'),
|
||
'description': ach_data.get('description', ''),
|
||
'category': ach_data.get('category', 'other'),
|
||
'earned_at': ach.earned_at.isoformat()
|
||
})
|
||
|
||
return result
|
||
|
||
async def get_all_achievements(self) -> Dict:
|
||
"""Получить все доступные достижения"""
|
||
return self.achievements
|
||
|
||
async def get_leaderboard(self, limit: int = 10) -> List[Dict]:
|
||
"""Получить топ пользователей по достижениям"""
|
||
from database.models import Achievement, User
|
||
|
||
stmt = (
|
||
select(
|
||
User.user_id,
|
||
User.full_name,
|
||
User.apartment,
|
||
func.count(Achievement.id).label('achievement_count')
|
||
)
|
||
.join(Achievement, User.user_id == Achievement.user_id, isouter=True)
|
||
.group_by(User.user_id)
|
||
.order_by(func.count(Achievement.id).desc())
|
||
.limit(limit)
|
||
)
|
||
result = await self.session.execute(stmt)
|
||
rows = list(result.all())
|
||
|
||
return [
|
||
{
|
||
'user_id': row.user_id,
|
||
'name': row.full_name or f'user{row.user_id}',
|
||
'apartment': row.apartment or '?',
|
||
'achievements': row.achievement_count
|
||
}
|
||
for row in rows
|
||
]
|
||
|
||
def get_achievement_info(self, achievement_type: str) -> Dict:
|
||
"""Получить информацию о достижении"""
|
||
return self.achievements.get(achievement_type, {})
|