kube-single-age/clusters/base/apps/photo-stats/configmap.yaml

apiVersion: v1
kind: ConfigMap
metadata:
  name: photo-stats-script
  namespace: photo-stats
data:
  scan.py: |
    import os
    import time
    import json
    import threading
    from datetime import datetime, timezone

    ROOT = "/mnt/sdb2/immich/uploads/library"
    OUT_DIR = "/www"
    current_date = datetime.now(timezone.utc)
    cutoff_date = current_date.replace(year=current_date.year - 1)
    CUTOFF = cutoff_date.timestamp()
    RESCAN_INTERVAL_SECONDS = 6 * 3600

    state_lock = threading.Lock()
    state = {
        "status": "starting",
        "cutoff_at": cutoff_date.isoformat(),
        "processed": 0,
        "total": 0,
        "percent": 0.0,
        "total_files": 0,
        "total_size": 0,
        "recent_files": 0,
        "recent_size": 0,
        "started_at": None,
        "finished_at": None,
        "error": None,
    }


    def human_size(n):
        n = float(n)
        for unit in ("B", "KiB", "MiB", "GiB", "TiB", "PiB"):
            if abs(n) < 1024.0:
                return f"{n:.1f}{unit}"
            n /= 1024.0
        return f"{n:.1f}EiB"


    def render_html(s):
        progress_html = ""
        if s["status"] == "counting":
            progress_html = "<p>Counting files...</p>"
        elif s["status"] == "scanning":
            pct = s["percent"]
            progress_html = f"""
    <p>Progress: {s['processed']} / {s['total']} ({pct:.1f}%)</p>
    <div class="bar"><div class="fill" style="width:{pct:.1f}%"></div></div>
    """
        error_html = f"<p class='err'>Error: {s['error']}</p>" if s["error"] else ""
        return f"""<!doctype html>
    <html><head><meta charset="utf-8"><title>Photo library stats</title>
    <meta http-equiv="refresh" content="5">
    <style>
    body {{ font-family: monospace; background: #111; color: #eee; padding: 2rem; }}
    table {{ border-collapse: collapse; }}
    td {{ padding: .25rem 1.5rem .25rem 0; }}
    .bar {{ background: #333; border-radius: 4px; overflow: hidden; width: 300px; height: 14px; }}
    .fill {{ background: #4caf50; height: 100%; }}
    .err {{ color: #f66; }}
    </style></head><body>
    <h1>Photo library stats</h1>
    <p>Status: <b>{s['status']}</b></p>
    {progress_html}
    {error_html}
    <table>
    <tr><td>Total files</td><td>{s['total_files']}</td></tr>
    <tr><td>Total size</td><td>{human_size(s['total_size'])} ({s['total_size']} bytes)</td></tr>
    <tr><td>Files newer than {s['cutoff_at']}</td><td>{s['recent_files']}</td></tr>
    <tr><td>Size of files newer than {s['cutoff_at']}</td><td>{human_size(s['recent_size'])} ({s['recent_size']} bytes)</td></tr>
    <tr><td>Last scan started</td><td>{s['started_at'] or '-'}</td></tr>
    <tr><td>Last scan finished</td><td>{s['finished_at'] or '-'}</td></tr>
    </table>
    </body></html>"""


    def write_output():
        with state_lock:
            s = dict(state)
        html = render_html(s)

        tmp = os.path.join(OUT_DIR, "index.html.tmp")
        with open(tmp, "w") as f:
            f.write(html)
        os.replace(tmp, os.path.join(OUT_DIR, "index.html"))

        tmp = os.path.join(OUT_DIR, "stats.json.tmp")
        with open(tmp, "w") as f:
            json.dump(s, f, indent=2)
        os.replace(tmp, os.path.join(OUT_DIR, "stats.json"))


    def scan_once():
        with state_lock:
            state.update(status="counting", started_at=datetime.now(timezone.utc).isoformat(),
                         finished_at=None, error=None, processed=0, total=0, percent=0.0)
        write_output()

        total = 0
        try:
            for _dirpath, _dirs, files in os.walk(ROOT):
                total += len(files)
        except OSError as e:
            with state_lock:
                state.update(status="error", error=str(e))
            write_output()
            return

        with state_lock:
            state.update(total=total, status="scanning")
        write_output()

        processed = 0
        total_files = 0
        total_size = 0
        recent_files = 0
        recent_size = 0
        last_write = time.time()

        try:
            for dirpath, _dirs, files in os.walk(ROOT):
                for name in files:
                    fp = os.path.join(dirpath, name)
                    processed += 1
                    try:
                        st = os.stat(fp)
                    except OSError:
                        continue
                    total_files += 1
                    total_size += st.st_size
                    if st.st_mtime >= CUTOFF:
                        recent_files += 1
                        recent_size += st.st_size

                    now = time.time()
                    if now - last_write > 2:
                        with state_lock:
                            state.update(
                                processed=processed, total_files=total_files,
                                total_size=total_size, recent_files=recent_files,
                                recent_size=recent_size,
                                percent=(processed / total * 100) if total else 100.0,
                            )
                        write_output()
                        last_write = now
        except OSError as e:
            with state_lock:
                state.update(status="error", error=str(e))
            write_output()
            return

        with state_lock:
            state.update(
                processed=processed, total_files=total_files, total_size=total_size,
                recent_files=recent_files, recent_size=recent_size, percent=100.0,
                status="done", finished_at=datetime.now(timezone.utc).isoformat(),
            )
        write_output()


    def main():
        os.makedirs(OUT_DIR, exist_ok=True)
        write_output()
        while True:
            scan_once()
            time.sleep(RESCAN_INTERVAL_SECONDS)


    if __name__ == "__main__":
        main()