Coverage for src/ph/library.py: 97.2%

1"""Installed game discovery, deletion, and transactional replacement.""" 

2 

3import logging 

4import os 

5import re 

6from collections.abc import Iterable, Iterator, Sequence 

7from pathlib import Path 

8 

9from ph.models import DownloadResult, InstalledGame, Platform 

10from ph.organizer import OrganizeError, unique_destination 

11 

12LOGGER = logging.getLogger(__name__) 

13 

14IGNORED_SUFFIXES = { 

15 ".cfg", 

16 ".db", 

17 ".gif", 

18 ".jpeg", 

19 ".jpg", 

20 ".json", 

21 ".md", 

22 ".mp4", 

23 ".png", 

24 ".sav", 

25 ".srm", 

26 ".state", 

27 ".txt", 

28 ".xml", 

29} 

30IGNORED_DIRECTORIES = {"images", "manuals", "screenshots", "videos"} 

31PLAYLIST_SUFFIXES = {".cue", ".m3u"} 

32 

33 

34class LibraryError(RuntimeError): 

35 """An installed game could not be managed safely.""" 

36 

37 

38def scan_library( 

39 roms_directories: Iterable[Path], platforms: Sequence[Platform] 

40) -> list[InstalledGame]: 

41 """Scan known platform folders on every mounted ROM partition.""" 

42 

43 roots = tuple(roms_directories) 

44 LOGGER.debug("Scanning ROM library roots=%s platforms=%d", roots, len(platforms)) 

45 games: list[InstalledGame] = [] 

46 for root in roots: 

47 root = root.resolve() 

48 for platform in platforms: 

49 for folder_name in platform.rom_folders: 

50 folder = root / folder_name 

51 if not folder.is_dir(): 

52 continue 

53 candidates = list(_iter_game_candidates(folder)) 

54 referenced: set[Path] = set() 

55 grouped: dict[Path, tuple[Path, ...]] = {} 

56 for path in candidates: 

57 if path.suffix.casefold() in PLAYLIST_SUFFIXES: 

58 members = _referenced_files(path) 

59 grouped[path] = tuple(dict.fromkeys((path, *members))) 

60 referenced.update(members) 

61 for path in candidates: 

62 if path in referenced: 

63 continue 

64 members = grouped.get(path, (path,)) 

65 games.append( 

66 InstalledGame( 

67 title=search_title(path), 

68 platform=platform, 

69 roms_directory=root, 

70 primary_file=path, 

71 files=members, 

72 ) 

73 ) 

74 result = sorted(games, key=lambda game: (game.platform.name.casefold(), game.title.casefold())) 

75 LOGGER.info("ROM library scan completed games=%d roots=%d", len(result), len(roots)) 

76 return result 

77 

78 

79def platforms_with_installed_games( 

80 roms_directory: Path, 

81 platforms: Sequence[Platform], 

82) -> tuple[Platform, ...]: 

83 """Quickly find navigable platforms without fully indexing every installed game.""" 

84 

85 available: list[Platform] = [] 

86 for platform in platforms: 

87 if any( 

88 next(_iter_game_candidates(roms_directory / folder), None) is not None 

89 for folder in platform.rom_folders 

90 if (roms_directory / folder).is_dir() 

91 ): 

92 available.append(platform) 

93 return tuple(available) 

94 

95 

96def search_title(path: Path) -> str: 

97 """Turn a ROM filename into a useful remote search phrase.""" 

98 

99 title = path.stem.replace("_", " ").replace(".", " ") 

100 title = re.sub(r"\s*[\[(].*?[\])]", " ", title) 

101 title = re.sub(r"\s+-\s+(?:disc|disk|side)\s*\w+$", " ", title, flags=re.IGNORECASE) 

102 return " ".join(title.split()) or path.stem 

103 

104 

105def delete_game(game: InstalledGame) -> None: 

106 """Permanently remove all files grouped under an installed title.""" 

107 

108 errors: list[str] = [] 

109 for path in game.files: 

110 try: 

111 path.unlink() 

112 except FileNotFoundError: 

113 continue 

114 except OSError as error: 

115 errors.append(f"{path.name}: {error}") 

