diff --git a/dlphelper/sorter.py b/dlphelper/sorter.py index 97122d8..d4dca5c 100644 --- a/dlphelper/sorter.py +++ b/dlphelper/sorter.py @@ -86,7 +86,8 @@ def list_7z(path: Path, limit: int = 40) -> list[str]: try: proc = subprocess.run( ["7z", "l", "-ba", "-slt", str(path)], - stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, timeout=120, + stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, + encoding="utf-8", errors="replace", timeout=120, ) except (OSError, subprocess.TimeoutExpired) as exc: log.debug("7z listing failed for %s: %s", path.name, exc) diff --git a/dlphelper/zero807.py b/dlphelper/zero807.py index de7354d..d3b76ee 100644 --- a/dlphelper/zero807.py +++ b/dlphelper/zero807.py @@ -72,7 +72,8 @@ def download( cmd.append(url) log.info("Downloading %s -> %s", url, dest.name) - proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + encoding="utf-8", errors="replace") if proc.returncode != 0: tail = [line for line in proc.stderr.strip().splitlines() if line][-3:] stderr = " | ".join(tail) diff --git a/tools/extract_all.py b/tools/extract_all.py index 5f83303..931f29f 100644 --- a/tools/extract_all.py +++ b/tools/extract_all.py @@ -19,7 +19,9 @@ Behavior: - Failures (bad download/HTML page, wrong password, disk full) are moved to `/.broken/` and logged; the run never stops. -Re-runs are safe: destinations that already have content are skipped. +Re-runs are safe: destinations recorded in the manifest are skipped (a +partial dir from a crash is NOT skipped - it gets re-extracted); a final +sweep expands archives stranded by interrupted nested expansions. """ from __future__ import annotations @@ -78,9 +80,11 @@ def extract(archive: Path, dest: Path) -> tuple[bool, str]: dest.mkdir(parents=True, exist_ok=True) cmd = ["7z", "x", "-y", "-p", f"-o{dest}", str(archive)] try: + # 7z prints legacy-encoded inner filenames (non-UTF8 bytes); decode + # tolerantly or the whole run crashes (seen with byte 0x87). proc = subprocess.run( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - text=True, timeout=EXTRACT_TIMEOUT, + encoding="utf-8", errors="replace", timeout=EXTRACT_TIMEOUT, ) except subprocess.TimeoutExpired: return False, f"extraction timed out after {EXTRACT_TIMEOUT}s" @@ -103,9 +107,10 @@ def move_to_broken(path: Path, out_root: Path, src_root: Path, note: str) -> Pat return shutil.move(str(path), str(dest)) -def expand_nested(dest: Path, out_root: Path, manifest, src_root: Path) -> int: +def expand_nested(dest: Path, out_root: Path, manifest, src_root: Path, + done: set[str]) -> int: """Extract archives inside `dest` until none remain. Returns count.""" - done = 0 + done_count = 0 queue: deque[Path] = deque([dest]) while queue: cur = queue.popleft() @@ -113,7 +118,7 @@ def expand_nested(dest: Path, out_root: Path, manifest, src_root: Path) -> int: if not p.is_file() or not is_archive(p): continue stem_dir = p.with_name(display_unit(p.name)) - if stem_dir.exists() and any(stem_dir.iterdir()): + if str(stem_dir.relative_to(out_root)) in done: # already extracted in a previous run; drop the archive copy p.unlink() log.info("Nested %s already extracted - archive removed", p.name) @@ -127,10 +132,11 @@ def expand_nested(dest: Path, out_root: Path, manifest, src_root: Path) -> int: move_to_broken(p, out_root, src_root, err) continue manifest.write(f"{json_line(p.relative_to(out_root), stem_dir)}\n") + done.add(str(stem_dir.relative_to(out_root))) p.unlink() # archive copy consumed - done += 1 + done_count += 1 queue.append(stem_dir) - return done + return done_count def json_line(archive_rel: Path, dest: Path) -> str: @@ -145,7 +151,7 @@ def json_line(archive_rel: Path, dest: Path) -> str: _OUT_ROOT: Path = Path(".") # set in main; used by json_line -def process_root(src_root: Path, out_root: Path, manifest) -> tuple[int, int, int]: +def process_root(src_root: Path, out_root: Path, manifest, done: set[str]) -> tuple[int, int, int]: """Returns (extracted, mirrored, failed).""" idx = src_root.name extracted = mirrored = failed = 0 @@ -159,7 +165,9 @@ def process_root(src_root: Path, out_root: Path, manifest) -> tuple[int, int, in if is_archive(f): dest_rel = Path(idx) / Path(str(rel.with_name(display_unit(f.name)))) dest = out_root / dest_rel - if dest.exists() and any(dest.iterdir()): + # Skip on a manifest record, never on "dir non-empty": a crash + # mid-extraction leaves a partial dir that must be redone. + if str(dest_rel) in done: log.info("Skip (already extracted): %s", rel) continue if free_bytes(out_root) < MIN_FREE_BYTES: @@ -173,9 +181,10 @@ def process_root(src_root: Path, out_root: Path, manifest) -> tuple[int, int, in failed += 1 continue manifest.write(f"{json_line(Path(idx) / rel, dest)}\n") + done.add(str(dest_rel)) extracted += 1 log.info("Extracted %s -> %s", rel, dest_rel) - expand_nested(dest, out_root, manifest, src_root) + expand_nested(dest, out_root, manifest, src_root, done) else: link_dest = out_root / idx / rel if link_dest.exists(): @@ -189,6 +198,54 @@ def process_root(src_root: Path, out_root: Path, manifest) -> tuple[int, int, in return extracted, mirrored, failed +def load_done(manifest_path: Path) -> set[str]: + """Dest dirs already recorded in the manifest (relative to out root).""" + import json + done: set[str] = set() + if manifest_path.exists(): + with manifest_path.open(encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if not line: + continue + try: + done.add(json.loads(line)["dest"]) + except (json.JSONDecodeError, KeyError): + continue + return done + + +def sweep(out_root: Path, manifest, done: set[str]) -> int: + """Expand any archive left anywhere in the tree (interrupted nested + expansions leave stranded copies even when the parent has a record).""" + import json + swept = 0 + for p in sorted(out_root.rglob("*")): + if not p.is_file() or not is_archive(p): + continue + if ".broken" in p.relative_to(out_root).parts: + continue + stem_dir = p.with_name(display_unit(p.name)) + if str(stem_dir.relative_to(out_root)) in done: + p.unlink() + log.info("Sweep: %s already extracted - archive removed", p.name) + continue + if free_bytes(out_root) < MIN_FREE_BYTES: + log.error("Out of disk space - sweep skipping %s", p) + continue + ok, err = extract(p, stem_dir) + if not ok: + log.error("FAILED sweep extraction %s: %s", p, err) + move_to_broken(p, out_root, out_root, err) + continue + manifest.write(f"{json_line(p.relative_to(out_root), stem_dir)}\n") + done.add(str(stem_dir.relative_to(out_root))) + p.unlink() + swept += 1 + expand_nested(stem_dir, out_root, manifest, out_root, done) + return swept + + def main() -> None: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("sources", nargs="+", type=Path) @@ -204,16 +261,19 @@ def main() -> None: args.out.mkdir(parents=True, exist_ok=True) totals = [0, 0, 0] + done = load_done(args.out / "extract_manifest.jsonl") + log.info("%d destinations already recorded in manifest", len(done)) with (args.out / "extract_manifest.jsonl").open("a", encoding="utf-8") as manifest: for src in args.sources: log.info("== processing %s", src) - counts = process_root(src, args.out, manifest) + counts = process_root(src, args.out, manifest, done) log.info("== %s: %d extracted, %d mirrored, %d failed", src.name, *counts) for i, c in enumerate(counts): totals[i] += c - log.info("TOTAL: %d archives extracted, %d files mirrored, %d failed", - *totals) + swept = sweep(args.out, manifest, done) + log.info("TOTAL: %d archives extracted, %d files mirrored, %d failed, %d swept", + *totals, swept) if totals[2]: log.warning("Failures were moved under %s - inspect and re-run", args.out / ".broken")