kodo/utils/qiniucdn.py

# -*- coding: utf-8 -*-

import pngquant
import qiniu
from django.conf import settings


QINIU = settings.QINIU
auth = qiniu.Auth(QINIU['access_key'], QINIU['secret_key'])
pngquant.config(settings.PNG_QUANT_FILE)


def upload(data, key=None, mime_type='application/octet-stream', bucket=QINIU['bucket_default'], compress=True):
    if not data:
        return ''
    if compress:
        try:
            data = pngquant.quant_data(data, ndeep=3)[1]
        except Exception as e:
            pass
    token = auth.upload_token(bucket, key=key)
    ret, _ = qiniu.put_data(token, key, data, mime_type=mime_type)
    return ret.get('key')


def upload_file_admin(obj, key=None, mime_type='application/octet-stream', bucket=QINIU['bucket_default'], compress=True):
    # Django Admin Upload
    if not obj.image:
        return ''
    obj.image.seek(0)
    return upload(obj.image.read(), key=key or obj.image.name, mime_type=mime_type, bucket=bucket, compress=compress)


def upload_file_req(file, key=None, mime_type='application/octet-stream', bucket=QINIU['bucket_default'], compress=True):
    # photo = request.FILES.get('photo', '')
    # <InMemoryUploadedFile: photo.png (image/png)>
    if not file:
        return ''
    file.seek(0)
    return upload(file.read(), key=key or file.name, mime_type=mime_type, bucket=bucket, compress=compress)


def upload_file_path(path, key=None, mime_type='application/octet-stream', bucket=QINIU['bucket_default'], compress=True):
    if not path:
        return ''
    token = auth.upload_token(bucket, key=key)
    ret, _ = qiniu.put_file(token, key, path, mime_type=mime_type)
    return ret.get('key')


def qiniu_file_url(key, bucket=QINIU['bucket_default']):
    if not key:
        return ''
    return '{}/{}'.format(QINIU['buckets'][bucket] if settings.QINIU_FILE_URL_HTTPS else settings.QINIU_FILE_URL_AFTER, key)