Coverage for src/ph/store.py: 100.0%

1"""Store contracts shared by the TUI, CLI, and concrete download sources.""" 

2 

3from abc import ABC, abstractmethod 

4from collections.abc import Callable 

5from pathlib import Path 

6from typing import ClassVar 

7 

8from ph.models import MediaDownload, Platform, SearchResult 

9from ph.store_cache import GameCatalogueCache, StoreCacheStatus 

10 

11type CatalogProgress = Callable[[int, int], None] 

12USER_AGENT = ( 

13 "Mozilla/5.0 (X11; Linux aarch64) AppleWebKit/537.36 " 

14 "(KHTML, like Gecko) Chrome/124.0 Safari/537.36" 

15) 

16 

17 

18class StoreError(RuntimeError): 

19 """A user-facing store network or response error.""" 

20 

21 def __init__(self, message: str, status_code: int | None = None) -> None: 

22 super().__init__(message) 

23 self.status_code = status_code 

24 

25 

26class GameStore(ABC): 

27 """A searchable source that can resolve its own game detail links.""" 

28 

29 store_id: ClassVar[str] 

30 display_name: ClassVar[str] 

31 description: ClassVar[str] 

32 timeout_seconds: float 

33 

34 def _configure_catalogue_cache( 

35 self, 

36 cache_directory: Path | None, 

37 ttl_seconds: int, 

38 ) -> None: 

39 self._catalogue_cache = ( 

40 GameCatalogueCache(cache_directory, self.store_id, self.base_url, ttl_seconds) 

41 if cache_directory is not None 

42 else None 

43 ) 

44 

45 def _load_catalogue( 

46 self, 

47 system_code: str, 

48 catalog_progress: CatalogProgress | None, 

49 *, 

50 force: bool = False, 

51 ) -> tuple[SearchResult, ...]: 

52 def fetcher() -> list[SearchResult]: 

53 return self._fetch_catalogue(system_code, catalog_progress) 

54 

55 cache = getattr(self, "_catalogue_cache", None) 

56 if cache is None: 

57 return tuple(fetcher()) 

58 return cache.get_or_fetch(system_code, fetcher, force=force) 

59 

60 def _fetch_catalogue( 

61 self, 

62 system_code: str, 

63 catalog_progress: CatalogProgress | None, 

64 ) -> list[SearchResult]: 

65 raise StoreError(f"{self.display_name} does not expose a cacheable catalogue.") 

66 

67 def refresh_catalogue( 

68 self, 

69 system_code: str, 

70 catalog_progress: CatalogProgress | None = None, 

71 ) -> list[SearchResult]: 

72 """Force-download and atomically replace one platform catalogue.""" 

73 

74 return list(self._load_catalogue(system_code, catalog_progress, force=True)) 

75 

76 def catalogue_cache_status(self, system_code: str) -> StoreCacheStatus | None: 

77 """Return local cache state for one platform without network access.""" 

78 

79 cache = getattr(self, "_catalogue_cache", None) 

80 if cache is None: 

81 return None 

82 return cache.status(system_code) 

83 

84 def catalogue_cache_file_count(self) -> int: 

85 """Return how many structured catalogue files exist for this store.""" 

86 

87 cache = getattr(self, "_catalogue_cache", None) 

88 if cache is None: 

89 return 0 

90 return len(cache.cached_files()) 

91 

92 def set_catalogue_ttl(self, ttl_seconds: int) -> None: 

93 """Apply a changed cache lifetime without rebuilding the store client.""" 

94 

95 cache = getattr(self, "_catalogue_cache", None) 

96 if cache is not None: 

97 cache.ttl_seconds = ttl_seconds 

98 

99 def set_network_timeout(self, timeout_seconds: float) -> None: 

100 """Apply a changed network timeout without rebuilding the store client.""" 

101 

102 self.timeout_seconds = timeout_seconds 

103 

104 @property 

105 @abstractmethod 

106 def base_url(self) -> str: 

107 """Return the configured root URL for this store.""" 

108 

109 @property 

110 @abstractmethod 

111 def download_referrer(self) -> str: 

112 """Return the HTTP referrer expected while downloading store media.""" 

113 

114 @abstractmethod 

115 def platform_code(self, platform: Platform) -> str: 

116 """Translate shared platform metadata to this store's identifier.""" 

117 

118 def supports_platform(self, platform: Platform) -> bool: 

119 """Return whether this store can search the shared platform.""" 

120 

121 return True 

122 

123 @abstractmethod 

124 def search( 

125 self, 

126 system_code: str, 

127 query: str, 

128 catalog_progress: CatalogProgress | None = None, 

129 ) -> list[SearchResult]: 

130 """Search this store using its platform identifier and title prefix.""" 

131 

132 @abstractmethod 

133 def validate_detail_url(self, url: str) -> bool: 

134 """Return whether a detail URL belongs to this store.""" 

135 

136 @abstractmethod 

137 def retrieve_download_url(self, detail_url: str) -> str: 

138 """Resolve one store detail page to its downloadable media URL.""" 

139 

140 def download_request(self, detail_url: str) -> MediaDownload: 

141 """Resolve download metadata; direct-download stores only need the URL.""" 

142 

143 return MediaDownload(self.retrieve_download_url(detail_url))