""" Архивация старых данных: 1. Ответы старше 30 дней → user_learning_answers_archive (агрегаты) 2. Данные архива старше 1 года → JSON файлы на диске 3. Удаление из основных таблиц Запуск: python3 archive_old_data.py Или через cron: 0 3 * * * cd /root/filwordspl/backend && source venv/bin/activate && python3 archive_old_data.py """ import asyncio import asyncpg import json import os import logging from datetime import datetime, timedelta, timezone from decimal import Decimal from pathlib import Path from config import DB_CONFIG logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # Настройки HOT_DAYS = 30 # Сколько дней хранить в основной таблице ARCHIVE_DIR = Path("/root/filwordspl/backend/archive") # Директория для JSON файлов def serialize_value(value): """Преобразует Decimal и datetime в JSON-совместимые типы.""" if value is None: return None if hasattr(value, 'isoformat'): return value.isoformat() if isinstance(value, Decimal): return float(value) if isinstance(value, (int, float, bool, str)): return value return str(value) async def archive_learning_answers(conn): """Переносит ответы старше 30 дней в архивную таблицу.""" # Переносим в архив moved = await conn.execute(""" INSERT INTO user_learning_answers_archive (user_id, skill_tag, correct, wrong_because, created_at) SELECT user_id, skill_tag, correct, wrong_because, created_at::date FROM user_learning_answers WHERE created_at < NOW() - INTERVAL '30 days' """) # Удаляем из основной таблицы deleted = await conn.execute(""" DELETE FROM user_learning_answers WHERE created_at < NOW() - INTERVAL '30 days' """) logger.info(f"Архивация ответов: перенесено={moved}, удалено={deleted}") return moved, deleted async def export_archive_to_json(conn): """Экспортирует данные архива старше 1 года в JSON файлы.""" # Получаем данные старше 1 года old_records = await conn.fetch(""" SELECT * FROM user_learning_answers_archive WHERE created_at < NOW() - INTERVAL '1 year' """) if not old_records: logger.info("Нет данных старше 1 года для экспорта") return 0 # Группируем по году и пользователю grouped = {} for record in old_records: year = record["created_at"].year if record["created_at"] else datetime.now().year user_id = record["user_id"] if year not in grouped: grouped[year] = {} if user_id not in grouped[year]: grouped[year][user_id] = [] grouped[year][user_id].append({ "id": record["id"], "user_id": record["user_id"], "skill_tag": record["skill_tag"], "correct": record["correct"], "wrong_because": record["wrong_because"], "created_at": serialize_value(record["created_at"]), "archived_at": serialize_value(record["archived_at"]), }) # Сохраняем в JSON файлы total_saved = 0 for year, users in grouped.items(): year_dir = ARCHIVE_DIR / str(year) year_dir.mkdir(parents=True, exist_ok=True) for user_id, records in users.items(): file_path = year_dir / f"user_{user_id}_answers.json" # Если файл существует — дополняем existing_data = [] if file_path.exists(): try: with open(file_path, 'r', encoding='utf-8') as f: existing_data = json.load(f) except: existing_data = [] all_data = existing_data + records with open(file_path, 'w', encoding='utf-8') as f: json.dump(all_data, f, ensure_ascii=False, indent=2) total_saved += len(records) # Удаляем экспортированные записи из архива if old_records: ids = [r["id"] for r in old_records] await conn.execute(""" DELETE FROM user_learning_answers_archive WHERE id = ANY($1) """, ids) logger.info(f"Экспортировано в JSON: {total_saved} записей") return total_saved async def archive_chat_history(conn): """Архивирует старую историю чата (старше 6 месяцев).""" # Экспортируем старую историю в JSON old_messages = await conn.fetch(""" SELECT * FROM chat_history WHERE created_at < NOW() - INTERVAL '6 months' """) if old_messages: # Группируем по году и пользователю grouped = {} for msg in old_messages: year = msg["created_at"].year if msg["created_at"] else datetime.now().year user_id = msg["user_id"] if year not in grouped: grouped[year] = {} if user_id not in grouped[year]: grouped[year][user_id] = [] grouped[year][user_id].append({ "id": msg["id"], "user_id": msg["user_id"], "message": msg["message"], "is_user": msg["is_user"], "created_at": serialize_value(msg["created_at"]), }) # Сохраняем в JSON total_saved = 0 for year, users in grouped.items(): year_dir = ARCHIVE_DIR / str(year) year_dir.mkdir(parents=True, exist_ok=True) for user_id, messages in users.items(): file_path = year_dir / f"user_{user_id}_chat.json" existing_data = [] if file_path.exists(): try: with open(file_path, 'r', encoding='utf-8') as f: existing_data = json.load(f) except: existing_data = [] all_data = existing_data + messages with open(file_path, 'w', encoding='utf-8') as f: json.dump(all_data, f, ensure_ascii=False, indent=2) total_saved += len(messages) # Удаляем старые сообщения await conn.execute(""" DELETE FROM chat_history WHERE created_at < NOW() - INTERVAL '6 months' """) logger.info(f"Архивация чата: {total_saved} сообщений в JSON") return total_saved logger.info("Нет старой истории чата для архивации") return 0 async def main(): logger.info("=== Начало архивации ===") conn = await asyncpg.connect(**DB_CONFIG) try: # 1. Архивация ответов moved, deleted = await archive_learning_answers(conn) # 2. Экспорт архива в JSON exported = await export_archive_to_json(conn) # 3. Архивация чата chat_archived = await archive_chat_history(conn) logger.info(f"=== Архивация завершена: ответы={moved}, JSON={exported}, чат={chat_archived} ===") except Exception as e: logger.error(f"Ошибка архивации: {str(e)}") finally: await conn.close() if __name__ == "__main__": asyncio.run(main())