File size: 3,069 Bytes
6f5feff 66eca06 6f5feff 66eca06 6f5feff | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | #!/usr/bin/env python3
"""Cleanup utility for GAIT_API generated files.
This script removes files under the runs directory that are older than a
configurable age (default: 30 minutes). It is designed to be triggered by cron.
"""
from __future__ import annotations
import argparse
import os
import time
from pathlib import Path
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Cleanup old files from runs directory")
parser.add_argument(
"--path",
default="/app/runs",
help="Runs directory path (default: /app/runs)",
)
parser.add_argument(
"--max-age-minutes",
type=float,
default=30,
help="Delete files older than this many minutes (default: 30)",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Only print what would be deleted",
)
return parser.parse_args()
def cleanup_runs(root: Path, max_age_minutes: float, dry_run: bool = False) -> tuple[int, int]:
if not root.exists():
return 0, 0
now = time.time()
cutoff = now - (max_age_minutes * 60)
deleted_files = 0
deleted_dirs = 0
# Delete eligible files first
for file_path in root.rglob("*"):
if not file_path.is_file():
continue
try:
mtime = file_path.stat().st_mtime
if mtime <= cutoff:
if dry_run:
print(f"[DRY-RUN] file: {file_path}")
else:
file_path.unlink(missing_ok=True)
deleted_files += 1
except FileNotFoundError:
# Might have been removed by another process
continue
except PermissionError:
print(f"[WARN] Permission denied: {file_path}")
# Remove empty subdirectories bottom-up (excluding root and outputs/)
preserve_dirs = {root, root / "outputs"}
for dir_path in sorted((p for p in root.rglob("*") if p.is_dir()), key=lambda p: len(p.parts), reverse=True):
# Skip preserving root structure
if dir_path in preserve_dirs or dir_path.parent in preserve_dirs:
continue
try:
if any(dir_path.iterdir()):
continue
if dry_run:
print(f"[DRY-RUN] dir: {dir_path}")
else:
dir_path.rmdir()
deleted_dirs += 1
except (FileNotFoundError, PermissionError, OSError):
# OSError when directory isn't empty anymore
continue
return deleted_files, deleted_dirs
def main() -> int:
args = parse_args()
root = Path(args.path)
deleted_files, deleted_dirs = cleanup_runs(
root=root,
max_age_minutes=args.max_age_minutes,
dry_run=args.dry_run,
)
print(
f"Cleanup complete for {root} | "
f"deleted_files={deleted_files} | deleted_dirs={deleted_dirs} | "
f"max_age_minutes={args.max_age_minutes}"
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
|