影刀RPA 压缩与解压：ZIP文件的处理
作者：林焱
什么情况用
你的影刀流程每天采集大量数据文件，磁盘空间不够用了，需要把历史数据压缩归档？你从网站下载了一个ZIP文件，需要自动解压后处理里面的数据？你想把多个采集结果文件打包成一个ZIP发送邮件？
ZIP文件处理是文件管理的常见需求。Python的zipfile模块提供了完整的ZIP压缩和解压功能，在影刀RPA的Python代码块中可以直接使用。
本文讲清楚：怎么创建ZIP、怎么解压ZIP、怎么处理中文文件名乱码。
怎么做
一、创建ZIP文件
importzipfileimportos# ===== 压缩单个文件 =====defzip_single_file(filepath,zip_path):"""把单个文件压缩成zip"""withzipfile.ZipFile(zip_path,'w',zipfile.ZIP_DEFLATED)aszf:zf.write(filepath,os.path.basename(filepath))print(f"已压缩{filepath}→{zip_path}")# ===== 压缩多个文件 =====defzip_multiple_files(filepaths,zip_path):"""把多个文件压缩成一个zip"""withzipfile.ZipFile(zip_path,'w',zipfile.ZIP_DEFLATED)aszf:forfilepathinfilepaths:ifos.path.exists(filepath):zf.write(filepath,os.path.basename(filepath))print(f" 添加:{filepath}")else:print(f" 跳过（不存在）:{filepath}")print(f"已压缩{len(filepaths)}个文件 →{zip_path}")# ===== 压缩整个文件夹 =====defzip_directory(dir_path,zip_path):"""把整个文件夹压缩成zip"""withzipfile.ZipFile(zip_path,'w',zipfile.ZIP_DEFLATED)aszf:forroot,dirs,filesinos.walk(dir_path):forfileinfiles:filepath=os.path.join(root,file)# 保持相对路径结构arcname=os.path.relpath(filepath,dir_path)zf.write(filepath,arcname)print(f"已压缩文件夹{dir_path}→{zip_path}")# 使用zip_single_file("data/report.xlsx","data/report.zip")zip_multiple_files(["data/a.xlsx","data/b.xlsx","data/c.csv"],"data/multiple.zip")zip_directory("data/output","data/output_backup.zip")二、解压ZIP文件
importzipfileimportos# ===== 解压到指定目录 =====defunzip_file(zip_path,extract_dir):"""解压zip文件到指定目录"""withzipfile.ZipFile(zip_path,'r')aszf:zf.extractall(extract_dir)print(f"已解压{zip_path}→{extract_dir}")# ===== 解压并获取文件列表 =====defunzip_and_list(zip_path,extract_dir):"""解压并返回文件列表"""extracted_files=[]withzipfile.ZipFile(zip_path,'r')aszf:forinfoinzf.infolist():# 跳过目录ifinfo.is_dir():continue# 解压单个文件zf.extract(info,extract_dir)extracted_path=os.path.join(extract_dir,info.filename)extracted_files.append({"name":info.filename,"path":extracted_path,"size":info.file_size,"compressed_size":info.compress_size,})print(f"解压完成，共{len(extracted_files)}个文件:")forfinextracted_files:print(f"{f['name']}({f['size']}bytes)")returnextracted_files# ===== 只读取ZIP内的文件内容（不解压到磁盘） =====defread_zip_content(zip_path,filename):"""直接读取ZIP内文件内容，不解压"""withzipfile.ZipFile(zip_path,'r')aszf:withzf.open(filename)asf:content=f.read()returncontent# 读取ZIP内的文本文件text=read_zip_content("data.zip","readme.txt").decode("utf-8")print(text)# 读取ZIP内的CSV文件importioimportcsv csv_bytes=read_zip_content("data.zip","data.csv")csv_reader=csv.reader(io.TextIOWrapper(io.BytesIO(csv_bytes),encoding="utf-8"))forrowincsv_reader:print(row)三、查看ZIP内容（不解压）
deflist_zip_contents(zip_path):"""列出ZIP文件内容"""withzipfile.ZipFile(zip_path,'r')aszf:print(f"ZIP文件:{zip_path}")print(f"文件数量:{len(zf.namelist())}")print("-"*60)forinfoinzf.infolist():ifinfo.is_dir():print(f" [目录]{info.filename}")else:ratio=(1-info.compress_size/info.file_size)*100ifinfo.file_size>0else0print(f"{info.filename}")print(f" 原始:{info.file_size:,}bytes, 压缩后:{info.compress_size:,}bytes, 压缩率:{ratio:.1f}%")# 总计total_size=sum(f.file_sizeforfinzf.infolist()ifnotf.is_dir())total_compressed=sum(f.compress_sizeforfinzf.infolist()ifnotf.is_dir())print("-"*60)print(f" 总计:{total_size:,}bytes →{total_compressed:,}bytes")list_zip_contents("data/output.zip")四、处理中文文件名乱码
importzipfileimportosdefunzip_with_chinese_names(zip_path,extract_dir):""" 解压ZIP文件，正确处理中文文件名 解决中文文件名乱码问题 """withzipfile.ZipFile(zip_path,'r')aszf:forinfoinzf.infolist():# 检测文件名编码filename=info.filename# ZIP文件名编码问题：# Windows创建的ZIP用GBK编码文件名# 但Python的zipfile默认用UTF-8解码# 需要尝试不同编码try:# 先尝试UTF-8filename.encode('utf-8').decode('utf-8')except(UnicodeDecodeError,UnicodeEncodeError):try:# 尝试GBKfilename=filename.encode('cp437').decode('gbk')except:try:filename=filename.encode('cp437').decode('utf-8')except:pass# 保持原样ifinfo.is_dir():dir_path=os.path.join(extract_dir,filename)os.makedirs(dir_path,exist_ok=True)continue# 解压文件data=zf.read(info)output_path=os.path.join(extract_dir,filename)# 确保目录存在os.makedirs(os.path.dirname(output_path),exist_ok=True)withopen(output_path,'wb')asf:f.write(data)print(f"解压:{filename}")五、增量压缩与更新
defappend_to_zip(zip_path,files_to_add):"""向已有的ZIP文件中追加文件"""withzipfile.ZipFile(zip_path,'a',zipfile.ZIP_DEFLATED)aszf:forfilepathinfiles_to_add:ifos.path.exists(filepath):zf.write(filepath,os.path.basename(filepath))print(f"追加:{filepath}")else:print(f"跳过（不存在）:{filepath}")defupdate_zip(zip_path,file_to_update):""" 更新ZIP中的文件 zipfile不支持直接更新，需要重建 """importtempfile temp_zip=tempfile.mktemp(suffix=".zip")withzipfile.ZipFile(zip_path,'r')aszf_old:withzipfile.ZipFile(temp_zip,'w',zipfile.ZIP_DEFLATED)aszf_new:# 复制不需要更新的文件foriteminzf_old.infolist():ifitem.filename!=os.path.basename(file_to_update):zf_new.writestr(item,zf_old.read(item.filename))# 添加更新后的文件zf_new.write(file_to_update,os.path.basename(file_to_update))# 替换原文件os.replace(temp_zip,zip_path)print(f"已更新{zip_path}中的{file_to_update}")六、实战：自动归档旧文件
importzipfileimportosimporttimefromdatetimeimportdatetime,timedeltadefauto_archive_old_files(source_dir,archive_dir,days_old=30):""" 自动归档超过指定天数的文件 把旧文件按月份压缩归档 """ifnotos.path.exists(archive_dir):os.makedirs(archive_dir)cutoff_time=time.time()-days_old*86400archives={}# {月份: [文件列表]}# 找出需要归档的文件forfilenameinos.listdir(source_dir):filepath=os.path.join(source_dir,filename)ifnotos.path.isfile(filepath):continuefile_mtime=os.path.getmtime(filepath)iffile_mtime<cutoff_time:# 按月份分组file_date=datetime.fromtimestamp(file_mtime)month_key=file_date.strftime("%Y-%m")ifmonth_keynotinarchives:archives[month_key]=[]archives[month_key].append(filepath)# 按月份压缩归档formonth,filesinarchives.items():zip_path=os.path.join(archive_dir,f"archive_{month}.zip")# 如果归档文件已存在，追加ifos.path.exists(zip_path):append_to_zip(zip_path,files)else:zip_multiple_files(files,zip_path)# 归档后删除原文件forfilepathinfiles:os.remove(filepath)print(f"已归档并删除:{filepath}")print(f"归档完成:{len(archives)}个月份, 共{sum(len(f)forfinarchives.values())}个文件")# 使用：归档30天前的文件auto_archive_old_files("D:/data/output","D:/data/archive",days_old=30)七、带密码的ZIP文件
# Python标准库zipfile只支持解压带密码的ZIP，不支持创建# 解压带密码的ZIP：defunzip_with_password(zip_path,extract_dir,password):"""解压带密码的ZIP文件"""withzipfile.ZipFile(zip_path,'r')aszf:try:zf.extractall(extract_dir,pwd=password.encode())print("解压成功")returnTrueexceptRuntimeErrorase:if"password"instr(e).lower():print("密码错误")else:print(f"解压失败:{e}")returnFalse# 使用unzip_with_password("secret.zip","output","mypassword")# 如果需要创建带密码的ZIP，需要第三方库pyzipper# pip install pyzipper# import pyzipper# with pyzipper.AESZipFile('secret.zip', 'w', compression=pyzipper.ZIP_DEFLATED, encryption=pyzipper.WZ_AES) as zf:# zf.setpassword(b'mypassword')# zf.write('data.xlsx')有什么坑
坑1：ZIP_DEFLATED未安装报错
现象：zipfile.ZipFile(path, 'w', zipfile.ZIP_DEFLATED)报错RuntimeError: Compression requires the (missing) zlib module。
原因：Python安装时没有包含zlib库（某些精简版Python会出现）。
解决：用zipfile.ZIP_STORED代替ZIP_DEFLATED（只打包不压缩），或者安装完整版Python。影刀RPA自带的Python通常包含zlib，不会有此问题。
坑2：解压时路径穿越攻击
现象：ZIP文件中包含../../../etc/passwd这样的路径，解压时文件被写到ZIP目录外面。
解决：解压前检查每个文件的路径，拒绝包含..的路径：
defsafe_extract(zip_path,extract_dir):withzipfile.ZipFile(zip_path,'r')aszf:forinfoinzf.infolist():# 检查路径安全性target=os.path.join(extract_dir,info.filename)ifnotos.path.abspath(target).startswith(os.path.abspath(extract_dir)):print(f"安全警告：跳过可疑路径{info.filename}")continuezf.extract(info,extract_dir)坑3：大文件压缩内存占用高
现象：压缩一个2GB的文件，内存占用飙升。
原因：zf.write()默认会一次性读取整个文件到内存。
解决：对大文件使用zf.writestr()分块写入，或者直接使用zf.write()（它底层是流式写入，不会一次性读取全部）。如果还是内存高，检查是否在读取ZIP内容时用了zf.read()一次性读取。
坑4：影刀变量不能直接传文件路径列表
现象：在影刀中用【获取文件列表】获取了文件路径列表，传给Python代码块进行压缩，但路径格式不对。
解决：影刀的文件列表变量可能是字符串用分隔符拼接的，需要在Python代码块中解析。或者直接在Python代码块中用os.listdir()获取文件列表，不依赖影刀变量传递。
坑5：解压后文件被锁
现象：解压ZIP文件后，想删除解压的文件但提示"文件被占用"。
原因：解压后文件可能被其他程序（如Excel）打开，或者Python的文件句柄没关闭。
解决：确保所有文件操作都用了with语句或手动关闭。在删除前用time.sleep(1)等一下系统释放文件锁。在影刀中，确保没有其他指令正在使用这些文件。