Buckets:
| """cwm3/data/prune_local_baton.py — free local disk by deleting local BATON VIDEO files that are verified identical (size AND sha256) | |
| on the canonical remote copy (USER@REMOTE_HOST:/data/backup/BATON). Owner-authorised 2026-09-12 (local BATON may be pruned; the | |
| remote holds the full dataset and files can be re-downloaded). Never touches the remote; never deletes an unverified file; never | |
| deletes CSV/metadata, results/ or caches. | |
| Tier 1: local leaves NOT used by the 26,578-clip manifest (results/baton_leaves.json: n_clips_manifest == 0) -> dcamera.mp4, qcamera.mp4, | |
| *.hevc inside the leaf dir; plus raw <seg>--dcamera.hevc segment files at the route-id level whose route-id dir holds NO manifest leaf. | |
| Tier 2: manifest leaves' qcamera.mp4 only (dcamera.mp4 of the 484 manifest leaves is kept). | |
| Per file: local sha256 (runmeta.sha256_file) and remote `stat -c %s` + `sha256sum` (one ssh per leaf, ControlMaster /tmp/baton_bak_ctl); | |
| verified = remote exists AND sizes equal AND digests equal -> os.remove(local). Ledger (append-only) results/baton_local_prune_ledger.jsonl | |
| {tier, leaf, file, bytes, local_sha, remote_sha, remote_bytes, verified, deleted, ts}. On ANY verification failure the leaf is left intact, | |
| the failure is recorded, and the run stops (exit 5) so it can be reported. Resumable: files already recorded deleted are skipped.""" | |
| import argparse, json, os, subprocess, sys, time | |
| from concurrent.futures import ThreadPoolExecutor, as_completed | |
| from pathlib import Path | |
| ROOT = Path(__file__).resolve().parents[1]; sys.path.insert(0, str(ROOT / "common")) | |
| from runmeta import write_run_json, sha256_file | |
| LROOT = Path("/home/USER/BATON"); REMOTE = "USER@REMOTE_HOST"; RROOT = "/data/backup/BATON" | |
| SSH = ["ssh", "-S", "/tmp/baton_bak_ctl", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20"] | |
| LEAVES = ROOT / "results" / "baton_leaves.json"; LEDGER = ROOT / "results" / "baton_local_prune_ledger.jsonl" | |
| VIDEO_EXT = (".mp4", ".hevc") | |
| def ensure_master(): | |
| if subprocess.run(SSH + ["-O", "check", REMOTE], capture_output=True).returncode == 0: return | |
| subprocess.run(["bash", str(ROOT.parent / "cwm2" / "private" / "ssh_master.sh")], capture_output=True); time.sleep(2) | |
| if subprocess.run(SSH + ["-O", "check", REMOTE], capture_output=True).returncode != 0: raise RuntimeError("ssh master unavailable") | |
| def remote_stat_sha(rel_files): | |
| """-> {rel: (bytes, sha) or None} for files (relative to the BATON root) in ONE ssh call.""" | |
| cmd = "; ".join(f"f='{RROOT}/{r}'; if [ -f \"$f\" ]; then printf '%s\\t' '{r}'; stat -c '%s' \"$f\" | tr -d '\\n'; printf '\\t'; sha256sum \"$f\" | cut -d' ' -f1; else printf '%s\\tMISSING\\t-\\n' '{r}'; fi" for r in rel_files) | |
| r = subprocess.run(SSH + [REMOTE, cmd], capture_output=True, text=True, timeout=7200) | |
| out = {} | |
| for line in r.stdout.splitlines(): | |
| parts = line.split("\t") | |
| if len(parts) == 3: out[parts[0]] = None if parts[1] == "MISSING" else (int(parts[1]), parts[2]) | |
| return out | |
| def targets(tier, lv): | |
| manifest_leaves = {l for l, v in lv["leaves"].items() if v["n_clips_manifest"] > 0} | |
| local_leaves = sorted({str(p.parent.relative_to(LROOT)) for p in LROOT.glob("*/*/*/*/dcamera.mp4")} | {str(p.parent.relative_to(LROOT)) for p in LROOT.glob("*/*/*/*/qcamera.mp4")}) | |
| files = [] | |
| if tier == 1: | |
| for leaf in local_leaves: | |
| if leaf in manifest_leaves: continue | |
| for p in sorted((LROOT / leaf).iterdir()): | |
| if p.is_file() and p.suffix in VIDEO_EXT: files.append((leaf, p)) | |
| # raw segment files at the route-id level, only where no leaf of that route-id is in the manifest | |
| for seg in sorted(LROOT.glob("*/*/*/*.hevc")): | |
| rid = seg.parent; rid_rel = str(rid.relative_to(LROOT)) | |
| if any(l.startswith(rid_rel + "/") for l in manifest_leaves): continue | |
| files.append((rid_rel, seg)) | |
| else: | |
| for leaf in local_leaves: | |
| if leaf not in manifest_leaves: continue | |
| q = LROOT / leaf / "qcamera.mp4" | |
| if q.exists(): files.append((leaf, q)) | |
| return files, manifest_leaves, local_leaves | |
| def main(): | |
| ap = argparse.ArgumentParser(); ap.add_argument("--tier", type=int, default=1); ap.add_argument("--dry-run", action="store_true"); ap.add_argument("--limit", type=int, default=0) | |
| ap.add_argument("--remote-workers", type=int, default=3); ap.add_argument("--local-workers", type=int, default=4); a = ap.parse_args(); t_start = time.time() | |
| lv = json.load(open(LEAVES)); files, man, loc = targets(a.tier, lv) | |
| done = set() | |
| if LEDGER.exists(): | |
| for l in open(LEDGER): | |
| if l.strip(): | |
| r = json.loads(l) | |
| if r.get("deleted"): done.add(r["file"]) | |
| files = [(g, p) for g, p in files if str(p.relative_to(LROOT)) not in done] | |
| if a.limit: files = files[:a.limit] | |
| tot = sum(p.stat().st_size for _, p in files); groups = {} | |
| for g, p in files: groups.setdefault(g, []).append(p) | |
| print(f"tier {a.tier}: manifest leaves {len(man)}, local leaves {len(loc)}, target groups {len(groups)}, files {len(files)}, bytes {tot/1e9:.1f} GB (already deleted per ledger: {len(done)})", flush=True) | |
| if a.dry_run: | |
| ext = {} | |
| for _, p in files: ext[p.suffix] = ext.get(p.suffix, 0) + p.stat().st_size | |
| print("by extension GB:", {k: round(v / 1e9, 1) for k, v in ext.items()}); return | |
| ensure_master(); free0 = os.statvfs(str(LROOT)).f_bavail * os.statvfs(str(LROOT)).f_frsize | |
| lock = __import__("threading").Lock(); stats = dict(verified=0, deleted=0, failed=0, bytes_deleted=0); failures = [] | |
| def process_group(g, paths): | |
| rels = [str(p.relative_to(LROOT)) for p in paths] | |
| with ThreadPoolExecutor(max_workers=a.local_workers) as ex: local = dict(zip(rels, ex.map(sha256_file, paths))) | |
| remote = remote_stat_sha(rels); rows = []; ok_all = True | |
| for p, rel in zip(paths, rels): | |
| lb = p.stat().st_size; rs = remote.get(rel); ok = rs is not None and rs[0] == lb and rs[1] == local[rel] | |
| rows.append(dict(tier=a.tier, leaf=g, file=rel, bytes=lb, local_sha=local[rel], remote_sha=(rs[1] if rs else None), remote_bytes=(rs[0] if rs else None), verified=ok, deleted=False, ts=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))) | |
| ok_all &= ok | |
| if ok_all: | |
| for row, p in zip(rows, paths): | |
| try: os.remove(p); row["deleted"] = True | |
| except FileNotFoundError: row["deleted"] = True | |
| return g, rows, ok_all | |
| stop = False | |
| with ThreadPoolExecutor(max_workers=a.remote_workers) as ex: | |
| futs = {ex.submit(process_group, g, ps): g for g, ps in groups.items()} | |
| for f in as_completed(futs): | |
| g, rows, ok_all = f.result() | |
| with lock, open(LEDGER, "a") as fh: | |
| for r in rows: fh.write(json.dumps(r) + "\n") | |
| stats["verified"] += sum(r["verified"] for r in rows); stats["deleted"] += sum(r["deleted"] for r in rows); stats["bytes_deleted"] += sum(r["bytes"] for r in rows if r["deleted"]) | |
| if not ok_all: stats["failed"] += sum(not r["verified"] for r in rows); failures.append(dict(leaf=g, files=[dict(file=r["file"], bytes=r["bytes"], remote_bytes=r["remote_bytes"], sha_equal=(r["local_sha"] == r["remote_sha"])) for r in rows if not r["verified"]])) | |
| if (stats["deleted"] + stats["failed"]) % 50 < len(rows): print(f" deleted {stats['deleted']} files {stats['bytes_deleted']/1e9:.1f} GB, failed {stats['failed']} {time.time()-t_start:.0f}s", flush=True) | |
| if not ok_all and not stop: | |
| stop = True; print(f"VERIFICATION_FAILED leaf {g}: {failures[-1]} -> stopping (no deletion for this leaf)", flush=True) | |
| for fu in futs: fu.cancel() | |
| break | |
| free1 = os.statvfs(str(LROOT)).f_bavail * os.statvfs(str(LROOT)).f_frsize | |
| summ = dict(tier=a.tier, groups=len(groups), files=len(files), verified=stats["verified"], deleted=stats["deleted"], failed=stats["failed"], GB_deleted=round(stats["bytes_deleted"] / 1e9, 2), disk_free_before_GB=round(free0 / 1e9, 1), disk_free_after_GB=round(free1 / 1e9, 1), failures=failures[:20], stopped_on_failure=stop, wall_s=round(time.time() - t_start)) | |
| write_run_json(ROOT / "results", f"prune_local_baton_tier{a.tier}", [Path(__file__)], [LEAVES], None, t_start, extra={"summary": summ}, input_listing=False) | |
| print(("PRUNE_STOPPED " if stop else "PRUNE_DONE ") + json.dumps({k: summ[k] for k in ("tier", "groups", "files", "verified", "deleted", "failed", "GB_deleted", "disk_free_before_GB", "disk_free_after_GB")}), flush=True) | |
| sys.exit(5 if stop else 0) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 8.73 kB
- Xet hash:
- a213d8363c91e7aa998546a03b5fd36ab4ccd0183afea6b30d07b353a2a3dcab
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.