# +====================================================================== # | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ] # +---------------------------------------------------------------------- # | Copyright (c) 2021-2026 一米阳光团队 版权所有 # +---------------------------------------------------------------------- # | 开源协议: Apache-2.0 # | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。 # +---------------------------------------------------------------------- # | 官方网站: https://www.djangoadmin.cn # +---------------------------------------------------------------------- # | 研发团队: 一米阳光团队 荣誉出品 # +====================================================================== # | 版权和免责声明: # | 一、知识产权声明 # | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、 # | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权， # | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。 # | 二、使用限制与禁止条款 # | 您理解并同意，禁止将本产品用于以下任何行为： # | 1. 任何违反中华人民共和国法律法规的用途； # | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为； # | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。 # | 三、责任豁免 # | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失 # | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三 # | 方提起索赔或行政机关进行调查，您应独立承担全部责任。 # | 四、违约与法律责任 # | 如发现任何违反本声明的行为，本团队有权： # | 1. 无条件配合公安机关及其他有权机关的调查取证工作； # | 2. 依法追究相关方的民事责任，并要求赔偿全部损失； # | 3. 保留进一步采取刑事控告等法律措施的权利。 # | 五、许可范围说明 # | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包 # | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通 # | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、 # | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造 # | 成的一切经济损失及商誉损害。 # | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件 # | 许可协议》附件为准。 # +====================================================================== """ 文件上传服务模块 ============================================================= 该模块实现了文件上传相关的业务逻辑。 支持图片和各类文件的上传，自动进行文件类型验证、存储路径生成、 文件重命名等操作，并返回标准化的文件信息。 """ import logging import os import re import random import time import mimetypes from flask import request from config import FLASK_TEMP_PATH, FLASK_UPLOAD_DIR, UPLOAD_ALLOWED_EXTS, UPLOAD_MAX_SIZE_MB from utils import R from utils.file import mkdir, verify_file_magic from utils.rich_text import get_file_url # 获取日志记录器 logger = logging.getLogger(__name__) # ============================================================ # 文件上传 # ============================================================ def upload_file(file_exts=UPLOAD_ALLOWED_EXTS): """ 上传文件（兼容图片和各类文件） Args: file_exts: 允许的文件扩展名列表或字符串，如 ['.jpg', '.png'] 或 ".jpg|.png|.gif" 默认使用 UPLOAD_ALLOWED_EXTS 全局配置 Returns: dict: 包含文件信息的字典，格式为： { 'originalName': 原始文件名, 'fileExtension': 文件扩展名, 'fileType': 文件MIME类型, 'fileSize': 文件大小, 'fileName': 生成的文件名, 'filePath': 文件相对路径, 'fileUrl': 文件访问URL } """ try: # 步骤1：获取上传的文件对象 file = request.files.get('file') if not file: logger.warning("未选择上传文件") return R.failed(msg="请选择要上传的文件") # 步骤2：获取文件基本信息 original_name = file.filename file_size = _get_file_size(file) # 步骤2.5：校验文件大小 max_size_bytes = UPLOAD_MAX_SIZE_MB * 1024 * 1024 if file_size and file_size > max_size_bytes: logger.warning(f"文件大小超限: {file_size} bytes, 上限: {max_size_bytes} bytes") return R.failed(msg=f"文件大小超过限制，最大允许 {UPLOAD_MAX_SIZE_MB}MB") # 步骤3：获取并处理文件扩展名 file_ext_with_dot = os.path.splitext(original_name)[1].lower() file_ext = file_ext_with_dot[1:] if file_ext_with_dot.startswith('.') else '' # 步骤4：验证文件扩展名是否在允许的范围内 if file_exts: allowed_exts = _parse_allowed_extensions(file_exts) if allowed_exts and file_ext_with_dot not in allowed_exts: logger.warning(f"文件格式不支持，允许: {allowed_exts}，实际: {file_ext_with_dot}") return R.failed(msg=f"文件格式不支持，仅支持 {', '.join(allowed_exts)} 格式") # 步骤4.5：校验文件魔术字节，防止扩展名伪造 is_valid, magic_msg = verify_file_magic(file, file_ext_with_dot) if not is_valid: logger.warning(f"文件魔术字节校验失败: {magic_msg}，文件名: {original_name}") return R.failed(msg=magic_msg) # 步骤5：获取文件的MIME类型 mime_type = _get_mime_type(original_name, file_ext_with_dot) # 步骤6：生成新的文件名（避免文件名冲突） generated_name = _generate_filename(file_ext_with_dot) # 步骤7：生成存储路径（按日期分目录存储） date_dir = time.strftime('%Y%m%d') save_dir = os.path.join(FLASK_TEMP_PATH, date_dir) # 步骤8：创建存放目录（如果目录不存在） mkdir(save_dir) # 步骤9：生成完整的文件保存路径 file_path = os.path.join(save_dir, generated_name) # 步骤10：保存文件 file.save(file_path) # 步骤10.5：SVG文件清洗，移除可执行脚本防止XSS攻击 if file_ext_with_dot == '.svg': _sanitize_svg(file_path) # 步骤11：获取相对路径（用于存储和返回） relative_path = file_path.replace(FLASK_UPLOAD_DIR, "").replace("\\", "/") # 步骤12：生成文件的访问URL file_url = get_file_url(relative_path) # 步骤13：构建标准化的返回数据 result = { 'originalName': original_name, 'fileExtension': file_ext, 'fileType': mime_type, 'fileSize': file_size, 'fileName': generated_name, 'filePath': relative_path, 'fileUrl': file_url, } return R.ok(msg="上传成功", data=result) except Exception as e: logger.error(f"文件上传异常: {str(e)}") return R.failed(msg=f"上传失败: {str(e)}") # ============================================================ # 内部辅助函数 # ============================================================ def _parse_allowed_extensions(exts): """ 解析允许的扩展名参数（内部工具函数） 将各种格式的扩展名参数统一转换为带点的列表格式。 支持的输入格式： - 列表：['.jpg', '.png'] 或 ['jpg', 'png'] - 字符串：".jpg|.png|.gif" 或 "jpg,png,gif" 或 "jpg" Args: exts: 扩展名参数，可以是字符串或列表 Returns: list: 允许的扩展名列表，格式为 ['.jpg', '.png'] """ if not exts: return [] if isinstance(exts, list): return [ext if ext.startswith('.') else f'.{ext}' for ext in exts] if isinstance(exts, str): if '|' in exts: parts = exts.split('|') elif ',' in exts: parts = exts.split(',') else: parts = [exts] result = [] for part in parts: part = part.strip().lower() if part: if not part.startswith('.'): part = f'.{part}' result.append(part) return result return [] def _get_mime_type(filename, ext_with_dot): """ 获取文件的MIME类型（内部工具函数） Args: filename: 原始文件名 ext_with_dot: 带点的扩展名，如 '.jpg' Returns: str: MIME类型 """ mime_type, _ = mimetypes.guess_type(filename) if not mime_type: mime_type = _guess_mime_type(ext_with_dot) return mime_type def _guess_mime_type(ext_with_dot): """ 根据扩展名推断MIME类型（内部工具函数） 当mimetypes.guess_type无法识别时，使用此函数手动推断MIME类型。 Args: ext_with_dot: 带点的扩展名，如 '.jpg' Returns: str: MIME类型，如果无法识别则返回默认的'application/octet-stream' """ mime_map = { # 图片类型 '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', '.png': 'image/png', '.gif': 'image/gif', '.bmp': 'image/bmp', '.webp': 'image/webp', '.svg': 'image/svg+xml', '.ico': 'image/x-icon', # 文本类型 '.txt': 'text/plain', '.html': 'text/html', '.htm': 'text/html', '.css': 'text/css', '.js': 'application/javascript', '.json': 'application/json', '.xml': 'application/xml', # 文档类型 '.pdf': 'application/pdf', '.doc': 'application/msword', '.docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', '.xls': 'application/vnd.ms-excel', '.xlsx': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', '.ppt': 'application/vnd.ms-powerpoint', '.pptx': 'application/vnd.openxmlformats-officedocument.presentationml.presentation', # 音频视频类型 '.mp3': 'audio/mpeg', '.mp4': 'video/mp4', '.avi': 'video/x-msvideo', '.mov': 'video/quicktime', # 压缩包类型 '.zip': 'application/zip', '.rar': 'application/x-rar-compressed', '.7z': 'application/x-7z-compressed', '.tar': 'application/x-tar', '.gz': 'application/gzip', } return mime_map.get(ext_with_dot, 'application/octet-stream') def _generate_filename(ext_with_dot): """ 生成唯一的文件名（内部工具函数） 格式：年月日时分秒 + 5位随机数 + 扩展名 Args: ext_with_dot: 带点的扩展名，如 '.jpg' Returns: str: 生成的文件名 """ timestamp = time.strftime('%Y%m%d%H%M%S') random_num = '%05d' % random.randint(0, 99999) return f"{timestamp}{random_num}{ext_with_dot}" def _get_file_size(file): """ 获取文件大小（内部工具函数） Flask中可以通过多种方式获取文件大小： - file.content_length: 内容长度 - 保存后使用 os.path.getsize Args: file: Flask文件对象 Returns: int: 文件大小（字节），如果无法获取则返回0 """ try: # 尝试获取 content_length if hasattr(file, 'content_length') and file.content_length: return file.content_length # 保存文件位置后获取大小 if hasattr(file, 'filename') and file.filename: # 临时保存到内存流中获取大小 file.seek(0, 2) # 移动到文件末尾 size = file.tell() # 获取当前位置（文件大小） file.seek(0) # 重置到文件开头 return size except Exception as e: logger.warning(f"获取文件大小异常: {str(e)}") return 0 # SVG中需要移除的危险标签 _SVG_DANGEROUS_TAGS = re.compile( r'<\s*/?(?:script|iframe|object|embed|applet|form|input|button|textarea|select|link|meta|base)' r'(?:\s[^>]*)?\s*/?>', re.IGNORECASE | re.DOTALL, ) # SVG中需要移除的危险事件属性（on*） _SVG_EVENT_ATTRS = re.compile( r'\s+on\w+\s*=\s*(?:"[^"]*"|\'[^\']*\'|[^\s>]+)', re.IGNORECASE, ) # SVG中需要移除的危险href（javascript: / data: 协议） _SVG_DANGEROUS_HREF = re.compile( r'(\s+(?:href|xlink:href)\s*=\s*)(?:"javascript:[^"]*"|\'javascript:[^\']*\'|javascript:[^\s>]+' r'|"data:[^"]*"|\'data:[^\']*\'|data:[^\s>]+)', re.IGNORECASE, ) def _sanitize_svg(file_path): """ 清洗SVG文件，移除可执行脚本内容防止XSS攻击 SVG是文本格式的XML，可内嵌