#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Скрипт для получения аналитики использования AI-КАЛОРИЯ бота """ import logging from datetime import datetime, timedelta from typing import Dict, Any, List from collections import defaultdict from sqlalchemy import create_engine, func from sqlalchemy.orm import sessionmaker import json from database import Base, FoodRecord from config import Config # Настройка логирования logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class BotAnalytics: """Класс для получения аналитики бота""" def __init__(self): """Инициализация аналитики""" db_config = Config.get_database_config() self.engine = create_engine(db_config['url']) self.SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=self.engine) def get_session(self): """Получение сессии базы данных""" return self.SessionLocal() def get_general_stats(self) -> Dict[str, Any]: """Получение общей статистики""" session = self.get_session() try: # Общее количество пользователей total_users = session.query(func.count(func.distinct(FoodRecord.user_id))).scalar() # Общее количество записей total_records = session.query(func.count(FoodRecord.id)).scalar() # Общая сумма калорий total_calories = session.query(func.sum(FoodRecord.calories)).scalar() or 0 # Первая и последняя записи first_record = session.query(FoodRecord).order_by(FoodRecord.created_at.asc()).first() last_record = session.query(FoodRecord).order_by(FoodRecord.created_at.desc()).first() # Средняя активность avg_records_per_user = total_records / total_users if total_users > 0 else 0 avg_calories_per_record = total_calories / total_records if total_records > 0 else 0 return { 'total_users': total_users, 'total_records': total_records, 'total_calories': float(total_calories), 'avg_records_per_user': round(avg_records_per_user, 2), 'avg_calories_per_record': round(avg_calories_per_record, 2), 'first_record_date': first_record.created_at.strftime('%d.%m.%Y %H:%M') if first_record else None, 'last_record_date': last_record.created_at.strftime('%d.%m.%Y %H:%M') if last_record else None, } finally: session.close() def get_top_users(self, limit: int = 10) -> List[Dict[str, Any]]: """Получение топ пользователей по активности""" session = self.get_session() try: users = session.query( FoodRecord.user_id, FoodRecord.username, func.count(FoodRecord.id).label('records_count'), func.sum(FoodRecord.calories).label('total_calories'), func.min(FoodRecord.created_at).label('first_record'), func.max(FoodRecord.created_at).label('last_record') ).group_by( FoodRecord.user_id, FoodRecord.username ).order_by( func.count(FoodRecord.id).desc() ).limit(limit).all() return [ { 'user_id': user.user_id, 'username': user.username or f"user_{user.user_id}", 'records_count': user.records_count, 'total_calories': float(user.total_calories) if user.total_calories else 0, 'first_record': user.first_record.strftime('%d.%m.%Y'), 'last_record': user.last_record.strftime('%d.%m.%Y'), 'days_active': (user.last_record - user.first_record).days + 1 } for user in users ] finally: session.close() def get_top_foods(self, limit: int = 20) -> List[Dict[str, Any]]: """Получение топ продуктов""" session = self.get_session() try: foods = session.query( FoodRecord.food_name, func.count(FoodRecord.id).label('count'), func.avg(FoodRecord.calories).label('avg_calories') ).group_by( FoodRecord.food_name ).order_by( func.count(FoodRecord.id).desc() ).limit(limit).all() return [ { 'food_name': food.food_name, 'count': food.count, 'avg_calories': round(float(food.avg_calories), 2) if food.avg_calories else 0 } for food in foods ] finally: session.close() def get_daily_activity(self, days: int = 30) -> List[Dict[str, Any]]: """Получение активности по дням""" session = self.get_session() try: start_date = datetime.utcnow() - timedelta(days=days) records = session.query(FoodRecord).filter( FoodRecord.created_at >= start_date ).all() # Группируем по дням daily_stats = defaultdict(lambda: {'records': 0, 'calories': 0, 'users': set()}) for record in records: date_key = record.created_at.date().strftime('%Y-%m-%d') daily_stats[date_key]['records'] += 1 daily_stats[date_key]['calories'] += record.calories daily_stats[date_key]['users'].add(record.user_id) # Формируем результат result = [] for i in range(days): date = (datetime.utcnow() - timedelta(days=days-i-1)).date() date_key = date.strftime('%Y-%m-%d') stats = daily_stats.get(date_key, {'records': 0, 'calories': 0, 'users': set()}) result.append({ 'date': date.strftime('%d.%m.%Y'), 'records_count': stats['records'], 'total_calories': round(float(stats['calories']), 2), 'active_users': len(stats['users']) }) return result finally: session.close() def get_hourly_activity(self) -> List[Dict[str, Any]]: """Получение активности по часам""" session = self.get_session() try: records = session.query(FoodRecord).all() # Группируем по часам hourly_stats = defaultdict(int) for record in records: hour = record.created_at.hour hourly_stats[hour] += 1 # Формируем результат result = [] for hour in range(24): result.append({ 'hour': f"{hour:02d}:00", 'records_count': hourly_stats.get(hour, 0) }) return result finally: session.close() def get_user_retention(self) -> Dict[str, Any]: """Получение статистики по удержанию пользователей""" session = self.get_session() try: users = session.query( FoodRecord.user_id, func.min(FoodRecord.created_at).label('first_record'), func.max(FoodRecord.created_at).label('last_record'), func.count(func.distinct(func.date(FoodRecord.created_at))).label('active_days') ).group_by(FoodRecord.user_id).all() if not users: return { 'total_users': 0, 'returning_users': 0, 'one_time_users': 0, 'avg_active_days': 0 } returning_users = sum(1 for u in users if (u.last_record - u.first_record).days > 0) one_time_users = len(users) - returning_users avg_active_days = sum(u.active_days for u in users) / len(users) return { 'total_users': len(users), 'returning_users': returning_users, 'one_time_users': one_time_users, 'avg_active_days': round(avg_active_days, 2), 'returning_rate': round((returning_users / len(users) * 100), 2) if len(users) > 0 else 0 } finally: session.close() def get_full_analytics(self, days: int = 30) -> Dict[str, Any]: """Получение полной аналитики""" logger.info("Сбор аналитики...") analytics = { 'generated_at': datetime.utcnow().strftime('%d.%m.%Y %H:%M:%S'), 'period_days': days, 'general_stats': self.get_general_stats(), 'user_retention': self.get_user_retention(), 'top_users': self.get_top_users(10), 'top_foods': self.get_top_foods(20), 'daily_activity': self.get_daily_activity(days), 'hourly_activity': self.get_hourly_activity() } logger.info("Аналитика собрана успешно") return analytics def print_analytics(self, days: int = 30): """Вывод аналитики в консоль""" analytics = self.get_full_analytics(days) print("\n" + "="*80) print("📊 АНАЛИТИКА AI-КАЛОРИЯ БОТА") print("="*80) print(f"Дата генерации: {analytics['generated_at']}") print(f"Период анализа: {analytics['period_days']} дней") # Общая статистика print("\n" + "-"*80) print("📈 ОБЩАЯ СТАТИСТИКА") print("-"*80) stats = analytics['general_stats'] print(f"👥 Всего пользователей: {stats['total_users']}") print(f"📝 Всего записей: {stats['total_records']}") print(f"🔥 Всего калорий: {stats['total_calories']:,.0f} ккал") print(f"📊 Средние записи на пользователя: {stats['avg_records_per_user']}") print(f"🍽️ Средние калории на запись: {stats['avg_calories_per_record']:.0f} ккал") if stats['first_record_date']: print(f"📅 Первая запись: {stats['first_record_date']}") if stats['last_record_date']: print(f"📅 Последняя запись: {stats['last_record_date']}") # Удержание пользователей print("\n" + "-"*80) print("🔄 УДЕРЖАНИЕ ПОЛЬЗОВАТЕЛЕЙ") print("-"*80) retention = analytics['user_retention'] print(f"👥 Всего пользователей: {retention['total_users']}") print(f"🔁 Возвращающихся: {retention['returning_users']} ({retention['returning_rate']:.1f}%)") print(f"1️⃣ Разовых: {retention['one_time_users']}") print(f"📅 Среднее активных дней: {retention['avg_active_days']}") # Топ пользователей print("\n" + "-"*80) print("🏆 ТОП-10 АКТИВНЫХ ПОЛЬЗОВАТЕЛЕЙ") print("-"*80) for i, user in enumerate(analytics['top_users'], 1): print(f"{i}. {user['username']} (ID: {user['user_id']})") print(f" 📝 Записей: {user['records_count']} | 🔥 Калорий: {user['total_calories']:,.0f} | " f"📅 Дней активности: {user['days_active']}") # Топ продуктов print("\n" + "-"*80) print("🍽️ ТОП-20 ПРОДУКТОВ") print("-"*80) for i, food in enumerate(analytics['top_foods'], 1): print(f"{i}. {food['food_name']}: {food['count']} раз (сред. {food['avg_calories']:.0f} ккал)") # Активность по часам print("\n" + "-"*80) print("⏰ АКТИВНОСТЬ ПО ЧАСАМ") print("-"*80) hourly = analytics['hourly_activity'] max_records = max(h['records_count'] for h in hourly) if hourly else 1 for hour_data in hourly: bar_length = int((hour_data['records_count'] / max_records) * 40) if max_records > 0 else 0 bar = '█' * bar_length print(f"{hour_data['hour']}: {bar} {hour_data['records_count']}") # Активность по дням (последние 7 дней) print("\n" + "-"*80) print("📅 АКТИВНОСТЬ ПО ДНЯМ (последние 7 дней)") print("-"*80) recent_days = analytics['daily_activity'][-7:] for day in recent_days: print(f"{day['date']}: 📝 {day['records_count']} записей | " f"👥 {day['active_users']} пользователей | " f"🔥 {day['total_calories']:,.0f} ккал") print("\n" + "="*80) def save_analytics_to_file(self, filename: str = None, days: int = 30): """Сохранение аналитики в файл JSON""" if filename is None: filename = f"analytics_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.json" analytics = self.get_full_analytics(days) with open(filename, 'w', encoding='utf-8') as f: json.dump(analytics, f, ensure_ascii=False, indent=2) logger.info(f"Аналитика сохранена в файл: {filename}") print(f"\n✅ Аналитика сохранена в файл: {filename}") return filename def main(): """Главная функция""" import argparse parser = argparse.ArgumentParser(description='Аналитика AI-КАЛОРИЯ бота') parser.add_argument('--days', type=int, default=30, help='Количество дней для анализа (по умолчанию: 30)') parser.add_argument('--save', action='store_true', help='Сохранить аналитику в JSON файл') parser.add_argument('--output', type=str, help='Имя файла для сохранения (если --save)') args = parser.parse_args() try: analytics = BotAnalytics() analytics.print_analytics(days=args.days) if args.save: analytics.save_analytics_to_file(filename=args.output, days=args.days) except Exception as e: logger.error(f"Ошибка при получении аналитики: {e}") print(f"\n❌ Ошибка: {e}") return 1 return 0 if __name__ == '__main__': exit(main())