本地ipynb - 执行日志
[15:51:51] 服务器名 ['oracle-arm', 'oracle-trilium', 'max2-alpine', 'kc-v100']
[15:51:51] ✅ 加载配置文件: vps_backup_ssh_config.json
============================================================
备份服务器: oracle-trilium
============================================================
[15:51:58] 执行备份脚本: ~/backupGhost.sh
[15:51:58] 上传脚本到 ~/backupGhost.sh
[15:52:04] cat 上传成功
[15:52:11] ✅ 脚本已上传: /home/ghost/backupGhost.sh
[15:52:32] 执行命令: sudo -n /home/ghost/backupGhost.sh
[15:55:38] ✅ 远程备份: /home/ghost/ghost_20260829155300.tar.gz
[15:55:38] 本地: D:\fileJob\1606atibm\2105oracle\instance-20210526-1342\bak\ghost_20260829155300.tar.gz
[15:55:45] 文件大小: 613.10 MB
下载 ghost_20260829155300.tar.gz: 100%|██████████| 613M/613M [877kB/s]
[16:07:58] ✅ ghost_20260829155300.tar.gz (613.10 MB)
[16:08:06] ✅ 清理远程：/home/ghost/ghost_20260829155300.tar.gz
[16:08:16] ✅ 清理远程：/home/ghost/backupGhost.sh
本地ipynb - 执行任务
# 备份所有服务器
results = run_backup(config_file=config_json, script_file=script_sh)
# 备份指定服务器
results = run_backup(config_file=config_json, script_file=script_sh, server_name="oracle-trilium")
本地ipynb - 调度代码
#!/usr/bin/env python3
"""
VPS 备份工具 - 核心对象设计
- VPS: 远程服务器对象，负责执行备份脚本
- LocalPC: 本地计算机对象，负责下载和存储备份
使用方式:
from backup_manager import run_backup, list_servers
# 查看服务器列表
servers = list_servers("vps_backup_ssh_config.json")
print(servers)
# 备份所有服务器
results = run_backup(
config_file="vps_backup_ssh_config.json",
script_file="vps_backup_backupGhost.sh"
)
# 备份指定服务器
results = run_backup(
config_file="vps_backup_ssh_config.json",
script_file="vps_backup_backupGhost.sh",
server_name="oracle-arm"
)
"""
import os
import re
import json
import subprocess
import platform
import stat
import time
import tempfile
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Optional, Tuple, Union
from tqdm import tqdm
# ==============================================
# 工具类
# ==============================================
class PathUtils:
"""跨平台路径工具"""
@staticmethod
def to_local(path: str) -> str:
"""转换为本地路径格式"""
if not path:
return path
path = os.path.expanduser(path)
if platform.system() == "Windows":
# WSL: /mnt/d/path -> D:\path
if path.startswith('/mnt/'):
drive = path[5].upper()
rest = path[7:]
return f"{drive}:\\{rest.replace('/', '\\')}"
# Cygwin: /cygdrive/d/path -> D:\path
if path.startswith('/cygdrive/'):
drive = path[10].upper()
rest = path[12:]
return f"{drive}:\\{rest.replace('/', '\\')}"
# 已是 Windows 路径
if re.match(r'^[A-Za-z]:[/\\]', path):
return path.replace('/', '\\')
return str(Path(path)).replace('\\', '/')
@staticmethod
def fix_ssh_key_permission(key_path: str) -> bool:
"""修复 SSH 私钥权限（Windows/Linux）"""
try:
path = Path(key_path)
if not path.exists():
return False
if platform.system() == "Windows":
subprocess.run(["icacls", str(path), "/reset"], capture_output=True, check=False)
subprocess.run(["icacls", str(path), "/inheritance:r"], capture_output=True, check=False)
username = os.environ.get("USERNAME", "UNKNOWN")
subprocess.run(["icacls", str(path), "/grant", f"{username}:F"], capture_output=True, check=False)
return True
else:
os.chmod(path, stat.S_IRUSR | stat.S_IWUSR)
return True
except Exception as e:
log.debug(f"修复权限失败: {e}")
return False
# ==============================================
# Logger
# ==============================================
class Logger:
"""简单日志"""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._init()
return cls._instance
def _init(self):
self._debug = False
def set_debug(self, debug: bool):
self._debug = debug
def debug(self, msg):
if self._debug:
print(f"[DEBUG] {msg}")
def info(self, msg):
print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}")
def ok(self, msg):
print(f"[{datetime.now().strftime('%H:%M:%S')}] ✅ {msg}")
def warn(self, msg):
print(f"[{datetime.now().strftime('%H:%M:%S')}] ⚠️ {msg}")
def err(self, msg):
print(f"[{datetime.now().strftime('%H:%M:%S')}] ❌ {msg}")
def section(self, title):
print(f"\n{'='*60}\n {title}\n{'='*60}")
log = Logger()
# ==============================================
# 配置工具函数
# ==============================================
def list_servers(config_file: str) -> List[str]:
"""查看配置文件中的服务器名称列表"""
config_path = PathUtils.to_local(config_file)
if not Path(config_path).exists():
log.err(f"配置文件不存在: {config_path}")
return []
with open(config_path, 'r', encoding='utf-8') as f:
config_dict = json.load(f)
servers = []
for server in config_dict.get('servers', []):
if server.get('enabled', True):
servers.append(server['name'])
return servers
# ==============================================
# VPS 对象 - 远程服务器
# ==============================================
class VPS:
def __init__(self, config: Dict, script_path: str):
self.server_config = config
self.name = config['name']
self.user = config['user']
self.host = config['host']
self.ssh_key = PathUtils.to_local(config['ssh_key'])
self.remote_script = script_path
self.port = config.get('port', 22)
self.timeout = config.get('timeout', 30)
self.download_timeout = config.get('download_timeout', 7200)
self.use_proxy = config.get('use_proxy', False)
self.proxy = config.get('proxy', {})
PathUtils.fix_ssh_key_permission(self.ssh_key)
self._ssh_args = self._build_ssh_args()
self._backup_file = None
self._remote_script_name = 'backupGhost.sh'
def _build_ssh_args(self) -> List[str]:
args = [
"-x",
"-p", str(self.port),
"-o", "BatchMode=yes",
"-o", "StrictHostKeyChecking=no",
"-o", f"ConnectTimeout={self.timeout}",
"-i", self.ssh_key
]
if self.use_proxy:
proxy_cmd = (
f"{self.proxy.get('bin', '/usr/bin/connect-proxy')} "
f"-{self.proxy.get('type', 'S')} "
f"{self.proxy.get('addr', '127.0.0.1:7890')} %h %p"
)
args.extend(["-o", f"ProxyCommand={proxy_cmd}"])
return args
def _target(self) -> str:
return f"{self.user}@{self.host}"
def _run(self, command: str) -> Tuple[int, str, str]:
cmd = ["ssh"] + self._ssh_args + [self._target(), command]
log.debug(f"执行: {' '.join(cmd)}")
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=self.timeout + 60
)
return result.returncode, result.stdout.strip(), result.stderr.strip()
except subprocess.TimeoutExpired:
return -1, "", "Timeout"
except Exception as e:
return -1, "", str(e)
def ensure_package(self, package: str) -> bool:
log.debug(f"检查 {package}...")
code, _, _ = self._run(f"which {package} || apk info -e {package}")
if code == 0:
return True
log.info(f"安装 {package}...")
code, _, _ = self._run(f"sudo apk add {package}")
if code != 0:
code, _, _ = self._run(f"sudo apt update && sudo apt install -y {package}")
return code == 0
def _upload_script(self) -> bool:
log.info(f"上传脚本到 ~/{self._remote_script_name}")
try:
with open(self.remote_script, 'r', encoding='utf-8') as f:
script_content = f.read()
except Exception as e:
log.err(f"读取脚本失败: {e}")
return False
# 转换 Windows 换行符为 Unix 换行符
script_content = script_content.replace('\r\n', '\n').replace('\r', '\n')
log.debug(f"转换换行符: Windows -> Unix")
remote_path = f"/home/{self.user}/{self._remote_script_name}"
with tempfile.NamedTemporaryFile(mode='w', suffix='.sh', delete=False, encoding='utf-8', newline='\n') as tmp:
tmp.write(script_content)
tmp_path = tmp.name
try:
log.debug("尝试 scp 上传...")
scp_cmd = [
"scp",
"-P", str(self.port),
"-i", self.ssh_key,
"-o", "StrictHostKeyChecking=no",
"-o", "BatchMode=yes",
"-o", "ConnectTimeout=30",
tmp_path,
f"{self._target()}:{remote_path}"
]
if self.use_proxy:
proxy_cmd = (
f"{self.proxy.get('bin', '/usr/bin/connect-proxy')} "
f"-{self.proxy.get('type', 'S')} "
f"{self.proxy.get('addr', '127.0.0.1:7890')} %h %p"
)
scp_cmd.extend(["-o", f"ProxyCommand={proxy_cmd}"])
log.debug(f"scp 命令: {' '.join(scp_cmd)}")
result = subprocess.run(scp_cmd, timeout=30)
if result.returncode != 0:
log.debug(f"scp 失败 (code={result.returncode})，尝试 cat 方式...")
upload_cmd = f"cat > {remote_path} << 'EOF'\n{script_content}\nEOF\nchmod +x {remote_path}"
code, _, stderr = self._run(upload_cmd)
if code != 0:
log.err(f"上传脚本失败: {stderr}")
return False
log.info("cat 上传成功")
else:
log.info("scp 上传成功")
# 验证脚本是否存在
check_cmd = f"test -f {remote_path} && echo 'exists' || echo 'not exists'"
code, stdout, _ = self._run(check_cmd)
log.debug(f"验证结果: {stdout}")
if 'not exists' in stdout:
log.err(f"脚本上传后验证失败: {remote_path}")
return False
log.ok(f"脚本已上传: {remote_path}")
return True
except Exception as e:
log.err(f"上传脚本异常: {e}")
return False
finally:
try:
os.unlink(tmp_path)
except:
pass
def run_backup(self) -> bool:
log.info(f"执行备份脚本: ~/{self._remote_script_name}")
if not self._upload_script():
return False
remote_path = f"/home/{self.user}/{self._remote_script_name}"
# 执行前详细检查 (debug)
log.debug("执行前详细检查...")
# 1. 检查文件是否存在
check_cmd = f"test -f {remote_path} && echo 'exists' || echo 'not exists'"
code, stdout, _ = self._run(check_cmd)
log.debug(f"文件存在: {stdout}")
if 'not exists' in stdout:
log.err(f"脚本不存在: {remote_path}")
return False
# 2. 查看文件权限和详细信息
check_cmd = f"ls -la {remote_path}"
code, stdout, _ = self._run(check_cmd)
log.debug(f"文件详情:\n{stdout}")
# 3. 查看文件内容（前几行）
check_cmd = f"head -5 {remote_path}"
code, stdout, _ = self._run(check_cmd)
log.debug(f"文件内容开头:\n{stdout}")
# 4. 检查当前用户
check_cmd = "whoami"
code, stdout, _ = self._run(check_cmd)
log.debug(f"当前用户: {stdout}")
# ★ 直接使用 sudo 执行脚本，不加 shell 前缀
cmd = f"sudo -n {remote_path}"
log.info(f"执行命令: {cmd}")
code, stdout, stderr = self._run(cmd)
if code != 0:
log.err(f"备份脚本执行失败 (code={code})")
if stderr:
log.err(f"错误: {stderr[:200]}")
# 如果 sudo -n 失败，尝试不加 sudo
if code != 0:
log.debug("sudo 失败，尝试直接执行...")
cmd2 = f"{remote_path}"
code2, stdout2, stderr2 = self._run(cmd2)
if code2 == 0:
log.debug("直接执行成功")
stdout = stdout2
code = 0
else:
log.err(f"直接执行也失败 (code={code2})")
return False
# 提取备份文件路径
patterns = [
r"Backup complete:\s*(\S+\.tar\.gz)",
r"备份完成:\s*(\S+\.tar\.gz)",
r"(\S+\.tar\.gz)\s*$"
]
for pattern in patterns:
match = re.search(pattern, stdout, re.IGNORECASE | re.MULTILINE)
if match:
self._backup_file = match.group(1)
log.ok(f"远程备份: {self._backup_file}")
return True
matches = re.findall(r"(\S+\.tar\.gz)", stdout)
if matches:
self._backup_file = matches[-1]
log.ok(f"远程备份: {self._backup_file}")
return True
log.err("无法提取备份文件路径")
log.debug(stdout[:500])
return False
def get_backup_file(self) -> Optional[str]:
return self._backup_file
def cleanup(self):
if self._backup_file:
self._run(f"rm -f {self._backup_file}")
log.ok(f"清理远程：{self._backup_file}")
self._run(f"rm -f /home/{self.user}/{self._remote_script_name}")
log.ok(f"清理远程：/home/{self.user}/{self._remote_script_name}")
def __repr__(self):
return f"VPS({self.name}: {self.user}@{self.host})"
# ==============================================
# LocalPC 对象 - 本地计算机
# ==============================================
class LocalPC:
def __init__(self, config: Dict):
self.default_backup_dir = PathUtils.to_local(config.get('backup_dir', './backups'))
self.keep_temp = config.get('keep_temp', False)
self._downloaded_file = None
def get_backup_dir(self, server_config: Dict) -> str:
if server_config.get('local_dest'):
return PathUtils.to_local(server_config['local_dest'])
return self.default_backup_dir
def ensure_dir(self, path: str) -> bool:
try:
Path(path).mkdir(parents=True, exist_ok=True)
return True
except Exception as e:
log.err(f"创建目录失败: {e}")
return False
def download_from_vps(self, vps: VPS, remote_path: str) -> Optional[Path]:
if not remote_path:
log.err("没有远程文件路径")
return None
backup_dir = self.get_backup_dir(vps.server_config)
if not self.ensure_dir(backup_dir):
return None
local_file = Path(backup_dir) / Path(remote_path).name
log.info(f"本地: {local_file}")
return self._download_with_scp(vps, remote_path, local_file)
def _download_with_scp(self, vps: VPS, remote_path: str, local_file: Path) -> Optional[Path]:
"""使用 scp 下载（带重试）"""
# 获取远程文件大小
size_cmd = f"ls -l {remote_path} | awk '{{print $5}}'"
code, size_str, _ = vps._run(size_cmd)
total_size = int(size_str.strip()) if code == 0 and size_str.strip().isdigit() else 0
if total_size > 0:
log.info(f"文件大小: {total_size / (1024*1024):.2f} MB")
# 最大重试次数
max_retries = 3
for attempt in range(max_retries):
if attempt > 0:
log.info(f"重试下载 ({attempt + 1}/{max_retries})...")
time.sleep(5)
result = self._scp_download_once(vps, remote_path, local_file, total_size)
if result is not None:
return result
# 如果下载失败，删除不完整的文件
if local_file.exists():
local_file.unlink()
log.err(f"下载失败，已重试 {max_retries} 次")
return None
def _scp_download_once(self, vps: VPS, remote_path: str, local_file: Path, total_size: int) -> Optional[Path]:
"""单次 scp 下载"""
scp_cmd = [
"scp",
"-P", str(vps.port),
"-i", vps.ssh_key,
"-o", "StrictHostKeyChecking=no",
"-o", "BatchMode=yes",
"-o", "ServerAliveInterval=60",
"-o", "ServerAliveCountMax=3",
"-o", "ConnectTimeout=30",
]
if vps.use_proxy:
proxy_cmd = (
f"{vps.proxy.get('bin', '/usr/bin/connect-proxy')} "
f"-{vps.proxy.get('type', 'S')} "
f"{vps.proxy.get('addr', '127.0.0.1:7890')} %h %p"
)
scp_cmd.extend(["-o", f"ProxyCommand={proxy_cmd}"])
log.debug(f"下载使用代理")
scp_cmd.extend([
f"{vps._target()}:{remote_path}",
str(local_file)
])
log.debug(f"scp 命令: {' '.join(scp_cmd)}")
# 进度条
progress_bar = tqdm(
total=total_size,
unit='B',
unit_scale=True,
unit_divisor=1024,
desc=f"下载 {Path(remote_path).name}",
bar_format='{desc}: {percentage:3.0f}%|{bar}| {n_fmt}/{total_fmt} [{rate_fmt}]'
)
try:
process = subprocess.Popen(
scp_cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
last_bytes = 0
start_time = time.time()
no_progress_count = 0
while process.poll() is None:
elapsed = time.time() - start_time
if elapsed > vps.download_timeout:
process.kill()
log.err(f"下载超时 ({vps.download_timeout}s)")
progress_bar.close()
return None
if local_file.exists():
current_size = local_file.stat().st_size
if current_size > last_bytes:
progress_bar.update(current_size - last_bytes)
last_bytes = current_size
no_progress_count = 0
else:
no_progress_count += 1
# 如果 30 秒没有进度，可能连接断了
if no_progress_count > 60: # 60 * 0.5s = 30s
log.warn("下载无进度，可能连接中断")
process.kill()
progress_bar.close()
return None
time.sleep(0.5)
progress_bar.close()
if process.returncode != 0:
_, stderr = process.communicate()
log.err(f"scp 失败 (code={process.returncode})")
if stderr:
log.debug(f"scp stderr: {stderr[:200]}")
return None
if local_file.exists() and local_file.stat().st_size > 0:
size_mb = local_file.stat().st_size / (1024 * 1024)
log.ok(f"{local_file.name} ({size_mb:.2f} MB)")
self._downloaded_file = local_file
return local_file
else:
log.err("下载文件为空")
return None
except Exception as e:
progress_bar.close()
log.err(f"scp 异常: {e}")
return None
def get_downloaded_file(self) -> Optional[Path]:
return self._downloaded_file
# ==============================================
# 主方法
# ==============================================
def run_backup(config_file: str, script_file: str, server_name: Optional[str] = None) -> Union[Dict[str, Tuple[bool, Optional[Path]]], Tuple[bool, Optional[Path]]]:
"""
执行备份
Args:
config_file: 配置文件路径
script_file: 备份脚本文件路径
server_name: 可选，指定备份的服务器名称
Returns:
- 如果指定 server_name: 返回 (是否成功, 本地文件路径)
- 如果不指定 server_name: 返回 {服务器名: (是否成功, 本地文件路径)}
"""
config_path = PathUtils.to_local(config_file)
if not Path(config_path).exists():
log.err(f"配置文件不存在: {config_path}")
return None
with open(config_path, 'r', encoding='utf-8') as f:
config_dict = json.load(f)
log.ok(f"加载配置文件: {config_path}")
log.set_debug(config_dict.get('global', {}).get('debug', False))
local = LocalPC(config_dict.get('global', {}))
vps_list = []
proxy_config = config_dict.get('global', {}).get('proxy', {})
script_path = PathUtils.to_local(script_file)
if not Path(script_path).exists():
log.err(f"备份脚本不存在: {script_path}")
return None
for server_config in config_dict.get('servers', []):
if server_config.get('enabled', True):
if server_name and server_config['name'] != server_name:
continue
if 'use_proxy' not in server_config:
server_config['use_proxy'] = proxy_config.get('enabled', False)
if 'proxy' not in server_config:
server_config['proxy'] = proxy_config
vps_list.append(VPS(server_config, script_path))
if not vps_list:
log.err("没有可用的服务器")
return None
if server_name:
vps = vps_list[0]
log.section(f"备份服务器: {vps.name}")
return _backup_one(vps, local)
else:
results = {}
log.section(f"开始备份 ({len(vps_list)} 个服务器)")
for vps in vps_list:
log.info(f"\n>>> {vps.name}")
success, path = _backup_one(vps, local)
results[vps.name] = (success, path)
log.section("备份汇总")
success_count = sum(1 for v in results.values() if v[0])
log.info(f"总计: {len(results)}, 成功: {success_count}, 失败: {len(results) - success_count}")
for name, (success, path) in results.items():
status = "✅" if success else "❌"
log.info(f" {status} {name}: {path if path else '失败'}")
return results
def _backup_one(vps: VPS, local: LocalPC) -> Tuple[bool, Optional[Path]]:
try:
vps.ensure_package("tar")
if not vps.run_backup():
return False, None
remote_file = vps.get_backup_file()
if not remote_file:
return False, None
local_file = local.download_from_vps(vps, remote_file)
vps.cleanup()
return local_file is not None, local_file
except Exception as e:
log.err(f"备份异常: {e}")
return False, None
# ==============================================
# 入口
# ==============================================
if __name__ == "__main__":
config_json = "vps_backup_ssh_config.json"
script_sh = "vps_backup_backupGhost.sh"
log.info('脚本加载完成')
servers = list_servers(config_json)
log.info(f'配置文件: {config_json} > 服务器: {servers}')
log.info(f'备份脚本: {script_sh}')
本地配置文件
{
"global": {
"proxy": {
"enabled": false,
"bin": "C:/Users/cat/scoop/apps/git/current/mingw64/bin/connect.exe",
"addr": "127.0.0.1:7890",
"type": "S"
},
"debug": false,
"keep_temp": false,
"max_retries": 3,
"retry_delay": 5,
"default_timeout": 30,
"download_timeout": 7200
},
"servers": [
{
"name": "oracle-arm",
"user": "ghost",
"host": "arm.xxx.xxx",
"ssh_key": "D:\\***\\max2-20250215",
"local_dest": "D:\\***\\bak",
"enabled": true,
"port": 22,
"timeout": 1200,
"download_timeout": 3600,
"install_package": "tar",
"use_proxy": true
},
{
"name": "oracle-trilium",
"user": "ghost",
"host": "ghost.xxx.xxx",
"ssh_key": "D:\\***\\max2-20250215",
"local_dest": "D:\\***\\bak",
"enabled": true,
"port": 22,
"timeout": 2400,
"download_timeout": 3600,
"install_package": "tar",
"use_proxy": true
},
{
"name": "max2-alpine",
"user": "ghost",
"host": "alpine.xxx.xxx",
"ssh_key": "D:\\***\\max2-20250215",
"local_dest": "D:\\***\\bak",
"enabled": true,
"port": 22,
"timeout": 600,
"download_timeout": 600,
"install_package": "tar",
"use_proxy": false
},
{
"name": "kc-v100",
"user": "x99",
"host": "x99.tail9e6317.ts.net",
"ssh_key": "D:\\***\\rsa4096-20260514",
"local_dest": "D:\\***\\bak",
"enabled": true,
"port": 22,
"timeout": 600,
"download_timeout": 7200,
"install_package": "tar",
"use_proxy": false
}
]
}
注入远端的脚本
#!/bin/sh
# ============================================
# 统一备份脚本 - 合并所有服务器的备份规则
# 所有服务器使用同一份配置，不存在的目录自动跳过
# ============================================
set -e
# ============================================
# 1. 自动识别环境
# ============================================
if [ -n "$SUDO_USER" ]; then
CURRENT_USER="$SUDO_USER"
else
CURRENT_USER=$(whoami)
fi
# ============================================
# 2. 日志清理（合并所有服务器的清理规则）
# ============================================
echo ">>> Cleaning up log files before backup ..."
# 清理各应用日志（不存在的目录自动忽略）
find /www/nginx/data/logs -type f \( -name "*.log" -o -name "*.log.*" \) -delete 2>/dev/null || true
find /www/trilium/data -type f \( -name "*.log" -o -name "*.log.*" \) -delete 2>/dev/null || true
find /www/ghost/data/content/logs -type f \( -name "*.log" -o -name "*.log.*" \) -delete 2>/dev/null || true
find /www/certbot/data/log/letsencrypt -type f \( -name "*.log" -o -name "*.log.*" \) -delete 2>/dev/null || true
# Nginx 日志重载（仅当容器存在时执行）
docker kill -s USR1 nginx 2>/dev/null || true
echo ">>> Log cleanup complete."
# ============================================
# 3. 准备备份目录
# ============================================
SCRIPT_PATH=$(readlink -f "$0")
SCRIPT_DIR=$(dirname "$SCRIPT_PATH")
cd "$SCRIPT_DIR"
FILE_NAME="${CURRENT_USER}_$(date +"%Y%m%d%H%M%S").tar.gz"
# ============================================
# 4. 合并的排除列表（去重后的完整配置）
# 来自所有服务器的 backupGhost.sh
# ============================================
EXCLUDES="\
--exclude=*/nginx/data/logs/* \
--exclude=*/nginx/docker-compose.yml.* \
--exclude=*/nginx/data/nginx.conf.bak \
--exclude=*/nginx/data/conf.d/*.bak \
--exclude=*/trilium/data/backup \
--exclude=*/trilium/data/log \
--exclude=*/trilium/data/sessions/* \
--exclude=*/trilium/data/*.db-shm \
--exclude=*/trilium/data/*.db-wal \
"
# ============================================
# 5. 打包目标（合并所有服务器的目录）
# ============================================
# 包含脚本自身和所有可能的业务目录
# 不存在的目录 tar 会自动跳过并提示
TARGETS="$SCRIPT_PATH /www"
# ============================================
# 6. 执行压缩
# ============================================
echo ">>> [User: $CURRENT_USER] Starting backup to $FILE_NAME ..."
echo ">>> 打包目录: /www 及所有子目录"
echo ">>> 不存在的目录将自动跳过"
sudo tar -czvPf "$FILE_NAME" $EXCLUDES $TARGETS 2>&1 | grep -v "Cannot stat" || true
# ============================================
# 7. 权限归还
# ============================================
if [ -f "$FILE_NAME" ] && [ "$CURRENT_USER" != "root" ]; then
sudo chown "$CURRENT_USER":"$CURRENT_USER" "$FILE_NAME" 2>/dev/null || true
fi
# ============================================
# 8. 显示结果
# ============================================
FILE_SIZE=$(du -h "$FILE_NAME" | cut -f1)
echo ">>> Backup complete: $SCRIPT_DIR/$FILE_NAME"
echo ">>> File size: $FILE_SIZE"