413 Request Entity Too Large错误如何用Python实现多文件分块上传

错误原因说明 #

• 413错误是服务器单请求的实体大小限制导致，你原代码单请求传两个2MB文件总大小超过限制，400KB文件总大小没超过所以可以正常运行
• 415错误是你将文件对象放入data参数导致的，requests只有传入files参数时才会自动构建multipart/form-data格式的请求体，手动设置Content-Type: application/pdf完全不符合请求体的实际格式，所以服务器无法识别

方案1：使用分块传输编码绕过限制（无需修改服务器接口） #

你之前使用requests_toolbelt的代码已经接近可用，只需增加分块传输编码的请求头，无需将整个请求加载到内存，同时可以绕过部分服务器的单请求大小限制：

from requests_toolbelt.multipart.encoder import MultipartEncoder
import requests

m = MultipartEncoder(
    fields={
        'user': user,
        'password': password,
        'configname': configname,
        'jobid': jobid,
        'reference': (os.path.basename(reference), open(reference, 'rb'), 'application/pdf'),
        'candidate': (os.path.basename(candidate), open(candidate, 'rb'), 'application/pdf')
    }
)

headers = {
    'Content-Type': m.content_type,
    'Transfer-Encoding': 'chunked'
}

r = requests.post(url, data=m, headers=headers, verify=False)

方案2：多文件分片上传（需要服务器支持分片接口） #

如果方案1无效，说明服务器严格限制单请求大小，需要拆分文件为多个小块分请求上传，实现逻辑如下：

步骤1：封装文件拆分工具函数 #

def split_file(file_path, chunk_size=300*1024):
    """将文件拆分为指定大小的块，默认单块300KB，小于你测试可用的400KB限制"""
    chunks = []
    with open(file_path, 'rb') as f:
        while chunk := f.read(chunk_size):
            chunks.append(chunk)
    return chunks, len(chunks)

步骤2：完整分片上传流程 #

import os
import requests

CHUNK_SIZE = 300 * 1024

# 拆分两个文件为块
ref_chunks, ref_total = split_file(reference, CHUNK_SIZE)
can_chunks, can_total = split_file(candidate, CHUNK_SIZE)

# 1. 预上传：通知服务器要上传的文件信息，获取上传任务标识
pre_data = {
    "user": user,
    "password": password,
    "configname": configname,
    "jobid": jobid,
    "ref_name": os.path.basename(reference),
    "ref_total_chunks": ref_total,
    "ref_size": os.path.getsize(reference),
    "can_name": os.path.basename(candidate),
    "can_total_chunks": can_total,
    "can_size": os.path.getsize(candidate)
}
pre_resp = requests.post(f"{url}/pre_upload", data=pre_data, verify=False)
task_id = pre_resp.json()["task_id"]

# 2. 上传reference所有分片
for idx, chunk in enumerate(ref_chunks):
    chunk_data = {
        "task_id": task_id,
        "file_type": "reference",
        "chunk_idx": idx,
        "total_chunks": ref_total
    }
    files = {"chunk": (f"ref_{idx}", chunk, "application/pdf")}
    requests.post(f"{url}/upload_chunk", data=chunk_data, files=files, verify=False)

# 3. 上传candidate所有分片
for idx, chunk in enumerate(can_chunks):
    chunk_data = {
        "task_id": task_id,
        "file_type": "candidate",
        "chunk_idx": idx,
        "total_chunks": can_total
    }
    files = {"chunk": (f"can_{idx}", chunk, "application/pdf")}
    requests.post(f"{url}/upload_chunk", data=chunk_data, files=files, verify=False)

# 4. 通知服务器合并所有分片
merge_resp = requests.post(f"{url}/merge", data={"task_id": task_id}, verify=False)

注：以上接口路径/pre_upload、/upload_chunk、/merge需要和服务器提供的分片接口路径对应。

方案3：压缩PDF后上传（无需修改接口） #

如果服务器不支持分片，可以先压缩PDF文件大小到限制范围内再上传：

from pypdf import PdfReader, PdfWriter

def compress_pdf(input_path, output_path):
    writer = PdfWriter()
    reader = PdfReader(input_path)
    for page in reader.pages:
        page.compress_content_streams()
        writer.add_page(page)
    with open(output_path, "wb") as f:
        writer.write(f)

# 压缩两个文件
compressed_ref = "./compressed_ref.pdf"
compressed_can = "./compressed_can.pdf"
compress_pdf(reference, compressed_ref)
compress_pdf(candidate, compressed_can)

# 用原逻辑上传压缩后的文件
multiple_files = [
    ('reference', (os.path.basename(reference), open(compressed_ref, 'rb'), 'application/pdf')),
    ('candidate', (os.path.basename(candidate), open(compressed_can, 'rb'), 'application/pdf'))
]
payload = {'user': user, 'password': password, 'configname': configname, 'jobid': jobid}
r = requests.post(url, files=multiple_files, data=payload, verify=False)

内容的提问来源于stack exchange，提问作者JoshHutch