影刀RPA Python节点中操作文件系统：读写文件、遍历目录、压缩解压
影刀的文件操作动作覆盖了基本的创建/复制/删除/移动。但遇到复杂场景——递归遍历子目录、按文件大小筛选、批量重命名、压缩解压——就得靠Python节点了。
这篇文章给出文件系统操作的标准代码片段，每一段都能直接拷过去用。
遍历目录
列出所有文件（含子目录）
importos folder=r'C:\RPA_data\reports'all_files=[]forroot,dirs,filesinos.walk(folder):forfilenameinfiles:full_path=os.path.join(root,filename)all_files.append({'path':full_path,'name':filename,'folder':root,'size':os.path.getsize(full_path),'modified':os.path.getmtime(full_path),})print(f'共找到{len(all_files)}个文件')只列出特定类型
importosimportglob# 方法1：glob模式匹配excel_files=glob.glob(r'C:\RPA_data\**\*.xlsx',recursive=True)# 方法2：手动过滤forroot,dirs,filesinos.walk(folder):forfinfiles:iff.endswith(('.xlsx','.xls')):# 处理Excel文件pass按文件修改时间筛选
拼多多店群自动化上架方案
importosimporttime# 找最近24小时修改过的文件now=time.time()recent_files=[]forroot,dirs,filesinos.walk(folder):forfinfiles:path=os.path.join(root,f)ifnow-os.path.getmtime(path)<86400:# 24小时 = 86400秒recent_files.append(path)文件读写
文本文件
# 读withopen(r'C:\data\log.txt','r',encoding='utf-8')asf:content=f.read()# 逐行读（大文件）withopen(r'C:\data\log.txt','r',encoding='utf-8')asf:forlineinf:print(line.strip())# 写（覆盖）withopen(r'C:\data\output.txt','w',encoding='utf-8')asf:f.write('第一行\n')f.write('第二行\n')# 追加withopen(r'C:\data\log.txt','a',encoding='utf-8')asf:f.write('新的一行\n')JSON文件
importjson# 读withopen(r'C:\data\config.json','r',encoding='utf-8')asf:config=json.load(f)# 写data={'name':'张三','orders':[1,2,3]}withopen(r'C:\data\output.json','w',encoding='utf-8')asf:json.dump(data,f,ensure_ascii=False,indent=2)文件批量处理
批量重命名
importos folder=r'C:\RPA_data\images'fori,filenameinenumerate(os.listdir(folder),1):iffilename.endswith('.png'):old_path=os.path.join(folder,filename)new_name=f'product_{i:04d}.png'# product_0001.pngnew_path=os.path.join(folder,new_name)os.rename(old_path,new_path)print(f'{filename}→{new_name}')批量移动文件（按条件分类）
importosimportshutil source=r'C:\RPA_data\incoming'dest_excel=r'C:\RPA_data\excel_files'dest_images=r'C:\RPA_data\images'dest_others=r'C:\RPA_data\others'os.makedirs(dest_excel,exist_ok=True)os.makedirs(dest_images,exist_ok=True)os.makedirs(dest_others,exist_ok=True)forfilenameinos.listdir(source):src=os.path.join(source,filename)ifnotos.path.isfile(src):continueiffilename.endswith(('.xlsx','.xls')):shutil.move(src,os.path.join(dest_excel,filename))eliffilename.endswith(('.png','.jpg','.jpeg')):shutil.move(src,os.path.join(dest_images,filename))else:shutil.move(src,os.path.join(dest_others,filename))压缩与解压
压缩文件/文件夹
importzipfileimportosdefzip_folder(folder_path,output_path):"""压缩整个文件夹"""withzipfile.ZipFile(output_path,'w',zipfile.ZIP_DEFLATED)aszf:forroot,dirs,filesinos.walk(folder_path):forfileinfiles:file_path=os.path.join(root,file)arcname=os.path.relpath(file_path,folder_path)zf.write(file_path,arcname)print(f'压缩完成：{output_path}')defzip_files(file_list,output_path):"""压缩指定文件列表"""withzipfile.ZipFile(output_path,'w',zipfile.ZIP_DEFLATED)aszf:forfilepathinfile_list:zf.write(filepath,os.path.basename(filepath))print(f'压缩完成：{output_path}（{len(file_list)}个文件）')# 使用zip_folder(r'C:\RPA_data\reports',r'C:\RPA_data\reports_backup.zip')解压
importzipfiledefunzip_file(zip_path,extract_to):"""解压zip文件"""withzipfile.ZipFile(zip_path,'r')aszf:zf.extractall(extract_to)[video(video-GZQFjvBh-1783622761443)(type-csdn)(url-https://live.csdn.net/v/embed/524993)(image-https://v-blog.csdnimg.cn/asset/a547123d88ad712dccba346c9217e237/cover/Cover0.jpg)(title-TEMU店群如何管理运营？)]# 列出解压的文件withzipfile.ZipFile(zip_path,'r')aszf:print(f'解压了{len(zf.namelist())}个文件到{extract_to}')unzip_file(r'C:\RPA_data\data.zip',r'C:\RPA_data\extracted')文件夹操作
importosimportshutil# 创建嵌套目录os.makedirs(r'C:\RPA_data\2026\07\01',exist_ok=True)# 复制文件夹（含所有内容）shutil.copytree(r'C:\RPA_data\source',r'C:\RPA_data\backup')# 删除文件夹（含所有内容）shutil.rmtree(r'C:\RPA_data\temp')# 获取文件夹大小defget_folder_size(folder):total=0forroot,dirs,filesinos.walk(folder):forfinfiles:total+=os.path.getsize(os.path.join(root,f))returntotal size_mb=get_folder_size(r'C:\RPA_data')/(1024*1024)print(f'文件夹大小：{size_mb:.2f}MB')文件清理（安全版）
importosimporttimedefsafe_cleanup(folder,days_old=30,dry_run=True):"""清理N天之前的文件（dry_run=True只列不删）"""now=time.time()cutoff=now-days_old*86400to_delete=[]forroot,dirs,filesinos.walk(folder):forfinfiles:path=os.path.join(root,f)ifos.path.getmtime(path)<cutoff:to_delete.append(path)print(f'将清理{len(to_delete)}个文件（{days_old}天前）：')forpathinto_delete[:10]:# 只显示前10个print(f'{path}')ifnotdry_runandto_delete:print(f'开始删除...')forpathinto_delete:try:os.remove(path)exceptExceptionase:print(f' 删除失败：{path}—{e}')print('清理完成')# 先做一次dry_run看看要删什么safe_cleanup(r'C:\RPA_data\temp',days_old=7,dry_run=True)# 确认OK后正式清理# safe_cleanup(r'C:\RPA_data\temp', days_old=7, dry_run=False)dry_run=True的安全机制是推荐的——先看一遍要删什么，确认没删错再正式执行。
总结：文件系统操作在Python里就是os、shutil、zipfile三个模块的事。遍历用os.walk，移动用shutil.move/copy，压缩用zipfile。批量操作前先用dry_run模式预览，避免删错。
作者：林焱
