# -*- coding: utf-8 -*- """收集 WorkBuddy 关键资产并打包 zip + manifest + HTML 说明。""" from __future__ import annotations import fnmatch import hashlib import json import os import shutil import zipfile from datetime import datetime from pathlib import Path import _bootstrap # noqa: F401 import backup_settings as S from backup_guide_html import categorize_files, render_backup_guide_html, write_category_guides from backup_remote import pull_remote_snapshot def load_manifest() -> dict: if not S.MANIFEST_PATH.is_file(): raise FileNotFoundError(f"缺少备份清单: {S.MANIFEST_PATH}") return json.loads(S.MANIFEST_PATH.read_text(encoding="utf-8")) def _sha256_file(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def _should_skip_name(name: str, manifest: dict) -> bool: if name in manifest.get("exclude_file_names", []): return True for pat in manifest.get("exclude_globs", []): if fnmatch.fnmatch(name, pat): return True return False def collect_files(root: Path, manifest: dict) -> tuple[list[dict], list[dict]]: """返回 (included_entries, skipped_entries)。""" included: list[dict] = [] skipped: list[dict] = [] seen: set[str] = set() max_bytes = int(manifest.get("max_file_bytes", 50 * 1024 * 1024)) exclude_dirs = set(manifest.get("exclude_dir_names", [])) def add_file(src: Path, rel_posix: str) -> None: rel_posix = rel_posix.replace("\\", "/") if rel_posix in seen: return if _should_skip_name(src.name, manifest): skipped.append({"path": rel_posix, "reason": "exclude_file_name"}) return try: size = src.stat().st_size except OSError as e: skipped.append({"path": rel_posix, "reason": str(e)}) return if size > max_bytes: skipped.append({"path": rel_posix, "reason": f"skipped_large>{max_bytes}"}) return seen.add(rel_posix) included.append( { "path": rel_posix, "size": size, "sha256": _sha256_file(src), } ) def walk_dir(rel_dir: str) -> None: base = root / rel_dir if not base.is_dir(): skipped.append({"path": rel_dir, "reason": "dir_missing"}) return for dirpath, dirnames, filenames in os.walk(base): dirnames[:] = [d for d in dirnames if d not in exclude_dirs] dp = Path(dirpath) for fn in filenames: fp = dp / fn rel = fp.relative_to(root).as_posix() add_file(fp, rel) for rel_dir in manifest.get("include_dirs", []): walk_dir(rel_dir) for rel_file in manifest.get("include_files", []): fp = root / rel_file if fp.is_file(): add_file(fp, rel_file.replace("\\", "/")) else: skipped.append({"path": rel_file, "reason": "file_missing"}) for pat in manifest.get("include_globs_root", []): for fp in root.glob(pat): if fp.is_file() and fp.parent == root: add_file(fp, fp.name) included.sort(key=lambda x: x["path"]) return included, skipped def stage_tree(root: Path, staging: Path, entries: list[dict]) -> None: staging.mkdir(parents=True, exist_ok=True) for ent in entries: src = root / ent["path"] dst = staging / ent["path"] dst.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src, dst) def make_zip(staging: Path, zip_path: Path) -> None: zip_path.parent.mkdir(parents=True, exist_ok=True) with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=6) as zf: for fp in sorted(staging.rglob("*")): if fp.is_file(): zf.write(fp, fp.relative_to(staging).as_posix()) def _rescan_tree(staging: Path) -> list[dict]: files: list[dict] = [] for fp in sorted(staging.rglob("*")): if not fp.is_file(): continue rel = fp.relative_to(staging).as_posix() files.append({"path": rel, "size": fp.stat().st_size, "sha256": _sha256_file(fp)}) return files def build_backup_bundle(*, root: Path | None = None, pull_remote: bool = True) -> dict: root = root or S.ROOT manifest_def = load_manifest() included, skipped = collect_files(root, manifest_def) ts = datetime.now().strftime("%Y%m%d_%H%M%S") bundle_id = f"workbuddy_backup_{ts}" work = S.STATE_DIR / bundle_id staging = work / "tree" stage_tree(root, staging, included) snap_summary: dict = {"enabled": False, "ok": [], "missing": [], "errors": []} if pull_remote and (manifest_def.get("remote_snapshot") or {}).get("enabled"): print("[INFO] 拉取 207 平台配置快照（Dify / Langfuse / Xinference / vLLM / 监控）…") snap_summary = pull_remote_snapshot(manifest_def, staging) print( f"[INFO] 快照完成: ok={len(snap_summary.get('ok') or [])} " f"missing={len(snap_summary.get('missing') or [])} " f"errors={len(snap_summary.get('errors') or [])}" ) # 重新扫描 staging（含平台快照） all_files = _rescan_tree(staging) categories = categorize_files(all_files) meta = { "bundle_id": bundle_id, "created_at": datetime.now().isoformat(timespec="seconds"), "root": str(root), "remote_folder": S.BAIDUPAN_FOLDER, "final_root": S.BAIDUPAN_FINAL_ROOT, "file_count": len(all_files), "total_bytes": sum(x["size"] for x in all_files), "files": all_files, "skipped": skipped, "categories": {k: len(v) for k, v in categories.items()}, "remote_snapshot": snap_summary, "manifest_source": str(S.MANIFEST_PATH), } # 分类子目录说明 HTML cat_guides = write_category_guides(staging, categories, meta) # 总说明（先写占位，zip 后再更新 sha） guide_path = work / "备份说明.html" guide_in_tree = staging / "备份说明.html" guide_html = render_backup_guide_html(meta, categories) guide_path.write_text(guide_html, encoding="utf-8") guide_in_tree.write_text(guide_html, encoding="utf-8") # 重新计入说明文件 all_files = _rescan_tree(staging) categories = categorize_files(all_files) meta["files"] = all_files meta["file_count"] = len(all_files) meta["total_bytes"] = sum(x["size"] for x in all_files) meta["categories"] = {k: len(v) for k, v in categories.items()} meta["guide_files"] = ["备份说明.html"] + cat_guides meta_path = work / "manifest.json" meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8") readme = work / "README_backup.txt" readme.write_text( "\n".join( [ "WorkBuddy AI Infra 备份包", f"时间: {meta['created_at']}", f"文件数: {meta['file_count']}", f"总大小: {meta['total_bytes']} bytes", f"网盘终盘: /{S.BAIDUPAN_FINAL_ROOT}/{bundle_id}/", f"staging: 我的应用数据/bypy/{S.BAIDUPAN_FOLDER}/{bundle_id}/（上传后会自动挪走）", "", "请先打开：备份说明.html", "恢复: 解压 workbuddy_backup_*.zip 到任意目录即可。", "注意: 含 .env / 密钥 / 平台配置，请妥善保管网盘访问权限。", ] ) + "\n", encoding="utf-8", ) zip_path = work / f"{bundle_id}.zip" make_zip(staging, zip_path) meta["zip_path"] = str(zip_path) meta["zip_bytes"] = zip_path.stat().st_size meta["zip_sha256"] = _sha256_file(zip_path) # 用最终 sha 刷新 HTML guide_html = render_backup_guide_html(meta, categories) guide_path.write_text(guide_html, encoding="utf-8") guide_in_tree.write_text(guide_html, encoding="utf-8") # HTML 变了，zip 需重打一次以保证包内说明含 sha（体积小，可接受） make_zip(staging, zip_path) meta["zip_bytes"] = zip_path.stat().st_size meta["zip_sha256"] = _sha256_file(zip_path) meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8") meta["guide_html_path"] = str(guide_path) return meta