From e80be564a5bd5f025ba51894df1c0fd37ceda63b Mon Sep 17 00:00:00 2001 From: Lan Date: Wed, 7 Jan 2026 19:50:02 +0800 Subject: [PATCH] fix: https://github.com/vastsa/FileCodeBox/issues/362 --- apps/base/migrations/migrations_004.py | 14 + apps/base/models.py | 6 +- apps/base/views.py | 349 +++++++++++++++---------- core/tasks.py | 32 +-- 4 files changed, 246 insertions(+), 155 deletions(-) create mode 100644 apps/base/migrations/migrations_004.py diff --git a/apps/base/migrations/migrations_004.py b/apps/base/migrations/migrations_004.py new file mode 100644 index 0000000..69f06fd --- /dev/null +++ b/apps/base/migrations/migrations_004.py @@ -0,0 +1,14 @@ +from tortoise import connections + + +async def add_save_path_to_uploadchunk(): + conn = connections.get("default") + await conn.execute_script( + """ + ALTER TABLE uploadchunk ADD COLUMN save_path VARCHAR(512) NULL; + """ + ) + + +async def migrate(): + await add_save_path_to_uploadchunk() diff --git a/apps/base/models.py b/apps/base/models.py index b38ce4c..428a057 100644 --- a/apps/base/models.py +++ b/apps/base/models.py @@ -49,6 +49,7 @@ class UploadChunk(models.Model): file_size = fields.BigIntField() chunk_size = fields.IntField() file_name = fields.CharField(max_length=255) + save_path = fields.CharField(max_length=512, null=True) created_at = fields.DatetimeField(auto_now_add=True) completed = fields.BooleanField(default=False) @@ -66,6 +67,7 @@ class KeyValue(Model): class PresignUploadSession(models.Model): """预签名上传会话模型""" + id = fields.IntField(pk=True) upload_id = fields.CharField(max_length=36, unique=True, index=True) file_name = fields.CharField(max_length=255) @@ -85,4 +87,6 @@ class PresignUploadSession(models.Model): file_codes_pydantic = pydantic_model_creator(FileCodes, name="FileCodes") upload_chunk_pydantic = pydantic_model_creator(UploadChunk, name="UploadChunk") key_value_pydantic = pydantic_model_creator(KeyValue, name="KeyValue") -presign_upload_session_pydantic = pydantic_model_creator(PresignUploadSession, name="PresignUploadSession") +presign_upload_session_pydantic = pydantic_model_creator( + PresignUploadSession, name="PresignUploadSession" +) diff --git a/apps/base/views.py b/apps/base/views.py index 7804222..453d520 100644 --- a/apps/base/views.py +++ b/apps/base/views.py @@ -5,13 +5,25 @@ import uuid from datetime import timedelta from urllib.parse import unquote +from typing import Optional, Tuple, Union + from fastapi import APIRouter, Form, UploadFile, File, Depends, HTTPException from starlette import status from apps.admin.dependencies import share_required_login from apps.base.models import FileCodes, UploadChunk, PresignUploadSession -from apps.base.schemas import SelectFileModel, InitChunkUploadModel, CompleteUploadModel, PresignUploadInitRequest -from apps.base.utils import get_expire_info, get_file_path_name, ip_limit, get_chunk_file_path_name +from apps.base.schemas import ( + SelectFileModel, + InitChunkUploadModel, + CompleteUploadModel, + PresignUploadInitRequest, +) +from apps.base.utils import ( + get_expire_info, + get_file_path_name, + ip_limit, + get_chunk_file_path_name, +) from core.response import APIResponse from core.settings import settings from core.storage import storages, FileStorageInterface @@ -25,7 +37,9 @@ class FileUploadService: """统一的文件上传服务""" @staticmethod - async def generate_file_path(file_name: str, upload_id: str = None) -> tuple[str, str, str, str, str]: + async def generate_file_path( + file_name: str, upload_id: Optional[str] = None + ) -> tuple[str, str, str, str, str]: """统一的路径生成""" today = datetime.datetime.now() storage_path = settings.storage_path.strip("/") @@ -44,10 +58,12 @@ class FileUploadService: file_path: str, expire_value: int, expire_style: str, - **extra_fields + **extra_fields, ) -> str: """统一创建FileCodes记录，返回code""" - expired_at, expired_count, used_count, code = await get_expire_info(expire_value, expire_style) + expired_at, expired_count, used_count, code = await get_expire_info( + expire_value, expire_style + ) prefix, suffix = os.path.splitext(file_name) await FileCodes.create( @@ -60,7 +76,7 @@ class FileUploadService: expired_at=expired_at, expired_count=expired_count, used_count=used_count, - **extra_fields + **extra_fields, ) return code @@ -68,8 +84,7 @@ class FileUploadService: async def validate_file_size(file: UploadFile, max_size: int) -> int: size = file.size if size is None: - # 读取流计算大小，保持指针复位 - await file.seek(0, 2) + await file.seek(0, 2) # type: ignore[arg-type] size = file.file.tell() await file.seek(0) if size > max_size: @@ -86,10 +101,10 @@ async def create_file_code(code, **kwargs): @share_api.post("/text/", dependencies=[Depends(share_required_login)]) async def share_text( - text: str = Form(...), - expire_value: int = Form(default=1, gt=0), - expire_style: str = Form(default="day"), - ip: str = Depends(ip_limit["upload"]), + text: str = Form(...), + expire_value: int = Form(default=1, gt=0), + expire_style: str = Form(default="day"), + ip: str = Depends(ip_limit["upload"]), ): text_size = len(text.encode("utf-8")) max_txt_size = 222 * 1024 @@ -114,15 +129,17 @@ async def share_text( @share_api.post("/file/", dependencies=[Depends(share_required_login)]) async def share_file( - expire_value: int = Form(default=1, gt=0), - expire_style: str = Form(default="day"), - file: UploadFile = File(...), - ip: str = Depends(ip_limit["upload"]), + expire_value: int = Form(default=1, gt=0), + expire_style: str = Form(default="day"), + file: UploadFile = File(...), + ip: str = Depends(ip_limit["upload"]), ): file_size = await validate_file_size(file, settings.uploadSize) if expire_style not in settings.expireStyle: raise HTTPException(status_code=400, detail="过期时间类型错误") - expired_at, expired_count, used_count, code = await get_expire_info(expire_value, expire_style) + expired_at, expired_count, used_count, code = await get_expire_info( + expire_value, expire_style + ) path, suffix, prefix, uuid_file_name, save_path = await get_file_path_name(file) file_storage: FileStorageInterface = storages[settings.file_storage]() await file_storage.save_file(file, save_path) @@ -141,7 +158,9 @@ async def share_file( return APIResponse(detail={"code": code, "name": file.filename}) -async def get_code_file_by_code(code, check=True): +async def get_code_file_by_code( + code: str, check: bool = True +) -> Tuple[bool, Union[FileCodes, str]]: file_code = await FileCodes.filter(code=code).first() if not file_code: return False, "文件不存在" @@ -150,7 +169,7 @@ async def get_code_file_by_code(code, check=True): return True, file_code -async def update_file_usage(file_code): +async def update_file_usage(file_code: FileCodes) -> None: file_code.used_count += 1 if file_code.expired_count > 0: file_code.expired_count -= 1 @@ -165,6 +184,7 @@ async def get_code_file(code: str, ip: str = Depends(ip_limit["error"])): ip_limit["error"].add_ip(ip) return APIResponse(code=404, detail=file_code) + assert isinstance(file_code, FileCodes) await update_file_usage(file_code) return await file_storage.get_file_response(file_code) @@ -177,6 +197,7 @@ async def select_file(data: SelectFileModel, ip: str = Depends(ip_limit["error"] ip_limit["error"].add_ip(ip) return APIResponse(code=404, detail=file_code) + assert isinstance(file_code, FileCodes) await update_file_usage(file_code) return APIResponse( detail={ @@ -201,6 +222,7 @@ async def download_file(key: str, code: str, ip: str = Depends(ip_limit["error"] has, file_code = await get_code_file_by_code(code, False) if not has: return APIResponse(code=404, detail="文件不存在") + assert isinstance(file_code, FileCodes) return ( APIResponse(detail=file_code.text) if file_code.text @@ -219,8 +241,7 @@ async def init_chunk_upload(data: InitChunkUploadModel): if max_possible_size > settings.uploadSize: max_size_mb = settings.uploadSize / (1024 * 1024) raise HTTPException( - status_code=403, - detail=f"文件大小超过限制，最大为 {max_size_mb:.2f} MB" + status_code=403, detail=f"文件大小超过限制，最大为 {max_size_mb:.2f} MB" ) # # 秒传检查 @@ -247,21 +268,25 @@ async def init_chunk_upload(data: InitChunkUploadModel): ).first() if existing_session: - # 复用已有会话，获取已上传的分片列表 - uploaded_chunks = await UploadChunk.filter( - upload_id=existing_session.upload_id, - completed=True - ).values_list('chunk_index', flat=True) - return APIResponse(detail={ - "existed": False, - "upload_id": existing_session.upload_id, - "chunk_size": existing_session.chunk_size, - "total_chunks": existing_session.total_chunks, - "uploaded_chunks": list(uploaded_chunks) - }) + if not existing_session.save_path: + await UploadChunk.filter(upload_id=existing_session.upload_id).delete() + else: + uploaded_chunks = await UploadChunk.filter( + upload_id=existing_session.upload_id, completed=True + ).values_list("chunk_index", flat=True) + return APIResponse( + detail={ + "existed": False, + "upload_id": existing_session.upload_id, + "chunk_size": existing_session.chunk_size, + "total_chunks": existing_session.total_chunks, + "uploaded_chunks": list(uploaded_chunks), + } + ) # 创建新的上传会话 upload_id = uuid.uuid4().hex + _, _, _, _, save_path = await get_chunk_file_path_name(data.file_name, upload_id) await UploadChunk.create( upload_id=upload_id, chunk_index=-1, @@ -270,21 +295,27 @@ async def init_chunk_upload(data: InitChunkUploadModel): chunk_size=data.chunk_size, chunk_hash=data.file_hash, file_name=data.file_name, + save_path=save_path, + ) + return APIResponse( + detail={ + "existed": False, + "upload_id": upload_id, + "chunk_size": data.chunk_size, + "total_chunks": total_chunks, + "uploaded_chunks": [], + } ) - return APIResponse(detail={ - "existed": False, - "upload_id": upload_id, - "chunk_size": data.chunk_size, - "total_chunks": total_chunks, - "uploaded_chunks": [] - }) -@chunk_api.post("/upload/chunk/{upload_id}/{chunk_index}", dependencies=[Depends(share_required_login)]) +@chunk_api.post( + "/upload/chunk/{upload_id}/{chunk_index}", + dependencies=[Depends(share_required_login)], +) async def upload_chunk( - upload_id: str, - chunk_index: int, - chunk: UploadFile = File(...), + upload_id: str, + chunk_index: int, + chunk: UploadFile = File(...), ): # 获取上传会话信息 chunk_info = await UploadChunk.filter(upload_id=upload_id, chunk_index=-1).first() @@ -297,12 +328,12 @@ async def upload_chunk( # 检查是否已上传（支持断点续传） existing_chunk = await UploadChunk.filter( - upload_id=upload_id, - chunk_index=chunk_index, - completed=True + upload_id=upload_id, chunk_index=chunk_index, completed=True ).first() if existing_chunk: - return APIResponse(detail={"chunk_hash": existing_chunk.chunk_hash, "skipped": True}) + return APIResponse( + detail={"chunk_hash": existing_chunk.chunk_hash, "skipped": True} + ) # 读取分片数据并计算哈希 chunk_data = await chunk.read() @@ -312,47 +343,49 @@ async def upload_chunk( if chunk_size > chunk_info.chunk_size: raise HTTPException( status.HTTP_400_BAD_REQUEST, - detail=f"分片大小超过声明值: 最大 {chunk_info.chunk_size}, 实际 {chunk_size}" + detail=f"分片大小超过声明值: 最大 {chunk_info.chunk_size}, 实际 {chunk_size}", ) # 计算已上传分片数，校验累计大小不超限（用分片数 * chunk_size 估算） uploaded_count = await UploadChunk.filter( - upload_id=upload_id, - completed=True + upload_id=upload_id, completed=True ).count() # 已上传分片的最大可能大小 + 当前分片 max_uploaded_size = uploaded_count * chunk_info.chunk_size + chunk_size if max_uploaded_size > settings.uploadSize: max_size_mb = settings.uploadSize / (1024 * 1024) raise HTTPException( - status_code=403, - detail=f"累计上传大小超过限制，最大为 {max_size_mb:.2f} MB" + status_code=403, detail=f"累计上传大小超过限制，最大为 {max_size_mb:.2f} MB" ) - + chunk_hash = hashlib.sha256(chunk_data).hexdigest() - # 获取文件路径 - _, _, _, _, save_path = await get_chunk_file_path_name(chunk_info.file_name, upload_id) - + save_path = chunk_info.save_path + # 保存分片到存储 storage = storages[settings.file_storage]() try: - await storage.save_chunk(upload_id, chunk_index, chunk_data, chunk_hash, save_path) + await storage.save_chunk( + upload_id, chunk_index, chunk_data, chunk_hash, save_path + ) except Exception as e: - raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"分片保存失败: {str(e)}") - + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"分片保存失败: {str(e)}" + ) + # 更新或创建分片记录（保存成功后再记录） await UploadChunk.update_or_create( upload_id=upload_id, chunk_index=chunk_index, defaults={ - 'chunk_hash': chunk_hash, - 'completed': True, - 'file_size': chunk_info.file_size, - 'total_chunks': chunk_info.total_chunks, - 'chunk_size': chunk_info.chunk_size, - 'file_name': chunk_info.file_name - } + "chunk_hash": chunk_hash, + "completed": True, + "file_size": chunk_info.file_size, + "total_chunks": chunk_info.total_chunks, + "chunk_size": chunk_info.chunk_size, + "file_name": chunk_info.file_name, + "save_path": chunk_info.save_path, + }, ) return APIResponse(detail={"chunk_hash": chunk_hash}) @@ -363,50 +396,56 @@ async def cancel_upload(upload_id: str): chunk_info = await UploadChunk.filter(upload_id=upload_id, chunk_index=-1).first() if not chunk_info: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="上传会话不存在") - - # 获取文件路径 - _, _, _, _, save_path = await get_chunk_file_path_name(chunk_info.file_name, upload_id) - + + save_path = chunk_info.save_path + # 清理存储中的临时文件 storage = storages[settings.file_storage]() - try: - await storage.clean_chunks(upload_id, save_path) - except Exception as e: - # 记录错误但不阻止删除数据库记录 - pass - + if save_path: + try: + await storage.clean_chunks(upload_id, save_path) + except Exception as e: + pass + # 清理数据库记录 await UploadChunk.filter(upload_id=upload_id).delete() - + return APIResponse(detail={"message": "上传已取消"}) -@chunk_api.get("/upload/status/{upload_id}", dependencies=[Depends(share_required_login)]) +@chunk_api.get( + "/upload/status/{upload_id}", dependencies=[Depends(share_required_login)] +) async def get_upload_status(upload_id: str): """获取上传状态""" chunk_info = await UploadChunk.filter(upload_id=upload_id, chunk_index=-1).first() if not chunk_info: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="上传会话不存在") - + # 获取已上传的分片列表 uploaded_chunks = await UploadChunk.filter( - upload_id=upload_id, - completed=True - ).values_list('chunk_index', flat=True) - - return APIResponse(detail={ - "upload_id": upload_id, - "file_name": chunk_info.file_name, - "file_size": chunk_info.file_size, - "chunk_size": chunk_info.chunk_size, - "total_chunks": chunk_info.total_chunks, - "uploaded_chunks": list(uploaded_chunks), - "progress": len(uploaded_chunks) / chunk_info.total_chunks * 100 - }) + upload_id=upload_id, completed=True + ).values_list("chunk_index", flat=True) + + return APIResponse( + detail={ + "upload_id": upload_id, + "file_name": chunk_info.file_name, + "file_size": chunk_info.file_size, + "chunk_size": chunk_info.chunk_size, + "total_chunks": chunk_info.total_chunks, + "uploaded_chunks": list(uploaded_chunks), + "progress": len(uploaded_chunks) / chunk_info.total_chunks * 100, + } + ) -@chunk_api.post("/upload/complete/{upload_id}", dependencies=[Depends(share_required_login)]) -async def complete_upload(upload_id: str, data: CompleteUploadModel, ip: str = Depends(ip_limit["upload"])): +@chunk_api.post( + "/upload/complete/{upload_id}", dependencies=[Depends(share_required_login)] +) +async def complete_upload( + upload_id: str, data: CompleteUploadModel, ip: str = Depends(ip_limit["upload"]) +): # 获取上传基本信息 chunk_info = await UploadChunk.filter(upload_id=upload_id, chunk_index=-1).first() if not chunk_info: @@ -415,8 +454,7 @@ async def complete_upload(upload_id: str, data: CompleteUploadModel, ip: str = D storage = storages[settings.file_storage]() # 验证所有分片 completed_chunks_list = await UploadChunk.filter( - upload_id=upload_id, - completed=True + upload_id=upload_id, completed=True ).all() if len(completed_chunks_list) != chunk_info.total_chunks: raise HTTPException(status.HTTP_400_BAD_REQUEST, detail="分片不完整") @@ -424,27 +462,29 @@ async def complete_upload(upload_id: str, data: CompleteUploadModel, ip: str = D # 用分片数 * chunk_size 校验最大可能大小 max_total_size = len(completed_chunks_list) * chunk_info.chunk_size if max_total_size > settings.uploadSize: - # 清理已上传的分片 - _, _, _, _, save_path = await get_chunk_file_path_name(chunk_info.file_name, upload_id) - try: - await storage.clean_chunks(upload_id, save_path) - except Exception: - pass + save_path = chunk_info.save_path + if save_path: + try: + await storage.clean_chunks(upload_id, save_path) + except Exception: + pass await UploadChunk.filter(upload_id=upload_id).delete() max_size_mb = settings.uploadSize / (1024 * 1024) raise HTTPException( - status_code=403, - detail=f"实际上传大小超过限制，最大为 {max_size_mb:.2f} MB" + status_code=403, detail=f"实际上传大小超过限制，最大为 {max_size_mb:.2f} MB" ) - # 获取文件路径 - path, suffix, prefix, _, save_path = await get_chunk_file_path_name(chunk_info.file_name, upload_id) - + save_path = chunk_info.save_path + path = os.path.dirname(save_path) if save_path else "" + prefix, suffix = os.path.splitext(chunk_info.file_name) + try: # 合并文件并计算哈希 _, file_hash = await storage.merge_chunks(upload_id, chunk_info, save_path) # 创建文件记录 - expired_at, expired_count, used_count, code = await get_expire_info(data.expire_value, data.expire_style) + expired_at, expired_count, used_count, code = await get_expire_info( + data.expire_value, data.expire_style + ) await FileCodes.create( code=code, file_hash=file_hash, # 使用合并后计算的哈希 @@ -457,7 +497,7 @@ async def complete_upload(upload_id: str, data: CompleteUploadModel, ip: str = D file_path=path, uuid_file_name=f"{prefix}{suffix}", prefix=prefix, - suffix=suffix + suffix=suffix, ) # 清理临时文件 await storage.clean_chunks(upload_id, save_path) @@ -473,7 +513,9 @@ async def complete_upload(upload_id: str, data: CompleteUploadModel, ip: str = D await storage.clean_chunks(upload_id, save_path) except Exception: pass - raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"文件合并失败: {str(e)}") + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"文件合并失败: {str(e)}" + ) # ============ 预签名上传API ============ @@ -482,7 +524,9 @@ presign_api = APIRouter(prefix="/presign", tags=["预签名上传"]) PRESIGN_SESSION_EXPIRES = 900 # 15分钟 -async def _get_valid_session(upload_id: str, expected_mode: str = None) -> PresignUploadSession: +async def _get_valid_session( + upload_id: str, expected_mode: Optional[str] = None +) -> PresignUploadSession: """获取并验证会话""" session = await PresignUploadSession.filter(upload_id=upload_id).first() if not session: @@ -496,18 +540,27 @@ async def _get_valid_session(upload_id: str, expected_mode: str = None) -> Presi @presign_api.post("/upload/init", dependencies=[Depends(share_required_login)]) -async def presign_upload_init(data: PresignUploadInitRequest, ip: str = Depends(ip_limit["upload"])): +async def presign_upload_init( + data: PresignUploadInitRequest, ip: str = Depends(ip_limit["upload"]) +): """初始化预签名上传，S3返回直传URL，其他存储返回代理URL""" if data.file_size > settings.uploadSize: - raise HTTPException(403, f"文件大小超过限制，最大为 {settings.uploadSize / (1024*1024):.2f} MB") + raise HTTPException( + 403, + f"文件大小超过限制，最大为 {settings.uploadSize / (1024 * 1024):.2f} MB", + ) if data.expire_style not in settings.expireStyle: raise HTTPException(400, "过期时间类型错误") upload_id = uuid.uuid4().hex - path, _, _, filename, save_path = await FileUploadService.generate_file_path(data.file_name, upload_id) + path, _, _, filename, save_path = await FileUploadService.generate_file_path( + data.file_name, upload_id + ) storage: FileStorageInterface = storages[settings.file_storage]() - presigned_url = await storage.generate_presigned_upload_url(save_path, PRESIGN_SESSION_EXPIRES) + presigned_url = await storage.generate_presigned_upload_url( + save_path, PRESIGN_SESSION_EXPIRES + ) mode = "direct" if presigned_url else "proxy" upload_url = presigned_url or f"/api/presign/upload/proxy/{upload_id}" @@ -524,16 +577,22 @@ async def presign_upload_init(data: PresignUploadInitRequest, ip: str = Depends( ) ip_limit["upload"].add_ip(ip) - return APIResponse(detail={ - "upload_id": upload_id, - "upload_url": upload_url, - "mode": mode, - "expires_in": PRESIGN_SESSION_EXPIRES, - }) + return APIResponse( + detail={ + "upload_id": upload_id, + "upload_url": upload_url, + "mode": mode, + "expires_in": PRESIGN_SESSION_EXPIRES, + } + ) -@presign_api.put("/upload/proxy/{upload_id}", dependencies=[Depends(share_required_login)]) -async def presign_upload_proxy(upload_id: str, file: UploadFile = File(...), ip: str = Depends(ip_limit["upload"])): +@presign_api.put( + "/upload/proxy/{upload_id}", dependencies=[Depends(share_required_login)] +) +async def presign_upload_proxy( + upload_id: str, file: UploadFile = File(...), ip: str = Depends(ip_limit["upload"]) +): """代理模式上传，服务器转存到存储后端""" session = await _get_valid_session(upload_id, expected_mode="proxy") @@ -548,8 +607,11 @@ async def presign_upload_proxy(upload_id: str, file: UploadFile = File(...), ip: raise HTTPException(500, f"文件保存失败: {str(e)}") code = await FileUploadService.create_file_record( - session.file_name, file_size, os.path.dirname(session.save_path), - session.expire_value, session.expire_style + session.file_name, + file_size, + os.path.dirname(session.save_path), + session.expire_value, + session.expire_style, ) await session.delete() @@ -557,7 +619,9 @@ async def presign_upload_proxy(upload_id: str, file: UploadFile = File(...), ip: return APIResponse(detail={"code": code, "name": session.file_name}) -@presign_api.post("/upload/confirm/{upload_id}", dependencies=[Depends(share_required_login)]) +@presign_api.post( + "/upload/confirm/{upload_id}", dependencies=[Depends(share_required_login)] +) async def presign_upload_confirm(upload_id: str, ip: str = Depends(ip_limit["upload"])): """直传确认，客户端完成S3直传后调用获取分享码""" session = await _get_valid_session(upload_id, expected_mode="direct") @@ -567,8 +631,11 @@ async def presign_upload_confirm(upload_id: str, ip: str = Depends(ip_limit["upl raise HTTPException(404, "文件未上传或上传失败") code = await FileUploadService.create_file_record( - session.file_name, session.file_size, os.path.dirname(session.save_path), - session.expire_value, session.expire_style + session.file_name, + session.file_size, + os.path.dirname(session.save_path), + session.expire_value, + session.expire_style, ) await session.delete() @@ -576,22 +643,26 @@ async def presign_upload_confirm(upload_id: str, ip: str = Depends(ip_limit["upl return APIResponse(detail={"code": code, "name": session.file_name}) -@presign_api.get("/upload/status/{upload_id}", dependencies=[Depends(share_required_login)]) +@presign_api.get( + "/upload/status/{upload_id}", dependencies=[Depends(share_required_login)] +) async def presign_upload_status(upload_id: str): """查询上传会话状态""" session = await PresignUploadSession.filter(upload_id=upload_id).first() if not session: raise HTTPException(404, "上传会话不存在") - return APIResponse(detail={ - "upload_id": session.upload_id, - "file_name": session.file_name, - "file_size": session.file_size, - "mode": session.mode, - "created_at": session.created_at.isoformat(), - "expires_at": session.expires_at.isoformat(), - "is_expired": await session.is_expired(), - }) + return APIResponse( + detail={ + "upload_id": session.upload_id, + "file_name": session.file_name, + "file_size": session.file_size, + "mode": session.mode, + "created_at": session.created_at.isoformat(), + "expires_at": session.expires_at.isoformat(), + "is_expired": await session.is_expired(), + } + ) @presign_api.delete("/upload/{upload_id}", dependencies=[Depends(share_required_login)]) diff --git a/core/tasks.py b/core/tasks.py index b6dd806..1064b00 100644 --- a/core/tasks.py +++ b/core/tasks.py @@ -48,36 +48,38 @@ async def delete_expire_files(): async def clean_incomplete_uploads(): """清理超时未完成的分片上传""" file_storage: FileStorageInterface = storages[settings.file_storage]() - # 默认 24 小时未完成的上传视为过期 - expire_hours = getattr(settings, 'chunk_expire_hours', 24) + expire_hours = getattr(settings, "chunk_expire_hours", 24) while True: try: - expire_time = datetime.datetime.now() - datetime.timedelta(hours=expire_hours) - # 查找所有过期的上传会话（chunk_index=-1 的记录） + expire_time = datetime.datetime.now() - datetime.timedelta( + hours=expire_hours + ) expired_sessions = await UploadChunk.filter( - chunk_index=-1, - created_at__lt=expire_time + chunk_index=-1, created_at__lt=expire_time ).all() for session in expired_sessions: try: - # 获取分片存储路径 - _, _, _, _, save_path = await get_chunk_file_path_name( - session.file_name, session.upload_id - ) - # 清理存储中的临时文件 + save_path = session.save_path + if not save_path: + _, _, _, _, save_path = await get_chunk_file_path_name( + session.file_name, session.upload_id + ) await file_storage.clean_chunks(session.upload_id, save_path) except Exception as e: - logging.error(f"清理分片文件失败 upload_id={session.upload_id}: {e}") + logging.error( + f"清理分片文件失败 upload_id={session.upload_id}: {e}" + ) try: - # 删除该会话的所有数据库记录 await UploadChunk.filter(upload_id=session.upload_id).delete() logging.info(f"已清理过期上传会话 upload_id={session.upload_id}") except Exception as e: - logging.error(f"删除分片记录失败 upload_id={session.upload_id}: {e}") + logging.error( + f"删除分片记录失败 upload_id={session.upload_id}: {e}" + ) except Exception as e: logging.error(f"清理未完成上传任务异常: {e}") finally: - await asyncio.sleep(3600) # 每小时执行一次 + await asyncio.sleep(3600)