import os import posixpath from datetime import datetime from urllib.parse import urljoin from io import BytesIO from django.core.files import File from django.core.files.storage import Storage from django.core.signals import setting_changed from django.core.exceptions import ImproperlyConfigured, SuspiciousOperation from django.utils.encoding import force_text, force_bytes from django.conf import settings import oss2 from oss2 import Auth, Bucket, ObjectIterator class AliyunBaseStorage(Storage): default_settings = { 'ACCESS_KEY_ID': '', 'ACCESS_KEY_SECRET': '', 'END_POINT': 'oss-cn-beijing.aliyuncs.com', 'BUCKET_NAME': '', # If you need to access static files by oss's doamin name directly. 'OSS_SERVER_STAIC': True } base_location = '/' base_url = '/' store_static_resouse = False def __init__(self, *args, **kwargs): self._setup() # Re-initialize the storage if settings changed setting_changed.connect(self._setting_changed_handler) super(AliyunBaseStorage, self).__init__(*args, **kwargs) def _setting_changed_handler(self, setting, **kawrgs): if setting == 'ALIYUN_OSS': self._setup() def _setup(self): self.settings = self.default_settings self.settings.update(getattr(settings, 'ALIYUN_OSS', {})) # Simply check settings for k, v in self.settings.items(): if not v: raise ImproperlyConfigured( 'ALIYUN_OSS\'s {setting} cannot be empty, ' 'please defined ALIYUN_OSS = {{\'{setting}\':' ' ...}}'.format(setting=k)) self.auth = Auth(self.settings['ACCESS_KEY_ID'], self.settings['ACCESS_KEY_SECRET']) self.bucket = Bucket(self.auth, self.settings['END_POINT'], self.settings['BUCKET_NAME']) def _clean_name(self, name): """ Cleans the name so that Windows style paths work """ # Normalize Windows style paths clean_name = posixpath.normpath(name).replace('\\', '/') # os.path.normpath() can strip trailing slashes so we implement # a workaround here. if name.endswith('/') and not clean_name.endswith('/'): # Add a trailing slash as it was stripped. return clean_name + '/' else: return clean_name def _normalize_name(self, name): """ Normalizes the name so that paths like /path/to/ignored/../foo.txt work. We check to make sure that the path pointed to is not outside the directory specified by the LOCATION setting. """ base_path = force_text(self.base_location) base_path = base_path.rstrip('/') final_path = urljoin(base_path.rstrip('/') + "/", name) base_path_len = len(base_path) if (not final_path.startswith(base_path) or final_path[base_path_len:base_path_len + 1] not in ('', '/')): raise SuspiciousOperation("Attempted access to '%s' denied." % name) return final_path.lstrip('/') def _get_target_name(self, name): name = self._normalize_name(self._clean_name(name)) return name def _open(self, name, mode='rb'): print('open: {}'.format(name)) if mode != 'rb': raise ValueError('Oss file can noly opened in `rb` mode') name = self._get_target_name(name) return AliyunFile(name, self, mode) def _save(self, name, content): """ Return relative path after save file """ target_name = self._get_target_name(name) content.open() content_str = b''.join(chunk for chunk in content.chunks()) permission = oss2.OBJECT_ACL_DEFAULT # Force set object as public_read if use oss to server static files if self.store_static_resouse and self.settings['OSS_SERVER_STAIC']: permission = oss2.OBJECT_ACL_PUBLIC_READ self.bucket.put_object(target_name, content_str, headers={'x-oss-object-acl': permission}) content.close() return self._clean_name(name) def delete(self, name): name = self._get_target_name(name) result = self.bucket.delete_object(name) if result.status >= 400: raise AliyunOperationError(result.resp) def exists(self, name): """ Input name may be a pathname or filename. """ name = self._normalize_name(self._clean_name(name)) if name.endswith('/') or not name: return self._path_exists(name) else: return self._file_exists(name) or self._path_exists(name) def _file_exists(self, absolute_name): return self.bucket.object_exists(absolute_name) def _path_exists(self, absolute_name): return bool(self.bucket.list_objects(absolute_name, '', max_keys=1).object_list) def listdir(self, name): name = self._normalize_name(self._clean_name(name)) # OSS has different apperance if path not endswith '/' if not name.endswith('/'): name += '/' files = [] dirs = set() for obj in ObjectIterator(self.bucket, prefix=name, delimiter='/'): if obj.is_prefix(): # OSS return absoulte-path, but os usually return a relative-path # To be consistent with the performance of the system file system dirs.add(os.path.relpath(obj.key, name)) else: files.append(os.path.relpath(obj.key, name)) return list(dirs), files def size(self, name): file_info = self.get_file_header(name) return file_info.content_length def url(self, name): return os.path.join(self.base_url, name) def sign_url(self, name, expire=5): return self.bucket.sign_url('GET', self._get_target_name(name), expire, slash_safe=True) def get_modified_time(self, name): file_info = self.get_file_header(name) return datetime.fromtimestamp(file_info.last_modified) class AliyunOperationError(Exception): def __init__(self, value): self.value = value def __str__(self): return repr(self.value) class AliyunFile(File): def __init__(self, name, storage, mode): self._storage = storage self._name = name self._mode = mode self.file = BytesIO() self._is_dirty = False self._is_read = False super(AliyunFile, self).__init__(self.file, self._name) def read(self, num_bytes=None): if not self._is_read: content = self._storage.bucket.get_object(self._name) self.file = BytesIO(content.read()) self._is_read = True if num_bytes is None: data = self.file.read() else: data = self.file.read(num_bytes) if 'b' in self._mode: return data else: return force_text(data) def write(self, content): if 'w' not in self._mode: raise AliyunOperationError("Operation write is not allowed.") self.file.write(force_bytes(content)) self._is_dirty = True self._is_read = True def close(self): if self._is_dirty: self.file.seek(0) self._storage._save(self._name, self.file) self.file.close() class AliyunMediaStorage(AliyunBaseStorage): base_location = settings.MEDIA_ROOT base_url = settings.MEDIA_URL class AliyunStaticStorage(AliyunBaseStorage): base_location = settings.STATIC_ROOT base_url = settings.STATIC_URL store_static_resouse = True def _setup(self): if not settings.STATIC_ROOT: raise ImproperlyConfigured( 'You must defined STATIC_ROOT when use ' 'university.file_storage.aliyun.AliyunStaticStorage') super(AliyunStaticStorage, self)._setup()