lou
e6d12e0417
- backup() 加 srcs 参数: None=预设/配置目录, 显式传=备份指定目录 - CLI 加 --dir: 目录 basename 作为备份项名, restore 同名恢复 - TUI 菜单 13: 输入目录路径直接备份指定目录, 预设项不变
263 lines
9.7 KiB
Python
263 lines
9.7 KiB
Python
#!/usr/bin/env python3
|
|
"""一键备份: 相册/聊天记录/文档 -> 私有加密服务器
|
|
|
|
整个文件夹打包一个文件: tar -> 32MB 卷压缩 (逐卷 7z + 逐卷加密) -> aria2 上传。
|
|
不分批次, 一个文件夹 = 一个文件 (file_id)。
|
|
|
|
手机 ZeroTermux/PRoot 可用: /sdcard/DCIM 等外部存储直接可见。
|
|
|
|
聊天记录注意: PRoot 无 root 读不到微信数据库 (/data/data), 需先把
|
|
微信"备份与迁移/导出"的文件放到手机存储, 再配置 backup.chat 目录。
|
|
|
|
用法:
|
|
sz-backup # 备份全部 (photos+documents+download+chat)
|
|
sz-backup photos # 只备份相册
|
|
sz-backup list # 已备份清单
|
|
sz-backup restore photos # 恢复相册
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tarfile
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from tqdm import tqdm
|
|
|
|
STATE_DIR = Path.home() / ".local/share/7z-encrypt" / "backup"
|
|
VOL_MB = 32 # 卷压缩粒度钉死 32MB (与上传一致)
|
|
|
|
|
|
def _cfg() -> dict[str, Any]:
|
|
"""读配置文件 backup 段 (XDG, 同客户端)"""
|
|
env = os.environ.get("SZ_CONFIG")
|
|
if env:
|
|
p = Path(env).expanduser()
|
|
else:
|
|
xdg = os.environ.get("XDG_CONFIG_HOME") or str(Path.home() / ".config")
|
|
p = Path(xdg) / "7z-encrypt" / "config.json"
|
|
try:
|
|
return dict(json.loads(p.read_text(encoding="utf-8")).get("backup", {}))
|
|
except (OSError, json.JSONDecodeError):
|
|
return {}
|
|
|
|
|
|
def _server() -> str:
|
|
"""服务器地址: 配置 server.url (备份复用客户端配置)"""
|
|
env = os.environ.get("SZ_CONFIG")
|
|
if env:
|
|
p = Path(env).expanduser()
|
|
else:
|
|
xdg = os.environ.get("XDG_CONFIG_HOME") or str(Path.home() / ".config")
|
|
p = Path(xdg) / "7z-encrypt" / "config.json"
|
|
try:
|
|
return str(json.loads(p.read_text(encoding="utf-8")).get("server", {}).get("url", ""))
|
|
except (OSError, json.JSONDecodeError):
|
|
return ""
|
|
|
|
|
|
def _token() -> str:
|
|
"""登录 token: 配置文件 server.token"""
|
|
env = os.environ.get("SZ_CONFIG")
|
|
if env:
|
|
p = Path(env).expanduser()
|
|
else:
|
|
xdg = os.environ.get("XDG_CONFIG_HOME") or str(Path.home() / ".config")
|
|
p = Path(xdg) / "7z-encrypt" / "config.json"
|
|
try:
|
|
return str(json.loads(p.read_text(encoding="utf-8")).get("server", {}).get("token", ""))
|
|
except (OSError, json.JSONDecodeError):
|
|
return ""
|
|
|
|
|
|
def _src_dirs(name: str) -> list[Path]:
|
|
"""备份源目录: 配置 override 默认"""
|
|
defaults = {
|
|
"photos": ["~/storage/dcim", "~/storage/pictures", "/sdcard/DCIM", "/sdcard/Pictures"],
|
|
"documents": ["/sdcard/Documents"],
|
|
"download": ["/sdcard/Download"],
|
|
"chat": [],
|
|
}
|
|
dirs = _cfg().get(name, defaults.get(name, []))
|
|
return [Path(d).expanduser() for d in dirs]
|
|
|
|
|
|
def _manifest_path(name: str) -> Path:
|
|
return STATE_DIR / f"{name}.json"
|
|
|
|
|
|
def _load_manifest(name: str) -> dict[str, Any]:
|
|
try:
|
|
return json.loads(_manifest_path(name).read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return {}
|
|
|
|
|
|
def _save_manifest(name: str, m: dict[str, Any]) -> None:
|
|
_manifest_path(name).parent.mkdir(parents=True, exist_ok=True)
|
|
_manifest_path(name).write_text(json.dumps(m, ensure_ascii=False, indent=1), encoding="utf-8")
|
|
|
|
|
|
def _cleanup(tar_path: Path) -> None:
|
|
"""清理 tar 及其卷压缩产物 (chunk/init/manifest)"""
|
|
for p in tar_path.parent.glob(tar_path.name + "*"):
|
|
try:
|
|
p.unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def backup(name: str, chunk_mb: int = VOL_MB, srcs: list[Path] | None = None) -> int:
|
|
"""整个文件夹 -> 单个 tar -> 32MB 卷压缩 (逐卷 7z + 逐卷加密) -> aria2 上传
|
|
|
|
一个文件夹 = 一个文件 (file_id), 不分批次。
|
|
srcs 为 None 时用预设/配置源目录 (photos/documents/...), 显式传入则
|
|
备份指定目录 (PC 端任意路径, 如 /home/lou/图片)。
|
|
"""
|
|
from metadata import _build_vol_init
|
|
import argparse as _ap
|
|
from transfer import TransferClient
|
|
|
|
if srcs is None:
|
|
srcs = [s for s in _src_dirs(name) if s.exists()]
|
|
else:
|
|
srcs = [s for s in srcs if s.is_dir()]
|
|
files = [p for s in srcs for p in s.rglob("*") if p.is_file()]
|
|
if not files:
|
|
print(f"[备份] {name}: 源目录为空")
|
|
return 1
|
|
total = sum(f.stat().st_size for f in files)
|
|
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
tar_path = STATE_DIR / f"{name}_{ts}.tar"
|
|
tar_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 1. 整个文件夹打包 (磁盘, 不落 tmpfs)
|
|
print(f"[备份] {name}: {len(files)} 个文件 ({total / 1048576:.1f} MB), 打包 tar ...")
|
|
with tarfile.open(tar_path, "w") as tf:
|
|
for f in files:
|
|
try:
|
|
tf.add(f, arcname=str(f))
|
|
except (OSError, tarfile.TarError) as e:
|
|
print(f" [跳过] {f}: {e}")
|
|
|
|
# 2. 32MB 卷压缩: 逐卷 7z (mx=1 快速, 备份照片视频为主) + 逐卷加密
|
|
ns = _ap.Namespace(
|
|
vol_compress=chunk_mb, vol_no_7z=False, vol_mx=1,
|
|
key_id="default_key", file_name=f"{name}_{ts}.tar", no_encrypt_name=False,
|
|
)
|
|
print(f"[备份] 卷压缩 ({chunk_mb}MB 切卷 + 逐卷 7z + 逐卷加密) ...")
|
|
if _build_vol_init(tar_path, ns) != 0:
|
|
print("[备份] 卷压缩/加密失败")
|
|
_cleanup(tar_path)
|
|
return 1
|
|
tar_path.unlink(missing_ok=True) # 明文 tar 用完即删
|
|
|
|
# 3. aria2 上传 (复用传输管线)
|
|
init_json = json.loads(Path(str(tar_path) + ".init.json").read_text(encoding="utf-8"))
|
|
manifest = json.loads(Path(str(tar_path) + "_manifest.json").read_text(encoding="utf-8"))
|
|
chunk_files = {ch["index"]: Path(ch["filename"]).resolve() for ch in manifest}
|
|
client = TransferClient(_server(), token=_token())
|
|
try:
|
|
receipt = client.transfer(init_json, chunk_files)
|
|
except Exception as e:
|
|
print(f"[备份] 上传失败: {e}")
|
|
_cleanup(tar_path)
|
|
return 1
|
|
fid = str(receipt["file_id"])
|
|
|
|
# 4. manifest 记录 (一个备份集 = 一个 file_id)
|
|
_save_manifest(name, {
|
|
"file_id": fid, "name": name, "size": total, "time": ts,
|
|
"files": [str(f) for f in files],
|
|
})
|
|
_cleanup(tar_path)
|
|
print(f"[备份] {name} 完成: {len(files)} 个文件 -> file_id={fid} (卷压缩 mx=1)")
|
|
return 0
|
|
|
|
|
|
def restore(name: str, out_root: Path | None = None) -> int:
|
|
"""下载单文件 (卷模式自动解密+7z x+拼接) -> tar 解包还原"""
|
|
from transfer import TransferClient
|
|
|
|
m = _load_manifest(name)
|
|
fid = m.get("file_id")
|
|
if not fid:
|
|
print(f"[恢复] {name}: 无备份记录 (先 sz-backup {name})")
|
|
return 1
|
|
out_dir = out_root or Path.cwd()
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
client = TransferClient(_server(), token=_token())
|
|
print(f"[恢复] {name}: 下载 + 解密 + 拼接 (file_id={fid}) ...")
|
|
try:
|
|
enc_path, enc_params = client.download(fid, out_dir / ".sz-restore.tar")
|
|
except Exception as e:
|
|
print(f"[恢复] 下载失败: {e}")
|
|
return 1
|
|
tar_src = out_dir / str(enc_params.get("file_name", f"{name}.tar"))
|
|
shutil.move(enc_path, tar_src)
|
|
print(f"[恢复] 解包 {tar_src} ...")
|
|
with tarfile.open(tar_src) as tf:
|
|
for member in tf.getmembers():
|
|
if out_root is not None:
|
|
# 解到 out_root 下 (member.name 是绝对路径, lstrip 后成相对路径)
|
|
tf.extract(member, path=out_root, filter="data")
|
|
else:
|
|
tf.extract(member, path=Path(member.name).parent, filter="data")
|
|
tar_src.unlink(missing_ok=True)
|
|
print(f"[恢复] {name} 完成")
|
|
return 0
|
|
|
|
|
|
def list_backups() -> int:
|
|
if not STATE_DIR.exists():
|
|
print("[备份] 暂无备份记录")
|
|
return 0
|
|
for mp in sorted(STATE_DIR.glob("*.json")):
|
|
name = mp.stem
|
|
m = _load_manifest(name)
|
|
fid = m.get("file_id", "-")
|
|
print(f" {name}: file_id={fid}, {len(m.get('files', []))} 文件, "
|
|
f"{m.get('size', 0) / 1048576:.1f} MB, {m.get('time', '-')}")
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(description="一键备份: 相册/聊天记录/文档/指定目录 -> 加密服务器")
|
|
ap.add_argument("action", nargs="?", default="all", help="备份项: photos/documents/download/chat/all/list/restore")
|
|
ap.add_argument("target", nargs="?", help="restore 的备份项")
|
|
ap.add_argument("--dir", help="备份指定目录 (不限于预设项, 如 /home/lou/图片)")
|
|
ap.add_argument("--chunk-size", type=int, default=VOL_MB, help=f"卷压缩 MB (默认 {VOL_MB})")
|
|
args = ap.parse_args()
|
|
|
|
if args.action == "list":
|
|
return list_backups()
|
|
if args.action == "restore":
|
|
if not args.target:
|
|
print("[错误] 用法: sz-backup restore <备份项名>")
|
|
return 1
|
|
return restore(args.target)
|
|
if args.dir:
|
|
p = Path(args.dir).expanduser()
|
|
if not p.is_dir():
|
|
print(f"[错误] 目录不存在: {p}")
|
|
return 1
|
|
return backup(p.name, args.chunk_size, srcs=[p])
|
|
names = ["photos", "documents", "download", "chat"] if args.action == "all" else [args.action]
|
|
if args.action not in ("photos", "documents", "download", "chat", "all"):
|
|
print(f"[错误] 未知备份项: {args.action} (photos/documents/download/chat/all)")
|
|
return 1
|
|
ok = True
|
|
for n in names:
|
|
if not _src_dirs(n):
|
|
print(f"[备份] {n}: 无源目录 (config backup.{n} 为空, 跳过)")
|
|
continue
|
|
ok = backup(n, args.chunk_size) == 0 and ok
|
|
return 0 if ok else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|