""" MELODIX - Script de Limpieza Automática Este script limpia archivos temporales y procesados antiguos. Se recomienda ejecutarlo periódicamente (ej. diariamente). Uso: python cleanup.py [--days 7] [--dry-run] """ import os import sys import shutil import argparse import logging from datetime import datetime, timedelta from dotenv import load_dotenv # ========================================== # CONFIGURACIÓN # ========================================== load_dotenv() # Configurar logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # ========================================== # CONFIGURACIÓN DE DIRECTORIOS # ========================================== BASE_DIR = os.getenv("BASE_DIR", os.path.dirname(os.path.abspath(__file__))) UPLOADS_DIR = os.getenv("UPLOADS_DIR", os.path.join(BASE_DIR, "uploads")) PROCESSED_DIR = os.getenv("PROCESSED_DIR", os.path.join(BASE_DIR, "processed")) LOGS_DIR = os.path.join(BASE_DIR, "logs") # ========================================== # FUNCIONES DE LIMPIEZA # ========================================== def get_directory_size(path): """Calcula el tamaño total de un directorio en MB""" total_size = 0 if os.path.exists(path): for dirpath, dirnames, filenames in os.walk(path): for f in filenames: fp = os.path.join(dirpath, f) if os.path.exists(fp): total_size += os.path.getsize(fp) return total_size / (1024 * 1024) def clean_uploads(max_age_days=1, dry_run=False): """ Limpia archivos de uploads más antiguos que X días. Los archivos de uploads deberían eliminarse después de procesarse, pero esto es un respaldo por si algo falla. """ logger.info(f"Limpiando uploads más antiguos de {max_age_days} días...") if not os.path.exists(UPLOADS_DIR): logger.info(f"Directorio uploads no existe: {UPLOADS_DIR}") return 0 cutoff = datetime.now() - timedelta(days=max_age_days) cleaned = 0 freed_mb = 0 for filename in os.listdir(UPLOADS_DIR): filepath = os.path.join(UPLOADS_DIR, filename) if os.path.isfile(filepath): mtime = datetime.fromtimestamp(os.path.getmtime(filepath)) if mtime < cutoff: size_mb = os.path.getsize(filepath) / (1024 * 1024) if dry_run: logger.info(f" [DRY-RUN] Eliminaría: {filename} ({size_mb:.2f}MB)") else: try: os.remove(filepath) logger.info(f" Eliminado: {filename} ({size_mb:.2f}MB)") except Exception as e: logger.error(f" Error eliminando {filename}: {e}") continue cleaned += 1 freed_mb += size_mb logger.info(f"Uploads limpiados: {cleaned} archivos, {freed_mb:.2f}MB liberados") return cleaned def clean_processed(max_age_days=7, dry_run=False): """ Limpia archivos procesados más antiguos que X días. Los usuarios tienen una semana para descargar sus resultados. """ logger.info(f"Limpiando procesados más antiguos de {max_age_days} días...") if not os.path.exists(PROCESSED_DIR): logger.info(f"Directorio processed no existe: {PROCESSED_DIR}") return 0 cutoff = datetime.now() - timedelta(days=max_age_days) cleaned = 0 freed_mb = 0 for task_folder in os.listdir(PROCESSED_DIR): task_path = os.path.join(PROCESSED_DIR, task_folder) if os.path.isdir(task_path): # Obtener la fecha más reciente de cualquier archivo en la carpeta latest_mtime = 0 for root, dirs, files in os.walk(task_path): for f in files: fp = os.path.join(root, f) if os.path.exists(fp): mtime = os.path.getmtime(fp) if mtime > latest_mtime: latest_mtime = mtime if latest_mtime > 0: folder_mtime = datetime.fromtimestamp(latest_mtime) if folder_mtime < cutoff: size_mb = get_directory_size(task_path) if dry_run: logger.info(f" [DRY-RUN] Eliminaría: {task_folder} ({size_mb:.2f}MB)") else: try: shutil.rmtree(task_path) logger.info(f" Eliminado: {task_folder} ({size_mb:.2f}MB)") except Exception as e: logger.error(f" Error eliminando {task_folder}: {e}") continue cleaned += 1 freed_mb += size_mb logger.info(f"Procesados limpiados: {cleaned} carpetas, {freed_mb:.2f}MB liberados") return cleaned def clean_logs(max_age_days=30, dry_run=False): """ Limpia archivos de log más antiguos que X días. El logging rotativo ya maneja el tamaño, esto es por seguridad. """ logger.info(f"Limpiando logs más antiguos de {max_age_days} días...") if not os.path.exists(LOGS_DIR): logger.info(f"Directorio logs no existe: {LOGS_DIR}") return 0 cutoff = datetime.now() - timedelta(days=max_age_days) cleaned = 0 for filename in os.listdir(LOGS_DIR): filepath = os.path.join(LOGS_DIR, filename) if os.path.isfile(filepath) and filename.endswith('.log'): mtime = datetime.fromtimestamp(os.path.getmtime(filepath)) if mtime < cutoff: size_mb = os.path.getsize(filepath) / (1024 * 1024) if dry_run: logger.info(f" [DRY-RUN] Eliminaría: {filename} ({size_mb:.2f}MB)") else: # Mantener siempre el archivo actual if filename == "melodix.log": logger.info(f" Saltando log actual: {filename}") continue try: os.remove(filepath) logger.info(f" Eliminado: {filename} ({size_mb:.2f}MB)") except Exception as e: logger.error(f" Error eliminando {filename}: {e}") continue cleaned += 1 logger.info(f"Logs limpiados: {cleaned} archivos") return cleaned def show_stats(): """Muestra estadísticas de uso de disco""" logger.info("=" * 50) logger.info("ESTADÍSTICAS DE ALMACENAMIENTO") logger.info("=" * 50) uploads_mb = get_directory_size(UPLOADS_DIR) processed_mb = get_directory_size(PROCESSED_DIR) logs_mb = get_directory_size(LOGS_DIR) logger.info(f"Uploads: {uploads_mb:>10.2f} MB") logger.info(f"Processed: {processed_mb:>10.2f} MB") logger.info(f"Logs: {logs_mb:>10.2f} MB") logger.info(f"Total: {uploads_mb + processed_mb + logs_mb:>10.2f} MB") logger.info("=" * 50) # ========================================== # EXPIRACIÓN DE ALMACENAMIENTO EN HUGGING FACE # ========================================== def clean_expired_sync_queue(max_age_days=7, dry_run=False): """ Ejecuta la limpieza de archivos expirados en Hugging Face Dataset y la purga mensual de usuarios normales si corresponde. """ logger.info("Ejecutando limpieza de archivos expirados en Hugging Face Datasets...") if dry_run: logger.info(" [DRY-RUN] Ejecutaría la tarea de limpieza en Hugging Face.") return 0 current_dir = os.path.dirname(os.path.abspath(__file__)) if current_dir not in sys.path: sys.path.insert(0, current_dir) try: from tasks import limpiar_stems_expirados_task res = limpiar_stems_expirados_task() cleaned = res.get("cleaned", 0) logger.info(f"Limpieza en Hugging Face completada. Tareas limpiadas: {cleaned}") return cleaned except Exception as e: logger.error(f"Error ejecutando limpiar_stems_expirados_task desde cleanup.py: {e}") return 0 # ========================================== # MAIN # ========================================== def main(): parser = argparse.ArgumentParser(description="Limpieza automática de Melodix") parser.add_argument("--days", type=int, default=7, help="Días de antigüedad para limpiar (default: 7)") parser.add_argument("--dry-run", action="store_true", help="Solo mostrar qué se eliminaría") parser.add_argument("--stats", action="store_true", help="Mostrar estadísticas de uso") args = parser.parse_args() logger.info("=" * 50) logger.info("MELODIX - LIMPIEZA AUTOMÁTICA") logger.info("=" * 50) logger.info(f"Fecha: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") logger.info(f"Modo: {'DRY-RUN' if args.dry_run else 'NORMAL'}") logger.info(f"Días de antigüedad: {args.days}") logger.info("=" * 50) if args.stats: show_stats() logger.info("") cleaned_uploads = clean_uploads(max_age_days=1, dry_run=args.dry_run) # Uploads: 1 día logger.info("") cleaned_processed = clean_processed(max_age_days=args.days, dry_run=args.dry_run) # Procesados: X días logger.info("") cleaned_logs = clean_logs(max_age_days=30, dry_run=args.dry_run) # Logs: 30 días logger.info("") cleaned_sync = clean_expired_sync_queue(max_age_days=args.days, dry_run=args.dry_run) # HF/Expira logger.info("") logger.info("=" * 50) logger.info("RESUMEN") logger.info("=" * 50) logger.info(f"Uploads limpiados: {cleaned_uploads}") logger.info(f"Procesados limpiados: {cleaned_processed}") logger.info(f"Logs limpiados: {cleaned_logs}") logger.info(f"HF/Nube limpiados: {cleaned_sync}") logger.info("=" * 50) if args.dry_run: logger.info("") logger.info("[DRY-RUN] No se eliminó ningún archivo") logger.info("Ejecuta sin --dry-run para aplicar la limpieza") if __name__ == "__main__": main()