Script to find orophaned cache

scripts/run_find_step1_cache.py

@@ -0,0 +1,112 @@
+#!/usr/bin/python3
+
+# Finds lidar_odometry_step_1 cache folders (<session>/cache/*.cache) under a given
+# folder and writes a deletion script for review. This script never deletes anything.
+#
+# A cache folder is listed only when its session still has the raw lidar*.laz input,
+# so the cache can be regenerated by running step 1 again.
+
+import argparse
+import datetime
+import os
+import shlex
+import sys
+
+SKIP_DIRECTORIES = {'.git', '.Trash', 'Library', '$RECYCLE.BIN', 'System Volume Information'}
+
+
+def directory_size(path):
+    total = 0
+    for root, _, files in os.walk(path):
+        for name in files:
+            try:
+                total += os.lstat(os.path.join(root, name)).st_size
+            except OSError:
+                pass
+    return total
+
+
+def human_size(size):
+    for unit in ['B', 'K', 'M', 'G', 'T']:
+        if size < 1024 or unit == 'T':
+            return f'{size:.1f}{unit}'
+        size /= 1024
+
+
+def find_cache_directories(root_directory):
+    for root, directories, files in os.walk(root_directory, onerror=lambda e: None):
+        directories[:] = [d for d in directories if d not in SKIP_DIRECTORIES]
+        if os.path.basename(root) == 'cache' and any(f.endswith('.cache') for f in files):
+            yield root
+
+
+def quote(path, windows):
+    return f'"{path}"' if windows else shlex.quote(path)
+
+
+def main():
+    parser = argparse.ArgumentParser(description='Prepare a deletion script for lidar_odometry_step_1 cache folders.')
+    parser.add_argument('folder', help='folder to search')
+    parser.add_argument('-o', '--output', help='deletion script to write (default: ./delete_step1_cache_<timestamp>.sh/.bat)')
+    args = parser.parse_args()
+
+    if not os.path.isdir(args.folder):
+        sys.exit(f"'{args.folder}' is not a directory")
+
+    windows = os.name == 'nt'
+    timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
+    output = args.output or f'delete_step1_cache_{timestamp}.{"bat" if windows else "sh"}'
+    comment = 'REM' if windows else '#'
+
+    root_directory = os.path.abspath(args.folder)
+    found = 0
+    listed = 0
+    total_size = 0
+    commands = []
+    skipped = []
+
+    for cache_directory in sorted(find_cache_directories(root_directory)):
+        found += 1
+        session_directory = os.path.dirname(cache_directory)
+        entries = os.listdir(cache_directory)
+        cache_files = [e for e in entries if e.endswith('.cache')]
+        laz_files = [f for f in os.listdir(session_directory) if f.startswith('lidar') and f.endswith('.laz')]
+        size = directory_size(cache_directory)
+
+        if not laz_files:
+            skipped.append(f'{comment}   [{human_size(size)}] {cache_directory}  (no lidar*.laz in session, cache may be the only copy)')
+            print(f'{human_size(size):>8}  {cache_directory}\n          SKIPPED: no lidar*.laz in session')
+            continue
+
+        listed += 1
+        total_size += size
+        quoted = quote(cache_directory, windows)
+        commands.append(f'{comment} [{human_size(size)}] {len(cache_files)} cache files, session has {len(laz_files)} lidar*.laz')
+        if len(cache_files) == len(entries):
+            commands.append(f'rmdir /s /q {quoted}' if windows else f'rm -rf -- {quoted}')
+        else:
+            commands.append(f'del /q {quote(os.path.join(cache_directory, "*.cache"), windows)}' if windows
+                            else f"find {quoted} -maxdepth 1 -type f -name '*.cache' -delete")
+        commands.append('')
+        print(f'{human_size(size):>8}  {cache_directory}  ({len(cache_files)} files)')
+
+    with open(output, 'w', newline='\r\n' if windows else '\n') as f:
+        f.write('@echo off\n' if windows else '#!/usr/bin/env bash\n')
+        f.write(f'{comment} Generated {datetime.datetime.now():%Y-%m-%d %H:%M:%S} by {os.path.basename(__file__)} for: {root_directory}\n')
+        f.write(f'{comment} {listed} of {found} cache folders, total {human_size(total_size)}\n')
+        f.write(f'{comment} REVIEW BEFORE RUNNING. Comment out any line you want to keep.\n')
+        if skipped:
+            f.write(f'{comment} Skipped (not scheduled):\n')
+            f.write('\n'.join(skipped) + '\n')
+        if not windows:
+            f.write('set -euo pipefail\n')
+        f.write('\n' + '\n'.join(commands))
+    if not windows:
+        os.chmod(output, 0o755)
+
+    print(f'\nFound {found} cache folders, {listed} listed for deletion ({human_size(total_size)}).')
+    print(f'Deletion script (NOT executed): {os.path.abspath(output)}')
+
+
+if __name__ == '__main__':
+    main()