#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
备份完整性检查器
功能：
1. 定期检查指定备份目录中的文件是否齐全（与清单对比）
2. 校验文件哈希（MD5/SHA256），确保数据未被篡改或损坏
3. 支持生成新的校验清单，用于首次初始化
4. 检查结果通过日志记录，并可选发送邮件告警
5. 通过cron或任务计划程序实现定时执行
用法：
python backup_integrity_checker.py --init # 首次生成校验清单
python backup_integrity_checker.py --check # 执行完整性检查
python backup_integrity_checker.py --check --email # 检查并发送邮件通知
"""
import os
import sys
import hashlib
import json
import argparse
import logging
from datetime import datetime
from pathlib import Path
# 尝试导入邮件模块（若无则跳过邮件功能）
try:
import smtplib
from email.mime.text import MIMEText
from email.header import Header
EMAIL_AVAILABLE = True
except ImportError:
EMAIL_AVAILABLE = False
# ------------------------- 配置区域 -------------------------
BACKUP_ROOT = "/path/to/your/backup" # 备份根目录（请修改）
MANIFEST_FILE = "manifest.json" # 校验清单文件名
HASH_ALGO = "sha256" # 哈希算法: md5, sha1, sha256
LOG_FILE = "backup_check.log" # 日志文件
# 邮件配置（可选）
EMAIL_ENABLED = False
SMTP_SERVER = "smtp.example.com"
SMTP_PORT = 465
SMTP_USER = "your_username"
SMTP_PASSWORD = "your_password"
MAIL_FROM = "checker@example.com"
MAIL_TO = ["admin@example.com"]
MAIL_SUBJECT = "备份完整性检查告警"
# ------------------------------------------------------------
# 日志配置
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(LOG_FILE, encoding='utf-8'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger("backup-checker")
def compute_file_hash(filepath: Path, algo: str = HASH_ALGO) -> str:
"""计算单个文件的哈希值"""
h = hashlib.new(algo)
try:
with open(filepath, "rb") as f:
# 分块读取，避免大文件占用过多内存
for chunk in iter(lambda: f.read(65536), b""):
h.update(chunk)
return h.hexdigest()
except Exception as e:
logger.error(f"无法计算文件哈希 {filepath}: {e}")
return None
def generate_manifest(root: Path, manifest_path: Path) -> bool:
"""生成当前备份目录的校验清单"""
manifest = {}
try:
logger.info(f"正在扫描目录: {root}")
for dirpath, dirnames, filenames in os.walk(root):
# 跳过隐藏目录
dirnames[:] = [d for d in dirnames if not d.startswith('.')]
for fname in filenames:
# 跳过清单文件自身
if fname == manifest_path.name:
continue
full_path = Path(dirpath) / fname
rel_path = str(full_path.relative_to(root))
hash_val = compute_file_hash(full_path)
if hash_val:
manifest[rel_path] = hash_val
logger.debug(f"已记录: {rel_path} -> {hash_val[:8]}...")
# 写入清单文件
with open(manifest_path, "w", encoding="utf-8") as f:
json.dump(manifest, f, indent=2, ensure_ascii=False)
logger.info(f"校验清单已生成: {manifest_path}")
return True
except Exception as e:
logger.error(f"生成清单失败: {e}")
return False
def check_integrity(root: Path, manifest_path: Path) -> dict:
"""执行备份完整性检查，返回结果统计"""
result = {
"total": 0, # 总文件数
"ok": 0, # 正常文件数
"missing": [], # 缺失的文件
"corrupted": [], # 哈希不匹配的文件
"extra": [] # 额外的新文件（不在清单中）
}
if not manifest_path.exists():
logger.error(f"清单文件不存在: {manifest_path}，请先执行 --init")
return None
with open(manifest_path, "r", encoding="utf-8") as f:
manifest = json.load(f)
# 获取当前所有文件相对路径
current_files = set()
for dirpath, dirnames, filenames in os.walk(root):
dirnames[:] = [d for d in dirnames if not d.startswith('.')]
for fname in filenames:
full_path = Path(dirpath) / fname
if full_path.name == manifest_path.name:
continue
rel_path = str(full_path.relative_to(root))
current_files.add(rel_path)
manifest_files = set(manifest.keys())
result["total"] = len(manifest_files)
# 检查缺失文件
result["missing"] = list(manifest_files - current_files)
# 检查新增文件
result["extra"] = list(current_files - manifest_files)
# 检查哈希匹配（仅针对存在于现有清单中的文件）
for fname in manifest_files & current_files:
full_path = root / fname
current_hash = compute_file_hash(full_path)
if current_hash is None:
result["corrupted"].append(fname)
elif current_hash != manifest[fname]:
result["corrupted"].append(fname)
else:
result["ok"] += 1
return result
def send_email_alert(subject: str, body: str):
"""发送邮件告警（需配置SMTP）"""
if not EMAIL_AVAILABLE or not EMAIL_ENABLED:
logger.warning("邮件功能未启用或依赖未安装，跳过发送")
return False
msg = MIMEText(body, "plain", "utf-8")
msg["Subject"] = Header(subject, "utf-8")
msg["From"] = MAIL_FROM
msg["To"] = ", ".join(MAIL_TO)
try:
with smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT) as server:
server.login(SMTP_USER, SMTP_PASSWORD)
server.sendmail(MAIL_FROM, MAIL_TO, msg.as_string())
logger.info("告警邮件发送成功")
return True
except Exception as e:
logger.error(f"邮件发送失败: {e}")
return False
def format_report(result: dict) -> str:
"""格式化检查结果为文本报告"""
lines = []
lines.append("=" * 50)
lines.append(f"备份完整性检查报告")
lines.append(f"检查时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
lines.append(f"备份目录: {BACKUP_ROOT}")
lines.append(f"总文件数: {result['total']}")
lines.append(f"正常文件: {result['ok']}")
if result["missing"]:
lines.append(f"\n❌ 缺失文件 ({len(result['missing'])}):")
for f in result["missing"]:
lines.append(f" - {f}")
if result["corrupted"]:
lines.append(f"\n🔴 损坏/篡改文件 ({len(result['corrupted'])}):")
for f in result["corrupted"]:
lines.append(f" - {f}")
if result["extra"]:
lines.append(f"\n🆕 新增文件 (不在清单中) ({len(result['extra'])}):")
for f in result["extra"]:
lines.append(f" + {f}")
lines.append("=" * 50)
return "\n".join(lines)
def main():
parser = argparse.ArgumentParser(description="备份完整性检查工具")
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("--init", action="store_true", help="生成新的校验清单")
group.add_argument("--check", action="store_true", help="执行完整性检查")
parser.add_argument("--email", action="store_true", help="检查后发送邮件通知（需配置SMTP）")
args = parser.parse_args()
# 检查备份目录是否存在
backup_path = Path(BACKUP_ROOT)
if not backup_path.exists() or not backup_path.is_dir():
logger.error(f"备份目录不存在或不可用: {BACKUP_ROOT}")
sys.exit(1)
manifest_path = backup_path / MANIFEST_FILE
if args.init:
# 初始化模式：生成清单并退出
if generate_manifest(backup_path, manifest_path):
logger.info("清单初始化完成，请定期执行 --check 进行验证。")
else:
sys.exit(1)
elif args.check:
# 检查模式
logger.info("开始备份完整性检查...")
result = check_integrity(backup_path, manifest_path)
if result is None:
sys.exit(1)
report = format_report(result)
logger.info("\n" + report)
# 判断是否出现异常
has_issues = bool(result["missing"] or result["corrupted"])
if has_issues:
logger.warning("⚠️ 检测到备份异常！")
if args.email and EMAIL_ENABLED:
send_email_alert(MAIL_SUBJECT, report)
else:
logger.info("✅ 备份完整性正常")
# 额外提醒存在新增文件（非错误）
if result["extra"]:
logger.info("提醒：检测到清单之外的新增文件，可考虑更新清单。")
# 根据检查结果设置进程退出码（便于外部监控）
sys.exit(1 if has_issues else 0)
if __name__ == "__main__":
main()
使用方法与核心功能
这个脚本通过对比备份文件与预先生成的“清单”来验证数据完整性,并能通过邮件及时告警。
首次使用（生成清单）
python backup_integrity_checker.py --init
此命令会扫描您配置的备份目录，并生成一个包含所有文件哈希值的 manifest.json 清单。
定时检查（添加到cron）
# 每天凌晨2点执行检查 0 2 * * * /usr/bin/python3 /path/to/backup_integrity_checker.py --check
您也可以通过 --email 参数,在发现异常时自动发送邮件通知。
核心检查能力
- 缺失检测：对比清单,找出已被删除的关键备份文件。
- 篡改/损坏检测：重新计算哈希并与清单比对,识别被修改或损坏的文件。
- 新增文件提醒：列出不在清单中的新文件,提示您是否需要将其纳入备份范围。
