xiaoxia-saas/scripts/cleanup_generated_files.py

"""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())