定时检查备份完整性的脚本

#!/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 参数,在发现异常时自动发送邮件通知。

核心检查能力

• 缺失检测：对比清单,找出已被删除的关键备份文件。
• 篡改/损坏检测：重新计算哈希并与清单比对,识别被修改或损坏的文件。
• 新增文件提醒：列出不在清单中的新文件,提示您是否需要将其纳入备份范围。