影刀RPA 临时文件管理：创建与清理

影刀RPA 临时文件管理创建与清理作者林焱什么情况用什么自动化流程运行中会产生临时文件——中间结果、下载的临时文件、处理过程的缓存。这些文件不及时清理会占满磁盘空间但又不能在流程结束前删掉因为还在用。在影刀RPA里需要合理管理临时文件的创建、使用和清理。适用场景流程中间文件管理、下载缓存清理、系统临时文件清理、磁盘空间监控。怎么做拼多多店群自动化报活动上架创建临时文件importosimporttempfileimportuuidfromdatetimeimportdatetime# 方式1用tempfile模块推荐defcreate_temp_file(suffix.tmp,contentNone):创建临时文件fd,filepathtempfile.mkstemp(suffixsuffix)ifcontent:withos.fdopen(fd,w,encodingutf-8)asf:f.write(content)else:os.close(fd)returnfilepath# 方式2在指定目录创建带时间戳的临时文件defcreate_temp_in_dir(temp_dir,prefixtemp,suffix.xlsx):在指定目录创建临时文件os.makedirs(temp_dir,exist_okTrue)timestampdatetime.now().strftime(%Y%m%d_%H%M%S)unique_iduuid.uuid4().hex[:8]filenamef{prefix}_{timestamp}_{unique_id}{suffix}filepathos.path.join(temp_dir,filename)returnfilepath# 使用temp_filecreate_temp_file(suffix.json,content{status: processing})temp_excelcreate_temp_in_dir(rC:\Temp,prefixreport,suffix.xlsx)临时文件管理器classTempFileManager:临时文件管理器def__init__(self,base_dirNone):ifbase_dirisNone:base_diros.path.join(tempfile.gettempdir(),yingdao_rpa)self.base_dirbase_dir os.makedirs(base_dir,exist_okTrue)self.files[]defcreate(self,prefixtemp,suffix.tmp,contentNone):创建临时文件timestampdatetime.now().strftime(%Y%m%d_%H%M%S)filenamef{prefix}_{timestamp}_{uuid.uuid4().hex[:6]}{suffix}filepathos.path.join(self.base_dir,filename)ifcontent:withopen(filepath,w,encodingutf-8)asf:f.write(content)else:# 创建空文件open(filepath,w).close()self.files.append(filepath)returnfilepathdefcleanup(self):清理所有临时文件cleaned0forfilepathinself.files:try:ifos.path.exists(filepath):ifos.path.isfile(filepath):os.remove(filepath)elifos.path.isdir(filepath):importshutil shutil.rmtree(filepath)cleaned1exceptExceptionase:print(f清理失败:{filepath}-{e})self.files.clear()print(f已清理{cleaned}个临时文件)returncleaneddefcleanup_old(self,max_age_hours24):清理超过指定时间的临时文件importtime cutofftime.time()-max_age_hours*3600cleaned0forfinos.listdir(self.base_dir):filepathos.path.join(self.base_dir,f)ifos.path.isfile(filepath):mtimeos.path.getmtime(filepath)ifmtimecutoff:os.remove(filepath)cleaned1print(f清理{cleaned}个超期临时文件)returncleaned# 使用managerTempFileManager(rC:\Temp\RPA)# 流程中创建临时文件temp1manager.create(prefixdownload,suffix.xlsx)temp2manager.create(prefixprocess,suffix.csv)# ... 使用临时文件 ...# 流程结束时清理manager.cleanup()系统临时文件清理defclean_system_temp():清理系统临时文件夹temp_dirs[os.environ.get(TEMP,rC:\Users\Public\Temp),os.environ.get(TMP,rC:\Users\Public\Temp),rC:\Windows\Temp,]total_cleaned0total_freed0fortemp_dirintemp_dirs:ifnotos.path.exists(temp_dir):continueforfinos.listdir(temp_dir):filepathos.path.join(temp_dir,f)try:ifos.path.isfile(filepath):sizeos.path.getsize(filepath)os.remove(filepath)total_cleaned1total_freedsizeelifos.path.isdir(filepath):importshutil sizeget_dir_size(filepath)shutil.rmtree(filepath,ignore_errorsTrue)total_cleaned1total_freedsizeexcept:pass# 跳过无法删除的文件print(f清理临时文件:{total_cleaned}个)print(f释放空间:{total_freed/1024/1024:.1f}MB)returntotal_cleaned,total_freeddefget_dir_size(path):获取目录大小total0fordirpath,_,filesinos.walk(path):forfinfiles:try:totalos.path.getsize(os.path.join(dirpath,f))except:passreturntotal磁盘空间监控importshutilasshdefcheck_disk_space(pathC:\\):检查磁盘空间total,used,freesh.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_gbfree/1024**3iffree_gb5:print(f\n⚠ 警告磁盘可用空间不足5GB)returnresult# 使用check_disk_space(C:\\)check_disk_space(D:\\)影刀RPA临时文件管理流程【流程开始】 【执行Python代码】 # 创建临时文件管理器 manager TempFileManager(rC:\Temp\RPA) yd_var[temp_manager] manager 【下载文件】→ manager.create(prefixdownload, suffix.xlsx) 【处理数据】→ 产生中间文件 【生成报表】→ 最终输出 【流程结束前】 【执行Python代码】 # 清理临时文件 manager.cleanup() # 检查磁盘空间 check_disk_space(C:\\)有什么坑TEMU店群矩阵自动化运营核价报活动坑1流程异常退出临时文件残留# 问题流程中途出错临时文件没清理# 解决用try-finally确保清理managerTempFileManager()try:temp_filemanager.create(suffix.xlsx)# ... 处理逻辑 ...finally:manager.cleanup()# 无论成功失败都清理# 或在流程开始时先清理上次残留managerTempFileManager()manager.cleanup_old(max_age_hours1)# 清理1小时前的坑2删除正在使用的临时文件# 问题Excel文件正被打开删除失败# 解决先关闭再删除defsafe_delete(filepath,max_retries3):安全删除文件importtimeforiinrange(max_retries):try:ifos.path.exists(filepath):os.remove(filepath)returnTrueexceptPermissionError:time.sleep(1)print(f无法删除可能被占用:{filepath})returnFalse坑3临时文件名冲突# 问题多个流程同时运行临时文件名相同# 解决用UUID确保唯一filenameftemp_{uuid.uuid4().hex}{suffix}# 或用时间戳进程IDimportos filenameftemp_{datetime.now().strftime(%Y%m%d_%H%M%S)}_{os.getpid()}{suffix}坑4Temp目录在不同电脑上路径不同# 问题硬编码C:\Temp在没这个目录的电脑上报错# 解决用系统临时目录temp_dirtempfile.gettempdir()# 或temp_diros.environ.get(TEMP,os.environ.get(TMP,/tmp))坑5清理时误删重要文件# 问题清理临时目录时把不该删的文件也删了# 解决只清理特定前缀的文件SAFE_PREFIXyingdao_# 只清理这个前缀的defsafe_cleanup(temp_dir,prefixSAFE_PREFIX):安全清理只删指定前缀的文件forfinos.listdir(temp_dir):iff.startswith(prefix):filepathos.path.join(temp_dir,f)os.remove(filepath)print(f 删除:{f})# 不匹配前缀的文件不动