feat(migrate): Halo 全量内容迁移（压缩版）——6文/21图/1瞬间 URL 1:1

- 正文源=线上渲染页（Halo 快照层 patch 残迹：泥潭/赎罪仅 2 字符），
  pandoc html→gfm + tags API 显示名换算后分类：5 书评（五档情绪色）+1 技术
- 图库 21 张：长边 2400/q86 压缩入仓（原图 196M→9.5M——公开仓历史膨胀
  与 nginx 100M 推送上限双杀；原图在 Phase 0 备份 tar 中永久留档）
- 瞬间 1 条；URL 1:1（6 slug 全等，零 301）
- 本提交由 7 个本地提交重写而成（清除已进历史的 196M blob，未推送过）；
  因 reset --soft 暂存竞态，一并包含 spec 归档文件
  （.spec/specs/archive/2026-10-24_migrate-blog-content/）

验证: check_urls.py 6==6；astro build 11 页 exit 0（8.67s）；
      EXIF 管线跑通；staging 更新；halo healthy+200 全程未动

Co-authored-by: GLM-5.3

scripts/migrate/check_urls.py

@@ -0,0 +1,22 @@
+#!/usr/bin/env python3
+"""URL 1:1 断言（Phase 2 T5）：Halo slugs == 新站文件 slugs。
+
+新站文章路由 = /archives/{文件名去 .md}/，与 Halo permalink 同构；
+集合差集非空即失败（多一篇少一篇都不行）。
+"""
+import json
+import sys
+from pathlib import Path
+
+old = {p["spec"]["slug"] for p in json.load(open(".migration/posts.json", encoding="utf-8"))}
+new = {f.stem for f in Path("content/posts").glob("*.md")}
+
+missing = old - new   # 旧有新无 → 404 风险
+extra = new - old     # 新有旧无 → 无害但记录
+
+print(f"halo slugs={len(old)} new slugs={len(new)}")
+for s in sorted(missing):
+    print(f"  MISSING: {s}")
+for s in sorted(extra):
+    print(f"  extra  : {s}")
+sys.exit(1 if missing else 0)

