# -*- coding: utf-8 -*- """生成备份说明 HTML(每个时间戳子目录一份)。""" from __future__ import annotations import html from collections import defaultdict from pathlib import Path def _cat_for(path: str) -> str: p = path.replace("\\", "/").lower() if p.startswith("platform_snapshot/dify/") or "/dify_" in p or p.startswith("dify") or "dify_" in Path(p).name: if "langfuse" in p: return "langfuse" return "dify" if p.startswith("platform_snapshot/langfuse/") or "langfuse" in p: return "langfuse" if p.startswith("platform_snapshot/xinference/") or "xinference" in p or "bge-m3" in p or "rerank" in p: return "xinference" if p.startswith("platform_snapshot/vllm/") or "vllm" in p or "qwen" in p: return "vllm" if p.startswith("platform_snapshot/monitoring/") or "作业8" in path or "prometheus" in p or "grafana" in p: return "monitoring" if p.startswith("platform_snapshot/"): return "platform_snapshot" if p.startswith("docs/"): return "docs" if p.startswith("scripts_cursor/") or p.startswith("scripts_bash_cursor/") or p.startswith("patches_cursor/"): return "programs" if p.startswith("data/"): return "data" if p.endswith(".env") or p.endswith(".yaml") or p.endswith(".yml") or "config" in Path(p).name.lower(): return "configs" if p.startswith("tools/") or p.startswith("keep/") or p.startswith(".cursor/"): return "tools" return "other" CATEGORY_META = { "dify": ("Dify 相关", "控制台、Agent 快照、Docker 编排、诊断与修复脚本"), "langfuse": ("Langfuse 相关", "观测平台 .env、compose、连接 Dify 的脚本与密钥文件"), "xinference": ("Xinference 相关", "嵌入/重排模型接口探测与相关脚本"), "vllm": ("vLLM / GPU 推理", "Qwen 推理服务探测、启动脚本与 GPU 快照"), "monitoring": ("SLA 监控", "Prometheus / Grafana / Exporter 配置"), "platform_snapshot": ("207 平台快照", "从服务器拉取的运行时配置与状态"), "docs": ("文档", "Markdown / HTML 专题与手册"), "programs": ("程序与补丁", "scripts_cursor / scripts_bash_cursor / patches_cursor"), "configs": ("配置与密钥", "本地 .env、token、密钥文件(请妥善保管)"), "data": ("运行数据", "作业产出与本地 data/"), "tools": ("工具与其它", "tools / keep / Cursor rules"), "other": ("其它", "未归类文件"), } def categorize_files(files: list[dict]) -> dict[str, list[dict]]: buckets: dict[str, list[dict]] = defaultdict(list) for ent in files: buckets[_cat_for(ent["path"])].append(ent) for k in buckets: buckets[k].sort(key=lambda x: x["path"]) return dict(buckets) def _fmt_size(n: int) -> str: if n < 1024: return f"{n} B" if n < 1024 * 1024: return f"{n / 1024:.1f} KB" return f"{n / (1024 * 1024):.2f} MB" def render_backup_guide_html(meta: dict, categories: dict[str, list[dict]]) -> str: bundle_id = html.escape(meta.get("bundle_id", "")) created = html.escape(meta.get("created_at", "")) remote = html.escape(meta.get("remote_folder", "AIinfra_Cursor")) file_count = meta.get("file_count", 0) total_bytes = meta.get("total_bytes", 0) zip_bytes = meta.get("zip_bytes", 0) zip_sha = html.escape(meta.get("zip_sha256", "") or "") snap = meta.get("remote_snapshot") or {} cat_sections = [] order = [ "dify", "langfuse", "xinference", "vllm", "monitoring", "platform_snapshot", "docs", "programs", "configs", "data", "tools", "other", ] for key in order: items = categories.get(key) or [] if not items: continue title, desc = CATEGORY_META.get(key, (key, "")) rows = [] for ent in items[:200]: rows.append( "%s%s" % (html.escape(ent["path"]), _fmt_size(int(ent.get("size", 0)))) ) more = "" if len(items) > 200: more = f"

