使用Python拆分超过1MB的Zip文件：现有代码拆分功能失效问题求助

你的拆分功能失效的核心原因是 SplitFileWriter 的设计目标是将数据拆分为多个磁盘文件（比如 output.001、output.002 这样的分片），而不是直接在内存中生成多个字节块。你把单个 BytesIO 对象传给它，它只会把所有压缩后的数据写入到这个单一的内存缓冲区里，自然不会产生拆分效果。另外，你用 zipfile.ZipFile 的 "a" 模式也不合适——因为你是从头创建Zip文件，应该用 "w" 模式，而且 ZipFile 会在上下文管理器退出时才将所有数据写入目标对象，这时候 SplitFileWriter 已经关闭，可能导致数据写入不完整。

下面给出两种针对性的解决方案：

方案1：内存中直接拆分Zip数据（推荐，无磁盘IO） #

先完整生成Zip文件到内存缓冲区，再手动将其拆分为指定大小的字节块，全程在内存中完成：

from io import BytesIO
import zipfile
import xml.etree.ElementTree as etree

# 生成XML数据（你的原有逻辑）
tree = etree.ElementTree(batch_element)
xml_object = BytesIO()
tree.write(xml_object, pretty_print=True, xml_declaration=False, encoding="utf-8")
xml_file = xml_object.getvalue()

# 第一步：生成完整的Zip文件到内存
zip_buffer = BytesIO()
# 启用ZIP_DEFLATED压缩，可选但建议开启以减小体积
with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zip_file:
    zip_file.writestr('Batch.xml', xml_file)
zip_buffer.seek(0)  # 将缓冲区指针移到开头
zip_data = zip_buffer.getvalue()

# 第二步：拆分Zip数据为1MB的字节块
CHUNK_SIZE = 1 * 1024 * 1024  # 1MB
split_chunks = []
remaining_data = zip_data

while len(remaining_data) > 0:
    chunk = remaining_data[:CHUNK_SIZE]
    split_chunks.append(chunk)
    remaining_data = remaining_data[CHUNK_SIZE:]

# 验证分片结果
for idx, chunk in enumerate(split_chunks):
    print(f"分片 {idx+1} 大小：{len(chunk)} 字节")

方案2：用SplitFileWriter生成磁盘分片再读取为字节 #

如果你一定要用SplitFileWriter生成标准的分片文件格式，可以先让它输出到磁盘，再读取这些文件的字节内容：

from split_file_reader.split_file_writer import SplitFileWriter
import zipfile
from io import BytesIO
import xml.etree.ElementTree as etree
import os

# 生成XML数据（你的原有逻辑）
tree = etree.ElementTree(batch_element)
xml_object = BytesIO()
tree.write(xml_object, pretty_print=True, xml_declaration=False, encoding="utf-8")
xml_file = xml_object.getvalue()

# 第一步：用SplitFileWriter生成磁盘分片
output_prefix = "batch_zip"
CHUNK_SIZE = 1 * 1024 * 1024

with SplitFileWriter(output_prefix, CHUNK_SIZE) as sfw:
    with zipfile.ZipFile(sfw, "w", zipfile.ZIP_DEFLATED) as zip_file:
        zip_file.writestr('Batch.xml', xml_file)

# 第二步：读取所有分片文件的字节内容
split_chunks = []
idx = 1
while True:
    chunk_filename = f"{output_prefix}.{idx:03d}"
    if not os.path.exists(chunk_filename):
        break
    with open(chunk_filename, "rb") as f:
        split_chunks.append(f.read())
    # 可选：读取后删除临时分片文件
    os.remove(chunk_filename)
    idx += 1

注意事项 #

• 方案1更适合内存操作，避免磁盘IO，效率更高；
• 方案2适合需要生成标准分片文件（供其他工具读取）的场景；
• 创建ZipFile时指定ZIP_DEFLATED可以启用压缩，大幅减小Zip文件体积，建议开启。

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