脚本概述
- 功能描述：
- 将 macOS 构建机指定目录进行“项目级”归档备份，支持每周全量、其余日增量。
- 备份输出为 per-project 的 .tar.gz 文件，生成并校验 sha256 摘要。
- 支持保留最近 7 天快照并自动清理超期。
- 支持按日期恢复，支持 dry-run 模式与权限保留；增量恢复包含删除变更（按 deletions 列表）。
- 备份前后支持执行构建 Agent 停止/恢复钩子。
- 通过环境变量读取 SMB 凭证，自动挂载/卸载远端共享。
- 提供失败重试（最多重试 2 次）、日志记录（/var/log/op_backup.log），可安装为 launchd 定时任务（每日 02:30）。
- 适用环境：
- 操作系统：macOS 12+（Monterey 及以上）
- 需要 Python 3.8+ 环境
- SMB 共享可达且可挂载
- 执行权限：
- 建议以 root/sudo 运行（写 /var/log、挂载/卸载、权限还原等需要高权限）
完整代码
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
op_backup.py - macOS 构建机增量/全量项目级备份与恢复工具
主要特性：
- 每周一次全量（自动判断距上次全量 >= 6 天或 --force-full），其余为增量
- 备份对象：/Users/build/Projects 下每个项目目录（一级子目录）与 /Users/build/.ssh
- 排除：**/DerivedData/**, **/Library/Caches/**, **/*.tmp, **/node_modules/**
- 备份输出：per-project tar.gz；写入 sha256 并立即校验
- 快照保留：最近 N(7) 天
- 恢复：指定日期（YYYYMMDD）；dry-run；可保留权限与时间戳；处理删除集
- SMB 凭证：env BACKUP_USER/BACKUP_PASS；通过 AppleScript 安全地从 stdin 传参挂载
- 钩子：备份前停止 Agent，完成后恢复（支持 --agent-label 或自定义命令）
- 日志：/var/log/op_backup.log；失败重试 2 次（总 3 次尝试）
- 定时：支持安装 launchd plist（每日 02:30）
安全说明：
- 不在脚本中硬编码任何敏感信息；凭证从环境变量或外部 env 文件加载
- 删除与清理仅在目标备份根目录下进行，路径校验防误删
"""
import argparse
import datetime as dt
import fnmatch
import hashlib
import json
import logging
import os
import pathlib
import platform
import shutil
import socket
import subprocess
import sys
import tarfile
import tempfile
import time
from typing import Dict, List, Tuple, Iterable, Optional
# =========================
# 全局默认配置（可通过参数覆盖）
# =========================
DEFAULT_SOURCES = [
"/Users/build/Projects",
"/Users/build/.ssh",
]
DEFAULT_EXCLUDES = [
"**/DerivedData/**",
"**/Library/Caches/**",
"**/*.tmp",
"**/node_modules/**",
]
DEFAULT_SMB_URL = "smb://10.0.30.20/backup/build"
DEFAULT_MOUNT_POINT = "/Volumes/build_backup"
DEFAULT_LOG_PATH = "/var/log/op_backup.log"
DEFAULT_RETENTION_DAYS = 7
DEFAULT_FULL_INTERVAL_DAYS = 7 # 每周一次全量（≥6天未全量则全量）
DEFAULT_RETRIES = 2 # 失败重试次数
SCRIPT_LABEL = "com.ops.backup"
# =========================
# 日志
# =========================
def setup_logger(log_path: str) -> logging.Logger:
logger = logging.getLogger("op_backup")
logger.setLevel(logging.INFO)
fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
# 文件日志
try:
fh = logging.FileHandler(log_path)
fh.setFormatter(fmt)
fh.setLevel(logging.INFO)
logger.addHandler(fh)
except Exception:
# 回退到 stdout
sh = logging.StreamHandler(sys.stdout)
sh.setFormatter(fmt)
logger.addHandler(sh)
logger.warning("无法写入日志文件 %s，已回退到标准输出", log_path)
return logger
logger = setup_logger(DEFAULT_LOG_PATH)
# =========================
# 工具函数
# =========================
def retry(max_retries: int, pause_seconds: float = 3.0):
def deco(func):
def wrapper(*args, **kwargs):
attempt = 0
last_exc = None
while attempt <= max_retries:
try:
return func(*args, **kwargs)
except Exception as e:
last_exc = e
attempt += 1
if attempt > max_retries:
logger.error("函数 %s 最终失败：%s", func.__name__, e)
raise
logger.warning("函数 %s 失败（第 %d 次），%ss 后重试... 错误：%s",
func.__name__, attempt, pause_seconds, e)
time.sleep(pause_seconds)
return wrapper
return deco
def load_env_file(env_file: Optional[str]):
if not env_file:
return
p = pathlib.Path(env_file)
if not p.exists():
raise FileNotFoundError(f"env 文件不存在：{env_file}")
for line in p.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
k, v = line.split("=", 1)
os.environ[k.strip()] = v.strip()
def ensure_dir(path: pathlib.Path):
path.mkdir(parents=True, exist_ok=True)
def sha256_file(path: pathlib.Path, chunk: int = 2 * 1024 * 1024) -> str:
h = hashlib.sha256()
with path.open("rb") as f:
while True:
b = f.read(chunk)
if not b:
break
h.update(b)
return h.hexdigest()
def is_mounted(mount_point: str) -> bool:
try:
out = subprocess.check_output(["mount"], text=True)
for line in out.splitlines():
if f" on {mount_point} " in line and "smbfs" in line:
return True
except Exception:
pass
return False
def safe_join(base: pathlib.Path, rel: str) -> pathlib.Path:
# 防止 路径穿越
final = (base / rel).resolve()
if not str(final).startswith(str(base.resolve())):
raise ValueError(f"越界路径：{rel}")
return final
def date_str(d: Optional[dt.date] = None) -> str:
d = d or dt.date.today()
return d.strftime("%Y%m%d")
def list_subdirs(path: pathlib.Path) -> List[pathlib.Path]:
return [p for p in path.iterdir() if p.is_dir()]
def parse_date_dir_name(name: str) -> Optional[dt.date]:
try:
return dt.datetime.strptime(name, "%Y%m%d").date()
except Exception:
return None
def now_local() -> dt.datetime:
return dt.datetime.now()
def hostname() -> str:
try:
return socket.gethostname()
except Exception:
return platform.node() or "unknown-host"
# =========================
# 排除规则
# =========================
def normalize_rel(path: pathlib.Path, base: pathlib.Path) -> str:
rel = path.relative_to(base).as_posix()
return rel if rel != "." else ""
def should_exclude(rel_path: str, exclude_patterns: List[str]) -> bool:
# rel_path 使用 posix 风格
if rel_path is None:
return False
rel = rel_path if rel_path else ""
# 目录以 / 结尾，以便 **/dir/** 命中
candidates = {rel, rel + "/" if not rel.endswith("/") else rel}
for pat in exclude_patterns:
# 统一使用 posix 风格
if any(fnmatch.fnmatch(c, pat) for c in candidates):
return True
return False
# =========================
# SMB 挂载/卸载
# =========================
class SMBMounter:
def __init__(self, smb_url: str, mount_point: str):
self.smb_url = smb_url
self.mount_point = mount_point
def mount(self, user: Optional[str], password: Optional[str]):
mp = pathlib.Path(self.mount_point)
ensure_dir(mp)
if is_mounted(self.mount_point):
logger.info("SMB 已挂载：%s", self.mount_point)
return
if not user or not password:
raise RuntimeError("挂载 SMB 需要 BACKUP_USER/BACKUP_PASS 环境变量或 --env-file 提供")
# 通过 AppleScript 从 stdin 传参，避免密码出现在命令行
applescript = f'''
on run
mount volume "{self.smb_url}" as user name "{user}" with password "{password}"
end run
'''
try:
logger.info("尝试挂载 SMB：%s -> %s", self.smb_url, self.mount_point)
subprocess.check_call(["osascript", "-"], input=applescript.encode("utf-8"))
# 等待 Finder 挂载到 /Volumes/<ShareName>，然后将其绑定/链接到指定 mount_point
time.sleep(2.0)
except subprocess.CalledProcessError as e:
raise RuntimeError(f"SMB 挂载失败：{e}")
# macOS Finder 通常挂载到 /Volumes/<ShareName>，但我们需要稳定路径：
# 如果用户指定了自定义 mount_point，尝试将其软链接到 Finder 挂载点。
if not is_mounted(self.mount_point):
# 查找最新 smbfs 挂载点
out = subprocess.check_output(["mount"], text=True)
mounted = None
for line in out.splitlines():
if "smbfs on /Volumes/" in line:
mounted = line.split(" on ")[1].split(" (")[0].strip()
break
if mounted:
try:
if mp.exists() and not mp.is_dir():
raise RuntimeError(f"mount_point 不是目录：{mp}")
if mp.exists():
# 若已有同名目录，清空后重建软链接
if mp.is_symlink() or mp.is_dir():
try:
os.unlink(mp) if mp.is_symlink() else None
except Exception:
pass
ensure_dir(mp.parent)
if not mp.exists():
os.symlink(mounted, mp)
logger.info("SMB 已通过符号链接映射到 %s -> %s", mp, mounted)
except Exception as e:
logger.warning("创建符号链接失败：%s", e)
if not is_mounted(self.mount_point):
logger.warning("无法确认 %s 已挂载，但 Finder 可能已挂载成功。", self.mount_point)
def unmount(self):
if not is_mounted(self.mount_point):
return
try:
logger.info("卸载 SMB：%s", self.mount_point)
subprocess.check_call(["diskutil", "unmount", "force", self.mount_point])
except Exception as e:
logger.warning("卸载失败（忽略）：%s", e)
# =========================
# 备份/恢复核心
# =========================
class BackupManager:
def __init__(
self,
sources: List[str],
exclude_patterns: List[str],
mount_point: str,
smb_url: str,
retention_days: int,
full_interval_days: int,
agent_label: Optional[str] = None,
pre_stop_cmd: Optional[str] = None,
post_start_cmd: Optional[str] = None,
host_name: Optional[str] = None,
):
self.sources = [pathlib.Path(s).resolve() for s in sources]
self.exclude_patterns = exclude_patterns
self.mount_point = pathlib.Path(mount_point)
self.smb_url = smb_url
self.retention_days = retention_days
self.full_interval_days = full_interval_days
self.agent_label = agent_label
self.pre_stop_cmd = pre_stop_cmd
self.post_start_cmd = post_start_cmd
self.host_name = host_name or hostname()
# 备份目录结构：
# /Volumes/build_backup/<host>/
# snapshots/YYYYMMDD/
# TYPE: FULL | INCR
# <project>-<type>-YYYYMMDD.tar.gz
# <project>-<type>-YYYYMMDD.tar.gz.sha256
# deletions-<project>-YYYYMMDD.txt (增量时可能存在)
# meta/manifests/<project>.json （最新清单）
self.base_dir = self.mount_point / self.host_name
self.snapshots_dir = self.base_dir / "snapshots"
self.meta_dir = self.base_dir / "meta" / "manifests"
def _project_list(self) -> List[Tuple[str, pathlib.Path]]:
projects: List[Tuple[str, pathlib.Path]] = []
for src in self.sources:
if not src.exists():
logger.warning("源目录不存在：%s（跳过）", src)
continue
if src.name == ".ssh" and src.is_dir():
projects.append(("ssh", src))
continue
if src.name == "Projects" and src.is_dir():
for child in sorted(src.iterdir()):
if child.is_dir():
projects.append((child.name, child))
else:
# 其它单独目录作为一个项目
projects.append((src.name, src))
return projects
def _load_manifest(self, project: str) -> Dict:
p = self.meta_dir / f"{project}.json"
if not p.exists():
return {"project": project, "files": {}, "timestamp": ""}
try:
return json.loads(p.read_text(encoding="utf-8"))
except Exception:
logger.warning("读取 manifest 失败：%s（重置为空）", p)
return {"project": project, "files": {}, "timestamp": ""}
def _save_manifest(self, project: str, manifest: Dict):
ensure_dir(self.meta_dir)
p = self.meta_dir / f"{project}.json"
p.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
def _walk_files(self, base: pathlib.Path) -> Iterable[Tuple[str, os.stat_result]]:
# 遍历文件并应用排除规则；返回 (rel_path, stat)
for root, dirs, files in os.walk(base, followlinks=False):
root_p = pathlib.Path(root)
# 目录级排除（修改 dirs 就能剪枝）
keep_dirs = []
for d in dirs:
rel = normalize_rel(root_p / d, base)
if should_exclude(rel + "/", self.exclude_patterns):
continue
keep_dirs.append(d)
dirs[:] = keep_dirs
# 文件
for f in files:
p = root_p / f
rel = normalize_rel(p, base)
if should_exclude(rel, self.exclude_patterns):
continue
try:
st = p.lstat()
except FileNotFoundError:
continue
yield rel, st
def _scan_current_state(self, base: pathlib.Path) -> Dict[str, Dict]:
state: Dict[str, Dict] = {}
for rel, st in self._walk_files(base):
state[rel] = {"mtime": st.st_mtime, "size": st.st_size}
return state
def _compute_delta(
self, prev: Dict[str, Dict], current: Dict[str, Dict]
) -> Tuple[List[str], List[str]]:
changed: List[str] = []
for rel, meta in current.items():
pv = prev.get(rel)
if not pv or pv.get("mtime") != meta["mtime"] or pv.get("size") != meta["size"]:
changed.append(rel)
deletions: List[str] = []
for rel in prev:
if rel not in current:
# 只有当此前文件不属于排除时才记录删除（避免因排除导致虚假删除）
if not should_exclude(rel, self.exclude_patterns):
deletions.append(rel)
return sorted(changed), sorted(deletions)
def _need_full(self) -> bool:
# 若最近一次 FULL 距今 >= full_interval_days-1 则做全量；否则增量
if not self.snapshots_dir.exists():
return True
latest_full_date: Optional[dt.date] = None
for d in list_subdirs(self.snapshots_dir):
ds = parse_date_dir_name(d.name)
if not ds:
continue
type_file = d / "TYPE"
if type_file.exists():
try:
t = type_file.read_text(encoding="utf-8").strip().upper()
if t == "FULL":
if (latest_full_date is None) or (ds > latest_full_date):
latest_full_date = ds
except Exception:
continue
if latest_full_date is None:
return True
delta_days = (dt.date.today() - latest_full_date).days
return delta_days >= (self.full_interval_days - 1)
def _snapshot_dir(self, day: dt.date) -> pathlib.Path:
return self.snapshots_dir / day.strftime("%Y%m%d")
def _write_type_marker(self, snap_dir: pathlib.Path, snap_type: str):
(snap_dir / "TYPE").write_text(snap_type.upper(), encoding="utf-8")
def _archive_project(
self,
project: str,
base_dir: pathlib.Path,
rel_list: List[str],
snap_dir: pathlib.Path,
snap_type: str,
day: dt.date,
) -> Tuple[pathlib.Path, pathlib.Path]:
ensure_dir(snap_dir)
tar_name = f"{project}-{snap_type.lower()}-{day.strftime('%Y%m%d')}.tar.gz"
tar_path = snap_dir / tar_name
sha_path = snap_dir / f"{tar_name}.sha256"
logger.info("打包 %s（%s，%d 个条目）-> %s", project, snap_type, len(rel_list), tar_path)
# 创建 tar.gz
with tarfile.open(tar_path, mode="w:gz", format=tarfile.PAX_FORMAT, compresslevel=6) as tf:
for rel in rel_list:
abs_path = safe_join(base_dir, rel)
# 加入档案，保留相对路径结构
try:
tf.add(abs_path, arcname=rel, recursive=False)
except FileNotFoundError:
# 避免并发写导致的短暂缺失
logger.warning("文件消失（忽略）：%s", abs_path)
# 计算 sha256 并写入
digest = sha256_file(tar_path)
sha_path.write_text(f"{digest} {tar_name}\n", encoding="utf-8")
# 立刻重新校验一次
digest2 = sha256_file(tar_path)
if digest != digest2:
raise RuntimeError(f"SHA256 校验不一致：{tar_path}")
logger.info("归档与校验完成：%s (sha256=%s)", tar_path.name, digest)
return tar_path, sha_path
def _write_deletions(
self, project: str, deletions: List[str], snap_dir: pathlib.Path, day: dt.date
) -> Optional[pathlib.Path]:
if not deletions:
return None
path = snap_dir / f"deletions-{project}-{day.strftime('%Y%m%d')}.txt"
path.write_text("\n".join(deletions) + "\n", encoding="utf-8")
logger.info("记录删除集：%s（%d 条）", path.name, len(deletions))
return path
def _run_hook_stop(self):
if self.pre_stop_cmd:
logger.info("执行预停止命令：%s", self.pre_stop_cmd)
subprocess.check_call(self.pre_stop_cmd, shell=True)
elif self.agent_label:
# 尝试停止构建 Agent（可能需 root）
logger.info("停止构建 Agent（launchctl stop）：%s", self.agent_label)
subprocess.call(["launchctl", "stop", self.agent_label])
else:
logger.info("未配置停止钩子，跳过")
def _run_hook_start(self):
if self.post_start_cmd:
logger.info("执行恢复命令：%s", self.post_start_cmd)
subprocess.check_call(self.post_start_cmd, shell=True)
elif self.agent_label:
logger.info("恢复构建 Agent（launchctl start）：%s", self.agent_label)
subprocess.call(["launchctl", "start", self.agent_label])
else:
logger.info("未配置恢复钩子，跳过")
def _cleanup_retention(self):
if not self.snapshots_dir.exists():
return
entries = list_subdirs(self.snapshots_dir)
items: List[Tuple[dt.date, pathlib.Path]] = []
for d in entries:
ds = parse_date_dir_name(d.name)
if ds:
items.append((ds, d))
items.sort()
# 保留最近 retention_days 天
keep_after = (dt.date.today() - dt.timedelta(days=self.retention_days - 1))
removed = 0
for ds, d in items:
if ds < keep_after:
logger.info("清理超期快照：%s", d)
shutil.rmtree(d, ignore_errors=True)
removed += 1
if removed:
logger.info("保留策略执行完成，删除 %d 个快照目录", removed)
@retry(max_retries=DEFAULT_RETRIES, pause_seconds=3.0)
def backup_once(self, force_full: bool = False):
# 确保目标目录存在
ensure_dir(self.snapshots_dir)
ensure_dir(self.meta_dir)
# 挂载 SMB
mounter = SMBMounter(self.smb_url, str(self.mount_point))
mounter.mount(os.environ.get("BACKUP_USER"), os.environ.get("BACKUP_PASS"))
today = dt.date.today()
snap_dir = self._snapshot_dir(today)
ensure_dir(snap_dir)
snap_type = "FULL" if force_full or self._need_full() else "INCR"
self._write_type_marker(snap_dir, snap_type)
logger.info("开始备份（%s）：%s", snap_type, today.strftime("%Y-%m-%d"))
# 钩子：停止 Agent
try:
self._run_hook_stop()
except Exception as e:
logger.warning("停止钩子执行失败（继续）：%s", e)
try:
# 遍历项目
for project, base in self._project_list():
prev_manifest = self._load_manifest(project)
current_state = self._scan_current_state(base)
if snap_type == "FULL":
changed = sorted(current_state.keys())
deletions: List[str] = []
else:
changed, deletions = self._compute_delta(prev_manifest.get("files", {}), current_state)
# 即便 changed 为空，也创建一个最小 tar 以标记项目（可选）
tar_path, sha_path = self._archive_project(project, base, changed, snap_dir, snap_type, today)
# 写删除集（仅增量）
self._write_deletions(project, deletions, snap_dir, today)
# 更新 manifest（成为下一次基线）
new_manifest = {
"project": project,
"timestamp": now_local().isoformat(),
"files": current_state,
}
self._save_manifest(project, new_manifest)
# 快照中也保存一份项目 manifest（可选，辅助定位）
(snap_dir / f"{project}-manifest.json").write_text(
json.dumps(new_manifest, ensure_ascii=False, indent=2),
encoding="utf-8",
)
finally:
# 钩子：恢复 Agent
try:
self._run_hook_start()
except Exception as e:
logger.warning("恢复钩子执行失败：%s", e)
# 不强制卸载，留给系统/调度使用；如需可以在外层调用 unmount
# 保留策略清理
self._cleanup_retention()
logger.info("备份完成（%s）：%s", snap_type, today.strftime("%Y-%m-%d"))
def list_snapshots(self) -> List[Tuple[str, str]]:
res: List[Tuple[str, str]] = []
if not self.snapshots_dir.exists():
return res
for d in sorted(list_subdirs(self.snapshots_dir)):
t = (d / "TYPE").read_text(encoding="utf-8").strip().upper() if (d / "TYPE").exists() else "UNKNOWN"
res.append((d.name, t))
return res
def _locate_restore_chain(self, target_date: dt.date) -> List[Tuple[dt.date, pathlib.Path, str]]:
# 找到 <= target_date 的快照链：最近的 FULL 及之后直到 target_date 的所有快照
items: List[Tuple[dt.date, pathlib.Path, str]] = []
for d in list_subdirs(self.snapshots_dir):
ds = parse_date_dir_name(d.name)
if not ds or ds > target_date:
continue
t = (d / "TYPE").read_text(encoding="utf-8").strip().upper() if (d / "TYPE").exists() else "UNKNOWN"
items.append((ds, d, t))
items.sort()
# 找最近的 FULL
last_full_idx = -1
for i in range(len(items) - 1, -1, -1):
if items[i][2] == "FULL":
last_full_idx = i
break
if last_full_idx == -1:
raise RuntimeError("未找到可用的 FULL 快照")
chain = items[last_full_idx:]
return chain
def _list_project_archives(self, snap_dir: pathlib.Path, project: str) -> List[pathlib.Path]:
return sorted(snap_dir.glob(f"{project}-*.tar.gz"))
def _read_deletions(self, snap_dir: pathlib.Path, project: str) -> List[str]:
res: List[str] = []
for p in snap_dir.glob(f"deletions-{project}-*.txt"):
try:
res += [line.strip() for line in p.read_text(encoding="utf-8").splitlines() if line.strip()]
except Exception:
continue
return res
def _safe_extract(self, tf: tarfile.TarFile, dest: pathlib.Path, preserve_perms: bool):
# 安全解压，防止路径穿越
for m in tf.getmembers():
# 仅允许相对路径
if m.name.startswith("/") or ".." in pathlib.PurePosixPath(m.name).parts:
logger.warning("检测到可疑条目（忽略）：%s", m.name)
continue
target = dest / m.name
target_resolved = target.resolve().absolute()
if not str(target_resolved).startswith(str(dest.resolve().absolute())):
logger.warning("越界解压（忽略）：%s", m.name)
continue
# 解压
tf.extract(m, path=dest)
if preserve_perms and hasattr(os, "chmod"):
try:
if m.mode is not None:
os.chmod(target_resolved, m.mode)
# 尝试保留 mtime
if m.mtime is not None:
os.utime(target_resolved, (m.mtime, m.mtime), follow_symlinks=False)
except Exception:
pass
@retry(max_retries=DEFAULT_RETRIES, pause_seconds=3.0)
def restore(
self,
target_yyyymmdd: str,
dest_override: Optional[str],
projects: Optional[List[str]],
dry_run: bool,
preserve_perms: bool = True,
):
mounter = SMBMounter(self.smb_url, str(self.mount_point))
mounter.mount(os.environ.get("BACKUP_USER"), os.environ.get("BACKUP_PASS"))
target_date = dt.datetime.strptime(target_yyyymmdd, "%Y%m%d").date()
chain = self._locate_restore_chain(target_date)
logger.info("恢复目标日期：%s；包含 %d 个快照", target_yyyymmdd, len(chain))
# 选择项目
all_projects = dict(self._project_list())
selected: List[Tuple[str, pathlib.Path]] = []
if projects:
for p in projects:
base = all_projects.get(p)
if not base:
raise RuntimeError(f"未知项目：{p}")
selected.append((p, base))
else:
selected = list(all_projects.items())
for project, base in selected:
dest_base = pathlib.Path(dest_override).resolve() if dest_override else base
ensure_dir(dest_base)
logger.info("恢复项目：%s -> %s", project, dest_base)
# 依序应用 FULL + INCR
archives: List[pathlib.Path] = []
deletions_accum: List[str] = []
for ds, d, t in chain:
for arc in self._list_project_archives(d, project):
archives.append(arc)
# 收集删除集
deletions_accum += self._read_deletions(d, project)
if dry_run:
logger.info("[dry-run] 将解压 %d 个归档：%s", len(archives), ", ".join(a.name for a in archives))
if deletions_accum:
logger.info("[dry-run] 将删除 %d 个文件（若存在）", len(deletions_accum))
continue
# 解压归档
for arc in archives:
logger.info("解压：%s", arc)
with tarfile.open(arc, "r:gz") as tf:
self._safe_extract(tf, dest_base, preserve_perms)
# 应用删除集
unique_deletions = sorted(set(deletions_accum))
removed = 0
for rel in unique_deletions:
target = dest_base / rel
try:
if target.is_file() or target.is_symlink():
target.unlink(missing_ok=True)
removed += 1
elif target.is_dir():
shutil.rmtree(target, ignore_errors=True)
removed += 1
except Exception:
pass
if unique_deletions:
logger.info("应用删除集完成：删除 %d", removed)
logger.info("恢复完成：%s", target_yyyymmdd)
def verify(self, target_yyyymmdd: Optional[str]):
mounter = SMBMounter(self.smb_url, str(self.mount_point))
mounter.mount(os.environ.get("BACKUP_USER"), os.environ.get("BACKUP_PASS"))
# 选择快照
if target_yyyymmdd:
snap_dir = self.snapshots_dir / target_yyyymmdd
if not snap_dir.exists():
raise RuntimeError(f"快照不存在：{snap_dir}")
else:
snapshots = sorted(list_subdirs(self.snapshots_dir))
if not snapshots:
raise RuntimeError("无可用快照")
snap_dir = snapshots[-1]
logger.info("校验快照：%s", snap_dir.name)
for sha in snap_dir.glob("*.sha256"):
line = sha.read_text(encoding="utf-8").strip()
if " " in line:
expect, name = line.split(" ", 1)
else:
parts = line.split()
expect = parts[0]
name = parts[-1]
arc = snap_dir / name
if not arc.exists():
logger.error("缺少归档：%s", arc)
continue
got = sha256_file(arc)
if got != expect:
raise RuntimeError(f"校验失败：{arc} （期望 {expect}，实际 {got}）")
logger.info("OK: %s", name)
logger.info("校验完成：%s", snap_dir.name)
def install_schedule(self, label: str, run_time: str, env_file: Optional[str], script_path: str):
# 生成 launchd plist（/Library/LaunchDaemons/）
hh, mm = run_time.split(":")
plist = f"""<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
<key>Label</key><string>{label}</string>
<key>ProgramArguments</key>
<array>
<string>{sys.executable}</string>
<string>{script_path}</string>
<string>backup</string>
</array>
<key>StartCalendarInterval</key>
<dict>
<key>Hour</key><integer>{int(hh)}</integer>
<key>Minute</key><integer>{int(mm)}</integer>
</dict>
<key>RunAtLoad</key><false/>
<key>StandardOutPath</key><string>{DEFAULT_LOG_PATH}</string>
<key>StandardErrorPath</key><string>{DEFAULT_LOG_PATH}</string>
<key>EnvironmentVariables</key>
<dict>
<!-- 若使用 env_file，请在外部管理并由 launchd 的 wrapper 加载；这里通常留空 -->
</dict>
</dict>
</plist>
"""
plist_path = f"/Library/LaunchDaemons/{label}.plist"
pathlib.Path(plist_path).write_text(plist, encoding="utf-8")
logger.info("已写入 launchd plist：%s", plist_path)
# 加载
subprocess.check_call(["launchctl", "load", "-w", plist_path])
logger.info("已加载定时任务（每日 %s）", run_time)
# =========================
# CLI
# =========================
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(description="macOS 项目级增量/全量备份与恢复工具")
sub = p.add_subparsers(dest="cmd", required=True)
common = argparse.ArgumentParser(add_help=False)
common.add_argument("--sources", nargs="+", default=DEFAULT_SOURCES, help="源目录（默认：/Users/build/Projects /Users/build/.ssh）")
common.add_argument("--exclude", nargs="+", default=DEFAULT_EXCLUDES, help="排除模式（支持 ** 通配）")
common.add_argument("--smb-url", default=DEFAULT_SMB_URL, help="SMB 目标，如 smb://10.0.30.20/backup/build")
common.add_argument("--mount-point", default=DEFAULT_MOUNT_POINT, help="挂载点")
common.add_argument("--retention-days", type=int, default=DEFAULT_RETENTION_DAYS, help="保留天数（默认7）")
common.add_argument("--full-interval-days", type=int, default=DEFAULT_FULL_INTERVAL_DAYS, help="全量间隔天数（默认7）")
common.add_argument("--agent-label", default=os.environ.get("BUILD_AGENT_LABEL"), help="构建Agent的 launchd Label（可选）")
common.add_argument("--pre-stop-cmd", default=os.environ.get("HOOK_STOP"), help="备份前停止命令（可选）")
common.add_argument("--post-start-cmd", default=os.environ.get("HOOK_START"), help="备份后恢复命令（可选）")
common.add_argument("--env-file", default=os.environ.get("OP_BACKUP_ENV_FILE"), help="包含 BACKUP_USER/BACKUP_PASS 的安全 env 文件路径")
common.add_argument("--host-name", default=None, help="覆盖主机名，用于备份路径分隔")
p_backup = sub.add_parser("backup", parents=[common], help="执行一次备份")
p_backup.add_argument("--force-full", action="store_true", help="强制全量备份")
p_backup.add_argument("--no-mount", action="store_true", help="不自动挂载（假定已挂载）")
p_restore = sub.add_parser("restore", parents=[common], help="恢复到指定日期")
p_restore.add_argument("--date", required=True, help="目标日期 YYYYMMDD")
p_restore.add_argument("--project", nargs="*", help="指定项目（默认全部）")
p_restore.add_argument("--dest", default=None, help="恢复目标根目录（默认恢复到原路径）")
p_restore.add_argument("--dry-run", action="store_true", help="仅预演，不落盘")
p_restore.add_argument("--no-preserve-perms", action="store_true", help="不保留权限/时间戳")
p_list = sub.add_parser("list", parents=[common], help="列出快照")
p_verify = sub.add_parser("verify", parents=[common], help="校验快照 sha256")
p_verify.add_argument("--date", default=None, help="目标日期 YYYYMMDD（默认最新）")
p_sched = sub.add_parser("install-schedule", parents=[common], help="安装每日 02:30 定时任务（launchd）")
p_sched.add_argument("--time", default="02:30", help="HH:MM，默认 02:30")
p_sched.add_argument("--label", default=SCRIPT_LABEL, help="launchd Label")
p_sched.add_argument("--script-path", default=os.path.abspath(sys.argv[0]), help="脚本绝对路径")
return p
def main():
parser = build_parser()
args = parser.parse_args()
# 读取 env 文件（如提供）
if getattr(args, "env_file", None):
load_env_file(args.env_file)
mgr = BackupManager(
sources=args.sources,
exclude_patterns=args.exclude,
mount_point=args.mount_point,
smb_url=args.smb_url,
retention_days=args.retention_days,
full_interval_days=args.full_interval_days,
agent_label=args.agent_label,
pre_stop_cmd=args.pre_stop_cmd,
post_start_cmd=args.post_start_cmd,
host_name=args.host_name,
)
if args.cmd == "backup":
mgr.backup_once(force_full=args.force_full)
elif args.cmd == "restore":
mgr.restore(
target_yyyymmdd=args.date,
dest_override=args.dest,
projects=args.project,
dry_run=args.dry_run,
preserve_perms=not args.no_preserve_perms,
)
elif args.cmd == "list":
snaps = mgr.list_snapshots()
for d, t in snaps:
print(f"{d} {t}")
elif args.cmd == "verify":
mgr.verify(args.date)
elif args.cmd == "install-schedule":
mgr.install_schedule(args.label, args.time, args.env_file, args.script_path)
else:
parser.print_help()
if __name__ == "__main__":
main()
参数说明
- --sources
- 含义：需要备份的源目录列表
- 默认：/Users/build/Projects /Users/build/.ssh
- --exclude
- 含义：备份/遍历时的排除通配模式（支持 **）
- 默认：/DerivedData/, /Library/Caches/, **/*.tmp, /node_modules/
- --smb-url
- --mount-point
- 含义：本地挂载点（脚本会尝试将 Finder 挂载映射到此路径）
- 默认：/Volumes/build_backup
- --retention-days
- --full-interval-days
- 含义：全量间隔天数，距最近全量 >= (N-1) 天则当日执行全量
- 默认：7（约每周一次全量）
- --agent-label
- 含义：备份前后通过 launchctl stop/start 控制的构建 Agent 标签（可选）
- 默认：读取环境变量 BUILD_AGENT_LABEL
- --pre-stop-cmd / --post-start-cmd
- 含义：自定义停止/恢复命令（可选，优先于 --agent-label）
- 默认：读取 HOOK_STOP / HOOK_START 环境变量
- --env-file
- 含义：包含 BACKUP_USER/BACKUP_PASS 的外部安全 env 文件路径（避免在 plist 中明文）
- 默认：读取 OP_BACKUP_ENV_FILE 环境变量
- --host-name
- 含义：覆盖用于备份路径分隔的主机名（默认取当前主机名）
- backup 子命令
- restore 子命令
- --date：恢复目标日期（YYYYMMDD）
- --project：指定项目名（默认恢复全部）
- --dest：恢复目标根目录（默认恢复到原目录）
- --dry-run：只输出将执行的操作，不实际写入
- --no-preserve-perms：不保留文件权限与时间戳
- verify 子命令
- install-schedule 子命令
- --time：每日运行时间（HH:MM）
- --label：launchd 任务 Label
- --script-path：脚本绝对路径
环境变量：
- BACKUP_USER / BACKUP_PASS：SMB 认证账号与密码
- BUILD_AGENT_LABEL：构建 Agent 的 launchd Label（可选）
- HOOK_STOP / HOOK_START：自定义钩子命令（可选）
- OP_BACKUP_ENV_FILE：env 文件路径（可选）
使用示例
- 手动执行一次备份（自动判定增量/全量）
- 前置：确保已导出 BACKUP_USER/BACKUP_PASS 或提供 --env-file
- 命令：
sudo ./op_backup.py backup --env-file /etc/op_backup.env
- 强制全量备份（例如每周一手动触发）
- 命令：
sudo ./op_backup.py backup --force-full --env-file /etc/op_backup.env
- 恢复到指定日期，先 dry-run 查看动作，再实际恢复
- 预演：
sudo ./op_backup.py restore --date 20250115 --dry-run
- 实际恢复（保留权限），仅恢复某个项目到临时目录：
sudo ./op_backup.py restore --date 20250115 --project MyApp --dest /tmp/restore/MyApp
- 校验最新快照 sha256
- 命令：
sudo ./op_backup.py verify
- 安装每日 02:30 定时任务（launchd）
- 写入 env 文件（权限 600）：
sudo sh -c 'echo "BACKUP_USER=buildsvc" > /etc/op_backup.env'
sudo sh -c 'echo "BACKUP_PASS=********" >> /etc/op_backup.env'
sudo chmod 600 /etc/op_backup.env
- 安装 plist：
sudo OP_BACKUP_ENV_FILE=/etc/op_backup.env ./op_backup.py install-schedule --time 02:30
- 配置构建 Agent 钩子（例如 Jenkins）
- 通过 launchctl Label：
sudo BUILD_AGENT_LABEL=org.jenkins-ci ./op_backup.py backup
- 或自定义命令（优先于 Label）：
sudo HOOK_STOP="brew services stop jenkins" HOOK_START="brew services start jenkins" ./op_backup.py backup
注意事项
- 执行前的准备
- 确保 Python 3.8+ 与管理员权限（sudo）
- 目标 SMB 共享可达，确保 BACKUP_USER/BACKUP_PASS 有写权限
- 建议创建 /etc/op_backup.env 保存凭证（600 权限），通过 --env-file 或 OP_BACKUP_ENV_FILE 使用
- 首次运行会创建挂载点 /Volumes/build_backup 和备份目录结构
- 风险与防范
- 脚本对删除操作严格限制在备份目录（清理策略）和恢复目标根目录（删除集）内，并做路径越界校验
- SMB 挂载使用 AppleScript 从 stdin 传递密码，降低密码出现在进程列表的风险；仍建议使用受限账号
- 恢复含“删除集”时会删除目标目录中相对路径对应文件，建议先 dry-run 或恢复到临时目录验证
- 日志写入 /var/log/op_backup.log，需要 root；如失败将回退控制台输出
- 性能与资源
- 备份使用 tar.gz 压缩，CPU 占用取决于数据量与压缩等级（默认 6）；可调整代码中的 compresslevel
- sha256 校验会对归档进行二次顺序读，磁盘与网络 IO 会增加
- 增量是通过本地状态清单（mtime+size）判定，极端情况下相同 mtime+size 的少量变更可能无法检测，建议关键文件另行校验或开启更严格策略（可拓展为记录哈希）
- 其他
- “每周全量”通过距离最近 FULL 的天数判定（默认 >=6 天）；也可使用 --force-full 随时触发
- 项目识别规则：/Users/build/Projects 下一级目录每个为一个项目；/Users/build/.ssh 作为独立项目“ssh”
- 脚本恢复默认保留权限与时间戳；所有操作均记录日志，便于审计与追踪
- 本脚本不对 Keychain 写入凭证，不持久化敏感信息，符合不硬编码敏感信息的规范
