IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

定时转存文件到外部硬盘脚本

wen 实用脚本 2

本文目录导读：

定时转存文件到外部硬盘脚本

  1. 脚本使用说明
  2. 注意事项
#!/usr/bin/env python3
"""
定时转存文件到外部硬盘脚本
功能：监控指定源目录，将新增或修改的文件定时复制到外部硬盘目标目录。
支持：文件增量同步、日志记录、磁盘空间检查。
"""
import os
import sys
import shutil
import hashlib
import time
import logging
from datetime import datetime
from pathlib import Path
import argparse
# ============ 配置区（可修改） ============
SOURCE_DIR = "/path/to/source"          # 源目录（即要被备份的目录）
TARGET_DIR = "/mnt/external_drive"      # 外部硬盘挂载点或目标目录
INTERVAL_SECONDS = 3600                  # 检查间隔（秒），默认1小时
LOG_FILE = "./sync_log.txt"              # 日志文件路径
CHECK_DISK_SPACE = True                  # 是否检查磁盘空间
MIN_FREE_GB = 5                          # 最小剩余空间（GB），小于此值则报警
USE_MD5 = False                          # 是否使用MD5校验（False则比较文件大小+修改时间）
# =========================================
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(LOG_FILE, encoding='utf-8'),
        logging.StreamHandler(sys.stdout)
    ]
)
def get_file_hash(filepath):
    """计算文件的MD5哈希值（用于精确校验）"""
    hash_md5 = hashlib.md5()
    with open(filepath, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b""):
            hash_md5.update(chunk)
    return hash_md5.hexdigest()
def should_copy(source_path, target_path):
    """
    判断是否需要复制文件
    规则：目标不存在，或源文件比目标新/大小不同（USE_MD5=False时）
          目标不存在，或MD5不同（USE_MD5=True时）
    """
    if not os.path.exists(target_path):
        return True
    if USE_MD5:
        # MD5模式：比较哈希值
        src_hash = get_file_hash(source_path)
        tgt_hash = get_file_hash(target_path)
        return src_hash != tgt_hash
    else:
        # 快速模式：比较最后修改时间和文件大小
        src_stat = os.stat(source_path)
        tgt_stat = os.stat(target_path)
        return (src_stat.st_mtime != tgt_stat.st_mtime) or (src_stat.st_size != tgt_stat.st_size)
def check_disk_space(path, min_free_gb=MIN_FREE_GB):
    """检查目标路径所在磁盘的剩余空间"""
    if not CHECK_DISK_SPACE:
        return True
    try:
        stat = shutil.disk_usage(path)
        free_gb = stat.free / (1024**3)
        if free_gb < min_free_gb:
            logging.warning(f"磁盘空间不足！剩余 {free_gb:.2f} GB，低于 {min_free_gb} GB")
            return False
        else:
            logging.debug(f"磁盘空间充足：剩余 {free_gb:.2f} GB")
            return True
    except Exception as e:
        logging.error(f"检查磁盘空间时出错: {e}")
        return False
def sync_files(source_dir, target_dir):
    """
    同步源目录下的文件到目标目录（非递归复制所有内容）
    保持相同的目录结构。
    """
    if not os.path.exists(source_dir):
        logging.error(f"源目录不存在: {source_dir}")
        return
    if not os.path.exists(target_dir):
        logging.info(f"目标目录不存在，正在创建: {target_dir}")
        os.makedirs(target_dir, exist_ok=True)
    total_copied = 0
    total_errors = 0
    # 遍历源目录
    for root, dirs, files in os.walk(source_dir):
        # 计算相对路径，以便在目标目录重建结构
        relative_path = os.path.relpath(root, source_dir)
        target_root = os.path.join(target_dir, relative_path) if relative_path != "." else target_dir
        # 确保目标子目录存在
        os.makedirs(target_root, exist_ok=True)
        for file_name in files:
            source_path = os.path.join(root, file_name)
            target_path = os.path.join(target_root, file_name)
            # 跳过管道、设备文件等（仅处理普通文件）
            if not os.path.isfile(source_path):
                continue
            try:
                if should_copy(source_path, target_path):
                    if not check_disk_space(target_dir):
                        logging.warning("磁盘空间不足，暂停复制")
                        return total_copied, total_errors
                    # 执行复制（保留元数据）
                    shutil.copy2(source_path, target_path)
                    logging.info(f"已复制: {source_path} -> {target_path}")
                    total_copied += 1
                else:
                    logging.debug(f"跳过（无需更新）: {file_name}")
            except Exception as e:
                logging.error(f"复制失败 {source_path}: {e}")
                total_errors += 1
    # 可选：清理目标目录中在源目录已不存在的文件（删除），谨慎使用
    # clean_obsolete_files(source_dir, target_dir)
    return total_copied, total_errors
