from __future__ import annotations import hashlib import os import re import signal import shutil import subprocess import zipfile import zlib import sys import tempfile import threading import time import xml.etree.ElementTree as ET from collections import Counter, defaultdict from dataclasses import dataclass, field from pathlib import Path, PurePosixPath from typing import Callable, Dict, Iterable, List, Optional, Sequence, Tuple from PyQt6.QtCore import QThread, Qt, pyqtSignal from PyQt6.QtGui import QFont, QIcon from PyQt6.QtWidgets import ( QApplication, QAbstractItemView, QCheckBox, QComboBox, QDialog, QDialogButtonBox, QFileDialog, QFrame, QGridLayout, QHBoxLayout, QLabel, QLineEdit, QListWidget, QListWidgetItem, QMainWindow, QMessageBox, QPlainTextEdit, QProgressBar, QPushButton, QScrollArea, QSizePolicy, QSpacerItem, QVBoxLayout, QWidget, ) APP_NAME = "English Manga Vault Builder" VERSION = "2.4" HASH_RE = re.compile(r"^[0-9a-fA-F]{40}$") @dataclass(frozen=True) class ManifestEntry: sha1: str size: int rel_path: str @dataclass class ManifestData: entries: List[ManifestEntry] by_hash: Dict[str, List[ManifestEntry]] # Populated for a combined multi-DAT manifest. Ordinary parse_dat stays compatible. dat_counts: Dict[str, int] = field(default_factory=dict) dat_duplicates: Dict[str, int] = field(default_factory=dict) entry_source: Dict[str, str] = field(default_factory=dict) entry_bucket: Dict[str, str] = field(default_factory=dict) @dataclass(frozen=True) class BuildConflict: rel_path: str source_path: str expected_sha1: str existing_sha1: str action: str class ManifestError(Exception): pass def _xml_local_name(tag: str) -> str: """Return an XML tag name without an optional namespace prefix.""" return tag.rsplit("}", 1)[-1] if "}" in tag else tag def _validate_dat_relative_path(rel_path: str, context: str) -> str: """Normalize and validate a DAT-derived relative path.""" rel = rel_path.replace("\\", "/") p = PurePosixPath(rel) if not rel or p.is_absolute() or ".." in p.parts: raise ManifestError(f"Unsafe or empty path in {context}: {rel_path!r}") if re.match(r"^[A-Za-z]:[\\/]", rel) or rel.startswith("//"): raise ManifestError(f"Unsafe absolute path in {context}: {rel_path!r}") return str(p) def parse_dat(path: Path) -> ManifestData: """Parse an XML .DAT file into path, size, and SHA-1 entries. Each (or ) supplies the directory, and each child supplies the file metadata. The resulting relative path is game/name + rom/name. """ entries: List[ManifestEntry] = [] by_hash: Dict[str, List[ManifestEntry]] = defaultdict(list) seen_paths: Dict[str, str] = {} try: tree = ET.parse(path) except ET.ParseError as exc: raise ManifestError(f"Invalid XML DAT: {exc}") from exc except OSError as exc: raise ManifestError(f"Could not open DAT: {exc}") from exc root = tree.getroot() root_name = _xml_local_name(root.tag).lower() if root_name != "datafile": raise ManifestError( f"Expected a root element, found <{root_name}>." ) game_count = 0 rom_count = 0 for game in root: game_tag = _xml_local_name(game.tag).lower() if game_tag not in {"game", "machine"}: continue game_count += 1 game_name_raw = (game.get("name") or "").strip() if not game_name_raw: raise ManifestError(f"Game/machine #{game_count} has no name attribute.") game_name = _validate_dat_relative_path( game_name_raw, f"game/machine #{game_count} name" ) for rom in game: if _xml_local_name(rom.tag).lower() != "rom": continue rom_count += 1 rom_name_raw = (rom.get("name") or "").strip() if not rom_name_raw: raise ManifestError( f"ROM #{rom_count} in game {game_name!r} has no name attribute." ) rom_name = _validate_dat_relative_path( rom_name_raw, f"ROM #{rom_count} name in {game_name!r}" ) size_text = (rom.get("size") or "").strip() if not size_text: raise ManifestError( f"ROM {rom_name_raw!r} in {game_name!r} is missing its size attribute." ) try: size = int(size_text, 10) except ValueError as exc: raise ManifestError( f"Invalid size for ROM {rom_name_raw!r} in {game_name!r}: {size_text!r}" ) from exc if size < 0: raise ManifestError( f"Negative size for ROM {rom_name_raw!r} in {game_name!r}." ) digest = (rom.get("sha1") or "").strip().lower() if not HASH_RE.fullmatch(digest): raise ManifestError( f"Missing or invalid SHA-1 for ROM {rom_name_raw!r} in {game_name!r}." ) rel_path = str(PurePosixPath(game_name) / PurePosixPath(rom_name)) rel_path = _validate_dat_relative_path( rel_path, f"ROM {rom_name_raw!r} in {game_name!r}" ) previous = seen_paths.get(rel_path) if previous is not None: raise ManifestError( f"Duplicate destination path in DAT: {rel_path!r}\n" f"First seen under: {previous}\n" f"Seen again under: {game_name}" ) seen_paths[rel_path] = game_name entry = ManifestEntry(digest, size, rel_path) entries.append(entry) by_hash[entry.sha1].append(entry) if not entries: if game_count == 0: raise ManifestError("DAT contains no or entries.") raise ManifestError("DAT contains no entries with usable metadata.") return ManifestData(entries=entries, by_hash=dict(by_hash)) def merge_dats( paths: Sequence[Path], on_loaded: Optional[Callable[[int, int, Path, int], None]] = None, ) -> ManifestData: """Load several DATs as one SHA-1 index without losing DAT provenance. Identical duplicate paths are counted once. Conflicting entries at the same destination path are rejected BEFORE build, to avoid writing ambiguous files. """ if not paths: raise ManifestError("Select at least one XML DAT.") entries: List[ManifestEntry] = [] by_hash: Dict[str, List[ManifestEntry]] = defaultdict(list) dat_counts: Dict[str, int] = {} dat_duplicates: Dict[str, int] = {} entry_source: Dict[str, str] = {} entry_bucket: Dict[str, str] = {} seen: Dict[str, ManifestEntry] = {} total = len(paths) for i, path_value in enumerate(paths, 1): path = Path(path_value) source = str(path.resolve()) dat_bucket = bucket_from_dat_filename(path) if source in dat_counts: raise ManifestError(f"The same DAT was selected twice: {source}") try: parsed = parse_dat(path) except ManifestError as exc: raise ManifestError(f"DAT {path.name}: {exc}") from exc unique_count = 0 duplicates = 0 for entry in parsed.entries: previous = seen.get(entry.rel_path) if previous is not None: if previous.sha1 != entry.sha1 or previous.size != entry.size: first = entry_source[entry.rel_path] raise ManifestError( f"Conflicting DAT records for destination {entry.rel_path!r}:\n" f" First: {first} (size={previous.size}, sha1={previous.sha1})\n" f" Other: {source} (size={entry.size}, sha1={entry.sha1})\n" "Resolve the conflict before building. No files have been modified." ) duplicates += 1 continue seen[entry.rel_path] = entry entry_source[entry.rel_path] = source entry_bucket[entry.rel_path] = dat_bucket entries.append(entry) by_hash[entry.sha1].append(entry) unique_count += 1 dat_counts[source] = unique_count dat_duplicates[source] = duplicates if on_loaded is not None: on_loaded(i, total, path, unique_count) return ManifestData( entries=entries, by_hash=dict(by_hash), dat_counts=dat_counts, dat_duplicates=dat_duplicates, entry_source=entry_source, entry_bucket=entry_bucket, ) def bucket_from_dat_filename(path: Path) -> str: """Identify the collection for optional split-vault destinations. DAT provenance beats guessing a bucket from the title's first character. In particular, Oneshots and _Autouploads aren't letter collections. """ match = re.fullmatch( r"English Manga Vault - ([A-Z]|Oneshots|_Autouploads)", path.stem, flags=re.IGNORECASE, ) if not match: return "" value = match.group(1) if len(value) == 1: return value.upper() return "Oneshots" if value.casefold() == "oneshots" else "_Autouploads" def manifest_bucket(rel_path: str) -> str: """Legacy fallback for DATs without a recognized collection filename.""" parts = PurePosixPath(rel_path).parts if not parts or not parts[0]: return "" return parts[0][0].upper() def manifest_path( root: Path, rel_path: str, bucket_roots: Optional[Dict[str, Path]] = None, bucket: Optional[str] = None, ) -> Path: """Resolve a DAT entry against its *source DAT's collection*, not its title. English Manga Vault - A.dat: ROOT/A/ English Manga Vault - M.dat: ROOT/M/ English Manga Vault - Oneshots.dat: ROOT/Oneshots/ English Manga Vault - _Autouploads.dat ROOT/_Autouploads/ Preserve an existing explicit collection prefix in the DAT path rather than adding it twice. For legacy archives stored directly under ROOT, a file found at ROOT/ remains usable instead of being falsely reported missing. The collection folder is preferred when both exist. An optional per-collection override replaces ROOT, not the DAT-relative collection layout. DATs with unrecognized names use literal ROOT/. """ relative = _validate_dat_relative_path(rel_path, "DAT file destination") parts = PurePosixPath(relative).parts chosen_bucket = bucket if bucket is not None else "" base = Path(root) if bucket_roots: base = Path(bucket_roots.get(chosen_bucket, root)) # Do not guess the collection from a title such as "Manima ni!!". # Only the recognized source DAT filename authorizes this bucket prefix. recognized = bool(re.fullmatch(r"[A-Z]", chosen_bucket)) or chosen_bucket in { "Oneshots", "_Autouploads" } direct = base.joinpath(*parts) if not recognized or parts[0].casefold() == chosen_bucket.casefold(): return direct bucketed = base / chosen_bucket / Path(*parts) if bucketed.is_file(): return bucketed if direct.is_file(): return direct # Compatibility for old DATs generated by manga-sync before the root-file # fix. For a remote file stored directly inside Oneshots/, the old DAT # incorrectly emitted . # Its synthetic 'game' name was mistaken for a real series directory. # If that exact root-level file exists, accept it without moving anything. # A genuine Title/Title.zip directory always takes precedence above. # Never invent an alternate location solely from an unverified name: this # fallback is exclusively for physical files that actually exist on disk. if ( chosen_bucket == "Oneshots" and len(parts) == 2 and parts[0] == Path(parts[1]).stem ): root_level = base / chosen_bucket / parts[1] if root_level.is_file(): return root_level legacy_root_level = base / parts[1] if legacy_root_level.is_file(): return legacy_root_level # When neither exists, report and build in the canonical collection folder. return bucketed ARCHIVE_EXTENSIONS = frozenset({".zip", ".cbz", ".rar", ".cbr", ".7z"}) def _find_archive_tool(names: Sequence[str]) -> Optional[str]: """Find installed tools even when macOS Finder supplies a reduced PATH.""" for name in names: options = [shutil.which(name)] options.extend(str(Path(folder) / name) for folder in ( "/opt/homebrew/bin", "/usr/local/bin", "/usr/bin", "/bin" )) for candidate in options: if candidate and os.path.isfile(candidate) and os.access(candidate, os.X_OK): return candidate return None def _archive_external_test( command: Sequence[str], stop_event: Optional[threading.Event] ) -> Tuple[str, str]: """Test without extracting to disk; allow cancellation during long archives.""" try: proc = subprocess.Popen( list(command), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, text=True, errors="replace", start_new_session=(os.name == "posix"), ) except OSError as exc: return "unverified", f"Could not launch archive tester: {exc}" try: while True: if stop_event is not None and stop_event.is_set(): raise InterruptedError try: output, _ = proc.communicate(timeout=0.35) break except subprocess.TimeoutExpired: continue except BaseException: # On macOS/Linux terminate the entire tester process group. Otherwise # a child of a wrapper executable could keep the output pipe open and # make cancellation block until that child exits. if os.name == "posix": try: os.killpg(proc.pid, signal.SIGKILL) except ProcessLookupError: pass elif proc.poll() is None: proc.kill() proc.communicate() raise message = (output or "").strip() if proc.returncode == 0: if re.search(r"\b(?:files|folders):\s*0\b", message, re.I) and not re.search( r"\bfiles:\s*[1-9]", message, re.I ): return "failed", "Archive tester found no files to check." return "passed", "Archive contents verified by " + Path(command[0]).name if any(term in message.casefold() for term in ( "wrong password", "password required", "encrypted archive", "enter password", "unsupported method", "unsupported compression method", "not supported", )): return "unverified", "Archive needs a password or unsupported codec: " + message[-900:] return "failed", ( f"{Path(command[0]).name} reported exit status {proc.returncode}: " + (message[-1200:] or "Archive integrity test failed") ) def verify_archive_internal( path: Path, stop_event: Optional[threading.Event] = None ) -> Tuple[str, str]: """Verify member data against archive-embedded CRCs/checksums. Status: passed / failed / unverified. No archive is extracted to disk. ZIP and CBZ use Python's CRC32 checks. RAR/CBR and 7z use a real archive test program; absent tools never produce a false success. """ suffix = path.suffix.casefold() if suffix not in ARCHIVE_EXTENSIONS: return "unverified", "Not a supported archive extension" if stop_event is not None and stop_event.is_set(): raise InterruptedError if suffix in {".zip", ".cbz"}: try: with zipfile.ZipFile(path, "r") as archive: members = [info for info in archive.infolist() if not info.is_dir()] if not members: return "failed", "ZIP/CBZ archive contains no file entries" for info in members: if stop_event is not None and stop_event.is_set(): raise InterruptedError try: with archive.open(info, "r") as extracted: while True: if stop_event is not None and stop_event.is_set(): raise InterruptedError if not extracted.read(4 * 1024 * 1024): break # ZipExtFile validates the member's embedded CRC32 at EOF. except (RuntimeError, NotImplementedError) as exc: reason = str(exc).casefold() if any(word in reason for word in ( "password", "encrypted", "unsupported compression", "not supported" )) or isinstance(exc, NotImplementedError): return "unverified", f"Cannot test member {info.filename!r}: {exc}" return "failed", f"Could not decode member {info.filename!r}: {exc}" except (zipfile.BadZipFile, zlib.error, EOFError, OSError, ValueError) as exc: return "failed", f"Bad member {info.filename!r}: {exc}" return "passed", f"CRC32 verified for {len(members):,} ZIP/CBZ member(s)" except InterruptedError: raise except (zipfile.BadZipFile, zipfile.LargeZipFile, zlib.error, EOFError, OSError, ValueError) as exc: return "failed", f"Invalid or unreadable ZIP/CBZ archive: {exc}" if suffix in {".rar", ".cbr"}: unrar = _find_archive_tool(("unrar",)) if unrar: return _archive_external_test((unrar, "t", "-idq", "-p-", str(path)), stop_event) binary = _find_archive_tool(("7zz", "7z")) if binary: return _archive_external_test((binary, "t", "-bd", "-bsp0", "-bso0", str(path)), stop_event) return "unverified", "RAR/CBR verifier unavailable. Install 7-Zip (brew install sevenzip)." binary = _find_archive_tool(("7zz", "7z", "7za")) if binary: return _archive_external_test((binary, "t", "-bd", "-bsp0", "-bso0", str(path)), stop_event) return "unverified", "7z verifier unavailable. Install 7-Zip (brew install sevenzip)." def sha1_file(path: Path, stop_event: Optional[threading.Event] = None) -> str: h = hashlib.sha1() with path.open("rb", buffering=0) as fh: while True: if stop_event is not None and stop_event.is_set(): raise InterruptedError chunk = fh.read(8 * 1024 * 1024) if not chunk: break h.update(chunk) return h.hexdigest() def atomic_copy(source: Path, destination: Path) -> None: """Copy source to destination via a temporary sibling, then atomically replace.""" destination.parent.mkdir(parents=True, exist_ok=True) fd, tmp_name = tempfile.mkstemp( prefix=f".{destination.name}.", suffix=".emvtmp", dir=str(destination.parent) ) os.close(fd) tmp = Path(tmp_name) try: shutil.copy2(source, tmp) os.replace(tmp, destination) finally: try: tmp.unlink(missing_ok=True) except OSError: pass def filesystem_is_case_sensitive(root: Path) -> bool: """Probe case sensitivity with a short-lived empty file in the target root.""" with tempfile.TemporaryDirectory(prefix=".emv-case-check-", dir=str(root)) as directory: probe = Path(directory) / "MixedCaseProbe" probe.touch() return not (probe.parent / "mixedcaseprobe").exists() def format_eta(seconds: float) -> str: """Format an ETA duration as HH:MM:SS (or MM:SS under one hour).""" if seconds < 0 or not (seconds < float("inf")): return "calculating…" total = max(0, int(round(seconds))) hours, remainder = divmod(total, 3600) minutes, secs = divmod(remainder, 60) if hours: return f"{hours:02d}:{minutes:02d}:{secs:02d}" return f"{minutes:02d}:{secs:02d}" class WorkThread(QThread): progress = pyqtSignal(int, int, str) message = pyqtSignal(str) stats = pyqtSignal(dict) report_ready = pyqtSignal(str) failed = pyqtSignal(str) completed = pyqtSignal(str) def __init__( self, mode: str, manifest_paths_value: Sequence[str], vault_root_value: str, to_sort_value: str = "", remove_matched: bool = False, post_verify_mode: str = "quick", conflict_policy: str = "overwrite_verified", vault_root_overrides: Optional[Dict[str, str]] = None, manifest_display_paths: Optional[Dict[str, str]] = None, parent=None, ): super().__init__(parent) self.mode = mode self.manifest_files = tuple(Path(p) for p in manifest_paths_value) self.manifest_display_paths = dict(manifest_display_paths or {}) self.vault_root = Path(vault_root_value) self.vault_root_overrides: Dict[str, Path] = { str(bucket): Path(path) for bucket, path in (vault_root_overrides or {}).items() if str(bucket) and str(path) } self.to_sort = Path(to_sort_value) if to_sort_value else None self.remove_matched = remove_matched self.post_verify_mode = post_verify_mode self.conflict_policy = conflict_policy self.stop_event = threading.Event() self.entry_buckets: Dict[str, str] = {} def cancel(self) -> None: self.stop_event.set() def _check_cancel(self) -> None: if self.stop_event.is_set(): raise InterruptedError def _emit_progress(self, current: int, total: int, label: str) -> None: self.progress.emit(current, max(total, 1), label) def _manifest_path(self, rel_path: str) -> Path: return manifest_path( self.vault_root, rel_path, self.vault_root_overrides, bucket=self.entry_buckets.get(rel_path), ) def _vault_layout_lines(self, prefix: str = "") -> List[str]: lines = [f"{prefix}Default Vault root: {self.vault_root}"] if self.vault_root_overrides: lines.append( f"{prefix}Collection overrides: {len(self.vault_root_overrides):,} configured" ) for bucket in sorted(self.vault_root_overrides, key=lambda s: (s.casefold(), s)): lines.append( f"{prefix} {bucket} -> {self.vault_root_overrides[bucket]} " f"(source DAT collection prefix used unless already present; legacy flat files recognized)" ) else: lines.append(f"{prefix}Collection overrides: none (single-vault layout)") return lines def _validate_vault_roots(self) -> None: if not self.vault_root.exists(): raise RuntimeError(f"Default Vault Root does not exist: {self.vault_root}") if not self.vault_root.is_dir(): raise RuntimeError(f"Default Vault Root is not a directory: {self.vault_root}") for bucket, root in self.vault_root_overrides.items(): if not root.exists(): raise RuntimeError(f"Vault root for {bucket} does not exist: {root}") if not root.is_dir(): raise RuntimeError(f"Vault root for {bucket} is not a directory: {root}") def _dat_report_lines(self, manifest: ManifestData) -> List[str]: lines = [f"DAT files: {len(self.manifest_files):,}"] for path in self.manifest_files: source = str(path.resolve()) n = manifest.dat_counts.get(source, 0) dup = manifest.dat_duplicates.get(source, 0) detail = f"{n:,} unique files" if dup: detail += f", {dup:,} repeated path(s) ignored" lines.append(f" {path.name}: {detail}") lines.append(f" {self.manifest_display_paths.get(source, source)}") lines.append(f"Combined entries: {len(manifest.entries):,}") lines.append(f"Unique SHA-1s: {len(manifest.by_hash):,}") return lines def _per_dat_results( self, manifest: ManifestData, verified: Sequence[ManifestEntry], missing: Sequence[Tuple], mismatches: Sequence[Tuple], ) -> List[str]: """Compact per-DAT verification totals, with a combined detailed report below.""" def counter(rows: Sequence, unpack: bool) -> Counter: return Counter( manifest.entry_source[row[0].rel_path if unpack else row.rel_path] for row in rows ) ok_counts = counter(verified, False) miss_counts = counter(missing, True) bad_counts = counter(mismatches, True) lines = ["PER-DAT RESULTS", "-" * 80] lines.append("DAT file Expected Verified Missing Mismatch") for path in self.manifest_files: source = str(path.resolve()) lines.append( f"{path.name[:45]:45s} " f"{manifest.dat_counts.get(source, 0):10,d} " f"{ok_counts[source]:10,d} " f"{miss_counts[source]:10,d} " f"{bad_counts[source]:10,d}" ) lines.append("") return lines def run(self) -> None: try: def loaded(i: int, total: int, path: Path, count: int) -> None: self._check_cancel() self.message.emit( f"Loaded DAT {i:,}/{total:,}: {path.name} ({count:,} unique files)" ) self._emit_progress(i, total, f"Loading DAT {i:,}/{total:,}: {path.name}") manifest = merge_dats(self.manifest_files, loaded) self.entry_buckets = dict(manifest.entry_bucket) self.message.emit( f"Loaded {len(self.manifest_files):,} DAT files; " f"{len(manifest.entries):,} combined entries " f"({len(manifest.by_hash):,} unique SHA-1 values)." ) if self.mode == "quick_verify": self._run_quick_verify(manifest, emit_final=True) elif self.mode == "full_verify": self._run_full_verify(manifest, emit_final=True) elif self.mode == "internal_verify": self._run_internal_verify(manifest, emit_final=True) elif self.mode == "build": self._run_build(manifest) else: raise RuntimeError(f"Unknown worker mode: {self.mode}") except InterruptedError: self.report_ready.emit("Operation cancelled by user.\n") self.completed.emit("Cancelled") except Exception as exc: self.failed.emit(f"{type(exc).__name__}: {exc}") def _run_quick_verify(self, manifest: ManifestData, emit_final: bool) -> Tuple[dict, str]: """Quick Verify: require exact DAT path and exact DAT-listed file size. File contents are not hashed. This is intentionally much faster than Full Verify while still detecting truncated, oversized, or otherwise size-different files at the correct path. """ started = time.monotonic() total = len(manifest.entries) self._validate_vault_roots() present: List[ManifestEntry] = [] verified: List[ManifestEntry] = [] missing: List[Tuple[ManifestEntry, Path]] = [] size_mismatches: List[Tuple[ManifestEntry, Path, int]] = [] path_errors: List[Tuple[ManifestEntry, Path, str]] = [] self.message.emit( "Quick Verify: checking exact DAT paths and listed file sizes…" ) progress_started = time.monotonic() for i, entry in enumerate(manifest.entries, 1): self._check_cancel() expected_path = self._manifest_path(entry.rel_path) try: if not expected_path.is_file(): missing.append((entry, expected_path)) else: present.append(entry) actual_size = expected_path.stat().st_size if actual_size == entry.size: verified.append(entry) else: size_mismatches.append((entry, expected_path, actual_size)) except OSError as exc: path_errors.append((entry, expected_path, str(exc))) if i == 1 or i % 100 == 0 or i == total: elapsed_progress = max(0.001, time.monotonic() - progress_started) file_rate = i / elapsed_progress if i > 0 else 0.0 remaining_files = max(0, total - i) eta_seconds = remaining_files / file_rate if file_rate > 0 else float("inf") pct = (i / max(total, 1)) * 100.0 eta_text = "00:00" if i >= total else format_eta(eta_seconds) phase = ( "Post-build Quick Verify" if self.mode == "build" and not emit_final else "Quick Verify" ) self._emit_progress( i, total, f"{phase} {i:,} / {total:,} — {pct:04.1f}% — {eta_text} — {entry.rel_path}", ) self.stats.emit( { "expected": total, "present": len(present), "verified": len(verified), "missing": len(missing), "mismatch": len(size_mismatches), } ) elapsed = time.monotonic() - started stats = { "expected": total, "present": len(present), "verified": len(verified), "missing": len(missing), "mismatch": len(size_mismatches), } self.stats.emit(stats) lines: List[str] = [] lines.append(APP_NAME) lines.append("=" * len(APP_NAME)) lines.append("") lines.append("QUICK VERIFICATION SUMMARY") lines.extend(self._dat_report_lines(manifest)) lines.extend(self._vault_layout_lines()) if self.vault_root_overrides: lines.append( "Path rule: Prefix recognized DAT collection (A-Z, Oneshots, _Autouploads) once; allow existing flat files" ) else: lines.append( "Path rule: Default Root / DAT collection / relative path, or existing legacy flat file" ) lines.append("Check rule: Exact path + exact DAT-listed size — no SHA-1 calculation") lines.append(f"Expected files: {total:,}") lines.append(f"Present files: {len(present):,}") lines.append(f"Size verified: {len(verified):,}") lines.append(f"Missing files: {len(missing):,}") lines.append(f"Size mismatch: {len(size_mismatches):,}") if path_errors: lines.append(f"Path errors: {len(path_errors):,}") lines.append(f"Elapsed: {elapsed:.1f} seconds") lines.append("") lines.extend(self._per_dat_results(manifest, verified, missing, size_mismatches)) if missing: lines.append("MISSING FILES") lines.append("-" * 80) for entry, expected_path in missing: lines.append(entry.rel_path) lines.append(f" expected at: {expected_path}") lines.append("") if size_mismatches: lines.append("SIZE MISMATCHES") lines.append("-" * 80) for entry, expected_path, actual_size in size_mismatches: lines.append(entry.rel_path) lines.append(f" file: {expected_path}") lines.append(f" expected size: {entry.size:,} bytes") lines.append(f" actual size: {actual_size:,} bytes") lines.append("") if path_errors: lines.append("PATH ACCESS ERRORS") lines.append("-" * 80) for entry, expected_path, error in path_errors: lines.append(entry.rel_path) lines.append(f" file: {expected_path}") lines.append(f" error: {error}") lines.append("") if not missing and not size_mismatches and not path_errors: lines.append( "RESULT: QUICK VERIFY PASSED — every DAT file exists at its exact expected path " "and matches its listed size. File contents were not hashed." ) else: lines.append("RESULT: QUICK VERIFY FAILED. See the sections above.") report = "\n".join(lines) + "\n" if emit_final: self.report_ready.emit(report) self.completed.emit("Quick Verify complete") return stats, report def _run_full_verify(self, manifest: ManifestData, emit_final: bool) -> Tuple[dict, str]: """Full Verify: verify every manifest entry by exact path and SHA-1. Paths come from the DAT's named collection folder (A-Z, Oneshots, _Autouploads) unless they already include the collection prefix, with a fallback for pre-existing files stored flat beneath the selected root. Every found file is compared with its corresponding DAT SHA-1. """ started = time.monotonic() total = len(manifest.entries) self._validate_vault_roots() # Gather sizes using the exact manifest destinations. The displayed # percentage remains entry-based, while ETA uses bytes when possible so # large archives are weighted more realistically than tiny files. verify_sizes: Dict[str, int] = {} total_verify_bytes = 0 for entry in manifest.entries: expected_path = self._manifest_path(entry.rel_path) try: size = expected_path.stat().st_size if expected_path.is_file() else 0 except OSError: size = 0 size = max(0, size) verify_sizes[entry.rel_path] = size total_verify_bytes += size processed_verify_bytes = 0 verify_progress_started = time.monotonic() verified: List[ManifestEntry] = [] missing: List[Tuple[ManifestEntry, Path]] = [] mismatched: List[Tuple[ManifestEntry, Path, str]] = [] path_errors: List[Tuple[ManifestEntry, Path, str]] = [] hash_errors: List[Tuple[ManifestEntry, Path, str]] = [] present_count = 0 self.message.emit( "Full Verify: resolving each DAT entry, then calculating and comparing SHA-1…" ) for i, entry in enumerate(manifest.entries, 1): self._check_cancel() # This is the only path that can satisfy this manifest entry. expected_path = self._manifest_path(entry.rel_path) try: if not expected_path.is_file(): missing.append((entry, expected_path)) else: present_count += 1 try: actual_sha1 = sha1_file(expected_path, self.stop_event) except InterruptedError: raise except OSError as exc: hash_errors.append((entry, expected_path, str(exc))) else: # Compare against the SHA-1 from this exact DAT entry. if actual_sha1 == entry.sha1: verified.append(entry) else: mismatched.append((entry, expected_path, actual_sha1)) except InterruptedError: raise except OSError as exc: path_errors.append((entry, expected_path, str(exc))) processed_verify_bytes += verify_sizes.get(entry.rel_path, 0) if i == 1 or i % 10 == 0 or i == total: elapsed_progress = max(0.001, time.monotonic() - verify_progress_started) if total_verify_bytes > 0 and processed_verify_bytes > 0: byte_rate = processed_verify_bytes / elapsed_progress remaining_bytes = max(0, total_verify_bytes - processed_verify_bytes) eta_seconds = remaining_bytes / byte_rate if byte_rate > 0 else float("inf") else: file_rate = i / elapsed_progress if i > 0 else 0.0 remaining_files = max(0, total - i) eta_seconds = remaining_files / file_rate if file_rate > 0 else float("inf") pct = (i / max(total, 1)) * 100.0 eta_text = "00:00" if i >= total else format_eta(eta_seconds) phase = "Post-build Full Verify" if self.mode == "build" and not emit_final else "Full Verify" self._emit_progress( i, total, f"{phase} {i:,} / {total:,} — {pct:04.1f}% — ETA {eta_text} — {entry.rel_path}", ) self.stats.emit( { "expected": total, "present": present_count, "verified": len(verified), "missing": len(missing), "mismatch": len(mismatched), } ) elapsed = time.monotonic() - started stats = { "expected": total, "present": present_count, "verified": len(verified), "missing": len(missing), "mismatch": len(mismatched), } self.stats.emit(stats) lines: List[str] = [] lines.append(APP_NAME) lines.append("=" * len(APP_NAME)) lines.append("") lines.append("FULL VERIFICATION SUMMARY") lines.extend(self._dat_report_lines(manifest)) lines.extend(self._vault_layout_lines()) if self.vault_root_overrides: lines.append( "Path rule: Prefix recognized DAT collection (A-Z, Oneshots, _Autouploads) once; allow existing flat files" ) else: lines.append("Path rule: Default Root / DAT collection / relative path, or existing legacy flat file") lines.append("Hash rule: File SHA-1 must match sha1= on the same DAT entry") lines.append(f"Expected files: {total:,}") lines.append(f"Present files: {present_count:,}") lines.append(f"Verified files: {len(verified):,}") lines.append(f"Missing files: {len(missing):,}") lines.append(f"Hash mismatch: {len(mismatched):,}") if path_errors: lines.append(f"Path errors: {len(path_errors):,}") if hash_errors: lines.append(f"Read errors: {len(hash_errors):,}") lines.append(f"Elapsed: {elapsed:.1f} seconds") lines.append("") lines.extend(self._per_dat_results(manifest, verified, missing, mismatched)) if missing: lines.append("MISSING FILES") lines.append("-" * 80) for entry, expected_path in missing: lines.append(entry.rel_path) lines.append(f" expected at: {expected_path}") lines.append("") if mismatched: lines.append("HASH MISMATCHES") lines.append("-" * 80) for entry, expected_path, actual in mismatched: lines.append(entry.rel_path) lines.append(f" file: {expected_path}") lines.append(f" expected: {entry.sha1}") lines.append(f" actual: {actual}") lines.append("") if path_errors: lines.append("PATH ACCESS ERRORS") lines.append("-" * 80) for entry, expected_path, error in path_errors: lines.append(entry.rel_path) lines.append(f" file: {expected_path}") lines.append(f" error: {error}") lines.append("") if hash_errors: lines.append("FILE READ ERRORS") lines.append("-" * 80) for entry, expected_path, error in hash_errors: lines.append(entry.rel_path) lines.append(f" file: {expected_path}") lines.append(f" error: {error}") lines.append("") if not missing and not mismatched and not path_errors and not hash_errors: lines.append( "RESULT: COMPLETE SET VERIFIED — every DAT relative path exists " "under Vault Root and every file matches its listed SHA-1." ) else: lines.append("RESULT: SET IS NOT COMPLETE/VALID. See the sections above.") report = "\n".join(lines) + "\n" if emit_final: self.report_ready.emit(report) self.completed.emit("Full Verify complete") return stats, report def _run_internal_verify(self, manifest: ManifestData, emit_final: bool) -> Tuple[dict, str]: """Test archive-contained CRCs/checksums, independent of external DAT SHA-1. Zip/CBZ: all members read and CRC32 checked using the stdlib. RAR/CBR/7z: use archive utility test mode; no extraction to disk. Unknown extensions are excluded from the archive count and reported. """ started = time.monotonic() self._validate_vault_roots() archives = [e for e in manifest.entries if Path(e.rel_path).suffix.casefold() in ARCHIVE_EXTENSIONS] total = len(archives) skipped_count = len(manifest.entries) - total passed: List[ManifestEntry] = [] missing: List[Tuple[ManifestEntry, Path]] = [] failed: List[Tuple[ManifestEntry, Path, str]] = [] unverified: List[Tuple[ManifestEntry, Path, str]] = [] self.message.emit("Internal Verify: checking embedded archive CRCs/checksums…") for i, entry in enumerate(archives, 1): self._check_cancel() path = self._manifest_path(entry.rel_path) try: if not path.is_file(): missing.append((entry, path)) else: status, detail = verify_archive_internal(path, self.stop_event) if status == "passed": passed.append(entry) elif status == "failed": failed.append((entry, path, detail)) else: unverified.append((entry, path, detail)) except InterruptedError: raise except OSError as exc: failed.append((entry, path, f"Cannot read file: {exc}")) if i == 1 or i % 10 == 0 or i == total: elapsed = max(0.001, time.monotonic() - started) eta = (total - i) * elapsed / i self._emit_progress( i, total, f"Internal Verify {i:,} / {total:,} — {i / max(total, 1) * 100:04.1f}% " f"— ETA {format_eta(eta)} — {entry.rel_path}", ) self.stats.emit({ "expected": total, "present": i - len(missing), "verified": len(passed), "missing": len(missing), "mismatch": len(failed), }) stats = { "expected": total, "present": len(passed) + len(failed) + len(unverified), "verified": len(passed), "missing": len(missing), "mismatch": len(failed), } self.stats.emit(stats) lines = [ APP_NAME, "=" * len(APP_NAME), "", "INTERNAL VERIFICATION SUMMARY", *self._dat_report_lines(manifest), *self._vault_layout_lines(), "Check rule: Decompress/stream archive members and check embedded CRCs/checksums", "DAT SHA-1: Not checked by this mode (use Full Verify for that)", f"Archive files: {total:,}", f"Passed: {len(passed):,}", f"Missing: {len(missing):,}", f"Corrupt / error: {len(failed):,}", f"Unverified: {len(unverified):,}", f"Other extensions: {skipped_count:,} skipped", f"Elapsed: {time.monotonic() - started:.1f} seconds", "", "Format backends: ZIP/CBZ = Python CRC32; RAR/CBR = unrar or 7-Zip; 7z = 7-Zip.", "An unsupported or encrypted archive is not counted as passed.", "", ] lines.extend(self._per_dat_results(manifest, passed, missing, failed)) if missing: lines.extend(["MISSING ARCHIVES", "-" * 80]) for entry, path in missing: lines.extend([entry.rel_path, f" expected at: {path}"]) lines.append("") if failed: lines.extend(["CORRUPT / FAILED ARCHIVES", "-" * 80]) for entry, path, detail in failed: lines.extend([entry.rel_path, f" file: {path}", f" reason: {detail}"]) lines.append("") if unverified: lines.extend(["UNVERIFIED ARCHIVES (BACKEND / PASSWORD / CODEC)", "-" * 80]) for entry, path, detail in unverified: lines.extend([entry.rel_path, f" file: {path}", f" reason: {detail}"]) lines.append("") if total == 0: lines.append("RESULT: NO SUPPORTED ARCHIVES IN THE SELECTED DAT FILES.") elif not missing and not failed and not unverified: lines.append("RESULT: INTERNAL VERIFY PASSED — all archive contents tested successfully.") else: lines.append("RESULT: INTERNAL VERIFY INCOMPLETE/FAILED — review the sections above.") report = "\n".join(lines) + "\n" if emit_final: self.report_ready.emit(report) self.completed.emit("Internal Verify complete") return stats, report def _remove_empty_source_directories(self) -> Tuple[List[str], List[Tuple[str, str]]]: """Remove empty subdirectories beneath Source Files, deepest first. The selected Source Files root itself is never removed. A directory is removed only when it is actually empty at cleanup time. """ removed: List[str] = [] errors: List[Tuple[str, str]] = [] if self.to_sort is None or not self.to_sort.is_dir(): return removed, errors for root, dirs, files in os.walk(self.to_sort, topdown=False): self._check_cancel() folder = Path(root) if folder == self.to_sort: continue try: # Check again at deletion time because child folders may have # just been removed during this bottom-up walk. if any(folder.iterdir()): continue folder.rmdir() removed.append(str(folder)) except OSError as exc: errors.append((str(folder), str(exc))) return removed, errors def _run_build(self, manifest: ManifestData) -> None: if self.to_sort is None: raise RuntimeError("No Source Files directory selected.") if not self.to_sort.is_dir(): raise RuntimeError("The selected Source Files path is not a directory.") if self.conflict_policy not in {"skip", "overwrite_verified", "report"}: raise RuntimeError(f"Unknown conflict policy: {self.conflict_policy}") self._validate_vault_roots() # Source Files must be separate from every managed Vault tree. Otherwise # Remove Matched could accidentally delete an already-valid vault file. source_root = self.to_sort.resolve() for managed_root in (self.vault_root, *self.vault_root_overrides.values()): managed = managed_root.resolve() if (source_root == managed or source_root in managed.parents or managed in source_root.parents): raise RuntimeError( f"Source Files overlaps a managed Vault Root:\n" f" Source: {source_root}\n Vault: {managed}\n" "Choose a separate source directory to prevent accidental deletion." ) # macOS volumes are commonly case-insensitive. A few letter DATs contain # paths differing only in capitalization but with different SHA-1 values. # They cannot coexist on such a volume. Skip BOTH ambiguous paths so # rebuilding one cannot destroy the other or prematurely remove sources. seen_case_paths: Dict[str, ManifestEntry] = {} case_conflict_pairs: List[Tuple[ManifestEntry, ManifestEntry]] = [] for entry in manifest.entries: resolved = str(self._manifest_path(entry.rel_path)).casefold() previous = seen_case_paths.get(resolved) if previous is not None and previous.rel_path != entry.rel_path: case_conflict_pairs.append((previous, entry)) else: seen_case_paths[resolved] = entry insensitive_roots: Dict[Path, bool] = {} unsafe_case_paths: set[str] = set() unsafe_case_pairs: List[Tuple[str, str]] = [] for first, second in case_conflict_pairs: # Two different DAT paths that resolve to precisely the same target # are unsafe even on a case-sensitive filesystem. if self._manifest_path(first.rel_path) == self._manifest_path(second.rel_path): unsafe_case_paths.update((first.rel_path, second.rel_path)) unsafe_case_pairs.append((first.rel_path, second.rel_path)) continue bucket = self.entry_buckets.get(first.rel_path, manifest_bucket(first.rel_path)) root = self.vault_root_overrides.get(bucket, self.vault_root).resolve() if root not in insensitive_roots: try: insensitive_roots[root] = not filesystem_is_case_sensitive(root) except OSError as exc: raise RuntimeError( f"Could not determine filename case sensitivity for {root}: {exc}" ) from exc if insensitive_roots[root]: unsafe_case_paths.update((first.rel_path, second.rel_path)) unsafe_case_pairs.append((first.rel_path, second.rel_path)) if unsafe_case_paths: self.message.emit( f"Warning: {len(unsafe_case_pairs):,} case-only filename collision(s) " "at resolved destinations; ambiguous entries will be skipped." ) started = time.monotonic() expected_by_hash = manifest.by_hash self.message.emit("Scanning Source Files recursively…") selected_dat_paths = {p.resolve() for p in self.manifest_files} source_files: List[Path] = [] for root, dirs, files in os.walk(self.to_sort): self._check_cancel() root_path = Path(root) for name in files: p = root_path / name try: if p.resolve() in selected_dat_paths: continue except OSError: pass source_files.append(p) total_sources = len(source_files) if total_sources == 0: raise RuntimeError("The Source Files directory contains no files.") # Pre-calculate source sizes so build progress can use processed bytes, # which produces a much more useful ETA than file count when manga # archives vary greatly in size. Files whose size cannot be read fall # back to zero and still advance by file-count progress. source_sizes: Dict[Path, int] = {} total_source_bytes = 0 for source in source_files: try: size = max(0, source.stat().st_size) except OSError: size = 0 source_sizes[source] = size total_source_bytes += size build_progress_started = time.monotonic() processed_source_bytes = 0 matched_sources = 0 unmatched_sources: List[Tuple[Path, str]] = [] source_errors: List[Tuple[Path, str]] = [] restored_paths: List[str] = [] already_correct: List[str] = [] replaced_wrong: List[str] = [] conflicts: List[BuildConflict] = [] destination_errors: List[Tuple[str, str]] = [] removed_sources: List[str] = [] remove_errors: List[Tuple[str, str]] = [] conflict_report_path: Optional[Path] = None conflict_report_error: Optional[str] = None removed_empty_dirs: List[str] = [] empty_dir_errors: List[Tuple[str, str]] = [] fulfilled: set[str] = set() conflicted_paths: set[str] = set() self.message.emit("Hashing Source Files and rebuilding expected paths…") def emit_build_progress(i: int, source: Path) -> None: nonlocal processed_source_bytes processed_source_bytes += source_sizes.get(source, 0) elapsed = max(0.001, time.monotonic() - build_progress_started) # Prefer byte-weighted progress/ETA because archive sizes can differ # dramatically. Fall back to file-count progress for empty/unreadable # sources or when no byte total is available. if total_source_bytes > 0: fraction = min(1.0, processed_source_bytes / total_source_bytes) rate = processed_source_bytes / elapsed if processed_source_bytes > 0 else 0.0 remaining = max(0, total_source_bytes - processed_source_bytes) eta_seconds = (remaining / rate) if rate > 0 else float("inf") else: fraction = min(1.0, i / max(total_sources, 1)) rate = i / elapsed if i > 0 else 0.0 remaining = max(0, total_sources - i) eta_seconds = (remaining / rate) if rate > 0 else float("inf") pct = fraction * 100.0 eta_text = "00:00" if i >= total_sources else format_eta(eta_seconds) label = ( f"Building from Source Files {i:,} / {total_sources:,} " f"— {pct:04.1f}% — ETA {eta_text}" ) # Scale to 10,000 so the GUI can display tenths/hundredths smoothly # while retaining the existing integer progress signal. self.progress.emit(int(round(fraction * 10000)), 10000, label) for i, source in enumerate(source_files, 1): self._check_cancel() try: digest = sha1_file(source, self.stop_event) except InterruptedError: raise except OSError as exc: source_errors.append((source, str(exc))) emit_build_progress(i, source) continue targets = expected_by_hash.get(digest) if not targets: unmatched_sources.append((source, digest)) emit_build_progress(i, source) continue matched_sources += 1 source_success = True for entry in targets: self._check_cancel() if entry.rel_path in unsafe_case_paths: source_success = False continue dest = self._manifest_path(entry.rel_path) if entry.rel_path in fulfilled or entry.rel_path in conflicted_paths: continue try: if dest.is_file(): current = sha1_file(dest, self.stop_event) if current == entry.sha1: already_correct.append(entry.rel_path) fulfilled.add(entry.rel_path) continue if self.conflict_policy == "skip": conflicted_paths.add(entry.rel_path) conflicts.append(BuildConflict( entry.rel_path, str(source), entry.sha1, current, "Skipped; existing destination left unchanged." )) continue if self.conflict_policy == "report": conflicted_paths.add(entry.rel_path) conflicts.append(BuildConflict( entry.rel_path, str(source), entry.sha1, current, "Reported; existing destination left unchanged." )) continue # overwrite_verified: replace only when the Source File SHA-1 # exactly matches the DAT entry. if digest != entry.sha1: conflicted_paths.add(entry.rel_path) conflicts.append(BuildConflict( entry.rel_path, str(source), entry.sha1, current, "Skipped; source SHA-1 did not match DAT." )) source_success = False continue atomic_copy(source, dest) copied_hash = sha1_file(dest, self.stop_event) if copied_hash != entry.sha1: source_success = False destination_errors.append(( entry.rel_path, "Post-copy SHA-1 verification failed: " f"expected {entry.sha1}, got {copied_hash}", )) conflicted_paths.add(entry.rel_path) conflicts.append(BuildConflict( entry.rel_path, str(source), entry.sha1, current, "Overwrite attempted, but post-copy SHA-1 verification failed." )) continue replaced_wrong.append(entry.rel_path) fulfilled.add(entry.rel_path) conflicts.append(BuildConflict( entry.rel_path, str(source), entry.sha1, current, "Overwritten with SHA-1 verified source." )) else: atomic_copy(source, dest) restored_paths.append(entry.rel_path) fulfilled.add(entry.rel_path) except InterruptedError: raise except OSError as exc: source_success = False destination_errors.append((entry.rel_path, str(exc))) if self.remove_matched and source_success: if all(e.rel_path in fulfilled for e in targets): try: source.unlink() removed_sources.append(str(source)) except OSError as exc: remove_errors.append((str(source), str(exc))) emit_build_progress(i, source) self.message.emit("Cleaning empty folders from Source Files…") removed_empty_dirs, empty_dir_errors = self._remove_empty_source_directories() if self.conflict_policy == "report" and conflicts: timestamp = time.strftime("%Y-%m-%d_%H-%M-%S") candidate = self.to_sort / ( f"English Manga Vault Builder - Conflict Report - {timestamp}.txt" ) conflict_lines: List[str] = [ APP_NAME, "=" * len(APP_NAME), "", "CONFLICT REPORT", *self._dat_report_lines(manifest), f"Default root: {self.vault_root}", f"Source Files: {self.to_sort}", f"Conflicts: {len(conflicts):,}", "", "A conflict means the expected destination already existed but its", "SHA-1 did not match the DAT. No conflicting destination was", "overwritten under this policy.", "", ] if self.vault_root_overrides: conflict_lines.append("LETTER VAULT OVERRIDES") for bucket in sorted(self.vault_root_overrides, key=lambda s: (s.casefold(), s)): conflict_lines.append(f" {bucket} -> {self.vault_root_overrides[bucket]}") conflict_lines.append("") for n, conflict in enumerate(conflicts, 1): conflict_lines.extend([ f"[{n}] {conflict.rel_path}", f" source: {conflict.source_path}", f" expected sha1: {conflict.expected_sha1}", f" existing sha1: {conflict.existing_sha1}", f" action: {conflict.action}", "", ]) try: candidate.write_text("\n".join(conflict_lines), encoding="utf-8") conflict_report_path = candidate except OSError as exc: conflict_report_error = str(exc) build_elapsed = time.monotonic() - started policy_names = { "skip": "Skip conflicting destinations", "overwrite_verified": "Overwrite only with SHA-1 match", "report": "Record conflicts in separate report; do not overwrite", } skipped_conflicts = sum( 1 for c in conflicts if c.action.startswith(("Skipped", "Reported")) ) overwritten_conflicts = sum( 1 for c in conflicts if c.action.startswith("Overwritten") ) lines: List[str] = [] lines.append(APP_NAME) lines.append("=" * len(APP_NAME)) lines.append("") lines.append("BUILD SUMMARY") lines.extend(self._dat_report_lines(manifest)) lines.append(f"Default Vault root: {self.vault_root}") lines.append(f"Letter overrides: {len(self.vault_root_overrides):,}") for bucket in sorted(self.vault_root_overrides, key=lambda s: (s.casefold(), s)): lines.append(f" {bucket} -> {self.vault_root_overrides[bucket]}") lines.append(f"Source Files: {self.to_sort}") lines.append(f"Conflict policy: {policy_names[self.conflict_policy]}") lines.append(f"Files scanned: {total_sources:,}") lines.append(f"Matched source files: {matched_sources:,}") lines.append(f"New paths restored: {len(restored_paths):,}") lines.append(f"Wrong files replaced: {len(replaced_wrong):,}") lines.append(f"Already correct: {len(already_correct):,}") lines.append(f"Conflicts detected: {len(conflicts):,}") lines.append(f"Conflicts overwritten: {overwritten_conflicts:,}") lines.append(f"Conflicts left intact: {skipped_conflicts:,}") lines.append(f"Unmatched sources: {len(unmatched_sources):,}") lines.append(f"Unsafe case paths: {len(unsafe_case_paths):,} skipped") lines.append(f"Build elapsed: {build_elapsed:.1f} seconds") if self.remove_matched: lines.append(f"Sources removed: {len(removed_sources):,}") lines.append(f"Empty folders removed: {len(removed_empty_dirs):,}") if empty_dir_errors: lines.append(f"Folder cleanup errors: {len(empty_dir_errors):,}") if conflict_report_path is not None: lines.append(f"Conflict report: {conflict_report_path}") if conflict_report_error is not None: lines.append(f"Conflict report error: {conflict_report_error}") lines.append("") if unsafe_case_pairs: lines.append("AMBIGUOUS DESTINATION PATHS — SKIPPED") lines.append("-" * 80) lines.append("These DAT entries map to the same destination, or differ only by case on a case-insensitive volume.") lines.append("The builder did not copy either path or delete its matching Source Files.") for first, second in unsafe_case_pairs: lines.append(f" {first}") lines.append(f" {second}") lines.append("") if restored_paths: lines.append("RESTORED PATHS") lines.append("-" * 80) lines.extend(restored_paths) lines.append("") if replaced_wrong: lines.append("REPLACED HASH-MISMATCHED DESTINATIONS") lines.append("-" * 80) lines.extend(replaced_wrong) lines.append("") if conflicts: lines.append("DESTINATION CONFLICTS") lines.append("-" * 80) for conflict in conflicts: lines.append(conflict.rel_path) lines.append(f" source: {conflict.source_path}") lines.append(f" expected sha1: {conflict.expected_sha1}") lines.append(f" existing sha1: {conflict.existing_sha1}") lines.append(f" action: {conflict.action}") lines.append("") if unmatched_sources: lines.append("UNMATCHED FILES LEFT IN SOURCE FILES") lines.append("-" * 80) for source, digest in unmatched_sources: lines.append(f"{source}\n sha1: {digest}") lines.append("") if source_errors: lines.append("SOURCE READ ERRORS") lines.append("-" * 80) for source, error in source_errors: lines.append(f"{source}\n {error}") lines.append("") if destination_errors: lines.append("DESTINATION ERRORS") lines.append("-" * 80) for rel, error in destination_errors: lines.append(f"{rel}\n {error}") lines.append("") if remove_errors: lines.append("SOURCE REMOVAL ERRORS") lines.append("-" * 80) for source, error in remove_errors: lines.append(f"{source}\n {error}") lines.append("") if empty_dir_errors: lines.append("EMPTY FOLDER CLEANUP ERRORS") lines.append("-" * 80) for folder, error in empty_dir_errors: lines.append(f"{folder}\n {error}") lines.append("") if self.post_verify_mode == "quick": self.message.emit("Build finished — performing Quick Verify (path + DAT size)…") verify_stats, verify_report = self._run_quick_verify(manifest, emit_final=False) lines.append("POST-BUILD QUICK VERIFY") lines.append("=" * 80) lines.append(verify_report) self.stats.emit(verify_stats) elif self.post_verify_mode == "full": self.message.emit("Build finished — performing Full Verify (fresh SHA-1)…") verify_stats, verify_report = self._run_full_verify(manifest, emit_final=False) lines.append("POST-BUILD FULL VERIFY") lines.append("=" * 80) lines.append(verify_report) self.stats.emit(verify_stats) elif self.post_verify_mode == "internal": self.message.emit("Build finished — performing Internal Verify (archive CRC/checksums)…") verify_stats, verify_report = self._run_internal_verify(manifest, emit_final=False) lines.append("POST-BUILD INTERNAL VERIFY") lines.append("=" * 80) lines.append(verify_report) self.stats.emit(verify_stats) elif self.post_verify_mode == "none": self.stats.emit({ "expected": len(manifest.entries), "present": len(fulfilled), "verified": len(fulfilled), "missing": max(0, len(manifest.entries) - len(fulfilled)), "mismatch": len(conflicts) - overwritten_conflicts, }) else: raise RuntimeError(f"Unknown post-build verification mode: {self.post_verify_mode}") self.report_ready.emit("\n".join(lines) + "\n") self.completed.emit("Build complete") class PathRow(QFrame): def __init__(self, title: str, placeholder: str, button_text: str, directory: bool): super().__init__() self.directory = directory self.setObjectName("pathCard") layout = QVBoxLayout(self) layout.setContentsMargins(18, 14, 18, 14) layout.setSpacing(8) label = QLabel(title) label.setObjectName("fieldLabel") layout.addWidget(label) row = QHBoxLayout() row.setSpacing(10) self.edit = QLineEdit() self.edit.setPlaceholderText(placeholder) self.edit.setClearButtonEnabled(True) row.addWidget(self.edit, 1) self.button = QPushButton(button_text) self.button.setObjectName("browseButton") self.button.clicked.connect(self.browse) row.addWidget(self.button) layout.addLayout(row) def browse(self) -> None: if self.directory: value = QFileDialog.getExistingDirectory(self, "Select directory", self.edit.text()) else: value, _ = QFileDialog.getOpenFileName( self, "Select XML DAT", self.edit.text(), "DAT files (*.dat *.xml);;DAT files (*.dat);;XML files (*.xml);;All files (*)", ) if value: self.edit.setText(value) # Only DAT/XML members are staged, into app-owned temporary files. Never # extract archive-controlled paths directly or unpack the manga collection. DAT_ZIP_MAX_FILES = 4096 DAT_ZIP_MAX_MEMBER_SIZE = 1 * 1024**3 # 1 GiB per DAT DAT_ZIP_MAX_TOTAL_SIZE = 4 * 1024**3 # 4 GiB per selected DAT ZIP def stage_dat_zip(zip_path: Path, staging_dir: Path) -> List[Tuple[str, Path]]: """Stage DATs in a ZIP safely and return (archive member, local DAT) pairs. Basenames are retained so ``English Manga Vault - A.dat`` still identifies the A collection. Each member is isolated in a numbered subdirectory, avoiding collisions between identically named DATs in different ZIP folders. """ source = Path(zip_path) output = Path(staging_dir) if output.exists(): raise FileExistsError(f"DAT ZIP staging folder already exists: {output}") extracted: List[Tuple[str, Path]] = [] try: with zipfile.ZipFile(source, "r") as archive: members = [] total_size = 0 for info in archive.infolist(): if info.is_dir(): continue normalized = info.filename.replace("\\", "/") filename = PurePosixPath(normalized).name if (not filename or filename.startswith("._") or normalized.startswith("__MACOSX/") or Path(filename).suffix.lower() not in {".dat", ".xml"}): continue if filename in {".", ".."} or len(filename.encode("utf-8")) > 240: raise ManifestError(f"Unsafe DAT member filename: {info.filename!r}") if info.flag_bits & 0x1: raise ManifestError(f"Encrypted DAT ZIP member not supported: {info.filename!r}") if info.file_size > DAT_ZIP_MAX_MEMBER_SIZE: raise ManifestError(f"DAT ZIP member exceeds 1 GiB: {info.filename!r}") total_size += info.file_size if total_size > DAT_ZIP_MAX_TOTAL_SIZE: raise ManifestError("DAT ZIP exceeds 4 GiB of uncompressed DAT/XML data.") members.append((info, filename)) if len(members) > DAT_ZIP_MAX_FILES: raise ManifestError("DAT ZIP contains over 4,096 DAT/XML members.") if not members: raise ManifestError("Selected ZIP contains no .dat or .xml files.") output.mkdir(parents=True) for i, (info, filename) in enumerate(members, 1): member_dir = output / f"{i:06d}" member_dir.mkdir() target = member_dir / filename # Streaming through ZipExtFile validates the embedded CRC32. # Never use extract()/extractall(): archive paths are untrusted. copied = 0 with archive.open(info, "r") as src, target.open("wb") as dst: while True: chunk = src.read(8 * 1024 * 1024) if not chunk: break copied += len(chunk) if copied > DAT_ZIP_MAX_MEMBER_SIZE or copied > info.file_size: raise ManifestError(f"DAT ZIP member exceeds declared size: {info.filename!r}") dst.write(chunk) if copied != info.file_size: raise ManifestError(f"Truncated DAT ZIP member: {info.filename!r}") extracted.append((info.filename, target)) except Exception: shutil.rmtree(output, ignore_errors=True) raise return extracted class DatFilesRow(QFrame): """DAT picker supporting ZIP bundles, individual files, and folders.""" def __init__(self): super().__init__() self.setObjectName("pathCard") # Keep extracted DATs alive while any verification/build worker uses them. # TemporaryDirectory automatically cleans these up on application exit. self._dat_zip_temp = tempfile.TemporaryDirectory(prefix="emv-dat-zip-") self._next_zip_id = 0 layout = QVBoxLayout(self) layout.setContentsMargins(18, 14, 18, 14) layout.setSpacing(8) label = QLabel("XML DAT files") label.setObjectName("fieldLabel") layout.addWidget(label) self.file_list = QListWidget() self.file_list.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection) self.file_list.setMinimumHeight(100) self.file_list.setMaximumHeight(145) self.file_list.setToolTip("All selected DATs are combined for verification and rebuilding.") layout.addWidget(self.file_list) actions = QHBoxLayout() actions.setSpacing(8) self.add_zip_button = QPushButton("Add DAT Zip…") self.add_zip_button.setObjectName("browseButton") self.add_zip_button.setToolTip("Select dats.zip directly; DATs are loaded automatically without manual extraction.") self.add_zip_button.clicked.connect(self.add_dat_zip) actions.addWidget(self.add_zip_button) self.add_files_button = QPushButton("Add DAT files…") self.add_files_button.setObjectName("browseButton") self.add_files_button.clicked.connect(self.add_files) actions.addWidget(self.add_files_button) self.add_folder_button = QPushButton("Add DAT folder…") self.add_folder_button.setObjectName("browseButton") self.add_folder_button.clicked.connect(self.add_folder) actions.addWidget(self.add_folder_button) self.remove_button = QPushButton("Remove selected") self.remove_button.setObjectName("ghostButton") self.remove_button.clicked.connect(self.remove_selected) actions.addWidget(self.remove_button) self.clear_button = QPushButton("Clear") self.clear_button.setObjectName("ghostButton") self.clear_button.clicked.connect(self.clear) actions.addWidget(self.clear_button) actions.addStretch(1) self.count_label = QLabel("0 DAT files selected") self.count_label.setObjectName("subtitle") actions.addWidget(self.count_label) layout.addLayout(actions) def selected_paths(self) -> List[str]: return [ str(self.file_list.item(i).data(Qt.ItemDataRole.UserRole)) for i in range(self.file_list.count()) ] def source_labels(self) -> Dict[str, str]: """Friendly origin strings, including the containing DAT ZIP and member.""" labels = {} for i in range(self.file_list.count()): item = self.file_list.item(i) path = str(item.data(Qt.ItemDataRole.UserRole)) origin = item.data(Qt.ItemDataRole.UserRole + 2) if origin: labels[path] = str(origin) return labels def _add_paths(self, paths: Iterable[str]) -> None: known = {str(Path(p).resolve()) for p in self.selected_paths()} for path_text in paths: path = Path(path_text) if not path.is_file() or path.suffix.lower() not in {".dat", ".xml"}: continue canonical = str(path.resolve()) if canonical in known: continue item = QListWidgetItem(path.name) item.setData(Qt.ItemDataRole.UserRole, canonical) item.setToolTip(canonical) self.file_list.addItem(item) known.add(canonical) self._update_count() def add_dat_zip(self) -> None: zip_text, _ = QFileDialog.getOpenFileName( self, "Select DAT ZIP archive", str(Path.home()), "ZIP archives (*.zip);;All files (*)", ) if not zip_text: return zip_path = Path(zip_text).resolve() # Avoid adding the same bundle twice while its DATs are selected. for i in range(self.file_list.count()): if self.file_list.item(i).data(Qt.ItemDataRole.UserRole + 1) == str(zip_path): QMessageBox.information(self, APP_NAME, "This DAT ZIP is already selected.") return self._next_zip_id += 1 dest = Path(self._dat_zip_temp.name) / f"zip_{self._next_zip_id:06d}" try: staged = stage_dat_zip(zip_path, dest) except (OSError, ValueError, RuntimeError, zipfile.BadZipFile, zipfile.LargeZipFile, EOFError, ManifestError) as exc: QMessageBox.critical(self, APP_NAME, f"Could not load DAT ZIP:\n\n{exc}") return for member, local_path in staged: canonical = str(local_path.resolve()) item = QListWidgetItem(f"{local_path.name} [{zip_path.name}]") item.setData(Qt.ItemDataRole.UserRole, canonical) item.setData(Qt.ItemDataRole.UserRole + 1, str(zip_path)) origin = f"{zip_path} :: {member}" item.setData(Qt.ItemDataRole.UserRole + 2, origin) item.setToolTip(origin) self.file_list.addItem(item) self._update_count() def add_files(self) -> None: paths, _ = QFileDialog.getOpenFileNames( self, "Select English Manga Vault DAT files", str(Path.home()), "DAT files (*.dat *.xml);;DAT files (*.dat);;XML files (*.xml)", ) if paths: self._add_paths(paths) def add_folder(self) -> None: folder_text = QFileDialog.getExistingDirectory( self, "Select folder containing English Manga Vault DATs", str(Path.home()) ) if not folder_text: return folder = Path(folder_text) files = sorted( (p for p in folder.iterdir() if p.is_file() and p.suffix.lower() in {".dat", ".xml"}), key=lambda p: (p.name.casefold(), p.name), ) if not files: QMessageBox.information( self, APP_NAME, "No .dat or .xml files were found in that folder." ) return self._add_paths(str(p) for p in files) def remove_selected(self) -> None: for item in self.file_list.selectedItems(): self.file_list.takeItem(self.file_list.row(item)) self._update_count() def clear(self) -> None: self.file_list.clear() self._update_count() def _update_count(self) -> None: count = self.file_list.count() self.count_label.setText(f"{count:,} DAT file{'s' if count != 1 else ''} selected") class StatCard(QFrame): def __init__(self, title: str): super().__init__() self.setObjectName("statCard") box = QVBoxLayout(self) box.setContentsMargins(16, 13, 16, 13) box.setSpacing(3) self.value = QLabel("0") self.value.setObjectName("statValue") self.label = QLabel(title) self.label.setObjectName("statLabel") box.addWidget(self.value) box.addWidget(self.label) def set_value(self, value: int) -> None: self.value.setText(f"{value:,}") class VaultRootsDialog(QDialog): """Configure optional per-top-level-folder vault destinations.""" def __init__( self, buckets: Sequence[str], current: Dict[str, str], parent=None, ): super().__init__(parent) self.setWindowTitle("Collection Vault Locations") self.resize(900, 680) self.edits: Dict[str, QLineEdit] = {} outer = QVBoxLayout(self) outer.setContentsMargins(22, 20, 22, 20) outer.setSpacing(12) title = QLabel("Separate English Manga Vault folders") title.setObjectName("sectionLabel") outer.addWidget(title) info = QLabel( "Optional. Point an A-Z, Oneshots, or _Autouploads collection to a different " "vault root. The builder uses the collection identified by each DAT filename. " "For example, English Manga Vault - A.dat places A_/A3/file.zip under " "/A/A_/A3/file.zip. English Manga Vault - Oneshots.dat places " "Manima ni!!/File.cbz under /Oneshots/Manima ni!!/File.cbz, " "but also recognizes a pre-existing /Manima ni!!/File.cbz. " "A collection prefix already written inside a DAT is not repeated." ) info.setWordWrap(True) info.setObjectName("subtitle") outer.addWidget(info) scroll = QScrollArea() scroll.setWidgetResizable(True) scroll.setFrameShape(QFrame.Shape.NoFrame) content = QWidget() rows = QVBoxLayout(content) rows.setContentsMargins(0, 0, 0, 0) rows.setSpacing(8) for bucket in buckets: card = QFrame() card.setObjectName("pathCard") row = QHBoxLayout(card) row.setContentsMargins(12, 9, 12, 9) row.setSpacing(10) label = QLabel(bucket) label.setMinimumWidth(42) label.setObjectName("fieldLabel") row.addWidget(label) edit = QLineEdit(current.get(bucket, "")) edit.setPlaceholderText(f"Optional root for {bucket}") edit.setClearButtonEnabled(True) self.edits[bucket] = edit row.addWidget(edit, 1) button = QPushButton("Choose folder") button.setObjectName("browseButton") button.clicked.connect( lambda checked=False, b=bucket: self._browse_bucket(b) ) row.addWidget(button) rows.addWidget(card) rows.addStretch(1) scroll.setWidget(content) outer.addWidget(scroll, 1) bottom = QHBoxLayout() clear_button = QPushButton("Clear all overrides") clear_button.setObjectName("ghostButton") clear_button.clicked.connect(self._clear_all) bottom.addWidget(clear_button) bottom.addStretch(1) buttons = QDialogButtonBox( QDialogButtonBox.StandardButton.Save | QDialogButtonBox.StandardButton.Cancel ) buttons.accepted.connect(self.accept) buttons.rejected.connect(self.reject) bottom.addWidget(buttons) outer.addLayout(bottom) def _browse_bucket(self, bucket: str) -> None: edit = self.edits[bucket] value = QFileDialog.getExistingDirectory( self, f"Select English Manga Vault - {bucket}", edit.text(), ) if value: edit.setText(value) def _clear_all(self) -> None: for edit in self.edits.values(): edit.clear() def mappings(self) -> Dict[str, str]: return { bucket: edit.text().strip() for bucket, edit in self.edits.items() if edit.text().strip() } class MainWindow(QMainWindow): def __init__(self): super().__init__() self.worker: Optional[WorkThread] = None self.last_report = "" self.vault_overrides: Dict[str, str] = {} self.setWindowTitle(APP_NAME) self.resize(1200, 1080) self.setMinimumSize(1100, 850) self._build_ui() self._apply_style() def _build_ui(self) -> None: root = QWidget() self.setCentralWidget(root) outer = QVBoxLayout(root) outer.setContentsMargins(28, 24, 28, 24) outer.setSpacing(16) title_row = QHBoxLayout() title_box = QVBoxLayout() title_box.setSpacing(2) title = QLabel(APP_NAME) title.setObjectName("title") subtitle = QLabel("Visit https://emv.moe to download the latest English Manga Vault DATs!") subtitle.setObjectName("subtitle") title_box.addWidget(title) title_box.addWidget(subtitle) title_row.addLayout(title_box) title_row.addStretch(1) version = QLabel(f"v{VERSION} by rarenight") version.setObjectName("version") title_row.addWidget(version, 0, Qt.AlignmentFlag.AlignTop) outer.addLayout(title_row) self.manifest_row = DatFilesRow() self.vault_row = PathRow( "Default Vault Root Directory", "Select the root English Manga Vault folder that you wish to build", "Choose folder", True, ) self.sort_row = PathRow( "Source Files Directory (optional)", "Choose the folder containing your unsorted files to build a new English Manga Vault…", "Choose folder", True, ) outer.addWidget(self.manifest_row) outer.addWidget(self.vault_row) self.vault_overrides_card = QFrame() self.vault_overrides_card.setObjectName("pathCard") split_layout = QHBoxLayout(self.vault_overrides_card) split_layout.setContentsMargins(18, 14, 18, 14) split_layout.setSpacing(12) split_text = QVBoxLayout() split_text.setSpacing(3) split_label = QLabel("Collection vault locations (optional)") split_label.setObjectName("fieldLabel") self.vault_overrides_summary = QLabel( "Choose separate roots for letters, Oneshots, and _Autouploads." ) self.vault_overrides_summary.setObjectName("subtitle") self.vault_overrides_summary.setWordWrap(True) split_text.addWidget(split_label) split_text.addWidget(self.vault_overrides_summary) split_layout.addLayout(split_text, 1) self.configure_vaults_button = QPushButton("Configure folders") self.configure_vaults_button.setObjectName("browseButton") self.configure_vaults_button.clicked.connect(self.configure_vault_overrides) split_layout.addWidget(self.configure_vaults_button) outer.addWidget(self.vault_overrides_card) outer.addWidget(self.sort_row) conflict_row = QHBoxLayout() conflict_row.setSpacing(10) conflict_label = QLabel("Conflict handling") conflict_label.setObjectName("fieldLabel") self.conflict_policy = QComboBox() self.conflict_policy.addItem("Skip conflicting destinations", "skip") self.conflict_policy.addItem("Overwrite only with SHA-1 match", "overwrite_verified") self.conflict_policy.addItem("Record conflicts in separate report", "report") self.conflict_policy.setCurrentIndex(1) self.conflict_policy.setToolTip( "A conflict is an existing destination whose SHA-1 does not match the DAT. " "The overwrite option only uses a Source File whose calculated SHA-1 exactly " "matches the DAT entry." ) conflict_row.addWidget(conflict_label) conflict_row.addWidget(self.conflict_policy, 1) outer.addLayout(conflict_row) options = QHBoxLayout() options.setSpacing(18) post_verify_label = QLabel("Verification after rebuild") post_verify_label.setObjectName("fieldLabel") self.post_verify_mode = QComboBox() self.post_verify_mode.addItem("Quick Verify after Rebuild", "quick") self.post_verify_mode.addItem("Internal Verify after Rebuild", "internal") self.post_verify_mode.addItem("Full Verify after Rebuild", "full") self.post_verify_mode.addItem("No Verify after Rebuild", "none") self.post_verify_mode.setCurrentIndex(0) self.post_verify_mode.setToolTip( "Quick Verify checks each expected path and compares its file size with the DAT. " "Full Verify recalculates SHA-1 for every file. Internal Verify tests checksums " "embedded in ZIP/CBZ, RAR/CBR, and 7z archives. No Verify skips the check." ) self.remove_matched = QCheckBox("Remove matched files from Source Files after successful placement") self.remove_matched.setChecked(True) options.addWidget(post_verify_label) options.addWidget(self.post_verify_mode) options.addWidget(self.remove_matched) options.addStretch(1) outer.addLayout(options) actions = QHBoxLayout() actions.setSpacing(10) self.quick_verify_button = QPushButton("Quick Verify") self.quick_verify_button.setObjectName("primaryButton") self.quick_verify_button.setToolTip("Check exact DAT path and file size. Does not calculate SHA-1.") self.quick_verify_button.clicked.connect(self.start_quick_verify) self.full_verify_button = QPushButton("Full Verify") self.full_verify_button.setObjectName("primaryButton") self.full_verify_button.setToolTip("Calculate a fresh SHA-1 for every expected file and compare it with the DAT.") self.full_verify_button.clicked.connect(self.start_full_verify) self.internal_verify_button = QPushButton("Internal Verify") self.internal_verify_button.setObjectName("primaryButton") self.internal_verify_button.setToolTip( "Test embedded archive CRCs/checksums in ZIP/CBZ, RAR/CBR and 7z " "without unpacking files onto disk. Does not compare against DAT SHA-1." ) self.internal_verify_button.clicked.connect(self.start_internal_verify) self.build_button = QPushButton("Build from Source Files") self.build_button.setObjectName("secondaryButton") self.build_button.clicked.connect(self.start_build) self.cancel_button = QPushButton("Cancel") self.cancel_button.setObjectName("dangerButton") self.cancel_button.setEnabled(False) self.cancel_button.clicked.connect(self.cancel_work) self.export_button = QPushButton("Build Report") self.export_button.setObjectName("ghostButton") self.export_button.setEnabled(False) self.export_button.clicked.connect(self.export_report) actions.addWidget(self.quick_verify_button) actions.addWidget(self.internal_verify_button) actions.addWidget(self.full_verify_button) actions.addWidget(self.build_button) actions.addWidget(self.cancel_button) actions.addStretch(1) actions.addWidget(self.export_button) outer.addLayout(actions) stat_grid = QGridLayout() stat_grid.setSpacing(10) self.stat_expected = StatCard("Expected") self.stat_present = StatCard("Present") self.stat_verified = StatCard("Verified") self.stat_missing = StatCard("Missing") self.stat_mismatch = StatCard("Mismatch") cards = [ self.stat_expected, self.stat_present, self.stat_verified, self.stat_missing, self.stat_mismatch, ] for col, card in enumerate(cards): stat_grid.addWidget(card, 0, col) stat_grid.setColumnStretch(col, 1) outer.addLayout(stat_grid) progress_card = QFrame() progress_card.setObjectName("progressCard") pbox = QVBoxLayout(progress_card) pbox.setContentsMargins(16, 12, 16, 12) pbox.setSpacing(8) self.status_label = QLabel("Ready") self.status_label.setObjectName("statusLabel") self.progress = QProgressBar() self.progress.setRange(0, 10000) self.progress.setValue(0) self.progress.setTextVisible(True) self.progress.setFormat("00.0%") pbox.addWidget(self.status_label) pbox.addWidget(self.progress) outer.addWidget(progress_card) report_label = QLabel("Report") report_label.setObjectName("sectionLabel") outer.addWidget(report_label) self.report = QPlainTextEdit() self.report.setReadOnly(True) self.report.setPlaceholderText( "Verification results, missing files, hash mismatches, and build details will appear here." ) self.report.setLineWrapMode(QPlainTextEdit.LineWrapMode.NoWrap) outer.addWidget(self.report, 1) def _apply_style(self) -> None: self.setStyleSheet( """ QMainWindow, QWidget { background: #0c1017; color: #e8edf5; font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Inter, Arial, sans-serif; font-size: 13px; } QLabel#title { font-size: 28px; font-weight: 700; color: #f8fbff; } QLabel#subtitle { color: #8f9bad; font-size: 13px; } QLabel#version { color: #687386; padding-top: 5px; } QFrame#pathCard, QFrame#statCard, QFrame#progressCard { background: #121823; border: 1px solid #20293a; border-radius: 12px; } QLabel#fieldLabel, QLabel#sectionLabel { color: #c9d3e1; font-weight: 600; font-size: 13px; } QLineEdit { background: #0b111b; border: 1px solid #293448; border-radius: 8px; padding: 10px 11px; color: #eff4fb; selection-background-color: #316bff; } QLineEdit:focus { border: 1px solid #4d7cff; } QListWidget { background: #0b111b; border: 1px solid #293448; border-radius: 8px; color: #eff4fb; padding: 4px; font-family: "SFMono-Regular", Menlo, Monaco, Consolas, monospace; font-size: 12px; } QListWidget::item { padding: 3px 6px; } QListWidget::item:selected { background: #26395f; } QComboBox { background: #0b111b; border: 1px solid #293448; border-radius: 8px; padding: 9px 11px; color: #eff4fb; min-width: 300px; } QComboBox:focus { border: 1px solid #4d7cff; } QComboBox QAbstractItemView { background: #121823; border: 1px solid #293448; selection-background-color: #26395f; color: #eff4fb; } QPushButton { min-height: 20px; border-radius: 8px; padding: 9px 15px; font-weight: 600; } QPushButton#browseButton, QPushButton#ghostButton { background: #182131; border: 1px solid #2b384e; color: #dce5f1; } QPushButton#browseButton:hover, QPushButton#ghostButton:hover { background: #202c3f; } QPushButton#primaryButton { background: #4c79ff; border: 1px solid #4c79ff; color: white; } QPushButton#primaryButton:hover { background: #5b86ff; } QPushButton#secondaryButton { background: #1b6e59; border: 1px solid #25866e; color: #f3fffb; } QPushButton#secondaryButton:hover { background: #237c65; } QPushButton#dangerButton { background: #3b1e24; border: 1px solid #613039; color: #ffcbd1; } QPushButton#dangerButton:hover { background: #49252d; } QPushButton:disabled { background: #151a22; border-color: #222a35; color: #596474; } QCheckBox { color: #b7c2d0; spacing: 8px; } QCheckBox::indicator { width: 17px; height: 17px; } QLabel#statValue { font-size: 22px; font-weight: 700; color: #f4f8ff; } QLabel#statLabel { font-size: 11px; color: #798699; text-transform: uppercase; } QLabel#statusLabel { color: #a9b6c7; } QProgressBar { background: #0a0f16; border: 1px solid #20293a; border-radius: 6px; min-height: 18px; text-align: center; color: #f4f8ff; font-weight: 600; } QProgressBar::chunk { background: #4c79ff; border-radius: 4px; } QPlainTextEdit { background: #080d14; border: 1px solid #20293a; border-radius: 10px; color: #cad5e3; padding: 10px; font-family: "SFMono-Regular", Menlo, Monaco, Consolas, monospace; font-size: 11px; } """ ) def _manifest_buckets_for_dialog(self) -> List[str]: # Offer letter and named collections without parsing large DATs on the GUI thread. buckets = set(self.vault_overrides) buckets.update(chr(code) for code in range(ord("A"), ord("Z") + 1)) buckets.update(("Oneshots", "_Autouploads")) for selected in self.manifest_row.selected_paths(): dat_bucket = bucket_from_dat_filename(Path(selected)) if dat_bucket: buckets.add(dat_bucket) def sort_key(value: str): upper = value.upper() if len(value) == 1 and "A" <= upper <= "Z": return (0, ord(upper), value) return (1, 0, value.casefold()) return sorted(buckets, key=sort_key) def configure_vault_overrides(self) -> None: dialog = VaultRootsDialog( self._manifest_buckets_for_dialog(), self.vault_overrides, self, ) if dialog.exec() == QDialog.DialogCode.Accepted: self.vault_overrides = dialog.mappings() self._update_vault_override_summary() def _update_vault_override_summary(self) -> None: if not self.vault_overrides: self.vault_overrides_summary.setText( "Single-vault layout: DAT collections use their letter/named folder, with existing flat files supported." ) return buckets = sorted(self.vault_overrides, key=lambda s: (s.casefold(), s)) shown = ", ".join(buckets[:10]) if len(buckets) > 10: shown += f", +{len(buckets) - 10} more" self.vault_overrides_summary.setText( f"{len(buckets)} separate vault location(s): {shown}. " "Unassigned collections use the Default Vault Root." ) def _validate_vault_overrides(self) -> bool: seen_paths: Dict[str, str] = {} for bucket, raw_path in self.vault_overrides.items(): path = Path(raw_path) if not path.exists() or not path.is_dir(): QMessageBox.warning( self, APP_NAME, f"The configured Vault Root for {bucket} is not an existing directory:\n\n{path}", ) return False try: canonical = str(path.resolve()) except OSError: canonical = str(path.absolute()) previous = seen_paths.get(canonical) if previous is not None and previous != bucket: QMessageBox.warning( self, APP_NAME, f"{previous} and {bucket} are configured to use the same split Vault Root. " "Each override should point to its own root location.", ) return False seen_paths[canonical] = bucket return True def _validate_common(self) -> Optional[Tuple[List[str], str, Dict[str, str]]]: manifests = self.manifest_row.selected_paths() root = self.vault_row.edit.text().strip() if not manifests: QMessageBox.warning(self, APP_NAME, "Add at least one XML DAT file.") return None for manifest in manifests: if not Path(manifest).is_file(): QMessageBox.warning(self, APP_NAME, f"Selected DAT does not exist:\n{manifest}") return None if not root: QMessageBox.warning(self, APP_NAME, "Choose the Default Vault Root directory.") return None root_path = Path(root) if not root_path.exists(): try: root_path.mkdir(parents=True, exist_ok=True) except OSError as exc: QMessageBox.critical(self, APP_NAME, f"Could not create Default Vault Root:\n{exc}") return None if not root_path.is_dir(): QMessageBox.warning(self, APP_NAME, "Default Vault Root must be a directory.") return None # Large DAT XMLs are parsed and cross-checked by the background worker. # It rejects conflicting destination paths before any build can begin. if not self._validate_vault_overrides(): return None return manifests, root, dict(self.vault_overrides) def start_quick_verify(self) -> None: common = self._validate_common() if common is None: return manifest, root, overrides = common self._start_worker( WorkThread( "quick_verify", manifest, root, vault_root_overrides=overrides, manifest_display_paths=self.manifest_row.source_labels(), parent=self, ) ) def start_full_verify(self) -> None: common = self._validate_common() if common is None: return manifest, root, overrides = common self._start_worker( WorkThread( "full_verify", manifest, root, vault_root_overrides=overrides, manifest_display_paths=self.manifest_row.source_labels(), parent=self, ) ) def start_internal_verify(self) -> None: common = self._validate_common() if common is None: return manifest, root, overrides = common self._start_worker( WorkThread( "internal_verify", manifest, root, vault_root_overrides=overrides, manifest_display_paths=self.manifest_row.source_labels(), parent=self, ) ) def start_build(self) -> None: common = self._validate_common() if common is None: return manifest, root, overrides = common to_sort = self.sort_row.edit.text().strip() if not to_sort or not Path(to_sort).is_dir(): QMessageBox.warning(self, APP_NAME, "Choose a valid Source Files directory.") return self._start_worker( WorkThread( "build", manifest, root, to_sort_value=to_sort, remove_matched=self.remove_matched.isChecked(), post_verify_mode=str(self.post_verify_mode.currentData()), conflict_policy=str(self.conflict_policy.currentData()), vault_root_overrides=overrides, manifest_display_paths=self.manifest_row.source_labels(), parent=self, ) ) def _start_worker(self, worker: WorkThread) -> None: if self.worker is not None and self.worker.isRunning(): return self.worker = worker self.last_report = "" self.report.clear() self.export_button.setEnabled(False) self.progress.setRange(0, 10000) self.progress.setValue(0) self.progress.setFormat("00.0%") self.status_label.setText("Starting…") self._set_busy(True) worker.progress.connect(self.on_progress) worker.message.connect(self.status_label.setText) worker.stats.connect(self.on_stats) worker.report_ready.connect(self.on_report) worker.failed.connect(self.on_failed) worker.completed.connect(self.on_completed) worker.finished.connect(self.on_thread_finished) worker.start() def _set_busy(self, busy: bool) -> None: self.quick_verify_button.setEnabled(not busy) self.full_verify_button.setEnabled(not busy) self.internal_verify_button.setEnabled(not busy) self.build_button.setEnabled(not busy) self.cancel_button.setEnabled(busy) self.manifest_row.setEnabled(not busy) self.vault_row.button.setEnabled(not busy) self.sort_row.button.setEnabled(not busy) self.configure_vaults_button.setEnabled(not busy) self.vault_row.edit.setEnabled(not busy) self.sort_row.edit.setEnabled(not busy) self.post_verify_mode.setEnabled(not busy) self.remove_matched.setEnabled(not busy) self.conflict_policy.setEnabled(not busy) def on_progress(self, current: int, total: int, label: str) -> None: fraction = max(0.0, min(1.0, current / max(total, 1))) scaled = int(round(fraction * 10000)) self.progress.setRange(0, 10000) self.progress.setValue(scaled) self.progress.setFormat(f"{fraction * 100:04.1f}%") self.status_label.setText(label) def on_stats(self, data: dict) -> None: self.stat_expected.set_value(int(data.get("expected", 0))) self.stat_present.set_value(int(data.get("present", 0))) self.stat_verified.set_value(int(data.get("verified", 0))) self.stat_missing.set_value(int(data.get("missing", 0))) self.stat_mismatch.set_value(int(data.get("mismatch", 0))) def on_report(self, text: str) -> None: self.last_report = text self.report.setPlainText(text) self.export_button.setEnabled(bool(text)) def on_failed(self, text: str) -> None: self.status_label.setText("Error") self.report.setPlainText(f"ERROR\n=====\n{text}\n") self.last_report = self.report.toPlainText() self.export_button.setEnabled(True) QMessageBox.critical(self, APP_NAME, text) def on_completed(self, text: str) -> None: self.status_label.setText(text) if text != "Cancelled": self.progress.setRange(0, 10000) self.progress.setValue(10000) self.progress.setFormat("100.0%") def on_thread_finished(self) -> None: self._set_busy(False) def cancel_work(self) -> None: if self.worker is not None and self.worker.isRunning(): self.status_label.setText("Cancelling after the current read operation…") self.cancel_button.setEnabled(False) self.worker.cancel() def export_report(self) -> None: if not self.last_report: return default = str(Path.home() / "English Manga Vault Builder Report.txt") path, _ = QFileDialog.getSaveFileName( self, "Export report", default, "Text files (*.txt);;All files (*)", ) if not path: return try: Path(path).write_text(self.last_report, encoding="utf-8") except OSError as exc: QMessageBox.critical(self, APP_NAME, f"Could not save report:\n{exc}") return self.status_label.setText(f"Report exported to {path}") def closeEvent(self, event) -> None: # noqa: N802 (Qt naming) if self.worker is not None and self.worker.isRunning(): answer = QMessageBox.question( self, APP_NAME, "An operation is still running. Cancel it and close?", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.No, ) if answer != QMessageBox.StandardButton.Yes: event.ignore() return self.worker.cancel() self.worker.wait(3000) event.accept() def main() -> int: app = QApplication(sys.argv) app.setApplicationName(APP_NAME) app.setOrganizationName("English Manga Vault") window = MainWindow() window.show() return app.exec() if __name__ == "__main__": raise SystemExit(main())