django-vue-admin/backend/dvadmin/system/views/file_list.py

import datetime
import hashlib
import json
import os
import random

from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt
from rest_framework import serializers
from rest_framework.decorators import action
from application.settings import BASE_DIR
from application import dispatch
from dvadmin.system.models import FileList
from dvadmin.system.views.ueditor_settings import ueditor_upload_settings, ueditor_settings
from dvadmin.utils.serializers import CustomModelSerializer
from dvadmin.utils.string_util import format_bytes
from dvadmin.utils.viewset import CustomModelViewSet


class FileSerializer(CustomModelSerializer):
    url = serializers.SerializerMethodField(read_only=True)

    def get_url(self, instance):
        return instance.file_url or (f'media/{str(instance.url)}')

    class Meta:
        model = FileList
        fields = "__all__"

    def create(self, validated_data):
        file_engine = dispatch.get_system_config_values("fileStorageConfig.file_engine") or 'local'
        file_backup = dispatch.get_system_config_values("fileStorageConfig.file_backup")
        file = self.initial_data.get('file')
        file_size = file.size
        validated_data['name'] = file.name
        validated_data['size'] = file_size
        md5 = hashlib.md5()
        for chunk in file.chunks():
            md5.update(chunk)
        validated_data['md5sum'] = md5.hexdigest()
        validated_data['engine'] = file_engine
        validated_data['mime_type'] = file.content_type
        if file_backup:
            validated_data['url'] = file
        if file_engine == 'oss':
            from dvadmin_cloud_storage.views.aliyun import ali_oss_upload
            file_path = ali_oss_upload(file)
            if file_path:
                validated_data['file_url'] = file_path
            else:
                raise ValueError("上传失败")
        elif file_engine == 'cos':
            from dvadmin_cloud_storage.views.tencent import tencent_cos_upload
            file_path = tencent_cos_upload(file)
            if file_path:
                validated_data['file_url'] = file_path
            else:
                raise ValueError("上传失败")
        else:
            validated_data['url'] = file
        # 审计字段
        try:
            request_user = self.request.user
            validated_data['dept_belong_id'] = request_user.dept.id
            validated_data['creator'] = request_user.id
            validated_data['modifier'] = request_user.id
        except:
            pass
        return super().create(validated_data)


