"""Clean local generated-files storage with a dry-run-first workflow.""" from __future__ import annotations import argparse import logging import os import time from pathlib import Path logger = logging.getLogger(__name__) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--dir", default=os.getenv("GENERATED_FILES_DIR", "/app/generated"), help="Generated files root directory. Defaults to GENERATED_FILES_DIR or /app/generated.", ) parser.add_argument( "--days", type=int, default=7, help="Delete files older than this many days. Defaults to 7.", ) parser.add_argument( "--apply", action="store_true", help="Actually delete files. Without this flag the script only prints a dry run.", ) return parser.parse_args() def iter_expired_files(root: Path, cutoff_epoch: float): for path in root.rglob("*"): if path.is_file() and path.stat().st_mtime < cutoff_epoch: yield path def remove_empty_dirs(root: Path) -> None: for path in sorted((p for p in root.rglob("*") if p.is_dir()), key=lambda p: len(p.parts), reverse=True): try: path.rmdir() except OSError as e: logger.warning(f"Operation failed in scripts/cleanup_generated_files.py: {e}", exc_info=True) def main() -> int: args = parse_args() root = Path(args.dir).resolve() if not root.exists(): print(f"generated files directory does not exist: {root}") return 0 if not root.is_dir(): raise SystemExit(f"not a directory: {root}") if args.days < 1: raise SystemExit("--days must be >= 1") cutoff_epoch = time.time() - args.days * 24 * 60 * 60 expired_files = list(iter_expired_files(root, cutoff_epoch)) total_bytes = sum(path.stat().st_size for path in expired_files) mode = "APPLY" if args.apply else "DRY-RUN" print(f"mode={mode} root={root} days={args.days} files={len(expired_files)} bytes={total_bytes}") for path in expired_files: print(path) if args.apply: path.unlink() if args.apply: remove_empty_dirs(root) return 0 if __name__ == "__main__": raise SystemExit(main())