def clean_obsolete_files(source_dir, target_dir):
    """删除目标目录中源目录已不存在的文件（需谨慎启用）"""
    for root, dirs, files in os.walk(target_dir):
        relative_path = os.path.relpath(root, target_dir)
        source_root = os.path.join(source_dir, relative_path) if relative_path != "." else source_dir
        # 如果源对应子目录不存在，则考虑删除整个目录
        if not os.path.exists(source_root):
            logging.warning(f"源目录不存在，删除整个目标目录: {root}")
            shutil.rmtree(root, ignore_errors=True)
            continue
        for file_name in files:
            target_path = os.path.join(root, file_name)
            source_path = os.path.join(source_root, file_name)
            if not os.path.exists(source_path):
                try:
                    os.remove(target_path)
                    logging.info(f"已删除多余文件: {target_path}")
                except Exception as e:
                    logging.error(f"删除文件失败 {target_path}: {e}")
def main_loop():
    """主循环：定时执行同步"""
    logging.info("=== 文件定时转存脚本启动 ===")
    logging.info(f"源目录: {SOURCE_DIR}")
    logging.info(f"目标目录: {TARGET_DIR}")
    logging.info(f"检查间隔: {INTERVAL_SECONDS} 秒")
    logging.info(f"MD5校验: {'开启' if USE_MD5 else '关闭（使用大小+时间）'}")
    while True:
        logging.info("开始新一轮同步...")
        copied, errors = sync_files(SOURCE_DIR, TARGET_DIR)
        logging.info(f"本轮同步完成：复制 {copied} 个文件，错误 {errors} 个")
        # 等待下次执行
        logging.debug(f"等待 {INTERVAL_SECONDS} 秒后继续...")
        time.sleep(INTERVAL_SECONDS)
def run_once():
    """单次执行同步（用于cron或手动触发）"""
    logging.info("执行单次同步...")
    copied, errors = sync_files(SOURCE_DIR, TARGET_DIR)
    logging.info(f"同步结束：复制 {copied} 个文件，错误 {errors} 个")
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="定时转存文件到外部硬盘")
    parser.add_argument("--once", action="store_true", help="仅运行一次（不循环）")
    parser.add_argument("--source", help="覆盖默认源目录")
    parser.add_argument("--target", help="覆盖默认目标目录")
    parser.add_argument("--interval", type=int, help="覆盖默认检查间隔（秒）")
    args = parser.parse_args()
    # 命令行参数覆盖默认配置
    if args.source:
        SOURCE_DIR = args.source
    if args.target:
        TARGET_DIR = args.target
    if args.interval:
        INTERVAL_SECONDS = args.interval
    # 检查外部硬盘是否挂载（简单检查目标路径是否存在）
    if not os.path.exists(TARGET_DIR):
        logging.error(f"目标目录不存在，请检查外部硬盘是否已挂载: {TARGET_DIR}")
        sys.exit(1)
    if args.once:
        run_once()
    else:
        try:
            main_loop()
        except KeyboardInterrupt:
            logging.info("用户中断，脚本退出")
            sys.exit(0)
        except Exception as e:
            logging.exception(f"未预期的错误: {e}")
            sys.exit(1)

脚本使用说明

配置脚本

在脚本开头的 配置区 修改以下参数：

  • SOURCE_DIR：您需要备份的源文件夹路径。
  • TARGET_DIR：外部硬盘的挂载路径。
  • INTERVAL_SECONDS：检查间隔（秒），如 3600 为每小时。
  • LOG_FILE：日志保存位置。
  • CHECK_DISK_SPACE：是否在复制前检查硬盘剩余空间。
  • MIN_FREE_GB：最小剩余空间阈值（GB）。
  • USE_MD5：是否使用MD5精确校验（较慢但更准确，否则用文件大小+时间）。

运行方法

直接运行（带循环）

python3 backup_script.py

仅运行一次（适用于cron）

python3 backup_script.py --once

命令行覆盖配置

python3 backup_script.py --source /my/data --target /mnt/backup --interval 7200

设置定时任务（推荐）

Linux/macOS (cron)：

crontab -e
# 添加以下行：每天凌晨3点执行
0 3 * * * /usr/bin/python3 /path/to/backup_script.py --once

Windows (任务计划程序)：

  1. 打开“任务计划程序”
  2. 创建基本任务，设置触发器（如每天）
  3. 操作选择“启动程序”，程序为 python.exe，参数为 D:\scripts\backup_script.py --once

注意事项

  1. 硬件依赖：确保外部硬盘在脚本运行时已正确挂载，若用 --once 模式配合cron,建议在cron命令前加挂载检查。
  2. 权限问题：脚本需要对源目录有读权限，对目标目录有写权限，如遇权限错误，请使用 chmod 或 sudo。
  3. 文件冲突：默认只复制源目录比目标目录新的文件，若需双向同步或删除目标目录中已失效的文件，可启用 clean_obsolete_files 函数（默认注释）。
  4. 日志查看：运行日志会同时输出到控制台和 sync_log.txt,方便排查问题。

您只需修改配置区参数并设置定时任务,即可实现自动化备份。

上一篇如何用脚本生成简单思维导图

下一篇怎么用脚本获取内部地址范围

相关文章

抱歉，评论功能暂时关闭!