repository.py (15531B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\achievement_system\client\repository\activity\repository.py 2 import logging 3 from achievement_system.client.api.error import TrackedAchievementLimitReachedError, AchievementAlreadyTrackedError, TrackedAchievementNotFoundError, PinLimitReachedError, AlreadyPinnedError, PinNotFoundError, DefinitionNotFoundError 4 from achievement_system.client.api.signal import on_achievement_tracked, on_achievement_untracked, on_achievements_tracked_changed, on_achievement_pinned, on_achievement_unpinned, on_achievements_pinned_changed, on_achievement_progressed, on_achievement_completed, on_recently_progressed_achievements_changed, on_recently_completed_achievements_changed, on_definitions_loaded, on_admin_achievement_reset, on_admin_achievement_reset_all 5 from achievement_system.client.model.definition.achievement.identifier import AchievementDefinitionIdentifier 6 from achievement_system.client.repository.activity.settings import legacy_tracked_achievement_id_setting 7 from achievement_system.common.const import LOGGER_NAME 8 if False: 9 from typing import List 10 from carbon.common.script.sys.serviceManager import ServiceManager 11 from achievement_system.client.repository.test.setup import MockServiceManager 12 from clientsettings.setting import CharSettingUUID, CharSettingUUIDList 13 from achievement_system.client.repository.activity.test.settings_mock import TrackedAchievementSettingMock, PinnedAchievementSettingMock, RecentlyProgressedAchievementSettingMock, RecentlyCompletedAchievementSettingMock 14 logger = logging.getLogger(LOGGER_NAME) 15 _PINNED_ACHIEVEMENTS_LIMIT = 100 16 _TRACKED_ACHIEVEMENTS_LIMIT = 5 17 _RECENTLY_PROGRESSED_ACHIEVEMENTS_LIMIT = 10 18 _RECENTLY_COMPLETED_ACHIEVEMENTS_LIMIT = 10 19 20 class AchievementActivityRepository(object): 21 __notifyevents__ = ['OnSessionChanged'] 22 23 def __init__(self, service_manager, tracked_achievements_setting, pinned_achievements_setting, recently_progressed_achievements_setting, recently_completed_achievements_setting): 24 self._service_manager = service_manager 25 self._tracked_achievements_setting = tracked_achievements_setting 26 self._pinned_achievements_setting = pinned_achievements_setting 27 self._recently_progressed_achievements_setting = recently_progressed_achievements_setting 28 self._recently_completed_achievements_setting = recently_completed_achievements_setting 29 self._tracked_achievements = [] 30 self._pinned_achievements = [] 31 self._recently_progressed_achievements = [] 32 self._recently_completed_achievements = [] 33 self._service_manager.RegisterNotify(self) 34 on_achievement_progressed.connect(self._on_achievement_progressed) 35 on_achievement_completed.connect(self._on_achievement_completed) 36 on_admin_achievement_reset.connect(self._on_achievement_reset) 37 on_admin_achievement_reset_all.connect(self._on_achievement_reset_all) 38 on_definitions_loaded.connect(self._on_definitions_loaded) 39 40 def __del__(self): 41 self._service_manager.UnregisterNotify(self) 42 on_achievement_progressed.disconnect(self._on_achievement_progressed) 43 on_achievement_completed.disconnect(self._on_achievement_completed) 44 on_admin_achievement_reset.disconnect(self._on_achievement_reset) 45 on_admin_achievement_reset_all.disconnect(self._on_achievement_reset_all) 46 on_definitions_loaded.disconnect(self._on_definitions_loaded) 47 48 def _clear_cache(self): 49 self._tracked_achievements = [] 50 self._pinned_achievements = [] 51 self._recently_progressed_achievements = [] 52 self._recently_completed_achievements = [] 53 54 def _load_settings(self): 55 logger.info('loading achievement activity settings') 56 self._tracked_achievements = self._tracked_achievements_from_setting() 57 self.convert_legacy_tracked_achievement() 58 on_achievements_tracked_changed(self._tracked_achievements) 59 self._pinned_achievements = self._pinned_achievements_from_setting() 60 on_achievements_pinned_changed(self._pinned_achievements) 61 self._recently_progressed_achievements = self._recently_progressed_achievements_from_setting() 62 on_recently_progressed_achievements_changed(self._recently_progressed_achievements) 63 self._recently_completed_achievements = self._recently_completed_achievements_from_setting() 64 on_recently_completed_achievements_changed(self._recently_completed_achievements) 65 66 def convert_legacy_tracked_achievement(self): 67 legacy_tracked_achievement = self._legacy_tracked_achievement_from_setting() 68 if legacy_tracked_achievement is None: 69 legacy_tracked_achievement_id_setting.set(None) 70 return 71 if legacy_tracked_achievement in self._tracked_achievements: 72 legacy_tracked_achievement_id_setting.set(None) 73 return 74 self._tracked_achievements.append(legacy_tracked_achievement) 75 self._tracked_achievements_setting.set(self._tracked_achievements_to_setting(self._tracked_achievements)) 76 legacy_tracked_achievement_id_setting.set(None) 77 78 def get_tracked_achievements(self): 79 return self._tracked_achievements 80 81 def track_achievement(self, achievement_id): 82 if len(self._tracked_achievements) >= _TRACKED_ACHIEVEMENTS_LIMIT: 83 raise TrackedAchievementLimitReachedError 84 if achievement_id in self._tracked_achievements: 85 raise AchievementAlreadyTrackedError 86 self._tracked_achievements.append(achievement_id) 87 self._tracked_achievements_setting.set(self._tracked_achievements_to_setting(self._tracked_achievements)) 88 on_achievement_tracked(achievement_id) 89 on_achievements_tracked_changed(self._tracked_achievements) 90 logger.info('tracked achievement %s', achievement_id) 91 92 def untrack_achievement(self, achievement_id): 93 if achievement_id not in self._tracked_achievements: 94 raise TrackedAchievementNotFoundError 95 self._tracked_achievements.remove(achievement_id) 96 self._tracked_achievements_setting.set(self._tracked_achievements_to_setting(self._tracked_achievements)) 97 on_achievement_untracked(achievement_id) 98 on_achievements_tracked_changed(self._pinned_achievements) 99 logger.info('untracked achievement %s', achievement_id) 100 101 def is_tracked(self, achievement_id): 102 return achievement_id in self._tracked_achievements 103 104 def has_tracked_achievements(self): 105 return len(self._tracked_achievements) > 0 106 107 def get_pinned_achievements(self): 108 return self._pinned_achievements 109 110 def pin_achievement(self, achievement_id): 111 if len(self._pinned_achievements) >= _PINNED_ACHIEVEMENTS_LIMIT: 112 raise PinLimitReachedError 113 if achievement_id in self._pinned_achievements: 114 raise AlreadyPinnedError 115 self._pinned_achievements.append(achievement_id) 116 self._pinned_achievements_setting.set(self._pinned_achievements_to_setting(self._pinned_achievements)) 117 on_achievement_pinned(achievement_id) 118 on_achievements_pinned_changed(self._pinned_achievements) 119 logger.info('pinned achievement %s', achievement_id) 120 121 def unpin_achievement(self, achievement_id): 122 if achievement_id not in self._pinned_achievements: 123 raise PinNotFoundError 124 self._pinned_achievements.remove(achievement_id) 125 self._pinned_achievements_setting.set(self._pinned_achievements_to_setting(self._pinned_achievements)) 126 on_achievement_unpinned(achievement_id) 127 on_achievements_pinned_changed(self._pinned_achievements) 128 logger.info('unpinned achievement %s', achievement_id) 129 130 def is_pinned(self, achievement_id): 131 return achievement_id in self._pinned_achievements 132 133 def get_recently_progressed_achievements(self): 134 return self._recently_progressed_achievements 135 136 def get_recently_completed_achievements(self): 137 return self._recently_completed_achievements 138 139 def admin_reset_recently_progressed_and_completed(self): 140 self._recently_progressed_achievements = [] 141 self._recently_progressed_achievements_setting.set([]) 142 on_recently_progressed_achievements_changed(self._recently_progressed_achievements) 143 self._recently_completed_achievements = [] 144 self._recently_completed_achievements_setting.set([]) 145 on_recently_completed_achievements_changed(self._recently_completed_achievements) 146 147 def OnSessionChanged(self, _is_remote, _session, change): 148 if 'charid' in change: 149 self._load_settings() 150 151 def _on_definitions_loaded(self): 152 self._load_settings() 153 154 def _on_achievement_reset(self, achievement_id): 155 if achievement_id in self._recently_progressed_achievements: 156 self._recently_progressed_achievements.remove(achievement_id) 157 self._recently_progressed_achievements_setting.set(self._recently_progressed_achievements_to_setting(self._recently_progressed_achievements)) 158 on_recently_progressed_achievements_changed(self._recently_progressed_achievements) 159 if achievement_id in self._recently_completed_achievements: 160 self._recently_completed_achievements.remove(achievement_id) 161 self._recently_completed_achievements_setting.set(self._recently_completed_achievements_to_setting(self._recently_completed_achievements)) 162 on_recently_completed_achievements_changed(self._recently_completed_achievements) 163 164 def _on_achievement_reset_all(self): 165 self.admin_reset_recently_progressed_and_completed() 166 167 def _legacy_tracked_achievement_from_setting(self): 168 setting_value = legacy_tracked_achievement_id_setting.get() 169 if not setting_value: 170 return None 171 achievement_id = AchievementDefinitionIdentifier(setting_value) 172 try: 173 from achievement_system.client.api.repository import get_repository 174 get_repository().definitions.get_achievement(achievement_id) 175 return achievement_id 176 except DefinitionNotFoundError: 177 logger.info('%s was tracked but definition not found, skipping', achievement_id) 178 return None 179 180 def _tracked_achievements_from_setting(self): 181 achievements_list = [] 182 for x in self._tracked_achievements_setting.get(): 183 achievement_id = AchievementDefinitionIdentifier(x) 184 try: 185 from achievement_system.client.api.repository import get_repository 186 get_repository().definitions.get_achievement(achievement_id) 187 achievements_list.append(achievement_id) 188 except DefinitionNotFoundError: 189 logger.info('%s was tracked but definition not found, skipping', achievement_id) 190 191 return achievements_list 192 193 @staticmethod 194 def _tracked_achievements_to_setting(achievement_ids): 195 return [ x.uuid for x in achievement_ids ] 196 197 def _pinned_achievements_from_setting(self): 198 achievements_list = [] 199 for x in self._pinned_achievements_setting.get(): 200 achievement_id = AchievementDefinitionIdentifier(x) 201 try: 202 from achievement_system.client.api.repository import get_repository 203 get_repository().definitions.get_achievement(achievement_id) 204 achievements_list.append(achievement_id) 205 except DefinitionNotFoundError: 206 logger.info('%s was pinned but definition not found, skipping', achievement_id) 207 208 return achievements_list 209 210 @staticmethod 211 def _pinned_achievements_to_setting(achievement_ids): 212 return [ x.uuid for x in achievement_ids ] 213 214 def _recently_progressed_achievements_from_setting(self): 215 achievements_list = [] 216 for x in self._recently_progressed_achievements_setting.get(): 217 achievement_id = AchievementDefinitionIdentifier(x) 218 try: 219 from achievement_system.client.api.repository import get_repository 220 get_repository().definitions.get_achievement(achievement_id) 221 achievements_list.append(achievement_id) 222 except DefinitionNotFoundError: 223 logger.info('%s was marked as recently progressed but definition not found, skipping', achievement_id) 224 225 return achievements_list 226 227 @staticmethod 228 def _recently_progressed_achievements_to_setting(achievement_ids): 229 return [ x.uuid for x in achievement_ids ] 230 231 def _recently_completed_achievements_from_setting(self): 232 achievements_list = [] 233 for x in self._recently_completed_achievements_setting.get(): 234 achievement_id = AchievementDefinitionIdentifier(x) 235 try: 236 from achievement_system.client.api.repository import get_repository 237 get_repository().definitions.get_achievement(achievement_id) 238 achievements_list.append(achievement_id) 239 except DefinitionNotFoundError: 240 logger.info('%s was marked as recently completed but definition not found, skipping', achievement_id) 241 242 return achievements_list 243 244 @staticmethod 245 def _recently_completed_achievements_to_setting(achievement_ids): 246 return [ x.uuid for x in achievement_ids ] 247 248 def _on_achievement_progressed(self, achievement_id, _achievement_attr): 249 old_list = [ x for x in self._recently_progressed_achievements ] 250 if achievement_id in self._recently_progressed_achievements: 251 self._recently_progressed_achievements.remove(achievement_id) 252 if len(self._recently_progressed_achievements) == _RECENTLY_PROGRESSED_ACHIEVEMENTS_LIMIT: 253 self._recently_progressed_achievements.pop() 254 self._recently_progressed_achievements.insert(0, achievement_id) 255 self._recently_progressed_achievements_setting.set(self._recently_progressed_achievements_to_setting(self._recently_progressed_achievements)) 256 if self._recently_progressed_achievements != old_list: 257 on_recently_progressed_achievements_changed(self._recently_progressed_achievements) 258 259 def _on_achievement_completed(self, achievement_id): 260 old_list = [ x for x in self._recently_completed_achievements ] 261 if achievement_id in self._recently_completed_achievements: 262 self._recently_completed_achievements.remove(achievement_id) 263 if len(self._recently_completed_achievements) == _RECENTLY_COMPLETED_ACHIEVEMENTS_LIMIT: 264 self._recently_completed_achievements.pop() 265 self._recently_completed_achievements.insert(0, achievement_id) 266 self._recently_completed_achievements_setting.set(self._recently_completed_achievements_to_setting(self._recently_completed_achievements)) 267 if achievement_id in self._recently_progressed_achievements: 268 self._recently_progressed_achievements.remove(achievement_id) 269 self._recently_progressed_achievements_setting.set(self._recently_progressed_achievements_to_setting(self._recently_progressed_achievements)) 270 on_recently_progressed_achievements_changed(self._recently_progressed_achievements) 271 if self._recently_completed_achievements != old_list: 272 on_recently_completed_achievements_changed(self._recently_completed_achievements) 273 if achievement_id in self._pinned_achievements: 274 self.unpin_achievement(achievement_id=achievement_id)