… 另有 {len(items) - 200} 个文件,详见 manifest.json

" cat_sections.append( f"""

{html.escape(title)} {len(items)}

{html.escape(desc)}

{''.join(rows)}
路径大小
{more}
""" ) snap_html = "" if snap: ok_n = len(snap.get("ok") or []) miss_n = len(snap.get("missing") or []) err_n = len(snap.get("errors") or []) snap_html = f"""

207 平台配置快照

拉取时间:{html.escape(str(snap.get('pulled_at', '')))};成功 {ok_n},缺失 {miss_n},错误 {err_n}。

目录 platform_snapshot/ 内含 Dify / Langfuse / Xinference / vLLM / 监控相关配置与运行时探测结果。

""" skipped = meta.get("skipped") or [] skip_html = "" if skipped: skip_rows = "".join( "%s%s" % (html.escape(s.get("path", "")), html.escape(str(s.get("reason", "")))) for s in skipped[:50] ) skip_html = f"""

跳过项

{skip_rows}
路径原因
""" nav = "".join( f'{html.escape(CATEGORY_META.get(k, (k,))[0])}' for k in order if categories.get(k) ) return f""" 备份说明 · {bundle_id}

WorkBuddy AI Infra 备份说明

时间戳目录:{bundle_id} · 生成于 {created}

终盘路径:/{html.escape(meta.get('final_root') or remote)}/{bundle_id}/

流程:先传到「我的应用数据/bypy」→ 再移动到网盘根目录真正的 AIinfra_Cursor

{file_count}文件数
{_fmt_size(int(total_bytes))}原始大小
{_fmt_size(int(zip_bytes)) if zip_bytes else '-'}压缩包
{len(categories)}分类数
本备份含 Dify / Langfuse / Xinference / vLLM 相关配置与密钥文件,请勿公开分享网盘链接。

本次备份包含什么

解压 {bundle_id}.zip 后,目录结构与本地 WorkBuddy 根基本一致;平台快照在 platform_snapshot/。

zip sha256:{zip_sha or '-'}

{snap_html} {''.join(cat_sections)} {skip_html}
""" def write_category_guides(tree_root: Path, categories: dict[str, list[dict]], meta: dict) -> list[str]: """在 tree 下为各大类目录写简短 备份说明.html,返回写出的相对路径列表。""" written: list[str] = [] mapping = { "platform_snapshot/dify": "dify", "platform_snapshot/langfuse": "langfuse", "platform_snapshot/xinference": "xinference", "platform_snapshot/vllm": "vllm", "platform_snapshot/monitoring": "monitoring", "docs": "docs", "scripts_cursor": "programs", "patches_cursor": "programs", } all_files = meta.get("files") or [] for rel_dir, cat_key in mapping.items(): prefix = rel_dir.replace("\\", "/") + "/" items = [e for e in all_files if e["path"].replace("\\", "/").startswith(prefix)] dir_path = tree_root / rel_dir if not items and not dir_path.exists(): continue dir_path.mkdir(parents=True, exist_ok=True) title, desc = CATEGORY_META.get(cat_key, (rel_dir, "")) lis = "".join( f"
  • {html.escape(e['path'])} · {_fmt_size(int(e.get('size', 0)))}
  • " for e in items[:100] ) depth = rel_dir.count("/") + 1 back = "../" * depth + "备份说明.html" body = f""" {html.escape(title)} · 备份说明

    {html.escape(title)}

    {html.escape(desc)}

    所属备份:{html.escape(meta.get('bundle_id',''))} · {html.escape(meta.get('created_at',''))}

    本目录文件数:{len(items)}

    返回总说明

    """ out = dir_path / "备份说明.html" out.write_text(body, encoding="utf-8") written.append((rel_dir + "/备份说明.html").replace("\\", "/")) return written