影刀RPA 文件备份策略：增量备份与版本管理

影刀RPA 文件备份策略增量备份与版本管理作者林焱什么情况用什么重要文件需要定期备份但每次全量备份太耗时太占空间。增量备份只复制变化的文件版本管理保留历史版本可以回滚。在影刀RPA里可以用Python实现自动化的增量备份和版本管理。适用场景项目文件备份、报表版本管理、配置文件历史保存、数据安全策略。怎么做拼多多店群自动化报活动上架全量备份importosimportshutilfromdatetimeimportdatetimedeffull_backup(source,target):全量备份timestampdatetime.now().strftime(%Y%m%d_%H%M%S)backup_diros.path.join(target,fbackup_{timestamp})# 复制整个目录shutil.copytree(source,backup_dir)# 记录备份信息file_countsum(len(files)for_,_,filesinos.walk(backup_dir))total_sizeget_dir_size(backup_dir)print(f备份完成:{backup_dir})print(f 文件数:{file_count})print(f 总大小:{total_size/1024/1024:.1f}MB)returnbackup_dirdefget_dir_size(path):获取目录总大小total0fordirpath,_,filenamesinos.walk(path):forfinfilenames:fpos.path.join(dirpath,f)totalos.path.getsize(fp)returntotal增量备份importhashlibimportjsondefincremental_backup(source,target,record_filebackup_record.json):增量备份只备份新增或修改的文件# 1. 加载上次备份记录record_pathos.path.join(target,record_file)ifos.path.exists(record_path):withopen(record_path,r)asf:old_recordsjson.load(f)else:old_records{}# 2. 扫描当前文件current_records{}backup_files[]forroot,_,filesinos.walk(source):forfinfiles:filepathos.path.join(root,f)rel_pathos.path.relpath(filepath,source)# 计算文件MD5file_hashget_file_md5(filepath)mtimeos.path.getmtime(filepath)current_records[rel_path]{hash:file_hash,mtime:mtime}# 判断是否需要备份oldold_records.get(rel_path)ifoldisNone:# 新文件backup_files.append((rel_path,新增))elifold[hash]!file_hash:# 修改过的文件backup_files.append((rel_path,修改))# 3. 检查删除的文件deleted_filesset(old_records.keys())-set(current_records.keys())# 4. 执行备份ifbackup_filesordeleted_files:timestampdatetime.now().strftime(%Y%m%d_%H%M%S)backup_diros.path.join(target,fincremental_{timestamp})os.makedirs(backup_dir,exist_okTrue)forrel_path,actioninbackup_files:srcos.path.join(source,rel_path)dstos.path.join(backup_dir,rel_path)os.makedirs(os.path.dirname(dst),exist_okTrue)shutil.copy2(src,dst)# 保存备份记录withopen(record_path,w)asf:json.dump(current_records,f,indent2)# 保存备份日志log_pathos.path.join(backup_dir,backup_log.txt)withopen(log_path,w)asf:f.write(f备份时间:{timestamp}\n)f.write(f新增:{sum(1for_,ainbackup_filesifa新增)}个\n)f.write(f修改:{sum(1for_,ainbackup_filesifa修改)}个\n)f.write(f删除:{len(deleted_files)}个\n)f.write(f\n新增/修改文件:\n)forrel_path,actioninbackup_files:f.write(f [{action}]{rel_path}\n)ifdeleted_files:f.write(f\n删除的文件:\n)forrel_pathindeleted_files:f.write(f{rel_path}\n)print(f增量备份完成:{backup_dir})print(f 新增:{sum(1for_,ainbackup_filesifa新增)})print(f 修改:{sum(1for_,ainbackup_filesifa修改)})print(f 删除:{len(deleted_files)})else:print(无变化不需要备份)returnbackup_files,deleted_filesdefget_file_md5(filepath):计算文件MD5hash_md5hashlib.md5()withopen(filepath,rb)asf:forchunkiniter(lambda:f.read(8192),b):hash_md5.update(chunk)returnhash_md5.hexdigest()版本管理defversion_backup(filepath,version_dir,max_versions10):文件版本管理保留最近N个版本os.makedirs(version_dir,exist_okTrue)# 生成版本文件名timestampdatetime.now().strftime(%Y%m%d_%H%M%S)filenameos.path.basename(filepath)name,extos.path.splitext(filename)version_filenamef{name}_{timestamp}{ext}version_pathos.path.join(version_dir,version_filename)# 复制当前版本shutil.copy2(filepath,version_path)# 清理旧版本versions[fforfinos.listdir(version_dir)iff.startswith(name)]versions.sort(keylambdaf:os.path.getmtime(os.path.join(version_dir,f)),reverseTrue)# 删除超出数量的旧版本forold_versioninversions[max_versions:]:os.remove(os.path.join(version_dir,old_version))print(f 删除旧版本:{old_version})print(f版本保存:{version_filename}(当前保留{min(len(versions)1,max_versions)}个版本))returnversion_path影刀RPA备份流程【定时任务】每天凌晨2点执行 【设置变量】 source rC:\Projects\MyProject backup_root rD:\Backup\MyProject 【执行Python代码】 # 增量备份 backup_files, deleted incremental_backup( sourceyd_var[source], targetyd_var[backup_root]  ) 【输出信息】 备份完成新增X个修改Y个删除Z个 【清理旧备份】 # 保留最近30天的备份 clean_old_backups(yd_var[backup_root], keep_days30)自动清理旧备份defclean_old_backups(backup_root,keep_days30):清理超过指定天数的备份cutoffdatetime.now().timestamp()-keep_days*86400deleted_count0foriteminos.listdir(backup_root):item_pathos.path.join(backup_root,item)ifnotos.path.isdir(item_path):continue# 检查修改时间mtimeos.path.getmtime(item_path)ifmtimecutoff:shutil.rmtree(item_path)deleted_count1print(f 删除旧备份:{item})print(f共清理{deleted_count}个旧备份)有什么坑TEMU店群矩阵自动化运营核价报活动坑1大文件MD5计算慢# 问题1GB的文件算MD5要几十秒# 解决用文件大小修改时间代替MD5做快速判断defget_file_signature(filepath):快速文件签名大小修改时间statos.stat(filepath)returnf{stat.st_size}_{stat.st_mtime}# 或者用部分内容算MD5速度更快defget_file_md5_fast(filepath,sample_size65536):快速MD5只算文件头尾hash_md5hashlib.md5()withopen(filepath,rb)asf:hash_md5.update(f.read(sample_size))# 文件头f.seek(-sample_size,2)hash_md5.update(f.read(sample_size))# 文件尾hash_md5.update(str(os.path.getsize(filepath)).encode())returnhash_md5.hexdigest()坑2备份占用空间过大# 解决备份后压缩importzipfiledefbackup_and_compress(source,target):备份并压缩timestampdatetime.now().strftime(%Y%m%d_%H%M%S)zip_pathos.path.join(target,fbackup_{timestamp}.zip)withzipfile.ZipFile(zip_path,w,zipfile.ZIP_DEFLATED)aszf:forroot,_,filesinos.walk(source):forfinfiles:filepathos.path.join(root,f)arcnameos.path.relpath(filepath,source)zf.write(filepath,arcname)print(f压缩备份:{zip_path}({os.path.getsize(zip_path)//1024}KB))坑3备份过程中文件被修改# 问题复制文件时文件正被写入导致备份不完整# 解决用文件锁或先复制到临时位置importtempfiledefsafe_backup(filepath,backup_dir):安全备份先复制到临时文件再移动temp_dirtempfile.mkdtemp()temp_fileos.path.join(temp_dir,os.path.basename(filepath))try:# 先复制到临时位置shutil.copy2(filepath,temp_file)# 再移动到备份目录final_pathos.path.join(backup_dir,os.path.basename(filepath))shutil.move(temp_file,final_path)finally:os.rmdir(temp_dir)坑4路径太长# Windows路径超过260字符会报错# 解决使用长路径前缀deflong_path(path):Windows长路径支持ifos.namentandlen(path)250:ifnotpath.startswith(\\\\?\\):path\\\\?\\os.path.abspath(path)returnpath坑5备份到网络驱动器失败# 问题备份到NAS/网络驱动器时偶尔断连# 解决重试机制defbackup_with_retry(src,dst,max_retries3):带重试的备份forattemptinrange(max_retries):try:shutil.copy2(src,dst)returnTrueexceptExceptionase:print(f第{attempt1}次备份失败:{e})ifattemptmax_retries-1:importtime time.sleep(5)returnFalse