repository.py (12619B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\achievement_system\client\repository\flatbuffer\repository.py 2 from datetime import datetime 3 import logging 4 from achievement_system.client.api.error import DefinitionNotFoundError 5 from achievement_system.client.model.definition.achievement.attributes import AchievementDefinitionAttributes 6 from achievement_system.client.model.definition.achievement.identifier import AchievementDefinitionIdentifier 7 from achievement_system.client.model.definition.achievement.entry import AchievementDefinitionEntry 8 from achievement_system.client.model.definition.category.attributes import CategoryDefinitionAttributes 9 from achievement_system.client.model.definition.category.identifier import CategoryDefinitionIdentifier 10 from achievement_system.client.model.definition.category.roots import AchievementCategoryRoot 11 from achievement_system.client.model.definition.contribution_method.attributes import ContributionMethodAttributes 12 from achievement_system.client.model.definition.contribution_method.identifier import ContributionMethodIdentifier 13 from achievement_system.client.model.flatbuffer.generated import AchievementData as fb_model_root 14 from achievement_system.client.model.flatbuffer.generated import Category as FlatbufferCategory 15 from achievement_system.client.repository.flatbuffer.in_memory_cache import AchievementDefinitionsInMemoryCache 16 from achievement_system.common.const import LOGGER_NAME 17 from publicGateway.gateway import PublicGateway 18 from achievement_system.client.repository.flatbuffer.content_provider import AchievementDefinitionsProvider 19 from caching.client.storage.version_disk_storage import VersionedBinaryDiskStorage 20 from content.client.flatbuffer.repository import VersionCachedFlatbufferRepository 21 from semantic_version import SemanticVersion 22 from featuremanagement.client.flag import flag_bool 23 if False: 24 from typing import Dict, List 25 logger = logging.getLogger(LOGGER_NAME) 26 27 class AchievementDefinitionsRepository(VersionCachedFlatbufferRepository): 28 content_id = 'achievements' 29 provider_type = AchievementDefinitionsProvider 30 disk_storage_type = VersionedBinaryDiskStorage 31 in_memory_storage_type = AchievementDefinitionsInMemoryCache 32 in_memory_ttl = 86400 33 34 def __init__(self, public_gateway, are_unreleased_achievements_visible): 35 super(AchievementDefinitionsRepository, self).__init__(public_gateway=public_gateway) 36 self._featured_categories = [] 37 self._are_unreleased_achievements_visible = are_unreleased_achievements_visible 38 39 def loaded(self): 40 return self._in_memory_cache and len(self._in_memory_cache.root_categories) > 0 and len(self._in_memory_cache.categories_by_id) > 0 and len(self._in_memory_cache.achievements_by_id) > 0 and len(self._in_memory_cache.contribution_methods_by_id) > 0 41 42 def get_root_category(self, root): 43 if root not in self._in_memory_cache.root_categories: 44 raise DefinitionNotFoundError 45 return self._in_memory_cache.root_categories[root] 46 47 def get_category(self, category_identifier): 48 if category_identifier not in self._in_memory_cache.categories_by_id: 49 raise DefinitionNotFoundError 50 return self._in_memory_cache.categories_by_id[category_identifier] 51 52 def get_achievement(self, achievement_identifier): 53 if achievement_identifier not in self._in_memory_cache.achievements_by_id: 54 raise DefinitionNotFoundError 55 return self._in_memory_cache.achievements_by_id[achievement_identifier] 56 57 def get_contribution_method(self, contrib_method_identifier): 58 if contrib_method_identifier not in self._in_memory_cache.contribution_methods_by_id: 59 raise DefinitionNotFoundError 60 return self._in_memory_cache.contribution_methods_by_id[contrib_method_identifier] 61 62 def get_progress_mode(self, achievement_identifier): 63 definition = self.get_achievement(achievement_identifier=achievement_identifier) 64 contribution_method = self.get_contribution_method(contrib_method_identifier=definition.contribution_method.identifier) 65 return contribution_method.progress_config.progress_type 66 67 def get_all_achievements(self): 68 all_achievements = [] 69 for root_category_id in self._in_memory_cache.root_categories.values(): 70 all_achievements.extend(self.get_achievement_entries_recursive(root_category_id)) 71 72 return all_achievements 73 74 def get_all_historic_achievement_entries(self): 75 all_historic_achievements = [] 76 for root_category_id in self._in_memory_cache.root_categories.values(): 77 all_historic_achievements.extend(self.get_historic_achievement_entries_recursive(root_category_id)) 78 79 return all_historic_achievements 80 81 def get_achievement_entries(self, category_identifier): 82 return self._get_category_achievements(category_identifier) 83 84 def get_achievement_entries_recursive(self, parent_category_identifier): 85 return self._get_achievements_recursive(parent_category_identifier, []) 86 87 def get_historic_achievement_entries_recursive(self, parent_category_identifier): 88 return self._get_achievements_recursive(parent_category_identifier, [], include_non_historic=False, include_historic=True) 89 90 def is_root_category(self, category_identifier): 91 for root in AchievementCategoryRoot.ALL: 92 if self.get_root_category(root) == category_identifier: 93 return True 94 95 return False 96 97 def get_featured_categories(self): 98 return self._featured_categories 99 100 def get_current_version_in_memory(self): 101 return self._in_memory_cache.get_version() 102 103 def get_current_version_on_disk(self): 104 return self._disk_storage.get_version() 105 106 def admin_clear_disk_cache(self): 107 self._disk_storage.clear() 108 109 def admin_clear_memory_cache(self): 110 self._in_memory_cache.clear() 111 112 def _get_category_achievements(self, parent_category_identifier, include_non_historic = True, include_historic = False): 113 child_achievement_ids = self.get_category(parent_category_identifier).child_achievements 114 achievements_list = [] 115 for achievement_id in child_achievement_ids: 116 achievement_attributes = self.get_achievement(achievement_id) 117 if achievement_attributes.is_historic and include_historic: 118 achievements_list.append(AchievementDefinitionEntry(achievement_id, achievement_attributes)) 119 if not achievement_attributes.is_historic and include_non_historic: 120 achievements_list.append(AchievementDefinitionEntry(achievement_id, achievement_attributes)) 121 122 return achievements_list 123 124 def _get_achievements_recursive(self, parent_category_identifier, achievements_list, include_non_historic = True, include_historic = False): 125 category_achievements = self._get_category_achievements(parent_category_identifier, include_non_historic, include_historic) 126 achievements_list.extend(category_achievements) 127 category_attributes = self.get_category(parent_category_identifier) 128 for child_category in category_attributes.child_categories: 129 achievements_list = self._get_achievements_recursive(child_category, achievements_list, include_non_historic, include_historic) 130 131 return achievements_list 132 133 def check_version_and_prime(self, latest_version): 134 self._start_request_update_task(self._disk_storage.get_version(), latest_version) 135 136 def _construct_data_from_binary(self, binary_data): 137 root = fb_model_root.GetRootAs(binary_data) 138 root_categories = {} 139 categories_by_id = {} 140 achievements_by_id = {} 141 contribution_methods = {} 142 featured_categories_with_priority = [] 143 self._load_root_category(root.Challenges(), AchievementCategoryRoot.CHALLENGES, root_categories, categories_by_id, achievements_by_id, featured_categories_with_priority) 144 self._load_root_category(root.TimeLimited(), AchievementCategoryRoot.TIME_LIMITED, root_categories, categories_by_id, achievements_by_id, featured_categories_with_priority) 145 self._load_contribution_methods([ root.ContributionMethods(i) for i in range(root.ContributionMethodsLength()) ], contribution_methods) 146 self._save_featured_categories(featured_categories_with_priority) 147 return (root_categories, 148 categories_by_id, 149 achievements_by_id, 150 contribution_methods) 151 152 def _get_default_data(self): 153 return ({}, 154 {}, 155 {}, 156 {}) 157 158 def _load_root_category(self, fb_category_root, root_category, root_categories, categories_by_id, achievements_by_id, featured_categories_with_priority): 159 category_root_id = CategoryDefinitionIdentifier.from_flatbuffer(fb_category_root.Id()) 160 root_categories[root_category] = category_root_id 161 self._load_category_child_categories_recursive(fb_category_root, category_root_id, categories_by_id, achievements_by_id, featured_categories_with_priority) 162 child_achievements_to_skip = self._load_category_child_achievements(fb_category_root, category_root_id, achievements_by_id) 163 category_root_attr = CategoryDefinitionAttributes.from_flatbuffer(fb_category_root, parent_category_id=None, children_to_skip=child_achievements_to_skip) 164 categories_by_id[category_root_id] = category_root_attr 165 if category_root_attr.get_overview_feature_priority() > 0: 166 featured_categories_with_priority.append((category_root_attr.get_overview_feature_priority(), category_root_id)) 167 168 def _load_category_child_categories_recursive(self, fb_category, parent_category_id, categories_by_id, achievements_by_id, featured_categories_with_priority): 169 for i in range(fb_category.CategoriesLength()): 170 fb_child_category = fb_category.Categories(i) 171 child_category_identifier = CategoryDefinitionIdentifier.from_flatbuffer(fb_child_category.Id()) 172 self._load_category_child_categories_recursive(fb_child_category, child_category_identifier, categories_by_id, achievements_by_id, featured_categories_with_priority) 173 child_achievements_to_skip = self._load_category_child_achievements(fb_child_category, child_category_identifier, achievements_by_id) 174 child_category_attr = CategoryDefinitionAttributes.from_flatbuffer(fb_child_category, parent_category_id, child_achievements_to_skip) 175 categories_by_id[child_category_identifier] = child_category_attr 176 if child_category_attr.get_overview_feature_priority() > 0: 177 featured_categories_with_priority.append((child_category_attr.get_overview_feature_priority(), child_category_identifier)) 178 179 def _load_category_child_achievements(self, fb_category, category_id, achievements_by_id): 180 children_to_skip = [] 181 for i in range(fb_category.AchievementsLength()): 182 fb_child_achievement = fb_category.Achievements(i) 183 child_achievement_identifier = AchievementDefinitionIdentifier.from_flatbuffer(fb_child_achievement.Id()) 184 child_achievement_attributes = AchievementDefinitionAttributes.from_flatbuffer(fb_child_achievement, category_id) 185 if child_achievement_attributes.release_date is not None and child_achievement_attributes.release_date > datetime.now() and not self._are_unreleased_achievements_visible(): 186 logger.info('achievement %s is unreleased, skipping definition' % child_achievement_identifier) 187 children_to_skip.append(child_achievement_identifier) 188 continue 189 achievements_by_id[child_achievement_identifier] = child_achievement_attributes 190 191 return children_to_skip 192 193 def _load_contribution_methods(self, fb_contribution_methods, contribution_methods): 194 for fb_contribution_method in fb_contribution_methods: 195 contribution_method_identifier = ContributionMethodIdentifier.from_flatbuffer(fb_contribution_method.Id()) 196 contribution_methods[contribution_method_identifier] = ContributionMethodAttributes.from_flatbuffer(fb_contribution_method) 197 198 def _save_featured_categories(self, featured_categories_with_priority): 199 featured_categories_with_priority.sort(key=lambda c: c[0], reverse=True) 200 self._featured_categories = [ x[1] for x in featured_categories_with_priority ] 201 202 def print_definitions(self): 203 self._in_memory_cache.print_definitions() 204 205 def get(self, key, **kwargs): 206 pass 207 208 def get_all(self, **kwargs): 209 pass