大文件PDF翻译的流式上传方案：分片上传、断点续传与实时进度回调

2026/8/4 17:26:01

前言

在线PDF翻译服务通常限制单文件大小（如20MB），但对于开发者来说，集成翻译功能时还面临另一个挑战：如何在服务端实现大文件的可靠上传。本文以PDF翻译场景为例，深入探讨流式上传的三种核心技术方案：分片上传、断点续传和实时进度反馈。

这三种技术不仅适用于PDF翻译场景，任何需要上传大文件到云服务的应用都可以复用这些方案。

为什么需要流式上传？

传统的文件上传方式是"整块上传"——客户端一次性将整个文件发送到服务端。这种方式有以下问题：

内存压力大：服务端需要将整个文件加载到内存再处理
网络不稳定导致失败：上传到99%断网，需要从头开始
无法显示进度：用户不知道要等多久
超时风险高：大文件（如100页PDF，50MB+）可能在上传过程中超时

流式上传的核心思想是分而治之——将大文件切分为小块，逐块上传，最终在服务端拼接还原。

方案一：分片上传

原理

客户端将文件切分为固定大小的分片（如5MB/片），逐片上传到服务端。服务端接收所有分片后按顺序拼接。

客户端实现（Python）

importosimporthashlibimportrequestsfromtypingimportOptional,CallableclassChunkedUploader:"""分片上传器"""def__init__(self,upload_url:str,chunk_size:int=5*1024*1024,# 5MBmax_retries:int=3,):self.upload_url=upload_url self.chunk_size=chunk_size self.max_retries=max_retriesdefupload(self,file_path:str,progress_callback:Optional[Callable[[int,int],None]]=None,)->dict:"""分片上传文件 Args: file_path: 文件路径 progress_callback: 进度回调 (已上传字节, 总字节) Returns: API响应 """file_size=os.path.getsize(file_path)file_hash=self._compute_file_hash(file_path)total_chunks=(file_size+self.chunk_size-1)//self.chunk_size# Step 1: 初始化上传会话session_id=self._init_session(file_path,file_size,total_chunks,file_hash)# Step 2: 逐片上传withopen(file_path,'rb')asf:forchunk_idxinrange(total_chunks):offset=chunk_idx*self.chunk_size f.seek(offset)chunk_data=f.read(self.chunk_size)# 计算分片哈希（用于校验）chunk_hash=hashlib.md5(chunk_data).hexdigest()# 上传分片（带重试）self._upload_chunk(session_id,chunk_idx,chunk_data,chunk_hash)# 回调进度ifprogress_callback:uploaded=offset+len(chunk_data)progress_callback(uploaded,file_size)# Step 3: 通知服务端合并分片result=self._complete_session(session_id)returnresultdef_compute_file_hash(self,file_path:str,algorithm:str='md5')->str:"""计算文件哈希（用于去重和完整性校验）"""h=hashlib.new(algorithm)withopen(file_path,'rb')asf:whilechunk:=f.read(8192):h.update(chunk)returnh.hexdigest()def_init_session(self,file_path:str,file_size:int,total_chunks:int,file_hash:str)->str:"""初始化上传会话"""resp=requests.post(f"{self.upload_url}/session",json={"file_name":os.path.basename(file_path),"file_size":file_size,"total_chunks":total_chunks,"chunk_size":self.chunk_size,"file_hash":file_hash,},timeout=30,)resp.raise_for_status()returnresp.json()["session_id"]def_upload_chunk(self,session_id:str,chunk_idx:int,data:bytes,chunk_hash:str):"""上传单个分片（带指数退避重试）"""forattemptinrange(self.max_retries):try:resp=requests.put(f"{self.upload_url}/session/{session_id}/chunk/{chunk_idx}",data=data,headers={"Content-Type":"application/octet-stream","X-Chunk-Hash":chunk_hash,"X-Chunk-Index":str(chunk_idx),},timeout=60,)resp.raise_for_status()return# 上传成功exceptrequests.RequestExceptionase:ifattempt==self.max_retries-1:raisewait=2**attempt# 指数退避: 1s, 2s, 4simporttime time.sleep(wait)def_complete_session(self,session_id:str)->dict:"""通知服务端合并分片"""resp=requests.post(f"{self.upload_url}/session/{session_id}/complete",timeout=120,# 合并可能需要较长时间)resp.raise_for_status()returnresp.json()

服务端实现要点（伪代码）

# 服务端需要维护上传会话状态# 推荐使用Redis存储会话信息session_store={}# 生产环境使用Redis# POST /session - 创建上传会话defcreate_session(file_name,file_size,total_chunks,chunk_size,file_hash):session_id=generate_uuid()session_store[session_id]={"file_name":file_name,"file_size":file_size,"total_chunks":total_chunks,"chunk_size":chunk_size,"file_hash":file_hash,"received_chunks":set(),"chunk_dir":f"/tmp/uploads/{session_id}/",}os.makedirs(session_store[session_id]["chunk_dir"])return{"session_id":session_id}# PUT /session/{id}/chunk/{idx} - 接收分片defreceive_chunk(session_id,chunk_idx,chunk_data,chunk_hash):# 校验分片哈希ifhashlib.md5(chunk_data).hexdigest()!=chunk_hash:raiseHTTPException(400,"Chunk hash mismatch")# 保存分片chunk_path=f"{chunk_dir}/{chunk_idx:06d}"withopen(chunk_path,'wb')asf:f.write(chunk_data)session_store[session_id]["received_chunks"].add(chunk_idx)return{"received":len(session_store[session_id]["received_chunks"])}

方案二：断点续传

分片上传的一个直接扩展就是断点续传——当上传中断时，可以从上次中断的位置继续，而不是从头开始。

核心实现

classResumableUploader(ChunkedUploader):"""支持断点续传的上传器"""def_get_uploaded_chunks(self,session_id:str)->set:"""从服务端获取已上传的分片列表"""resp=requests.get(f"{self.upload_url}/session/{session_id}/status",timeout=10,)resp.raise_for_status()status=resp.json()returnset(status.get("received_chunks",[]))defupload(self,file_path:str,session_id:str=None,**kwargs):"""支持断点续传的上传方法"""file_size=os.path.getsize(file_path)total_chunks=(file_size+self.chunk_size-1)//self.chunk_sizeifsession_id:# 恢复已有会话uploaded_chunks=self._get_uploaded_chunks(session_id)print(f"恢复上传:{len(uploaded_chunks)}/{total_chunks}已完成")else:# 新建会话file_hash=self._compute_file_hash(file_path)session_id=self._init_session(file_path,file_size,total_chunks,file_hash)uploaded_chunks=set()# 只上传未完成的分片withopen(file_path,'rb')asf:forchunk_idxinrange(total_chunks):ifchunk_idxinuploaded_chunks:continue# 跳过已上传的分片offset=chunk_idx*self.chunk_size f.seek(offset)chunk_data=f.read(self.chunk_size)chunk_hash=hashlib.md5(chunk_data).hexdigest()self._upload_chunk(session_id,chunk_idx,chunk_data,chunk_hash)returnself._complete_session(session_id)

方案三：实时进度推送（WebSocket + Server-Sent Events）

上传过程中，客户端需要实时展示进度。两种主流方案：

SSE（Server-Sent Events）实现

适合"服务端→客户端"单向推送场景：

# 服务端（FastAPI）fromfastapiimportFastAPIfromfastapi.responsesimportStreamingResponseimportasyncioimportjson app=FastAPI()@app.get("/upload/session/{session_id}/progress")asyncdefupload_progress(session_id:str):"""通过SSE推送上传进度"""asyncdefevent_stream():whileTrue:# 从Redis获取当前进度progress=awaitget_session_progress(session_id)yieldf"data:{json.dumps(progress)}\n\n"ifprogress["status"]in("completed","failed"):breakawaitasyncio.sleep(1)# 每秒推送一次returnStreamingResponse(event_stream(),media_type="text/event-stream",headers={"Cache-Control":"no-cache","X-Accel-Buffering":"no",# 禁用Nginx缓冲})

// 客户端（JavaScript/React）functionuseUploadProgress(sessionId){const[progress,setProgress]=useState({percent:0,status:'uploading'});useEffect(()=>{consteventSource=newEventSource(`/upload/session/${sessionId}/progress`);eventSource.onmessage=(event)=>{constdata=JSON.parse(event.data);setProgress(data);if(data.status==='completed'){eventSource.close();}};eventSource.onerror=()=>{eventSource.close();setProgress(prev=>({...prev,status:'error'}));};return()=>eventSource.close();},[sessionId]);returnprogress;}

实际应用：PDF翻译上传流程整合

将以上方案整合到PDF翻译场景中：

classPDFTranslationUploader:"""PDF翻译专用上传器：分片上传 + 进度回调 + 翻译轮询"""def__init__(self,api_base_url:str):self.uploader=ResumableUploader(f"{api_base_url}/upload")self.api_base_url=api_base_urldeftranslate_pdf(self,pdf_path:str,source_lang:str="auto",target_lang:str="zh",on_upload_progress:Optional[Callable]=None,on_translation_progress:Optional[Callable]=None,)->dict:"""完整的PDF翻译流程"""# Phase 1: 流式上传print(f"📤 开始上传:{os.path.basename(pdf_path)}")upload_result=self.uploader.upload(pdf_path,progress_callback=on_upload_progress,)print(f"✅ 上传完成:{upload_result['file_id']}")# Phase 2: 提交翻译任务task_id=self._submit_translation(upload_result['file_id'],source_lang,target_lang,)print(f"🔄 翻译中... task_id={task_id}")# Phase 3: 轮询翻译进度result=self._poll_translation(task_id,on_translation_progress)print(f"✅ 翻译完成")returnresultdef_submit_translation(self,file_id:str,source_lang:str,target_lang:str)->str:resp=requests.post(f"{self.api_base_url}/translate",json={"file_id":file_id,"source_lang":source_lang,"target_lang":target_lang,},timeout=30,)resp.raise_for_status()returnresp.json()["task_id"]def_poll_translation(self,task_id:str,progress_callback=None,max_wait:int=600)->dict:"""轮询翻译状态，最多等待10分钟"""importtime start=time.time()whiletime.time()-start<max_wait:resp=requests.get(f"{self.api_base_url}/task/{task_id}",timeout=10,)data=resp.json()ifprogress_callback:progress_callback(data.get("progress",0))ifdata["status"]=="completed":returndataelifdata["status"]=="failed":raiseException(f"翻译失败:{data.get('error')}")time.sleep(2)# 每2秒查询一次raiseTimeoutError("翻译超时")