docs(deploy): document generated file retention
This commit is contained in:
@@ -0,0 +1,75 @@
|
||||
"""Clean local generated-files storage with a dry-run-first workflow."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
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:
|
||||
pass
|
||||
|
||||
|
||||
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())
|
||||
Reference in New Issue
Block a user