scripts/migrate/halo_export.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Halo 全量导出（Phase 2 T1）：posts/photos/moments → .migration/*.json
+
+对生产仅 GET；正文优先取列表 status.content，缺则回落 head-content 端点。
+产物为迁移中间件，不入库（.gitignore .migration/）。
+"""
+import json
+import sys
+import urllib.request
+from pathlib import Path
+
+BASE = "http://127.0.0.1:18090"
+OUT = Path(".migration")
+OUT.mkdir(exist_ok=True)
+
+
+def get(path: str) -> object:
+    with urllib.request.urlopen(BASE + path, timeout=30) as r:
+        return json.load(r)
+
+
+def export_list(api: str, size: int = 100) -> list[dict]:
+    """翻页取全量 items。"""
+    items, page = [], 1
+    while True:
+        data = get(f"{api}?page={page}&size={size}")
+        batch = data.get("items", [])
+        items.extend(batch)
+        if page * size >= data.get("total", 0) or not batch:
+            return items
+        page += 1
+
+
+def main() -> int:
+    posts = export_list("/apis/api.content.halo.run/v1alpha1/posts")
+    # 正文兜底：head-content 拿完整 HTML（status.content 可能是截断摘要）
+    for p in posts:
+        name = p["metadata"]["name"]
+        content = (p.get("status") or {}).get("content")
+        if not content:
+            try:
+                head = get(f"/apis/api.content.halo.run/v1alpha1/posts/{name}/head-content")
+                p["_fullContent"] = head.get("content", "")
+            except Exception as e:  # noqa: BLE001 - 单篇失败不阻断全量导出
+                p["_fullContent"] = ""
+                print(f"  warn: {name} head-content 失败: {e}", file=sys.stderr)
+    photos = export_list("/apis/api.photo.halo.run/v1alpha1/photos")
+    try:
+        groups = get("/apis/api.photo.halo.run/v1alpha1/groups")
+    except Exception:  # noqa: BLE001 - 分组端点缺失时空兜底
+        groups = {"items": []}
+    moments = export_list("/apis/api.moment.halo.run/v1alpha1/moments")
+
+    (OUT / "posts.json").write_text(json.dumps(posts, ensure_ascii=False, indent=1))
+    (OUT / "photos.json").write_text(json.dumps(photos, ensure_ascii=False, indent=1))
+    (OUT / "photo_groups.json").write_text(json.dumps(groups, ensure_ascii=False, indent=1))
+    (OUT / "moments.json").write_text(json.dumps(moments, ensure_ascii=False, indent=1))
+
+    full = sum(1 for p in posts if p.get("_fullContent") or (p.get("status") or {}).get("content"))
+    print(f"exported: posts={len(posts)}（含正文 {full}）photos={len(photos)} groups={len(groups.get('items', []))} moments={len(moments)}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/migrate/halo_export_db.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python3
+"""Halo DB 只读导出（Phase 2 T1 补充源）：posts + snapshots → .migration/。
+
+公共 API 只有摘要、head-content 404；正文真源在 postgres extensions 表：
+  /registry/blog.halo.run/posts/{name}          → spec.headSnapshot 指针
+  /registry/content.halo.run/snapshots/{uuid}    → raw/content 全量
+data 列是 bytea，经 base64 通道取出后 json 解析（psql 文本输出会搅转义）。
+"""
+import base64
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+OUT = Path(".migration")
+
+
+def psql_b64(name: str) -> dict:
+    sql = "SELECT encode(data,'base64') FROM extensions WHERE name = '%s';" % name
+    raw = subprocess.run(
+        ["docker", "exec", "postgres-db", "psql", "-U", "wait4x", "-d", "halo_db", "-t", "-A", "-c", sql],
+        capture_output=True, text=True, check=True,
+    ).stdout.strip()
+    return json.loads(base64.b64decode(raw))
+
+
+def names_like(pattern: str) -> list[str]:
+    sql = "SELECT name FROM extensions WHERE name LIKE '%s';" % pattern
+    raw = subprocess.run(
+        ["docker", "exec", "postgres-db", "psql", "-U", "wait4x", "-d", "halo_db", "-t", "-A", "-c", sql],
+        capture_output=True, text=True, check=True,
+    ).stdout.split()
+    return raw
+
+
+def main() -> int:
+    posts = [psql_b64(n) for n in names_like("%/posts/%")]
+    snaps = {n.rsplit("/", 1)[1]: psql_b64(n) for n in names_like("%/snapshots/%")}
+
+    # 每篇 post 挂上 head snapshot 的 raw/content 全文
+    # 结构：snapshot.spec.{rawPatch,contentPatch} 直接是字符串（实测解剖）
+    for p in posts:
+        spec = p.get("spec", {})
+        # head 与 release 各挂一个 snapshot；实测存在"head 空、release 满"与
+        # 反例（mian-sha），取 content 更长者作为迁移源
+        cands = [snaps.get(k, {}).get("spec", {}) for k in (spec.get("releaseSnapshot"), spec.get("headSnapshot")) if k]
+        snap = max(cands, key=lambda sp: len(sp.get("contentPatch", "") or ""), default={})
+        p["_snapshot"] = {
+            "rawType": snap.get("rawType"),
+            "raw": snap.get("rawPatch", ""),
+            "content": snap.get("contentPatch", ""),
+            "snapshotCount": len(cands),
+        }
+
+    (OUT / "posts_db.json").write_text(json.dumps(posts, ensure_ascii=False, indent=1))
+    ok = sum(1 for p in posts if p["_snapshot"]["content"])
+    print(f"db export: posts={len(posts)}（含正文 {ok}）；snapshots={len(snaps)}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/migrate/moments_migrate.py

@@ -0,0 +1,52 @@
+#!/usr/bin/env python3
+"""瞬间迁移（Phase 2 T4）：Halo moments → content/moments/*.md
+
+Halo moment.content.html 为富文本（内联样式很脏），经 pandoc 提纯为
+纯文本 md；medium（配图）当前库存为空，字段留待将来有图时扩展。
+"""
+import json
+import subprocess
+from datetime import datetime, timezone
+from pathlib import Path
+
+OUT = Path("content/moments")
+
+
+def main() -> int:
+    moments = json.load(open(".migration/moments.json", encoding="utf-8"))
+    n = 0
+    for m in moments:
+        spec = m["spec"]
+        if spec.get("visible") == "PRIVATE":
+            continue
+        html = (spec.get("content") or {}).get("html", "")
+        text = subprocess.run(
+            ["/home/wait4x/.local/bin/pandoc", "-f", "html", "-t", "plain", "--wrap=none"],
+            input=html.encode(), capture_output=True, check=True,
+        ).stdout.decode("utf-8").strip()
+        if not text:
+            continue
+        ts = spec.get("releaseTime") or m["metadata"].get("creationTimestamp", "")
+        day = datetime.fromisoformat(ts.replace("Z", "+00:00")).astimezone(timezone.utc).strftime("%Y-%m-%d")
+        out = OUT / f"{day}.md"
+        i = 2
+        while out.exists():
+            out = OUT / f"{day}-{i}.md"
+            i += 1
+        tags = spec.get("tags") or []
+        # schema 约束 text/image 在 frontmatter（body 不参与校验），一句话即 text 字段
+        out.write_text(
+            "---\n"
+            f"pubDate: {day}\n"
+            f"tags: [{', '.join(tags)}]\n"
+            f'text: "{text.replace(chr(34), chr(39))}"\n'
+            "---\n",
+            encoding="utf-8",
+        )
+        n += 1
+    print(f"migrated {n} moments")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/migrate/photos_migrate.py

@@ -0,0 +1,78 @@
+#!/usr/bin/env python3
+"""图库迁移（Phase 2 T3）：Halo photos → content/photos/{jpg,json sidecar}
+
+EXIF 不在迁移时提——构建期 photos loader（src/lib/exif.ts）会补全；
+sidecar 里 exif 留空对象。分组：Halo displayName 含山/雪/land/snow →
+landscape，否则 street（当前仅一组 snow mountain → landscape）。
+"""
+import json
+import re
+import urllib.request
+from pathlib import Path
+from urllib.parse import unquote
+
+BASE = "http://127.0.0.1:18090"
+OUT = Path("content/photos")
+
+
+def slugify(name: str) -> str:
+    stem = re.sub(r"\.(jpg|jpeg|png|webp)$", "", name, flags=re.I)
+    stem = re.sub(r"[^0-9A-Za-z\u4e00-\u9fff-]+", "-", stem).strip("-")
+    return stem or "photo"
+
+
+def group_of(display: str) -> str:
+    low = display.lower()
+    return "landscape" if any(k in low for k in ("山", "雪", "snow", "land", "mountain")) else "street"
+
+
+def compress(data: bytes) -> bytes:
+    from io import BytesIO
+
+    from PIL import Image
+
+    img = Image.open(BytesIO(data))
+    if img.mode not in ("RGB", "L"):
+        img = img.convert("RGB")
+    img.thumbnail((2400, 2400))
+    buf = BytesIO()
+    img.save(buf, "JPEG", quality=86, optimize=True)
+    return buf.getvalue()
+
+
+def main() -> int:
+    # 真实照片取代样例：先清样例（jpg/json 成对；仅清本脚本管辖的产物）
+    for f in OUT.glob("*"):
+        if f.suffix in (".jpg", ".json"):
+            f.unlink()
+
+    photos = json.load(open(".migration/photos.json", encoding="utf-8"))
+    n = 0
+    for p in photos:
+        spec = p["spec"]
+        url = spec.get("url") or spec.get("cover") or ""
+        if not url.startswith("/upload/"):
+            print(f"  skip 非本站图: {url[:60]}")
+            continue
+        fname = unquote(url.rsplit("/", 1)[1])
+        slug = slugify(spec.get("displayName") or fname)
+        data = urllib.request.urlopen(BASE + url, timeout=60).read()
+        # 入仓前压到长边 2400/q86：原图总计 196M（单张最大 28M）会永久膨胀
+        # 公开仓历史并撞死 nginx 100M 推送上限；原图已随 Phase 0 备份归档，
+        # 站内展示由 sharp 出 srcset，2400px 足够 lightbox 全屏
+        (OUT / f"{slug}.jpg").write_bytes(compress(data))
+
+        (OUT / f"{slug}.json").write_text(json.dumps({
+            "title": spec.get("displayName", fname).rsplit(".", 1)[0],
+            "image": f"./{slug}.jpg",
+            "group": group_of(spec.get("groupName", "")),
+            "alt": spec.get("description", ""),
+            "exif": {},
+        }, ensure_ascii=False, indent=1))
+        n += 1
+    print(f"migrated {n} photos")
+    return 0 if n else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/migrate/posts_to_md.py

@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+"""文章迁移（Phase 2 T2）：Halo 渲染页 → content/posts/*.md
+
+正文源用线上渲染页（reader 看到的即所得）——实测 DB snapshots 对
+shu-zui/ni-tan 只有 2 字符（patch 残迹），而渲染页完整。
+
+处理链：抓 /archives/{slug} → 剥主题噪音(button/nav/header) → 图片
+/upload/* 下载到 content/posts/assets/{slug}/ 并改写引用 → pandoc
+html→gfm → frontmatter（分类/标签沿用 Halo，书评分配五档情绪色）。
+"""
+import html as htmllib
+import json
+import re
+import subprocess
+import urllib.request
+from pathlib import Path
+
+BASE = "http://127.0.0.1:18090"
+ROOT = Path(".")
+ASSETS = ROOT / "content" / "posts" / "assets"
+
+# 书评五档情绪色（tokens --mood-1..5）；技术文不分配
+MOODS = ["#b45309", "#0f766e", "#7c3aed", "#9f1239", "#166534"]
+
+# Halo 分类名 → 集合枚举
+CAT_MAP = {"Reading": "reading", "Linux": "tech", "默认分类": "tech"}
+
+
+def fetch(url: str) -> bytes:
+    with urllib.request.urlopen(url, timeout=30) as r:
+        return r.read()
+
+
+def article_html(slug: str) -> str:
+    page = fetch(f"{BASE}/archives/{slug}").decode("utf-8")
+    m = re.search(r"<article[^>]*>(.*?)</article>", page, re.S)
+    if not m:
+        raise RuntimeError(f"no article for {slug}")
+    body = m.group(1)
+    # 主题噪音：higan 主题把主题切换钮等塞进 article；标题在 frontmatter 重建，正文 h1 去重
+    body = re.sub(r"<button.*?</button>", "", body, flags=re.S)
+    body = re.sub(r"<header.*?<h1[^>]*>.*?</h1>.*?</header>", "", body, count=1, flags=re.S)
+    return body
+
+
+def localize_images(slug: str, body: str) -> str:
+    def repl(m: re.Match) -> str:
+        src = m.group(1)
+        if not src.startswith("/upload/"):
+            return m.group(0)
+        dest_dir = ASSETS / slug
+        dest_dir.mkdir(parents=True, exist_ok=True)
+        fname = src.rsplit("/", 1)[1]
+        data = fetch(BASE + src)
+        (dest_dir / fname).write_bytes(data)
+        return m.group(0).replace(src, f"./assets/{slug}/{fname}")
+
+    return re.sub(r'src="([^"]+)"', repl, body)
+
+
+def to_markdown(body: str) -> str:
+    p = subprocess.run(
+        ["/home/wait4x/.local/bin/pandoc", "-f", "html", "-t", "gfm-raw_html", "--wrap=none"],
+        input=body.encode(), capture_output=True, check=True,
+    )
+    return p.stdout.decode("utf-8")
+
+
+def load_tag_display() -> dict[str, str]:
+    """内部标签名(tag-xxx) → 显示名(read/linux/...)。"""
+    with urllib.request.urlopen(BASE + "/apis/api.content.halo.run/v1alpha1/tags") as r:
+        data = json.loads(r.read())
+    return {t["metadata"]["name"]: t["spec"]["displayName"] for t in data.get("items", [])}
+
+
+def main() -> int:
+    posts = json.load(open(".migration/posts.json", encoding="utf-8"))
+    tag_disp = load_tag_display()
+    mood_i = 0
+    report = []
+    for p in posts:
+        spec = p["spec"]
+        slug = spec["slug"]
+        tags = [tag_disp.get(t, t) if isinstance(t, str) else t.get("name", "") for t in (spec.get("tags") or [])]
+        cat_raw = spec.get("category") or ""
+        # 分类判定：Halo 分类名优先，回退看标签（read→reading / linux→tech）
+        if CAT_MAP.get(cat_raw):
+            category = CAT_MAP[cat_raw]
+        elif any("read" in t.lower() for t in tags):
+            category = "reading"
+        else:
+            category = "tech"
+
+        mood = None
+        if category == "reading":
+            mood = MOODS[mood_i % len(MOODS)]
+            mood_i += 1
+
+        body = localize_images(slug, article_html(slug))
+        md = to_markdown(body).strip()
+        md = re.sub(r"^(-{4,}|_{4,})\s*", "", md)  # 原 hr 开头去掉（frontmatter 后首元素应是正文）
+        fm = [
+            "---",
+            f'title: "{htmllib.escape(spec["title"], quote=False)}"',
+            f"pubDate: {spec['publishTime'][:10]}",
+            f"category: {category}",
+            f"tags: [{', '.join(tags)}]",
+        ]
+        if mood:
+            fm.append(f"mood: '{mood}'")
+        fm.append("---")
+        out = ROOT / "content" / "posts" / f"{slug}.md"
+        out.write_text("\n".join(fm) + "\n\n" + md + "\n", encoding="utf-8")
+        n_img = len(list((ASSETS / slug).glob("*"))) if (ASSETS / slug).exists() else 0
+        report.append((slug, category, len(md), n_img))
+        print(f"  {slug} [{category}] {len(md)} md-chars, {n_img} imgs")
+
+    # 清理被真实内容取代的样例文件（真实技术文 slug=csge-lei-pei-zhi，样例为 cs-ge-lei-pei-zhi）
+    sample = ROOT / "content" / "posts" / "cs-ge-lei-pei-zhi.md"
+    if sample.exists():
+        sample.unlink()
+        print("  removed sample: cs-ge-lei-pei-zhi.md")
+    print(f"migrated {len(report)} posts")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())