199 lines
7.6 KiB
Python
199 lines
7.6 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""Limpia archivos y registros de SQLite que se acumulan con el uso del proyecto.
|
|
|
|
Categorías que limpia (todas configurables vía flags):
|
|
- generated/browser/screenshots/*.png → más viejos de --screenshots-days (default 30)
|
|
- generated/runtime/batch/_bulk_batch_*.json → más viejos de --batch-files-hours (default 1)
|
|
- error_log (SQLite) → filas más viejas de --error-log-days (default 90)
|
|
- sync_log (SQLite) → filas más viejas de --sync-log-days (default 60)
|
|
|
|
NO toca (auditoría):
|
|
- script_runs / script_change_log / script_run_control (auditoría de mutaciones).
|
|
|
|
Modo dry-run por defecto: solo reporta qué borraría. Para ejecutar, pasa --apply.
|
|
"""
|
|
|
|
import argparse
|
|
import glob
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import time
|
|
|
|
ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
if ROOT_DIR not in sys.path:
|
|
sys.path.insert(0, ROOT_DIR)
|
|
|
|
from paths import SCREENSHOTS_DIR as SCREENSHOT_DIR, BATCH_DIR, DB_PATH as SQLITE_DB # noqa: E402
|
|
|
|
|
|
def _human_bytes(n):
|
|
for unit in ("B", "KB", "MB", "GB"):
|
|
if n < 1024:
|
|
return f"{n:.1f} {unit}"
|
|
n /= 1024
|
|
return f"{n:.1f} TB"
|
|
|
|
|
|
def cleanup_screenshots(days, apply_changes, log=print):
|
|
"""Borra capturas más viejas de N días."""
|
|
if not os.path.isdir(SCREENSHOT_DIR):
|
|
return {"files_count": 0, "bytes": 0, "applied": apply_changes}
|
|
|
|
cutoff = time.time() - (days * 86400)
|
|
candidates = []
|
|
for f in os.listdir(SCREENSHOT_DIR):
|
|
path = os.path.join(SCREENSHOT_DIR, f)
|
|
try:
|
|
if os.path.isfile(path) and os.path.getmtime(path) < cutoff:
|
|
candidates.append((path, os.path.getsize(path)))
|
|
except OSError:
|
|
continue
|
|
|
|
total_bytes = sum(s for _, s in candidates)
|
|
log(f" Screenshots > {days} d: {len(candidates)} archivos, {_human_bytes(total_bytes)}")
|
|
|
|
if apply_changes:
|
|
for path, _ in candidates:
|
|
try:
|
|
os.remove(path)
|
|
except OSError as e:
|
|
log(f" ! No se pudo borrar {path}: {e}")
|
|
|
|
return {"files_count": len(candidates), "bytes": total_bytes, "applied": apply_changes}
|
|
|
|
|
|
def cleanup_batch_files(hours, apply_changes, log=print):
|
|
"""Borra archivos _bulk_batch_*.json más viejos de N horas."""
|
|
pattern = os.path.join(BATCH_DIR, "_bulk_batch_*.json")
|
|
cutoff = time.time() - (hours * 3600)
|
|
candidates = []
|
|
for path in glob.glob(pattern):
|
|
try:
|
|
if os.path.getmtime(path) < cutoff:
|
|
candidates.append((path, os.path.getsize(path)))
|
|
except OSError:
|
|
continue
|
|
|
|
total_bytes = sum(s for _, s in candidates)
|
|
log(f" Batch files > {hours} h: {len(candidates)} archivos, {_human_bytes(total_bytes)}")
|
|
|
|
if apply_changes:
|
|
for path, _ in candidates:
|
|
try:
|
|
os.remove(path)
|
|
except OSError as e:
|
|
log(f" ! No se pudo borrar {path}: {e}")
|
|
|
|
return {"files_count": len(candidates), "bytes": total_bytes, "applied": apply_changes}
|
|
|
|
|
|
def cleanup_sqlite_table(table, ts_column, days, apply_changes, log=print):
|
|
"""Borra filas de `table` cuyo `ts_column` sea más viejo de N días."""
|
|
if not os.path.exists(SQLITE_DB):
|
|
return {"table": table, "rows_count": 0, "applied": apply_changes}
|
|
|
|
cutoff_epoch = time.time() - (days * 86400)
|
|
# `created_at` y `started_at` se guardan como TEXT con datetime('now','localtime')
|
|
# → formato 'YYYY-MM-DD HH:MM:SS'. SQLite puede comparar como string en este formato.
|
|
cutoff_iso = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(cutoff_epoch))
|
|
|
|
conn = sqlite3.connect(SQLITE_DB)
|
|
try:
|
|
cur = conn.execute(f"SELECT COUNT(*) FROM {table} WHERE {ts_column} < ?", (cutoff_iso,))
|
|
n = cur.fetchone()[0]
|
|
log(f" {table} con {ts_column} < {cutoff_iso}: {n} filas")
|
|
if apply_changes and n > 0:
|
|
conn.execute(f"DELETE FROM {table} WHERE {ts_column} < ?", (cutoff_iso,))
|
|
conn.commit()
|
|
log(f" → borradas {n} filas de {table}.")
|
|
return {"table": table, "rows_count": n, "applied": apply_changes}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def cleanup_all(args, log=print):
|
|
"""Ejecuta todas las limpiezas. Devuelve un dict-resumen con métricas por categoría."""
|
|
if args.apply:
|
|
log("[APPLY] Ejecutando limpieza (borrado real)...")
|
|
else:
|
|
log("[DRY-RUN] Simulación — no se borra nada. Usa --apply para ejecutar.")
|
|
log("")
|
|
|
|
log("--- Archivos ---")
|
|
s = cleanup_screenshots(args.screenshots_days, args.apply, log=log)
|
|
b = cleanup_batch_files(args.batch_files_hours, args.apply, log=log)
|
|
|
|
log("")
|
|
log("--- SQLite ---")
|
|
el = cleanup_sqlite_table("error_log", "created_at", args.error_log_days, args.apply, log=log)
|
|
sl = cleanup_sqlite_table("sync_log", "started_at", args.sync_log_days, args.apply, log=log)
|
|
|
|
total_bytes = s["bytes"] + b["bytes"]
|
|
total_rows = el["rows_count"] + sl["rows_count"]
|
|
total_files = s["files_count"] + b["files_count"]
|
|
|
|
log("")
|
|
log("=== RESUMEN ===")
|
|
log(f" Archivos a borrar: {total_files} ({_human_bytes(total_bytes)})")
|
|
log(f" Filas SQLite a borrar: {total_rows}")
|
|
if not args.apply and (total_files > 0 or total_rows > 0):
|
|
log("")
|
|
log(" Para ejecutar la limpieza real, vuelve a correr con --apply")
|
|
|
|
if args.apply and total_rows > 0:
|
|
# Compactar el archivo SQLite después de borrar (libera espacio en disco).
|
|
log("")
|
|
log(" Compactando mp_manager.sqlite con VACUUM...")
|
|
conn = sqlite3.connect(SQLITE_DB)
|
|
try:
|
|
size_before = os.path.getsize(SQLITE_DB)
|
|
conn.execute("VACUUM")
|
|
size_after = os.path.getsize(SQLITE_DB)
|
|
log(f" Tamaño SQLite: {_human_bytes(size_before)} → {_human_bytes(size_after)} "
|
|
f"(liberados {_human_bytes(max(0, size_before - size_after))})")
|
|
except Exception as e:
|
|
log(f" ! VACUUM falló: {e}")
|
|
finally:
|
|
conn.close()
|
|
|
|
return {
|
|
"applied": args.apply,
|
|
"screenshots": s,
|
|
"batch_files": b,
|
|
"error_log": el,
|
|
"sync_log": sl,
|
|
"total_files": total_files,
|
|
"total_bytes": total_bytes,
|
|
"total_rows": total_rows,
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Limpia archivos y registros que se acumulan con el uso.")
|
|
parser.add_argument("--apply", action="store_true", help="Ejecuta el borrado. Sin esto es dry-run.")
|
|
parser.add_argument("--screenshots-days", type=int, default=30,
|
|
help="Borrar capturas más viejas de N días (default 30).")
|
|
parser.add_argument("--batch-files-hours", type=int, default=1,
|
|
help="Borrar _bulk_batch_*.json más viejos de N horas (default 1).")
|
|
parser.add_argument("--error-log-days", type=int, default=90,
|
|
help="Borrar filas de error_log más viejas de N días (default 90).")
|
|
parser.add_argument("--sync-log-days", type=int, default=60,
|
|
help="Borrar filas de sync_log más viejas de N días (default 60).")
|
|
args = parser.parse_args()
|
|
|
|
print("=== CONTROL DE SCRIPTS: cleanup_storage.py ===")
|
|
print(f"ROOT: {ROOT_DIR}")
|
|
print(f"Modo: {'APPLY (borra)' if args.apply else 'DRY-RUN (solo reporta)'}")
|
|
print(f"Umbrales: screenshots>{args.screenshots_days}d, batch>{args.batch_files_hours}h, "
|
|
f"error_log>{args.error_log_days}d, sync_log>{args.sync_log_days}d")
|
|
print("-" * 70)
|
|
|
|
result = cleanup_all(args)
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|