NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

image_collection_disk_storage.py (3050B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\caching\client\storage\image_collection_disk_storage.py
      2 import os
      3 from datetime import datetime
      4 import blue
      5 from caching.client.cache_entry import CacheEntry
      6 from caching.client.cache_key import CacheKey
      7 from caching.client.storage.base_storage import BaseCollectionStorage
      8 
      9 class ImageCollectionDiskStorage(BaseCollectionStorage):
     10 
     11     def __init__(self, content_id, file_extension = '.jpg'):
     12         super(ImageCollectionDiskStorage, self).__init__(content_id)
     13         self._file_extension = file_extension
     14         self._base_uri = u'cache:/Storage/%s' % content_id
     15 
     16     def _ensure_dir(self):
     17         base_resolved = blue.paths.ResolvePathForWriting(u'cache:/Storage')
     18         if not os.path.exists(base_resolved):
     19             os.makedirs(base_resolved)
     20         content_resolved = blue.paths.ResolvePathForWriting(self._base_uri)
     21         if not os.path.exists(content_resolved):
     22             os.makedirs(content_resolved)
     23 
     24     def _resolve(self, key):
     25         return blue.paths.ResolvePathForWriting(self.get_uri(key))
     26 
     27     def get_uri(self, key):
     28         return u'%s/%s%s' % (self._base_uri, key, self._file_extension)
     29 
     30     def get_resolved_dir(self):
     31         return blue.paths.ResolvePathForWriting(self._base_uri)
     32 
     33     def get(self, key):
     34         path = self._resolve(key)
     35         if os.path.exists(path):
     36             return self.get_uri(key)
     37 
     38     def get_entry(self, key):
     39         path = self._resolve(key)
     40         if not os.path.exists(path):
     41             return None
     42         stored_at = datetime.utcfromtimestamp(os.path.getmtime(path))
     43         cache_key = CacheKey(self.content_id, str(key))
     44         return CacheEntry(cache_key, self.get_uri(key), stored_at=stored_at)
     45 
     46     def get_all(self):
     47         result = {}
     48         base_path = self.get_resolved_dir()
     49         if os.path.exists(base_path):
     50             ext_len = len(self._file_extension)
     51             for file_name in os.listdir(base_path):
     52                 if file_name.endswith(self._file_extension):
     53                     key = file_name[:-ext_len]
     54                     result[key] = self.get_uri(key)
     55 
     56         return result
     57 
     58     def set(self, key, value):
     59         self._ensure_dir()
     60         path = self._resolve(key)
     61         with open(path, 'wb') as f:
     62             f.write(value)
     63 
     64     def set_all(self, values):
     65         for key, value in values.iteritems():
     66             self.set(key, value)
     67 
     68     def invalidate(self, key):
     69         uri = self.get_uri(key)
     70         path = self._resolve(key)
     71         if os.path.exists(path):
     72             os.remove(path)
     73             blue.motherLode.Delete(uri)
     74 
     75     def clear(self):
     76         base_path = self.get_resolved_dir()
     77         if os.path.exists(base_path):
     78             for file_name in os.listdir(base_path):
     79                 try:
     80                     os.remove(os.path.join(base_path, file_name))
     81                     uri = u'%s/%s' % (self._base_uri, file_name)
     82                     blue.motherLode.Delete(uri)
     83                 except OSError:
     84                     pass