影刀RPA 压缩与解压ZIP文件的处理作者林焱什么情况用你的影刀流程每天采集大量数据文件磁盘空间不够用了需要把历史数据压缩归档你从网站下载了一个ZIP文件需要自动解压后处理里面的数据你想把多个采集结果文件打包成一个ZIP发送邮件ZIP文件处理是文件管理的常见需求。Python的zipfile模块提供了完整的ZIP压缩和解压功能在影刀RPA的Python代码块中可以直接使用。本文讲清楚怎么创建ZIP、怎么解压ZIP、怎么处理中文文件名乱码。怎么做一、创建ZIP文件importzipfileimportos# 压缩单个文件 defzip_single_file(filepath,zip_path):把单个文件压缩成zipwithzipfile.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):把多个文件压缩成一个zipwithzipfile.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):把整个文件夹压缩成zipwithzipfile.ZipFile(zip_path,w,zipfile.ZIP_DEFLATED)aszf:forroot,dirs,filesinos.walk(dir_path):forfileinfiles:filepathos.path.join(root,file)# 保持相对路径结构arcnameos.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_pathos.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:contentf.read()returncontent# 读取ZIP内的文本文件textread_zip_content(data.zip,readme.txt).decode(utf-8)print(text)# 读取ZIP内的CSV文件importioimportcsv csv_bytesread_zip_content(data.zip,data.csv)csv_readercsv.reader(io.TextIOWrapper(io.BytesIO(csv_bytes),encodingutf-8))forrowincsv_reader:print(row)三、查看ZIP内容不解压deflist_zip_contents(zip_path):列出ZIP文件内容withzipfile.ZipFile(zip_path,r)aszf:print(fZIP文件:{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_size0else0print(f{info.filename})print(f 原始:{info.file_size:,}bytes, 压缩后:{info.compress_size:,}bytes, 压缩率:{ratio:.1f}%)# 总计total_sizesum(f.file_sizeforfinzf.infolist()ifnotf.is_dir())total_compressedsum(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():# 检测文件名编码filenameinfo.filename# ZIP文件名编码问题# Windows创建的ZIP用GBK编码文件名# 但Python的zipfile默认用UTF-8解码# 需要尝试不同编码try:# 先尝试UTF-8filename.encode(utf-8).decode(utf-8)except(UnicodeDecodeError,UnicodeEncodeError):try:# 尝试GBKfilenamefilename.encode(cp437).decode(gbk)except:try:filenamefilename.encode(cp437).decode(utf-8)except:pass# 保持原样ifinfo.is_dir():dir_pathos.path.join(extract_dir,filename)os.makedirs(dir_path,exist_okTrue)continue# 解压文件datazf.read(info)output_pathos.path.join(extract_dir,filename)# 确保目录存在os.makedirs(os.path.dirname(output_path),exist_okTrue)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_ziptempfile.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_old30): 自动归档超过指定天数的文件 把旧文件按月份压缩归档 ifnotos.path.exists(archive_dir):os.makedirs(archive_dir)cutoff_timetime.time()-days_old*86400archives{}# {月份: [文件列表]}# 找出需要归档的文件forfilenameinos.listdir(source_dir):filepathos.path.join(source_dir,filename)ifnotos.path.isfile(filepath):continuefile_mtimeos.path.getmtime(filepath)iffile_mtimecutoff_time:# 按月份分组file_datedatetime.fromtimestamp(file_mtime)month_keyfile_date.strftime(%Y-%m)ifmonth_keynotinarchives:archives[month_key][]archives[month_key].append(filepath)# 按月份压缩归档formonth,filesinarchives.items():zip_pathos.path.join(archive_dir,farchive_{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_old30)七、带密码的ZIP文件# Python标准库zipfile只支持解压带密码的ZIP不支持创建# 解压带密码的ZIPdefunzip_with_password(zip_path,extract_dir,password):解压带密码的ZIP文件withzipfile.ZipFile(zip_path,r)aszf:try:zf.extractall(extract_dir,pwdpassword.encode())print(解压成功)returnTrueexceptRuntimeErrorase:ifpasswordinstr(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, compressionpyzipper.ZIP_DEFLATED, encryptionpyzipper.WZ_AES) as zf:# zf.setpassword(bmypassword)# zf.write(data.xlsx)有什么坑坑1ZIP_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():# 检查路径安全性targetos.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)等一下系统释放文件锁。在影刀中确保没有其他指令正在使用这些文件。