#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Экспорт данных AI-КАЛОРИЯ бота в CSV формат Для анализа в Excel, Google Sheets и других программах """ import csv import logging from datetime import datetime from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from database import FoodRecord from config import Config logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def export_all_records_to_csv(filename=None): """Экспорт всех записей в CSV""" if filename is None: filename = f"food_records_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.csv" try: # Подключение к БД db_config = Config.get_database_config() engine = create_engine(db_config['url']) SessionLocal = sessionmaker(bind=engine) session = SessionLocal() # Получение всех записей records = session.query(FoodRecord).order_by(FoodRecord.created_at.desc()).all() if not records: logger.warning("Нет записей для экспорта") return None # Запись в CSV with open(filename, 'w', newline='', encoding='utf-8-sig') as csvfile: fieldnames = ['ID', 'User ID', 'Username', 'Food Name', 'Amount', 'Unit', 'Calories', 'Created At', 'Date', 'Time', 'Raw Text'] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() for record in records: writer.writerow({ 'ID': record.id, 'User ID': record.user_id, 'Username': record.username or f"user_{record.user_id}", 'Food Name': record.food_name, 'Amount': record.amount, 'Unit': record.unit, 'Calories': record.calories, 'Created At': record.created_at.strftime('%Y-%m-%d %H:%M:%S'), 'Date': record.created_at.strftime('%Y-%m-%d'), 'Time': record.created_at.strftime('%H:%M:%S'), 'Raw Text': record.raw_text or '' }) session.close() logger.info(f"✅ Экспортировано {len(records)} записей в {filename}") print(f"\n✅ Успешно экспортировано {len(records)} записей") print(f"📄 Файл: {filename}") return filename except Exception as e: logger.error(f"Ошибка экспорта: {e}") print(f"\n❌ Ошибка: {e}") return None def export_user_records_to_csv(user_id, filename=None): """Экспорт записей конкретного пользователя в CSV""" if filename is None: filename = f"user_{user_id}_records_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.csv" try: # Подключение к БД db_config = Config.get_database_config() engine = create_engine(db_config['url']) SessionLocal = sessionmaker(bind=engine) session = SessionLocal() # Получение записей пользователя records = session.query(FoodRecord).filter( FoodRecord.user_id == user_id ).order_by(FoodRecord.created_at.desc()).all() if not records: logger.warning(f"Нет записей для пользователя {user_id}") print(f"\n⚠️ Нет записей для пользователя {user_id}") return None # Запись в CSV with open(filename, 'w', newline='', encoding='utf-8-sig') as csvfile: fieldnames = ['ID', 'Food Name', 'Amount', 'Unit', 'Calories', 'Date', 'Time', 'Raw Text'] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() for record in records: writer.writerow({ 'ID': record.id, 'Food Name': record.food_name, 'Amount': record.amount, 'Unit': record.unit, 'Calories': record.calories, 'Date': record.created_at.strftime('%Y-%m-%d'), 'Time': record.created_at.strftime('%H:%M:%S'), 'Raw Text': record.raw_text or '' }) session.close() logger.info(f"✅ Экспортировано {len(records)} записей пользователя {user_id} в {filename}") print(f"\n✅ Успешно экспортировано {len(records)} записей пользователя {user_id}") print(f"📄 Файл: {filename}") return filename except Exception as e: logger.error(f"Ошибка экспорта: {e}") print(f"\n❌ Ошибка: {e}") return None def export_daily_summary_to_csv(filename=None): """Экспорт дневной сводки (калории по дням для каждого пользователя)""" if filename is None: filename = f"daily_summary_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.csv" try: # Подключение к БД db_config = Config.get_database_config() engine = create_engine(db_config['url']) SessionLocal = sessionmaker(bind=engine) session = SessionLocal() # Получение всех записей records = session.query(FoodRecord).order_by( FoodRecord.user_id, FoodRecord.created_at ).all() if not records: logger.warning("Нет записей для экспорта") return None # Группировка по пользователям и дням daily_data = {} for record in records: date = record.created_at.date() key = (record.user_id, record.username or f"user_{record.user_id}", date) if key not in daily_data: daily_data[key] = { 'records': 0, 'calories': 0 } daily_data[key]['records'] += 1 daily_data[key]['calories'] += record.calories # Запись в CSV with open(filename, 'w', newline='', encoding='utf-8-sig') as csvfile: fieldnames = ['User ID', 'Username', 'Date', 'Records Count', 'Total Calories'] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() for (user_id, username, date), data in sorted(daily_data.items()): writer.writerow({ 'User ID': user_id, 'Username': username, 'Date': date.strftime('%Y-%m-%d'), 'Records Count': data['records'], 'Total Calories': round(data['calories'], 2) }) session.close() logger.info(f"✅ Экспортирована дневная сводка в {filename}") print(f"\n✅ Успешно экспортирована дневная сводка") print(f"📄 Файл: {filename}") return filename except Exception as e: logger.error(f"Ошибка экспорта: {e}") print(f"\n❌ Ошибка: {e}") return None def main(): """Главная функция""" import argparse parser = argparse.ArgumentParser(description='Экспорт данных AI-КАЛОРИЯ в CSV') parser.add_argument('--all', action='store_true', help='Экспорт всех записей') parser.add_argument('--user', type=int, help='Экспорт записей конкретного пользователя (User ID)') parser.add_argument('--daily', action='store_true', help='Экспорт дневной сводки') parser.add_argument('--output', type=str, help='Имя выходного файла') args = parser.parse_args() if not (args.all or args.user or args.daily): print("Использование:") print(" python3 export_to_csv.py --all # Все записи") print(" python3 export_to_csv.py --user 23916354 # Записи пользователя") print(" python3 export_to_csv.py --daily # Дневная сводка") print(" python3 export_to_csv.py --all --output data.csv # С указанием имени файла") return 1 try: if args.all: export_all_records_to_csv(args.output) elif args.user: export_user_records_to_csv(args.user, args.output) elif args.daily: export_daily_summary_to_csv(args.output) return 0 except Exception as e: logger.error(f"Ошибка: {e}") print(f"\n❌ Ошибка: {e}") return 1 if __name__ == '__main__': exit(main())