feat: Add scripts

capture-tree.py

@@ -0,0 +1,204 @@
+#!/usr/bin/env python3
+"""Export surviving ext4 directories from an UNMOUNTED damaged LV using debugfs.
+
+Does not read file contents or write to the source. Unreachable trees must be
+supplemented manually; no software can infer names below missing dir metadata.
+"""
+
+import argparse
+import json
+import os
+import re
+import stat
+import subprocess
+import sys
+from pathlib import Path, PurePosixPath
+
+DEFAULT_LV = "/dev/mapper/plex--new-plex--new"
+DEFAULT_OUTPUT = "/mnt/recovery/plex-directory-tree.jsonl"
+INODE_RE = re.compile(r"\bInode:\s*(\d+)\s+Type:\s*([^\s]+)")
+MODE_RE = re.compile(r"\bMode:\s*([0-7]+)")
+OWNER_RE = re.compile(r"\bUser:\s*(\d+)\s+Group:\s*(\d+)")
+
+
+def fail(message):
+    sys.exit(f"ERROR: {message}")
+
+
+def normalize_path(value):
+    if not value.startswith("/"):
+        fail(f"Source directories must be absolute filesystem paths: {value!r}")
+    # Avoid guessing about special path syntax; '/' is meaningful within debugfs.
+    parts = value.split("/")
+    if any(part in (".", "..") for part in parts):
+        fail(f"Source paths must not contain '.' or '..': {value!r}")
+    return "/" + "/".join(p for p in parts if p)
+
+
+def quote_debugfs(value):
+    return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'
+
+
+def debugfs(lv, command):
+    env = os.environ.copy()
+    env["DEBUGFS_PAGER"] = "cat"
+    return subprocess.run(
+        ["debugfs", "-c", "-R", command, lv],
+        text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env,
+        errors="replace",
+    )
+
+
+def get_inode(lv, path):
+    r = debugfs(lv, f"stat {quote_debugfs(path)}")
+    match = INODE_RE.search(r.stdout)
+    if not match or match.group(2) != "directory":
+        return None, r
+    mode_match = MODE_RE.search(r.stdout)
+    owner_match = OWNER_RE.search(r.stdout)
+    return {
+        "inode": int(match.group(1)),
+        "mode": int(mode_match.group(1), 8) if mode_match else None,
+        "uid": int(owner_match.group(1)) if owner_match else None,
+        "gid": int(owner_match.group(2)) if owner_match else None,
+    }, r
+
+
+def list_dir(lv, inode):
+    r = debugfs(lv, f"ls -p <{inode}>")
+    entries = []
+    bad_lines = []
+    for raw in r.stdout.splitlines():
+        if not raw.startswith("/"):
+            continue
+        # debugfs ls -p: /inode/mode/uid/gid/name/size/
+        # Linux names cannot contain a slash. The last empty field is expected.
+        fields = raw.split("/")
+        if len(fields) < 7:
+            bad_lines.append(raw)
+            continue
+        try:
+            ino = int(fields[1])
+            mode = int(fields[2], 8)
+            uid = int(fields[3])
+            gid = int(fields[4])
+        except ValueError:
+            bad_lines.append(raw)
+            continue
+        name = fields[5]
+        if name in (".", "..") or ino == 0:
+            continue
+        entries.append((ino, mode, uid, gid, name))
+    # debugfs always writes a version banner to stderr; strip that banner.
+    stderr = "\n".join(
+        line for line in r.stderr.splitlines()
+        if not line.startswith("debugfs ")
+    ).strip()
+    return entries, stderr, bad_lines
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--source", action="append", default=[],
+                        help="Existing source directory to traverse; repeatable (default: /)")
+    parser.add_argument("--include", action="append", default=[],
+                        help="Add an expected directory even if its inode is unreadable; repeatable")
+    parser.add_argument("--lv", default=DEFAULT_LV)
+    parser.add_argument("--output", default=DEFAULT_OUTPUT,
+                        help="Write JSONL plus a .txt folder-path list and .warnings.txt")
+    args = parser.parse_args()
+
+    if os.geteuid() != 0:
+        fail("Run capture as root so debugfs can access the block device.")
+    ro = subprocess.run(["blockdev", "--getro", args.lv], capture_output=True, text=True)
+    if ro.returncode or ro.stdout.strip() != "1":
+        fail("Source LV must be kernel read-only. Set it read-only before capture.")
+
+    output = Path(args.output).absolute()
+    if output.suffix != ".jsonl":
+        fail("--output must end with .jsonl (two companion files are generated).")
+    if str(output).startswith("/mnt/recovery/") and not os.path.ismount("/mnt/recovery"):
+        fail("/mnt/recovery isn't mounted. Refusing to write the snapshot to an unmounted recovery drive.")
+    output.parent.mkdir(parents=True, exist_ok=True)
+    # A temporary JSONL is renamed only when the traversal is complete.
+    tmp = Path(str(output) + ".incomplete")
+    txt_path = output.with_suffix(".txt")
+    warning_path = output.with_suffix(".warnings.txt")
+    sources = [normalize_path(s) for s in (args.source or ["/"])]
+    additions = [normalize_path(s) for s in args.include]
+    records = {}
+    warnings = []
+    visited = set()
+    pending = []
+
+    def add_record(path, inode=None, mode=None, uid=None, gid=None, origin="debugfs"):
+        if path == "/":
+            return
+        if path not in records:
+            records[path] = {
+                "path": path, "inode": inode, "mode": mode, "uid": uid,
+                "gid": gid, "origin": origin,
+            }
+        elif origin == "debugfs":
+            records[path].update(
+                inode=inode, mode=mode, uid=uid, gid=gid, origin=origin,
+            )
+
+    for path in additions:
+        add_record(path, origin="manual")
+
+    for path in sources:
+        info, r = get_inode(args.lv, path)
+        if info is None:
+            warnings.append(f"UNREADABLE_SOURCE {path}: {r.stderr.strip()} {r.stdout.strip()}")
+            add_record(path, origin="unreadable-source")
+            continue
+        add_record(path, **info)
+        pending.append((info["inode"], path))
+
+    while pending:
+        inode, current_path = pending.pop()
+        if inode in visited:
+            continue
+        visited.add(inode)
+        entries, stderr, bad_lines = list_dir(args.lv, inode)
+        if stderr:
+            warnings.append(f"DIR_READ_WARNING {current_path} (inode {inode}): {stderr}")
+        for raw in bad_lines:
+            warnings.append(f"UNPARSEABLE_ENTRY {current_path}: {raw}")
+        for child_ino, mode, uid, gid, name in entries:
+            path = current_path.rstrip("/") + "/" + name
+            if stat.S_ISDIR(mode):
+                add_record(path, inode=child_ino, mode=stat.S_IMODE(mode),
+                           uid=uid, gid=gid)
+                pending.append((child_ino, path))
+            elif stat.S_IFMT(mode) == 0:
+                warnings.append(
+                    f"AMBIGUOUS_ENTRY {path} (inode {child_ino}): bad/missing type; "
+                    "might be a lost directory. Manually include if required."
+                )
+        if len(visited) % 200 == 0:
+            print(f"Visited {len(visited):,} directories; found {len(records):,} paths", flush=True)
+
+    with tmp.open("w", encoding="utf-8") as f:
+        f.write(json.dumps({"_meta": {
+            "format": "plex-directory-tree-v1", "lv": args.lv,
+            "sources": sources, "manual_includes": additions,
+            "warnings": len(warnings),
+            "note": "Only reachable ext4 metadata is exported; check warnings and missing share roots."
+        }}, ensure_ascii=False) + "\n")
+        for path in sorted(records):
+            f.write(json.dumps(records[path], ensure_ascii=False) + "\n")
+        f.flush()
+        os.fsync(f.fileno())
+    tmp.replace(output)
+    txt_path.write_text("".join(p + "\n" for p in sorted(records)), encoding="utf-8")
+    warning_path.write_text("\n".join(warnings) + ("\n" if warnings else ""), encoding="utf-8")
+    print(f"Exported {len(records):,} directories ({len(visited):,} visited)")
+    print(f"Manifest: {output}\nPaths:    {txt_path}\nWarnings: {warning_path}")
+    if warnings:
+        print(f"WARNING: {len(warnings)} warnings / ambiguous entries. Review before rebuilding.")
+
+
+if __name__ == "__main__":
+    main()