Coverage for src/ph/organizer.py: 92.5%
120 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 08:17 +0000
1"""Move completed downloads into a target's ROM library."""
3import logging
4import os
5import shutil
6import stat
7import zipfile
8from collections.abc import Callable, Iterable, Sequence
9from pathlib import Path, PurePosixPath
11from ph.models import DownloadResult, Platform
12from ph.targets import DARKOS
14LOGGER = logging.getLogger(__name__)
17class OrganizeError(RuntimeError):
18 """A completed file could not be placed in the ROM library."""
21MAX_BIOS_FILES = 128
22MAX_BIOS_FILE_SIZE = 64 * 1024 * 1024
23MAX_BIOS_TOTAL_SIZE = 256 * 1024 * 1024
24ROM_LOCAL_BIOS: dict[str, frozenset[str]] = {
25 "adventure-vision": frozenset({"advision.zip"}),
26 "astrocade": frozenset({"astrocde.zip"}),
27 "coco-3": frozenset({"coco.zip", "coco2.zip", "coco2b.zip", "coco3.zip", "coco3p.zip"}),
28}
29ROM_AND_SHARED_BIOS: dict[str, frozenset[str]] = {
30 "neo-geo": frozenset({"aes.zip", "neogeo.zip"}),
31}
34def detect_roms_directories(
35 configured: Path | Sequence[Path] | None = None,
36 candidates: Sequence[Path] = DARKOS.rom_roots,
37) -> tuple[Path, ...]:
38 """Find all active ROM partitions for the selected Linux target."""
40 if configured is not None: 40 ↛ 43line 40 didn't jump to line 43 because the condition on line 40 was always true
41 return (configured,) if isinstance(configured, Path) else tuple(configured)
43 return available_roms_directories(candidates)
46def available_roms_directories(candidates: Sequence[Path]) -> tuple[Path, ...]:
47 """Return every candidate that looks like a mounted, writable ROM partition."""
49 # Every official image exposes some combination of system folders. Avoid a
50 # short console allow-list so customized distribution images are recognized.
51 available = tuple(
52 candidate
53 for candidate in candidates
54 if candidate.is_dir()
55 and (
56 any(path.is_dir() for path in candidate.iterdir()) or os.access(str(candidate), os.W_OK)
57 )
58 )
59 return available
62def detect_roms_directory(
63 configured: Path | Sequence[Path] | None = None,
64 candidates: Sequence[Path] = DARKOS.rom_roots,
65) -> Path | None:
66 """Find the preferred target ROM partition for non-interactive use."""
68 directories = detect_roms_directories(configured, candidates)
69 return directories[0] if directories else None
72def install_downloads(
73 downloads: Iterable[DownloadResult],
74 platform: Platform,
75 roms_directory: Path,
76 bios_installed: Callable[[Path], None] | None = None,
77 bios_directory: str = "bios",
78) -> list[DownloadResult]:
79 """Move completed files to a platform folder without overwriting existing ROMs."""
81 if platform.rom_folder is None:
82 raise OrganizeError(f"{platform.name} does not have a supported ROM folder.")
83 existing_folder = next(
84 (folder for folder in platform.rom_folders if (roms_directory / folder).is_dir()),
85 platform.rom_folder,
86 )
87 destination_directory = roms_directory / existing_folder
88 LOGGER.debug(
89 "Installing downloads platform=%s destination=%s",
90 platform.alias,
91 destination_directory,
92 )
93 try:
94 destination_directory.mkdir(parents=True, exist_ok=True)
95 except OSError as error:
96 raise OrganizeError(f"Cannot create {destination_directory}: {error}") from error
98 moved: list[DownloadResult] = []
99 for download in downloads:
100 source = download.path
101 if not source.is_file():
102 raise OrganizeError(f"Completed file was not found: {source}")
103 for bios_path in install_bundled_bios(
104 source,
105 platform,
106 roms_directory,
107 bios_directory,
108 ):
109 if bios_installed is not None:
110 bios_installed(bios_path)
111 destination = unique_destination(destination_directory / source.name)
112 try:
113 final_path = Path(shutil.move(str(source), str(destination)))
114 except OSError as error:
115 raise OrganizeError(f"Could not move {source.name}: {error}") from error
116 moved.append(DownloadResult(url=download.url, path=final_path))
117 LOGGER.info("Installed ROM platform=%s path=%s", platform.alias, final_path)
118 return moved
121def install_bundled_bios(
122 archive_path: Path,
123 platform: Platform,
124 roms_directory: Path,
125 bios_directory: str = "bios",
126) -> tuple[Path, ...]:
127 """Safely install files explicitly stored under a BIOS directory in a game ZIP."""
129 if archive_path.suffix.casefold() != ".zip" or not zipfile.is_zipfile(archive_path):
130 return ()
131 try:
132 with zipfile.ZipFile(archive_path) as archive:
133 members = _bundled_bios_members(archive)
134 installed: list[Path] = []
135 for member, relative_path in members:
136 for destination in _bios_destinations(
137 relative_path,
138 platform,
139 roms_directory,
140 bios_directory,
141 ):
142 if destination.exists():
143 continue
144 destination.parent.mkdir(parents=True, exist_ok=True)
145 partial = destination.with_name(destination.name + ".part")
146 try:
147 with archive.open(member) as source, partial.open("wb") as output:
148 shutil.copyfileobj(source, output, length=1024 * 256)
149 os.replace(partial, destination)
150 except (OSError, RuntimeError, zipfile.BadZipFile) as error:
151 partial.unlink(missing_ok=True)
152 raise OrganizeError(
153 f"Could not install bundled BIOS {relative_path}: {error}"
154 ) from error
155 installed.append(destination)
156 LOGGER.info("Installed bundled BIOS path=%s", destination)
157 return tuple(installed)
158 except zipfile.BadZipFile as error:
159 raise OrganizeError(f"Could not inspect bundled BIOS files: {error}") from error
162def _bundled_bios_members(
163 archive: zipfile.ZipFile,
164) -> tuple[tuple[zipfile.ZipInfo, PurePosixPath], ...]:
165 selected: list[tuple[zipfile.ZipInfo, PurePosixPath]] = []
166 total_size = 0
167 for member in archive.infolist():
168 normalized = member.filename.replace("\\", "/")
169 parts = PurePosixPath(normalized).parts
170 bios_index = next(
171 (index for index, part in enumerate(parts) if part.casefold() == "bios"),
172 None,
173 )
174 if bios_index is None or member.is_dir():
175 continue
176 relative_parts = parts[bios_index + 1 :]
177 if not relative_parts or any(part in {"", ".", "..", "/"} for part in relative_parts):
178 raise OrganizeError(f"Unsafe bundled BIOS path: {member.filename}")
179 mode = member.external_attr >> 16
180 if stat.S_ISLNK(mode):
181 raise OrganizeError(f"Bundled BIOS cannot be a symbolic link: {member.filename}")
182 if member.file_size > MAX_BIOS_FILE_SIZE: 182 ↛ 183line 182 didn't jump to line 183 because the condition on line 182 was never true
183 raise OrganizeError(f"Bundled BIOS is unexpectedly large: {member.filename}")
184 total_size += member.file_size
185 if len(selected) >= MAX_BIOS_FILES or total_size > MAX_BIOS_TOTAL_SIZE: 185 ↛ 186line 185 didn't jump to line 186 because the condition on line 185 was never true
186 raise OrganizeError("Bundled BIOS payload exceeds the safe extraction limit.")
187 selected.append((member, PurePosixPath(*relative_parts)))
188 return tuple(selected)
191def _bios_destinations(
192 relative_path: PurePosixPath,
193 platform: Platform,
194 roms_directory: Path,
195 bios_directory: str = "bios",
196) -> tuple[Path, ...]:
197 folder = platform.rom_folder or ""
198 filename = relative_path.name.casefold()
199 local = roms_directory / folder / Path(*relative_path.parts)
200 shared = roms_directory / bios_directory / Path(*relative_path.parts)
201 if filename in ROM_LOCAL_BIOS.get(platform.slug, frozenset()):
202 return (local,)
203 if filename in ROM_AND_SHARED_BIOS.get(platform.slug, frozenset()):
204 return (shared, local)
205 return (shared,)
208def unique_destination(destination: Path) -> Path:
209 if not destination.exists():
210 return destination
211 suffixes = "".join(destination.suffixes)
212 stem = destination.name[: -len(suffixes)] if suffixes else destination.name
213 counter = 2
214 while True:
215 candidate = destination.with_name("%s (%d)%s" % (stem, counter, suffixes))
216 if not candidate.exists():
217 return candidate
218 counter += 1