""" 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, {})