from __future__ import annotations import hashlib import os import re import shutil import sys import tempfile import threading import time import xml.etree.ElementTree as ET from collections import defaultdict from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import Dict, Iterable, List, Optional, Sequence, Tuple from PyQt6.QtCore import QThread, Qt, pyqtSignal from PyQt6.QtGui import QFont, QIcon from PyQt6.QtWidgets import ( QApplication, QCheckBox, QComboBox, QDialog, QDialogButtonBox, QFileDialog, QFrame, QGridLayout, QHBoxLayout, QLabel, QLineEdit, QMainWindow, QMessageBox, QPlainTextEdit, QProgressBar, QPushButton, QScrollArea, QSizePolicy, QSpacerItem, QVBoxLayout, QWidget, ) APP_NAME = "English Manga Vault Builder" VERSION = "1.8" 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]] @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 a No-Intro-style XML DAT 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 No-Intro 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 manifest_bucket(rel_path: str) -> str: """Return the English Manga Vault top-level bucket for a DAT path. The bucket is the first character of the DAT path's first component, upper-cased. For example ``AZ/AZUR/...`` belongs to bucket ``A``. """ 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, ) -> Path: """Resolve a DAT path using the English Manga Vault letter layout. Current DATs already include their top-level letter, for example: A/AZ/AZUR/Series/File.cbz Such a path is preserved exactly beneath the selected root. If a compatible DAT omits that single-letter component (for example ``AZ/AZUR/...``), the bucket is derived and inserted once so it still becomes ``A/AZ/AZUR/...``. An A override changes only the root. It never removes or duplicates ``A/``. """ parts = PurePosixPath(rel_path).parts if not parts: return root bucket = manifest_bucket(rel_path) base = root if bucket_roots: override = bucket_roots.get(bucket) if override is not None: base = override already_bucketed = ( bool(bucket) and len(parts[0]) == 1 and parts[0].upper() == bucket.upper() ) if already_bucketed: return base.joinpath(*parts) if bucket: return base.joinpath(bucket, *parts) return base.joinpath(*parts) 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 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_path_value: 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, parent=None, ): super().__init__(parent) self.mode = mode self.manifest_file = Path(manifest_path_value) 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() 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) 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}Letter 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"(DAT path is preserved; the letter bucket is added only if absent)" ) else: lines.append(f"{prefix}Letter 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 run(self) -> None: try: manifest = parse_dat(self.manifest_file) self.message.emit( f"Loaded {len(manifest.entries):,} DAT 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 == "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.append(f"DAT: {self.manifest_file}") lines.extend(self._vault_layout_lines()) if self.vault_root_overrides: lines.append( "Path rule: Preserve the DAT relative path; configured letters use their override root" ) else: lines.append( "Path rule: Default Vault Root + DAT relative path (letter bucket added only if absent)" ) 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("") 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. The DAT relative path is preserved beneath the Default Vault Root, or beneath a configured letter override. The file at that exact destination is then hashed and compared with the SHA-1 recorded on the same DAT entry. """ 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.append(f"DAT: {self.manifest_file}") lines.extend(self._vault_layout_lines()) if self.vault_root_overrides: lines.append( "Path rule: Preserve the DAT relative path; configured letters use their override root" ) else: lines.append("Path rule: Default Vault Root + DAT relative path (letter bucket added only if absent)") 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("") 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 _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() started = time.monotonic() expected_by_hash = manifest.by_hash self.message.emit("Scanning Source Files recursively…") 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() == self.manifest_file.resolve(): 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() 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", f"DAT: {self.manifest_file}", 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.append(f"DAT: {self.manifest_file}") 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"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 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 == "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 No-Intro XML DAT", self.edit.text(), "No-Intro DAT files (*.dat *.xml);;DAT files (*.dat);;XML files (*.xml);;All files (*)", ) if value: self.edit.setText(value) 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("Letter 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. Assign a letter bucket (such as A or B) to a different root " "folder or drive. Current DATs already include that letter in each game path, " "so A/AA/Series/File.cbz remains A/AA/Series/File.cbz beneath the chosen root. " "For example, A/AA/Series/File.cbz normally becomes " "/A/AA/Series/File.cbz; with an A override it becomes " "/A/AA/Series/File.cbz." ) 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: English Manga Vault - {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(1200, 1080) 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 DAT!") 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 Aqua") version.setObjectName("version") title_row.addWidget(version, 0, Qt.AlignmentFlag.AlignTop) outer.addLayout(title_row) self.manifest_row = PathRow( "No-Intro XML DAT", "Select an English Manga Vault .dat file…", "Choose file", False, ) 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("Letter vault locations (optional)") split_label.setObjectName("fieldLabel") self.vault_overrides_summary = QLabel( "Choose the folders where you want specific letters of English Manga Vault to be stored." ) 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 letters") 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("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. No Verify skips the post-build 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.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.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; } 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]: buckets = set(self.vault_overrides) manifest_text = self.manifest_row.edit.text().strip() if manifest_text and Path(manifest_text).is_file(): try: manifest = parse_dat(Path(manifest_text)) except ManifestError: pass else: buckets.update( manifest_bucket(entry.rel_path) for entry in manifest.entries if manifest_bucket(entry.rel_path) ) if not buckets: buckets.update(chr(code) for code in range(ord("A"), ord("Z") + 1)) 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: every DAT path is preserved beneath the Default Vault Root." ) 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 letter buckets 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[str, str, Dict[str, str]]]: manifest = self.manifest_row.edit.text().strip() root = self.vault_row.edit.text().strip() if not manifest: QMessageBox.warning(self, APP_NAME, "Choose the No-Intro XML DAT first.") return None if not Path(manifest).is_file(): QMessageBox.warning(self, APP_NAME, "The selected DAT does not exist.") 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 try: parse_dat(Path(manifest)) except ManifestError as exc: QMessageBox.critical(self, APP_NAME, f"Invalid No-Intro XML DAT:\n\n{exc}") return None if not self._validate_vault_overrides(): return None return manifest, 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, 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, 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, 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.build_button.setEnabled(not busy) self.cancel_button.setEnabled(busy) self.manifest_row.button.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.manifest_row.edit.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())