class FileViewSet(CustomModelViewSet):
    """
    文件管理接口
    list:查询
    create:新增
    update:修改
    retrieve:单例
    destroy:删除
    """
    queryset = FileList.objects.all()
    serializer_class = FileSerializer
    filter_fields = ['name', ]
    permission_classes = []

    @csrf_exempt
    @action(methods=["GET", "POST"], detail=False, permission_classes=[])
    def ueditor(self, request):
        action = self.request.query_params.get("action", "")
        reponse_action = {
            "config": self.get_ueditor_settings,
            "uploadimage": self.upload_file,
            "uploadscrawl": self.upload_file,
            "uploadvideo": self.upload_file,
            "uploadfile": self.upload_file,
        }
        return reponse_action[action](request)

    def get_ueditor_settings(self, request):
        return HttpResponse(json.dumps(ueditor_upload_settings, ensure_ascii=False),
                            content_type="application/javascript")

    # 保存上传的文件
    def save_upload_file(self, file, file_path):
        with open(file_path, 'wb') as f:
            try:
                for chunk in file.chunks():
                    f.write(chunk)

            except Exception as e:
                return f"写入文件错误:{e}"
        return u"SUCCESS"

    def get_path_format_vars(self):
        return {
            "year": datetime.datetime.now().strftime("%Y"),
            "month": datetime.datetime.now().strftime("%m"),
            "day": datetime.datetime.now().strftime("%d"),
            "date": datetime.datetime.now().strftime("%Y%m%d"),
            "time": datetime.datetime.now().strftime("%H%M%S"),
            "datetime": datetime.datetime.now().strftime("%Y%m%d%H%M%S"),
            "rnd": random.randrange(100, 999)
        }

    def get_output_path(self, path_format_var):
        """
        取得输出文件的路径
        :param path_format_var:
        :return:
        """
        file_name = (ueditor_settings["defaultPathFormat"] % path_format_var).replace("\\", "/")
        # 分解OutputPathFormat
        output_path = os.path.join('media', 'ueditor', f'{self.request.user.id}')
        if not os.path.exists(output_path):
            os.makedirs(output_path)
        return (file_name, output_path)

    # 涂鸦功能上传处理
    def save_scrawl_file(self, request, file_path, file_name):
        import base64
        try:
            content = request.data.get(ueditor_upload_settings.get("scrawlFieldName", "upfile"))
            f = open(os.path.join(BASE_DIR, file_path, file_name), 'wb')
            f.write(base64.b64decode(content))
            f.close()
            state = "SUCCESS"
            FileList.save_file(request, file_path, file_name,mime_type='image/png')
        except Exception as e:
            state = f"写入图片文件错误:{e}"
        return state

    def upload_file(self, request):
        """上传文件"""
        state = "SUCCESS"
        action = self.request.query_params.get("action")
        # 上传文件
        upload_field_name_list = {
            "uploadfile": "fileFieldName",
            "uploadimage": "imageFieldName",
            "uploadscrawl": "scrawlFieldName",
            "catchimage": "catcherFieldName",
            "uploadvideo": "videoFieldName",
        }
        upload_field_name = self.request.query_params.get(upload_field_name_list[action],
                                                          ueditor_upload_settings.get(action, "upfile"))
        # 上传涂鸦，涂鸦是采用base64编码上传的，需要单独处理
        if action == "uploadscrawl":
            upload_file_name = "scrawl.png"
            upload_file_size = 0
        else:
            # 取得上传的文件
            file = request.FILES.get(upload_field_name, None)
            if file is None:
                return HttpResponse(json.dumps(u"{'state:'ERROR'}"), content_type="application/javascript")
            upload_file_name = file.name
            upload_file_size = file.size

        # 取得上传的文件的原始名称
        upload_original_name, upload_original_ext = os.path.splitext(upload_file_name)
        # 文件类型检验
        upload_allow_type = {
            "uploadfile": "fileAllowFiles",
            "uploadimage": "imageAllowFiles",
            "uploadvideo": "videoAllowFiles"
        }
        if action in upload_allow_type:
            allow_type = list(self.request.query_params.get(upload_allow_type[action],
                                                            ueditor_upload_settings.get(upload_allow_type[action], "")))
            if not upload_original_ext.lower() in allow_type:
                state = u"服务器不允许上传%s类型的文件。" % upload_original_ext
                return HttpResponse({"state": state}, content_type="application/javascript")

        # 大小检验
        upload_max_size = {
            "uploadfile": "filwMaxSize",
            "uploadimage": "imageMaxSize",
            "uploadscrawl": "scrawlMaxSize",
            "uploadvideo": "videoMaxSize"
        }
        max_size = int(self.request.query_params.get(upload_max_size[action],
                                                     ueditor_upload_settings.get(upload_max_size[action], 0)))
        if max_size != 0:
            if upload_file_size > max_size:
                state = u"上传文件大小不允许超过%s。" % format_bytes(max_size)
                return HttpResponse({"state": state}, content_type="application/javascript")

        path_format_var = self.get_path_format_vars()
        path_format_var.update({
            "basename": upload_original_name,
            "extname": upload_original_ext[1:],
            "filename": upload_file_name,
        })
        # 取得输出文件的路径
        format_file_name, output_path = self.get_output_path(path_format_var)
        # 所有检测完成后写入文件
        if state == "SUCCESS":
            if action == "uploadscrawl":
                state = self.save_scrawl_file(request, file_path=output_path,
                                              file_name=format_file_name)
            else:
                file = request.FILES.get(upload_field_name, None)
                # 保存到文件中，如果保存错误，需要返回ERROR
                state = self.save_upload_file(file, os.path.join(BASE_DIR, output_path, format_file_name))
                # 保存到附件管理中
                FileList.save_file(request, output_path, format_file_name, mime_type=file.content_type)

        # 返回数据
        return_info = {
            'url': os.path.join(output_path, format_file_name),  # 保存后的文件名称
            'original': upload_file_name,  # 原始文件名
            'type': upload_original_ext,
            'state': state,  # 上传状态，成功时返回SUCCESS,其他任何值将原样返回至图片上传框中
            'size': upload_file_size
        }
        return HttpResponse(json.dumps(return_info, ensure_ascii=False), content_type="application/javascript")