当前位置：首页 > news >正文

影刀RPA 数据备份验证：自动检查备份文件完整性

news 2026/7/17 19:38:58

影刀RPA 数据备份验证：自动检查备份文件完整性

作者：林焱

什么情况用什么

DBA每天做数据库备份，自以为万事大吉。真等哪天数据库崩了，打开备份文件一看——空的，或者只备份了一半。这不是段子，是真事。

备份的价值在于能恢复。而能恢复的前提是备份文件是完整的、可用的。影刀RPA可以自动做备份验证：检查备份文件是否存在 → 检查文件大小是否异常 → 尝试恢复到一个测试环境 → 验证恢复后的数据是否完整。

适合场景：

每天自动检查昨晚的数据库备份
每周做一次完整的备份恢复演练
监控备份文件大小趋势（突然变小=可能备份失败）
验证异地备份是否同步成功

店群矩阵自动化突破运营极限！

怎么做

第一步：检查备份文件

影刀操作步骤：

1. 【遍历文件夹】遍历 D:\db_backups\ 目录 2. 对每个备份文件： - 【获取文件大小】记录文件大小 - 【获取文件修改时间】检查是否在今天凌晨生成 - 对比：和昨天的备份大小是否差异超过30%

importosfromdatetimeimportdatetime backup_dir=r'D:\db_backups'expected_date=datetime.now().strftime('%Y%m%d')issues=[]forfinos.listdir(backup_dir):path=os.path.join(backup_dir,f)size=os.path.getsize(path)# 检查1：文件大小是否>0ifsize==0:issues.append(f'文件为空:{f}')# 检查2：文件日期是否匹配file_date=datetime.fromtimestamp(os.path.getmtime(path)).strftime('%Y%m%d')iffile_date!=expected_date:issues.append(f'文件日期不匹配:{f}(期望{expected_date}，实际{file_date})')# 检查3：大小波动是否异常（对比历史）# ...（下一段展开）ifissues:alert='备份异常：\n'+'\n'.join(issues)# 发送告警

第二步：大小波动检测

备份文件大小突然减少可能意味着备份不完整。

importjson# 读取历史大小记录withopen(r'D:\backup_history.json','r')asf:history=json.load(f)forf_name,filesizeincurrent_backups.items():iff_nameinhistory:last_size=history[f_name][-1]# 上一次的大小change_pct=abs(filesize-last_size)/last_size*100ifchange_pct>30:issues.append(f'{f_name}: 大小波动{change_pct:.1f}% ({last_size//1024//1024}MB →{filesize//1024//1024}MB)')# 更新历史记录forf_name,filesizeincurrent_backups.items():history.setdefault(f_name,[]).append(filesize)history[f_name]=history[f_name][-30:]# 只保留最近30天withopen(r'D:\backup_history.json','w')asf:json.dump(history,f)

第三步：恢复演练

这是最关键的一步——真正尝试恢复备份到测试库。

影刀操作步骤：

1. 【执行命令行】在测试服务器上恢复备份 # MySQL mysql -h test-db -u root -p test_db < backup.sql # PostgreSQL pg_restore -h test-db -U postgres -d test_db backup.dump 2. 【执行命令行】检查恢复结果 mysql -h test-db -u root -e "SELECT COUNT(*) FROM users" test_db 3. 【读取输出】对比生产库和测试库的行数是否在合理范围

importsubprocessdefrestore_and_verify(backup_file,test_db_config):# 恢复cmd=f'mysql -h{test_db_config["host"]}-u{test_db_config["user"]}'\f'-p{test_db_config["password"]}{test_db_config["db"]}<{backup_file}'result=subprocess.run(cmd,shell=True,capture_output=True,timeout=600)ifresult.returncode!=0:returnFalse,f'恢复失败:{result.stderr.decode()}'# 验证核心表记录数tables=['users','orders','products']fortableintables:count_cmd=f'mysql -h{test_db_config["host"]}-u{test_db_config["user"]}'\f'-p{test_db_config["password"]}-e "SELECT COUNT(*) FROM{table}"{test_db_config["db"]}'count_result=subprocess.run(count_cmd,shell=True,capture_output=True)count=int(count_result.stdout.decode().strip().split('\n')[1])ifcount==0:returnFalse,f'表{table}恢复后为空'returnTrue,'恢复验证通过'

第四步：日报生成

report=f""" # 数据备份验证日报 -{date}## 备份文件检查 - 应备份文件数：{expected_count}- 实际文件数：{actual_count}- 异常文件：{len(issues)}个 ## 大小趋势{format_size_trends()}## 恢复演练 - 演练数据库：test_db - 恢复状态：{'通过'ifrestore_okelse'失败'}- 核心表记录数：正常 ## 异地备份同步 - 状态：{remote_status}"""