影刀RPA 临时文件管理：创建与清理
影刀RPA 临时文件管理：创建与清理
作者：林焱
什么情况用什么
自动化流程运行中会产生临时文件——中间结果、下载的临时文件、处理过程的缓存。这些文件不及时清理会占满磁盘空间，但又不能在流程结束前删掉（因为还在用）。在影刀RPA里需要合理管理临时文件的创建、使用和清理。
适用场景：流程中间文件管理、下载缓存清理、系统临时文件清理、磁盘空间监控。
怎么做
拼多多店群自动化报活动上架！
创建临时文件
importosimporttempfileimportuuidfromdatetimeimportdatetime# 方式1：用tempfile模块（推荐）defcreate_temp_file(suffix='.tmp',content=None):"""创建临时文件"""fd,filepath=tempfile.mkstemp(suffix=suffix)ifcontent:withos.fdopen(fd,'w',encoding='utf-8')asf:f.write(content)else:os.close(fd)returnfilepath# 方式2：在指定目录创建带时间戳的临时文件defcreate_temp_in_dir(temp_dir,prefix='temp',suffix='.xlsx'):"""在指定目录创建临时文件"""os.makedirs(temp_dir,exist_ok=True)timestamp=datetime.now().strftime('%Y%m%d_%H%M%S')unique_id=uuid.uuid4().hex[:8]filename=f"{prefix}_{timestamp}_{unique_id}{suffix}"filepath=os.path.join(temp_dir,filename)returnfilepath# 使用temp_file=create_temp_file(suffix='.json',content='{"status": "processing"}')temp_excel=create_temp_in_dir(r"C:\Temp",prefix='report',suffix='.xlsx')临时文件管理器
classTempFileManager:"""临时文件管理器"""def__init__(self,base_dir=None):ifbase_dirisNone:base_dir=os.path.join(tempfile.gettempdir(),'yingdao_rpa')self.base_dir=base_dir os.makedirs(base_dir,exist_ok=True)self.files=[]defcreate(self,prefix='temp',suffix='.tmp',content=None):"""创建临时文件"""timestamp=datetime.now().strftime('%Y%m%d_%H%M%S')filename=f"{prefix}_{timestamp}_{uuid.uuid4().hex[:6]}{suffix}"filepath=os.path.join(self.base_dir,filename)ifcontent:withopen(filepath,'w',encoding='utf-8')asf:f.write(content)else:# 创建空文件open(filepath,'w').close()self.files.append(filepath)returnfilepathdefcleanup(self):"""清理所有临时文件"""cleaned=0forfilepathinself.files:try:ifos.path.exists(filepath):ifos.path.isfile(filepath):os.remove(filepath)elifos.path.isdir(filepath):importshutil shutil.rmtree(filepath)cleaned+=1exceptExceptionase:print(f"清理失败:{filepath}-{e}")self.files.clear()print(f"已清理{cleaned}个临时文件")returncleaneddefcleanup_old(self,max_age_hours=24):"""清理超过指定时间的临时文件"""importtime cutoff=time.time()-max_age_hours*3600cleaned=0forfinos.listdir(self.base_dir):filepath=os.path.join(self.base_dir,f)ifos.path.isfile(filepath):mtime=os.path.getmtime(filepath)ifmtime<cutoff:os.remove(filepath)cleaned+=1print(f"清理{cleaned}个超期临时文件")returncleaned# 使用manager=TempFileManager(r"C:\Temp\RPA")# 流程中创建临时文件temp1=manager.create(prefix='download',suffix='.xlsx')temp2=manager.create(prefix='process',suffix='.csv')# ... 使用临时文件 ...# 流程结束时清理manager.cleanup()系统临时文件清理
defclean_system_temp():"""清理系统临时文件夹"""temp_dirs=[os.environ.get('TEMP',r'C:\Users\Public\Temp'),os.environ.get('TMP',r'C:\Users\Public\Temp'),r"C:\Windows\Temp",]total_cleaned=0total_freed=0fortemp_dirintemp_dirs:ifnotos.path.exists(temp_dir):continueforfinos.listdir(temp_dir):filepath=os.path.join(temp_dir,f)try:ifos.path.isfile(filepath):size=os.path.getsize(filepath)os.remove(filepath)total_cleaned+=1total_freed+=sizeelifos.path.isdir(filepath):importshutil size=get_dir_size(filepath)shutil.rmtree(filepath,ignore_errors=True)total_cleaned+=1total_freed+=sizeexcept:pass# 跳过无法删除的文件print(f"清理临时文件:{total_cleaned}个")print(f"释放空间:{total_freed/1024/1024:.1f}MB")returntotal_cleaned,total_freeddefget_dir_size(path):"""获取目录大小"""total=0fordirpath,_,filesinos.walk(path):forfinfiles:try:total+=os.path.getsize(os.path.join(dirpath,f))except:passreturntotal磁盘空间监控
importshutilasshdefcheck_disk_space(path='C:\\'):"""检查磁盘空间"""total,used,free=sh.disk_usage(path)result={'磁盘':path,'总空间(GB)':round(total/1024**3,1),'已用(GB)':round(used/1024**3,1),'可用(GB)':round(free/1024**3,1),'使用率':f"{used/total*100:.1f}%",}fork,vinresult.items():print(f"{k}:{v}")# 空间不足预警free_gb=free/1024**3iffree_gb<5:print(f"\n⚠ 警告：磁盘可用空间不足5GB！")returnresult# 使用check_disk_space('C:\\')check_disk_space('D:\\')影刀RPA临时文件管理流程
【流程开始】 【执行Python代码】 # 创建临时文件管理器 manager = TempFileManager(r"C:\Temp\RPA") yd_var['temp_manager'] = manager 【下载文件】→ manager.create(prefix='download', suffix='.xlsx') 【处理数据】→ 产生中间文件 【生成报表】→ 最终输出 【流程结束前】 【执行Python代码】 # 清理临时文件 manager.cleanup() # 检查磁盘空间 check_disk_space('C:\\')有什么坑
TEMU店群矩阵自动化运营核价报活动
坑1：流程异常退出临时文件残留
# 问题：流程中途出错，临时文件没清理# 解决：用try-finally确保清理manager=TempFileManager()try:temp_file=manager.create(suffix='.xlsx')# ... 处理逻辑 ...finally:manager.cleanup()# 无论成功失败都清理# 或在流程开始时先清理上次残留manager=TempFileManager()manager.cleanup_old(max_age_hours=1)# 清理1小时前的坑2：删除正在使用的临时文件
# 问题：Excel文件正被打开，删除失败# 解决：先关闭再删除defsafe_delete(filepath,max_retries=3):"""安全删除文件"""importtimeforiinrange(max_retries):try:ifos.path.exists(filepath):os.remove(filepath)returnTrueexceptPermissionError:time.sleep(1)print(f"无法删除（可能被占用）:{filepath}")returnFalse坑3：临时文件名冲突
# 问题：多个流程同时运行，临时文件名相同# 解决：用UUID确保唯一filename=f"temp_{uuid.uuid4().hex}{suffix}"# 或用时间戳+进程IDimportos filename=f"temp_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{os.getpid()}{suffix}"坑4：Temp目录在不同电脑上路径不同
# 问题：硬编码C:\Temp在没这个目录的电脑上报错# 解决：用系统临时目录temp_dir=tempfile.gettempdir()# 或temp_dir=os.environ.get('TEMP',os.environ.get('TMP','/tmp'))坑5：清理时误删重要文件
# 问题：清理临时目录时把不该删的文件也删了# 解决：只清理特定前缀的文件SAFE_PREFIX='yingdao_'# 只清理这个前缀的defsafe_cleanup(temp_dir,prefix=SAFE_PREFIX):"""安全清理（只删指定前缀的文件）"""forfinos.listdir(temp_dir):iff.startswith(prefix):filepath=os.path.join(temp_dir,f)os.remove(filepath)print(f" 删除:{f}")# 不匹配前缀的文件不动