Coverage for src/ph/preferences.py: 96.9%
125 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 08:17 +0000
1"""Persistent user preferences for the controller-driven TUI."""
3import json
4import os
5from dataclasses import dataclass, field
6from pathlib import Path
8from ph.bittorrent import BitTorrentSettings
9from ph.cache_policy import DEFAULT_CATALOGUE_TTL_DAYS
10from ph.download_queue import (
11 DEFAULT_CONCURRENT_DOWNLOADS,
12 RateLimitRetrySettings,
13)
14from ph.i18n import DEFAULT_LANGUAGE, LanguageCode, normalize_language
16PREFERENCES_FILENAME = ".pocket-harbor.json"
19class PreferencesError(RuntimeError):
20 """Preferences could not be saved."""
23@dataclass(frozen=True, slots=True)
24class Preferences:
25 """Settings selected from the TUI options screen."""
27 store_id: str | None = None
28 minerva_bittorrent: BitTorrentSettings = field(default_factory=BitTorrentSettings)
29 catalogue_ttl_days: int = DEFAULT_CATALOGUE_TTL_DAYS
30 log_level: str | None = None
31 log_to_file: bool | None = None
32 language: LanguageCode = DEFAULT_LANGUAGE
33 max_concurrent_downloads: int = DEFAULT_CONCURRENT_DOWNLOADS
34 rate_limit_retry: RateLimitRetrySettings = field(default_factory=RateLimitRetrySettings)
35 default_roms_directory: str | None = None
36 network_timeout_seconds: float | None = None
37 store_rom_mappings: dict[str, dict[str, str]] = field(default_factory=dict)
38 bios_directory: str = "bios"
39 ask_store_each_time: bool = False
42def preference_path(download_directory: Path) -> Path:
43 """Keep settings beside staged downloads in the tool's writable directory."""
45 return download_directory / PREFERENCES_FILENAME
48def load_preferences(path: Path) -> Preferences:
49 """Read preferences, treating absent or malformed files as first-run state."""
51 try:
52 payload = json.loads(path.read_text(encoding="utf-8"))
53 except OSError, UnicodeError, json.JSONDecodeError:
54 return Preferences()
55 if not isinstance(payload, dict):
56 return Preferences()
57 store_id = payload.get("store")
58 if not isinstance(store_id, str) or not store_id.strip():
59 normalized_store_id = None
60 else:
61 normalized_store_id = store_id.strip().casefold()
62 ttl_days = _integer_setting(payload, "catalogue_ttl_days", DEFAULT_CATALOGUE_TTL_DAYS)
63 if ttl_days <= 0:
64 ttl_days = DEFAULT_CATALOGUE_TTL_DAYS
65 raw_log_level = payload.get("log_level")
66 log_level = (
67 raw_log_level.upper()
68 if isinstance(raw_log_level, str)
69 and raw_log_level.upper() in {"DEBUG", "INFO", "WARNING", "ERROR"}
70 else None
71 )
72 raw_log_to_file = payload.get("log_to_file")
73 log_to_file = raw_log_to_file if isinstance(raw_log_to_file, bool) else None
74 ask_store_each_time = payload.get("ask_store_each_time") is True
75 raw_default_roms_directory = payload.get("default_roms_directory")
76 default_roms_directory = (
77 raw_default_roms_directory.strip()
78 if isinstance(raw_default_roms_directory, str) and raw_default_roms_directory.strip()
79 else None
80 )
81 network_timeout_seconds = _optional_float_setting(payload, "network_timeout_seconds")
82 if network_timeout_seconds is not None and not 1 <= network_timeout_seconds <= 3600:
83 network_timeout_seconds = None
84 store_rom_mappings = _store_rom_mappings(payload.get("store_rom_mappings"))
85 bios_directory = _relative_directory(payload.get("bios_directory")) or "bios"
86 language = normalize_language(payload.get("language"))
87 max_concurrent_downloads = _integer_setting(
88 payload,
89 "max_concurrent_downloads",
90 DEFAULT_CONCURRENT_DOWNLOADS,
91 )
92 if not 1 <= max_concurrent_downloads <= 8:
93 max_concurrent_downloads = DEFAULT_CONCURRENT_DOWNLOADS
94 retry_defaults = RateLimitRetrySettings()
95 retry_payload = payload.get("rate_limit_retry")
96 if isinstance(retry_payload, dict):
97 try:
98 rate_limit_retry = RateLimitRetrySettings(
99 base_seconds=_float_setting(
100 retry_payload,
101 "base_seconds",
102 retry_defaults.base_seconds,
103 ),
104 max_seconds=_float_setting(
105 retry_payload,
106 "max_seconds",
107 retry_defaults.max_seconds,
108 ),
109 jitter_ratio=_float_setting(
110 retry_payload,
111 "jitter_ratio",
112 retry_defaults.jitter_ratio,
113 ),
114 )
115 except ValueError:
116 rate_limit_retry = retry_defaults
117 else:
118 rate_limit_retry = retry_defaults
119 minerva_payload = payload.get("minerva_bittorrent")
120 defaults = BitTorrentSettings()
121 if not isinstance(minerva_payload, dict):
122 return Preferences(
123 store_id=normalized_store_id,
124 minerva_bittorrent=defaults,
125 catalogue_ttl_days=ttl_days,
126 log_level=log_level,
127 log_to_file=log_to_file,
128 language=language,
129 max_concurrent_downloads=max_concurrent_downloads,
130 rate_limit_retry=rate_limit_retry,
131 default_roms_directory=default_roms_directory,
132 network_timeout_seconds=network_timeout_seconds,
133 store_rom_mappings=store_rom_mappings,
134 bios_directory=bios_directory,
135 ask_store_each_time=ask_store_each_time,
136 )
137 try:
138 settings = BitTorrentSettings(
139 udp_protocol_id=_integer_setting(
140 minerva_payload,
141 "udp_protocol_id",
142 defaults.udp_protocol_id,
143 ),
144 block_size=_integer_setting(minerva_payload, "block_size", defaults.block_size),
145 max_torrent_bytes=_integer_setting(
146 minerva_payload,
147 "max_torrent_bytes",
148 defaults.max_torrent_bytes,
149 ),
150 max_tracker_bytes=_integer_setting(
151 minerva_payload,
152 "max_tracker_bytes",
153 defaults.max_tracker_bytes,
154 ),
155 max_peer_attempts=_integer_setting(
156 minerva_payload,
157 "max_peer_attempts",
158 defaults.max_peer_attempts,
159 ),
160 peer_race_workers=_integer_setting(
161 minerva_payload,
162 "peer_race_workers",
163 defaults.peer_race_workers,
164 ),
165 max_peer_timeout_seconds=_float_setting(
166 minerva_payload,
167 "max_peer_timeout_seconds",
168 defaults.max_peer_timeout_seconds,
169 ),
170 max_tracker_queries=_integer_setting(
171 minerva_payload,
172 "max_tracker_queries",
173 defaults.max_tracker_queries,
174 ),
175 max_discovered_peers=_integer_setting(
176 minerva_payload,
177 "max_discovered_peers",
178 defaults.max_discovered_peers,
179 ),
180 )
181 except ValueError:
182 settings = defaults
183 return Preferences(
184 store_id=normalized_store_id,
185 minerva_bittorrent=settings,
186 catalogue_ttl_days=ttl_days,
187 log_level=log_level,
188 log_to_file=log_to_file,
189 language=language,
190 max_concurrent_downloads=max_concurrent_downloads,
191 rate_limit_retry=rate_limit_retry,
192 default_roms_directory=default_roms_directory,
193 network_timeout_seconds=network_timeout_seconds,
194 store_rom_mappings=store_rom_mappings,
195 bios_directory=bios_directory,
196 ask_store_each_time=ask_store_each_time,
197 )
200def save_preferences(path: Path, preferences: Preferences) -> None:
201 """Atomically persist preferences without leaving a partial JSON file."""
203 temporary = path.with_name(path.name + ".tmp")
204 try:
205 path.parent.mkdir(parents=True, exist_ok=True)
206 settings = preferences.minerva_bittorrent
207 temporary.write_text(
208 json.dumps(
209 {
210 "store": preferences.store_id,
211 "ask_store_each_time": preferences.ask_store_each_time,
212 "catalogue_ttl_days": preferences.catalogue_ttl_days,
213 "log_level": preferences.log_level,
214 "log_to_file": preferences.log_to_file,
215 "language": preferences.language,
216 "max_concurrent_downloads": preferences.max_concurrent_downloads,
217 "default_roms_directory": preferences.default_roms_directory,
218 "network_timeout_seconds": preferences.network_timeout_seconds,
219 "store_rom_mappings": preferences.store_rom_mappings,
220 "bios_directory": preferences.bios_directory,
221 "rate_limit_retry": {
222 "base_seconds": preferences.rate_limit_retry.base_seconds,
223 "max_seconds": preferences.rate_limit_retry.max_seconds,
224 "jitter_ratio": preferences.rate_limit_retry.jitter_ratio,
225 },
226 "minerva_bittorrent": {
227 "udp_protocol_id": settings.udp_protocol_id,
228 "block_size": settings.block_size,
229 "max_torrent_bytes": settings.max_torrent_bytes,
230 "max_tracker_bytes": settings.max_tracker_bytes,
231 "max_peer_attempts": settings.max_peer_attempts,
232 "peer_race_workers": settings.peer_race_workers,
233 "max_peer_timeout_seconds": settings.max_peer_timeout_seconds,
234 "max_tracker_queries": settings.max_tracker_queries,
235 "max_discovered_peers": settings.max_discovered_peers,
236 },
237 },
238 ensure_ascii=False,
239 indent=2,
240 )
241 + "\n",
242 encoding="utf-8",
243 )
244 os.replace(temporary, path)
245 except OSError as error:
246 temporary.unlink(missing_ok=True)
247 raise PreferencesError(f"Could not save TUI settings: {error}") from error
250def _integer_setting(payload: dict[object, object], key: str, default: int) -> int:
251 value = payload.get(key)
252 return value if isinstance(value, int) and not isinstance(value, bool) else default
255def _float_setting(payload: dict[object, object], key: str, default: float) -> float:
256 value = payload.get(key)
257 if isinstance(value, int | float) and not isinstance(value, bool):
258 return float(value)
259 return default
262def _optional_float_setting(payload: dict[object, object], key: str) -> float | None:
263 value = payload.get(key)
264 if isinstance(value, int | float) and not isinstance(value, bool):
265 return float(value)
266 return None
269def _relative_directory(value: object) -> str | None:
270 if not isinstance(value, str) or not value.strip():
271 return None
272 path = Path(value.strip())
273 if path.is_absolute() or any(part in {"", ".", ".."} for part in path.parts):
274 return None
275 return path.as_posix()
278def _store_rom_mappings(value: object) -> dict[str, dict[str, str]]:
279 if not isinstance(value, dict):
280 return {}
281 mappings: dict[str, dict[str, str]] = {}
282 for raw_store_id, raw_platforms in value.items():
283 if not isinstance(raw_store_id, str) or not isinstance(raw_platforms, dict):
284 continue
285 store_id = raw_store_id.strip().casefold()
286 if not store_id: 286 ↛ 287line 286 didn't jump to line 287 because the condition on line 286 was never true
287 continue
288 platforms: dict[str, str] = {}
289 for raw_slug, raw_directory in raw_platforms.items():
290 if not isinstance(raw_slug, str): 290 ↛ 291line 290 didn't jump to line 291 because the condition on line 290 was never true
291 continue
292 slug = raw_slug.strip().casefold()
293 directory = _relative_directory(raw_directory)
294 if slug and directory is not None:
295 platforms[slug] = directory
296 if platforms: 296 ↛ 282line 296 didn't jump to line 282 because the condition on line 296 was always true
297 mappings[store_id] = platforms
298 return mappings