123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142 |
- # -*- coding: utf-8 -*-
- # !/usr/bin/env python
- import os
- import uuid
- import hashlib
- import datetime
- from django.conf import settings
- from oss2.headers import OSS_OBJECT_TAGGING
- from apps.thirdparties import AliOSS
- from .exceptions import InvalidFileSize, InvalidFileName
- def byte_to_mega(bytes_):
- return bytes_ // 1024.0 // 1024.0
- class FileUploader(object):
- """
- 单机部署可以直接上传到本地目录
- 多机部署目前使用AliOssFileUploader上传到AliOSS
- 如果后续AliOSS成本高, 可以考虑使用minio部署文件服务
- """
- FILE_SIZE_LIMIT = 5
- filenameTemplate = '{filename}.{suffix}'
- def __init__(self, inputFile, uploadType):
- self.uploadType = uploadType
- self._inputFile = inputFile
- self.filename = self.filenameTemplate.format(filename = self._get_filename(), suffix = self._grab_suffix())
- @property
- def storagePath(self):
- return settings.MEDIA_ROOT
- @property
- def storageUrlPrefix(self):
- return '{}{}'.format(settings.FRONT_END_BASE_URL, settings.MEDIA_URL)
- @property
- def middlePath(self):
- return os.path.join(self.uploadType, datetime.datetime.now().strftime("%Y%m%d")).replace("\\", "/")
- def _get_filename(self):
- return hashlib.md5(str(uuid.uuid1())).hexdigest()
- def _grab_suffix(self):
- try:
- return self._inputFile.name.split('.')[1]
- except IndexError:
- raise InvalidFileName(u'文件名不合法,需要包含后缀')
- @property
- def byte_size_limit(self):
- return self.FILE_SIZE_LIMIT * 1024 * 1024
- @property
- def outputUrl(self):
- return self.storageUrlPrefix + self.filename
- def check(self):
- self._inputFile.seek(0, os.SEEK_END)
- size = self._inputFile.tell()
- if size > self.byte_size_limit:
- raise InvalidFileSize(u'文件过大(%sM),应该小于%dM!' % (byte_to_mega(size), self.FILE_SIZE_LIMIT),
- self.FILE_SIZE_LIMIT)
- def write(self):
- output_path = os.path.join(self.storagePath, self.middlePath)
- if not os.path.isdir(output_path):
- os.makedirs(output_path)
- output_file_path = os.path.join(output_path, self.filename)
- with open(output_file_path, 'wb') as destination:
- for chunk in self._inputFile.chunks():
- destination.write(chunk)
- def upload(self):
- self.check()
- self.write()
- return self.outputUrl
- def __repr__(self):
- return '<%s>with file(%s)' % (self.__class__.__name__, self.filename)
- class AliOssFileUploader(FileUploader):
- @property
- def storageUrlPrefix(self):
- return settings.MEDIA_URL
- @property
- def outputUrl(self):
- return settings.OSS_RESOURCE_URL + self.storageUrlPrefix + self.middlePath + '/' + self.filename
- def write(self):
- put_path = self.storageUrlPrefix[1:] + self.middlePath + '/' + self.filename
- AliOSS().put_object(put_path, self._inputFile.chunks())
- return self.outputUrl
- @classmethod
- def load(cls, url):
- get_path = url.replace(settings.OSS_RESOURCE_URL, '')
- return AliOSS().get_object(get_path[1:])
- class WechatSubscriptionAccountVerifyFileUploader(FileUploader):
- """
- 微信公众号自主配置时需要上传.txt文件来验证
- """
- @property
- def storageUrlPrefix(self):
- return '{}/'.format(settings.FRONT_END_BASE_URL)
- def _get_filename(self):
- return self._inputFile.name.split('.')[0]
- def write(self):
- put_path = self.filename
- AliOSS().put_object(put_path, self._inputFile.chunks(), headers = {
- 'Content-Disposition': 'attachment',
- OSS_OBJECT_TAGGING: 'lifecycle=7'
- })
- class SwapGroupPicFileUploader(AliOssFileUploader):
- """
- 互联互通用的充电站的图片上传
- """
- def __init__(self, inputFile, uploadType,groupId,tail):
- self.groupId = groupId
- self.tail = tail
- super(SwapGroupPicFileUploader, self).__init__(inputFile,uploadType)
-
-
- def _get_filename(self):
- return '%s-%s' % (self.groupId,self.tail)
-
|