116 if errors: 

117 LOGGER.error("Could not completely delete game title=%r errors=%s", game.title, errors) 

118 raise LibraryError("Some game files could not be deleted: {}".format("; ".join(errors))) 

119 LOGGER.info("Deleted installed game title=%r files=%d", game.title, len(game.files)) 

120 

121 

122def replace_game(game: InstalledGame, download: DownloadResult) -> DownloadResult: 

123 """Move a complete replacement beside the old game, then remove the old files.""" 

124 

125 source = download.path 

126 if not source.is_file(): 

127 raise LibraryError(f"The completed replacement file was not found: {source}") 

128 requested = game.primary_file.parent / source.name 

129 existing_files = set(game.files) 

130 destination = requested if requested in existing_files else unique_destination(requested) 

131 try: 

132 source.replace(destination) 

133 except OSError: 

134 try: 

135 import shutil 

136 

137 shutil.move(str(source), str(destination)) 

138 except OSError as error: 

139 raise OrganizeError(f"Could not install the replacement: {error}") from error 

140 

141 delete_errors: list[str] = [] 

142 for old_file in game.files: 

143 if old_file == destination: 143 ↛ 144line 143 didn't jump to line 144 because the condition on line 143 was never true

144 continue 

145 try: 

146 old_file.unlink() 

147 except FileNotFoundError: 

148 continue 

149 except OSError as error: 

150 delete_errors.append(f"{old_file.name}: {error}") 

151 if delete_errors: 

152 LOGGER.warning( 

153 "Replacement installed but old files remain title=%r errors=%s", 

154 game.title, 

155 delete_errors, 

156 ) 

157 raise LibraryError( 

158 "The new game was installed at {}, but old files remain: {}".format( 

159 destination, "; ".join(delete_errors) 

160 ) 

161 ) 

162 LOGGER.info("Replaced installed game title=%r path=%s", game.title, destination) 

163 return DownloadResult(url=download.url, path=destination) 

164 

165 

166def _is_game_candidate(path: Path) -> bool: 

167 if not path.is_file() or path.name.startswith("."): 

168 return False 

169 if any(part.casefold() in IGNORED_DIRECTORIES for part in path.parts): 

170 return False 

171 return path.suffix.casefold() not in IGNORED_SUFFIXES 

172 

173 

174def _iter_game_candidates(folder: Path) -> Iterator[Path]: 

175 """Yield game-like files while avoiding expensive media and metadata subtrees.""" 

176 

177 for directory, child_directories, filenames in os.walk(folder, followlinks=False): 

178 child_directories[:] = [ 

179 name 

180 for name in child_directories 

181 if not name.startswith(".") and name.casefold() not in IGNORED_DIRECTORIES 

182 ] 

183 parent = Path(directory) 

184 for filename in filenames: 

185 path = parent / filename 

186 if _is_game_candidate(path): 186 ↛ 184line 186 didn't jump to line 184 because the condition on line 186 was always true

187 yield path.resolve() 

188 

189 

190def _referenced_files(playlist: Path) -> tuple[Path, ...]: 

191 try: 

192 text = playlist.read_text(encoding="utf-8", errors="replace") 

193 except OSError: 

194 return () 

195 names: list[str] = [] 

196 if playlist.suffix.casefold() == ".cue": 

197 names.extend(re.findall(r'^\s*FILE\s+"([^"]+)"', text, flags=re.IGNORECASE | re.MULTILINE)) 

198 else: 

199 names.extend( 

200 line.strip() 

201 for line in text.splitlines() 

202 if line.strip() and not line.lstrip().startswith("#") 

203 ) 

204 members: list[Path] = [] 

205 for name in names: 

206 candidate = (playlist.parent / name).resolve() 

207 try: 

208 candidate.relative_to(playlist.parent.resolve()) 

209 except ValueError: 

210 continue 

211 if candidate.is_file(): 211 ↛ 205line 211 didn't jump to line 205 because the condition on line 211 was always true

212 members.append(candidate) 

213 if candidate.suffix.casefold() in PLAYLIST_SUFFIXES: 

214 members.extend(_referenced_files(candidate)) 

215 return tuple(dict.fromkeys(members))