""" 通用文件访问模块 提供文件下载、预览、缩略图、分享访问等接口 """ from __future__ import annotations from apiflask import APIBlueprint from flask import request, send_file from flask_jwt_extended import jwt_required from iti.applications.common.exceptions.biz_exp import BizException from iti.applications.common.utils import success from iti.applications.common.enums import StorageTypeEnum from iti_system.service.sys.sys_file import SysFileService bp = APIBlueprint( "common_file_access", __name__, url_prefix="/file", tag="通用.文件访问" ) # --------------------------------------------------------------------------- # 文件访问接口 # --------------------------------------------------------------------------- @bp.get("//download") @jwt_required(optional=True) def download_file(file_id: str): """下载文件""" try: file_obj, file_stream = SysFileService.download_file(file_id) # 对文件名进行 URL 编码以支持中文等特殊字符 from urllib.parse import quote encoded_filename = quote(file_obj.filename) response = send_file( file_stream, mimetype=file_obj.mime_type or "application/octet-stream", as_attachment=True, ) # 使用 RFC 5987 标准格式支持 UTF-8 文件名 response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}" return response except Exception as e: raise BizException(f"文件下载失败: {str(e)}") @bp.get("//preview") @jwt_required(optional=True) def preview_file(file_id: str): """预览文件""" try: file_obj, file_stream = SysFileService.download_file(file_id) except Exception as e: raise BizException(f"预览失败: {str(e)}") if file_obj.storage_type != StorageTypeEnum.LOCAL: from flask import redirect from iti.applications.common.storage import StorageManager storage = StorageManager.get_storage(file_obj.storage_type) oss_url = storage.get_url(file_obj.file_key, expires=3600) return redirect(oss_url) # 本地文件直接返回 try: response = send_file( file_stream, mimetype=file_obj.mime_type or "application/octet-stream", as_attachment=False, # 预览模式 ) # 对文件名进行 URL 编码以支持中文等特殊字符 from urllib.parse import quote encoded_filename = quote(file_obj.filename) # 统一设置预览响应头,防止 IDM 等下载管理器拦截 # 使用 RFC 5987 标准格式支持 UTF-8 文件名 response.headers["Content-Disposition"] = f"inline; filename*=UTF-8''{encoded_filename}" response.headers["X-Content-Type-Options"] = "nosniff" response.headers["Cache-Control"] = "public, max-age=3600" return response except Exception as e: raise BizException(f"文件预览失败: {str(e)}") @bp.get("//thumbnail") @jwt_required(optional=True) def thumbnail_file(file_id: str): """获取缩略图""" # 获取缩略图参数(参考阿里云OSS风格) width = request.args.get("w", type=int) or 200 height = request.args.get("h", type=int) or 200 mode = request.args.get("mode", "fit") # fit/fill/pad thumbnail_stream = SysFileService.get_thumbnail( file_id, width=width, height=height, mode=mode ) return send_file( thumbnail_stream, mimetype="image/jpeg", as_attachment=False, ) # --------------------------------------------------------------------------- # 分享文件访问接口 # --------------------------------------------------------------------------- @bp.get("/share/") def access_share(share_code: str): """ 访问分享文件 查询参数: - password: 分享密码(如果需要) 响应示例: ```json { "success": true, "data": { "id": "xxx", "filename": "example.pdf", "fileSize": 1024000, "url": "http://...", ... } } ``` """ password = request.args.get("password") file_obj = SysFileService.get_file_by_share_code(share_code, password) return success(file_obj) @bp.get("/share//download") def download_share(share_code: str): """下载分享文件""" try: password = request.args.get("password") file_obj = SysFileService.get_file_by_share_code(share_code, password) _, file_stream = SysFileService.download_file(file_obj.id) # 对文件名进行 URL 编码以支持中文等特殊字符 from urllib.parse import quote encoded_filename = quote(file_obj.filename) response = send_file( file_stream, mimetype=file_obj.mime_type or "application/octet-stream", as_attachment=True, ) # 使用 RFC 5987 标准格式支持 UTF-8 文件名 response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}" return response except Exception as e: raise BizException(f"分享文件下载失败: {str(e)}")