sz-backup: tar 后 7z 压缩 (与上传一致, 文本类 mx=9 照片 mx=1); 恢复端 7z x 解压; 修 manifest 保存 bug

This commit is contained in:
lou
2026-08-10 18:00:00 +08:00
parent 9d39437c8a
commit 96b6bd1af5
6 changed files with 54 additions and 14 deletions
+25 -5
View File
@@ -28,10 +28,17 @@ from tqdm import tqdm
STATE_DIR = Path.home() / ".local/share/7z-encrypt" / "backup" STATE_DIR = Path.home() / ".local/share/7z-encrypt" / "backup"
BATCH_MB = 500 # 每批 tar 上限 BATCH_MB = 500 # 每批 tar 上限
BATCH_FILES = 500 # 每批文件数上限 BATCH_FILES = 500 # 每批文件数上限
# 高压缩率类型 (聊天记录/文档) 用 gzip, 照片/视频直接存 # 高压缩率类型 (聊天记录/文档) 用 -mx=9, 照片/视频批次 -mx=1 快速
COMPRESS_SUFFIX = {".db", ".sqlite", ".sqlite3", ".sql", ".txt", ".json", ".xml", ".csv", ".log", ".md", ".html"} COMPRESS_SUFFIX = {".db", ".sqlite", ".sqlite3", ".sql", ".txt", ".json", ".xml", ".csv", ".log", ".md", ".html"}
def _7z(args: list[str]) -> None:
"""调系统 7z (p7zip-full), 失败抛异常"""
r = subprocess.run(["7z"] + args, capture_output=True, text=True, timeout=7200)
if r.returncode != 0:
raise RuntimeError(f"7z 失败: {r.stderr.strip()[-200:] or r.stdout.strip()[-200:]}")
def _cfg() -> dict[str, Any]: def _cfg() -> dict[str, Any]:
"""读配置文件 backup 段 (XDG, 同客户端)""" """读配置文件 backup 段 (XDG, 同客户端)"""
env = os.environ.get("SZ_CONFIG") env = os.environ.get("SZ_CONFIG")
@@ -201,13 +208,19 @@ def backup(name: str, chunk_mb: int = 10) -> int:
for i, batch in enumerate(batches, 1): for i, batch in enumerate(batches, 1):
tar_path = STATE_DIR / f"{name}_{ts}_{i:02d}.tar" tar_path = STATE_DIR / f"{name}_{ts}_{i:02d}.tar"
tar_path.parent.mkdir(parents=True, exist_ok=True) tar_path.parent.mkdir(parents=True, exist_ok=True)
mode = "w:gz" if any(_compress(f) for f in batch) else "w" with tarfile.open(tar_path, "w") as tf:
with tarfile.open(tar_path, mode) as tf:
for f in batch: for f in batch:
try: try:
tf.add(f, arcname=str(f)) tf.add(f, arcname=str(f))
except (OSError, tarfile.TarError) as e: except (OSError, tarfile.TarError) as e:
print(f" [跳过] {f}: {e}") print(f" [跳过] {f}: {e}")
# 7z 压缩 (与上传文件一致): 含文本类 -mx=9, 照片视频 -mx=1
mx = 9 if any(_compress(f) for f in batch) else 1
seven = tar_path.with_suffix(".tar.7z")
print(f"[备份] 7z 压缩 (mx={mx}) ...")
_7z(["a", "-y", "-bd", f"-mx={mx}", str(seven), str(tar_path)])
tar_path.unlink(missing_ok=True)
tar_path = seven
size_mb = tar_path.stat().st_size / 1048576 size_mb = tar_path.stat().st_size / 1048576
print(f"[备份] 批次 {i}/{len(batches)}: {tar_path.name} ({size_mb:.1f} MB), 上传中 ...") print(f"[备份] 批次 {i}/{len(batches)}: {tar_path.name} ({size_mb:.1f} MB), 上传中 ...")
try: try:
@@ -215,7 +228,7 @@ def backup(name: str, chunk_mb: int = 10) -> int:
except RuntimeError as e: except RuntimeError as e:
print(f"[错误] {e}") print(f"[错误] {e}")
return 1 return 1
files_in_batch = [str(f) for f in batch if tarfile.open(tar_path).getnames()] files_in_batch = [str(f) for f in batch]
for f in batch: for f in batch:
try: try:
st = f.stat() st = f.stat()
@@ -267,7 +280,14 @@ def restore(name: str, out_root: Path | None = None) -> int:
except Exception as e: except Exception as e:
print(f"[错误] 下载 {tar_name} 失败: {e}") print(f"[错误] 下载 {tar_name} 失败: {e}")
return 1 return 1
# 解包 (路径校验: 只解回 manifest 记录的文件) # 7z 解压 -> tar -> 解包 (路径校验: 只解回 manifest 记录的文件)
if tar_src.suffix == ".7z":
_7z(["x", "-y", "-bd", f"-o{dl}", str(tar_src)])
tar_src.unlink(missing_ok=True)
tar_src = dl / tar_name
if not tar_src.exists():
print(f"[错误] 未找到 tar: {dl}")
return 1
with tarfile.open(tar_src) as tf: with tarfile.open(tar_src) as tf:
for member in tf.getmembers(): for member in tf.getmembers():
if member.name not in info.get("files", []): if member.name not in info.get("files", []):
+1 -1
View File
@@ -3,7 +3,7 @@ Source: 7z-encrypt-client
Version: 1.0.0 Version: 1.0.0
Architecture: amd64 Architecture: amd64
Maintainer: edgevoid <edgevoid@users.noreply.gitee.com> Maintainer: edgevoid <edgevoid@users.noreply.gitee.com>
Installed-Size: 51850 Installed-Size: 51858
Depends: p7zip-full Depends: p7zip-full
Conflicts: 7z-encrypt-client Conflicts: 7z-encrypt-client
Replaces: 7z-encrypt-client Replaces: 7z-encrypt-client
+1 -1
View File
@@ -1,4 +1,4 @@
094c9453f9346a962da07a6e123cc5fc usr/bin/sz-backup b7ec4ffff510e79070edf4685f0ba9a1 usr/bin/sz-backup
61901c8e9af062eef5c386c715f5f3bb usr/bin/sz-config 61901c8e9af062eef5c386c715f5f3bb usr/bin/sz-config
02b57eea9c5ca54c63fd82da1d6d960b usr/bin/sz-transfer 02b57eea9c5ca54c63fd82da1d6d960b usr/bin/sz-transfer
8f9902ebfd228e08f7691abe6d471d86 usr/bin/sz-tui 8f9902ebfd228e08f7691abe6d471d86 usr/bin/sz-tui
+1 -1
View File
@@ -2,7 +2,7 @@ Package: 7z-encrypt-client
Version: 1.0.0 Version: 1.0.0
Architecture: all Architecture: all
Maintainer: edgevoid <edgevoid@users.noreply.gitee.com> Maintainer: edgevoid <edgevoid@users.noreply.gitee.com>
Installed-Size: 116 Installed-Size: 117
Depends: python3 (>= 3.10), python3-venv, p7zip-full Depends: python3 (>= 3.10), python3-venv, p7zip-full
Section: utils Section: utils
Priority: optional Priority: optional
+1 -1
View File
@@ -2,7 +2,7 @@
440c53f84f5722b5146959c8fc6d7a23 usr/bin/sz-config 440c53f84f5722b5146959c8fc6d7a23 usr/bin/sz-config
ec8cf0df841e479689ccb681f6168ccc usr/bin/sz-transfer ec8cf0df841e479689ccb681f6168ccc usr/bin/sz-transfer
0ba5b8147990daa7fd07344952c42847 usr/bin/sz-tui 0ba5b8147990daa7fd07344952c42847 usr/bin/sz-tui
3ed341805839f8dcee1fc8f719c5cc2f usr/share/7z-encrypt/backup.py 78aed672b874d47ecb3e9dc12b6951a8 usr/share/7z-encrypt/backup.py
dcb31e74e0a6ee8073337fc4fb2acf99 usr/share/7z-encrypt/config.py dcb31e74e0a6ee8073337fc4fb2acf99 usr/share/7z-encrypt/config.py
8a064c0b3c21d50812a3fd69a320d862 usr/share/7z-encrypt/crypto.py 8a064c0b3c21d50812a3fd69a320d862 usr/share/7z-encrypt/crypto.py
21062e04f8c7f4471881a39c489dfa73 usr/share/7z-encrypt/metadata.py 21062e04f8c7f4471881a39c489dfa73 usr/share/7z-encrypt/metadata.py
+25 -5
View File
@@ -28,10 +28,17 @@ from tqdm import tqdm
STATE_DIR = Path.home() / ".local/share/7z-encrypt" / "backup" STATE_DIR = Path.home() / ".local/share/7z-encrypt" / "backup"
BATCH_MB = 500 # 每批 tar 上限 BATCH_MB = 500 # 每批 tar 上限
BATCH_FILES = 500 # 每批文件数上限 BATCH_FILES = 500 # 每批文件数上限
# 高压缩率类型 (聊天记录/文档) 用 gzip, 照片/视频直接存 # 高压缩率类型 (聊天记录/文档) 用 -mx=9, 照片/视频批次 -mx=1 快速
COMPRESS_SUFFIX = {".db", ".sqlite", ".sqlite3", ".sql", ".txt", ".json", ".xml", ".csv", ".log", ".md", ".html"} COMPRESS_SUFFIX = {".db", ".sqlite", ".sqlite3", ".sql", ".txt", ".json", ".xml", ".csv", ".log", ".md", ".html"}
def _7z(args: list[str]) -> None:
"""调系统 7z (p7zip-full), 失败抛异常"""
r = subprocess.run(["7z"] + args, capture_output=True, text=True, timeout=7200)
if r.returncode != 0:
raise RuntimeError(f"7z 失败: {r.stderr.strip()[-200:] or r.stdout.strip()[-200:]}")
def _cfg() -> dict[str, Any]: def _cfg() -> dict[str, Any]:
"""读配置文件 backup 段 (XDG, 同客户端)""" """读配置文件 backup 段 (XDG, 同客户端)"""
env = os.environ.get("SZ_CONFIG") env = os.environ.get("SZ_CONFIG")
@@ -201,13 +208,19 @@ def backup(name: str, chunk_mb: int = 10) -> int:
for i, batch in enumerate(batches, 1): for i, batch in enumerate(batches, 1):
tar_path = STATE_DIR / f"{name}_{ts}_{i:02d}.tar" tar_path = STATE_DIR / f"{name}_{ts}_{i:02d}.tar"
tar_path.parent.mkdir(parents=True, exist_ok=True) tar_path.parent.mkdir(parents=True, exist_ok=True)
mode = "w:gz" if any(_compress(f) for f in batch) else "w" with tarfile.open(tar_path, "w") as tf:
with tarfile.open(tar_path, mode) as tf:
for f in batch: for f in batch:
try: try:
tf.add(f, arcname=str(f)) tf.add(f, arcname=str(f))
except (OSError, tarfile.TarError) as e: except (OSError, tarfile.TarError) as e:
print(f" [跳过] {f}: {e}") print(f" [跳过] {f}: {e}")
# 7z 压缩 (与上传文件一致): 含文本类 -mx=9, 照片视频 -mx=1
mx = 9 if any(_compress(f) for f in batch) else 1
seven = tar_path.with_suffix(".tar.7z")
print(f"[备份] 7z 压缩 (mx={mx}) ...")
_7z(["a", "-y", "-bd", f"-mx={mx}", str(seven), str(tar_path)])
tar_path.unlink(missing_ok=True)
tar_path = seven
size_mb = tar_path.stat().st_size / 1048576 size_mb = tar_path.stat().st_size / 1048576
print(f"[备份] 批次 {i}/{len(batches)}: {tar_path.name} ({size_mb:.1f} MB), 上传中 ...") print(f"[备份] 批次 {i}/{len(batches)}: {tar_path.name} ({size_mb:.1f} MB), 上传中 ...")
try: try:
@@ -215,7 +228,7 @@ def backup(name: str, chunk_mb: int = 10) -> int:
except RuntimeError as e: except RuntimeError as e:
print(f"[错误] {e}") print(f"[错误] {e}")
return 1 return 1
files_in_batch = [str(f) for f in batch if tarfile.open(tar_path).getnames()] files_in_batch = [str(f) for f in batch]
for f in batch: for f in batch:
try: try:
st = f.stat() st = f.stat()
@@ -267,7 +280,14 @@ def restore(name: str, out_root: Path | None = None) -> int:
except Exception as e: except Exception as e:
print(f"[错误] 下载 {tar_name} 失败: {e}") print(f"[错误] 下载 {tar_name} 失败: {e}")
return 1 return 1
# 解包 (路径校验: 只解回 manifest 记录的文件) # 7z 解压 -> tar -> 解包 (路径校验: 只解回 manifest 记录的文件)
if tar_src.suffix == ".7z":
_7z(["x", "-y", "-bd", f"-o{dl}", str(tar_src)])
tar_src.unlink(missing_ok=True)
tar_src = dl / tar_name
if not tar_src.exists():
print(f"[错误] 未找到 tar: {dl}")
return 1
with tarfile.open(tar_src) as tf: with tarfile.open(tar_src) as tf:
for member in tf.getmembers(): for member in tf.getmembers():
if member.name not in info.get("files", []): if member.name not in info.get("files", []):