feat(knowledge): 旧 memos 备忘搬进知识库的导入脚本

一次性的搬迁脚本：memo_app（MySQL）→ 知识库。放在 scripts/ 下是为了留个凭据，说明这 84 条备忘是怎么变成知识的，以及哪些东西刻意没搬。 - 走 HTTP 接口而不是直连库：直接 INSERT 会绕过版本号、审计和切片导入，导进去的知识搜不到——检索只认 published 且有 chunk 的文档。 - 分类照搬两层（接口上限三级），标签照搬，全部按 note 发布，归档那条保持归档。 - 正文里的图片文件在旧机器上已经找不到了，所以把 ![](path) 换成一行「原图未迁移：…（旧路径 …）」，不留碎图，也不丢掉原路径。 - 按标题去重、分类复用，中断后可以重跑；token 缓存到 .local/ 避免撞上验证码节流。 - 旧库时间戳是 UTC，接口不接受客户端指定 created_at，所以额外生成一条 SQL 把 item.created_at/updated_at 与 published_at 对回原值。实测：84 条（83 published + 1 archived）、18 个分类、语义检索可命中。
2026-10-31 00:26:14 +08:00
parent 8b6594bf10
commit 039e5b903c
1 changed files with 261 additions and 0 deletions
@@ -0,0 +1,261 @@
+"""把旧 memo_app（MySQL）里的备忘录导进达道Task的知识库。
+
+一次性数据迁移，但保留成脚本：同一套数据以后可能还要再导一次（换环境、重导），
+手点 84 次不现实，也不该让"导进去的东西为什么缺一块"变成口头传说。
+
+设计上的几个决定：
+
+1. **走 HTTP API，不直连数据库**。直接 INSERT 会绕过版本、审计和检索切片，
+   看起来更快，但导进去的知识搜不到——那等于把数据搬进了一个看不见的柜子。
+   发布（publish）会顺带建切片索引，所以每条都走一次 publish。
+2. **分类按层级先建父后建子**，挪过来仍是最多三层（旧库只有两层）。
+   已有同名分类就复用，重跑不会长出第二棵一模一样的树。
+3. **时间戳最后单独补**。API 不让人指定 created_at（本该如此），但备忘录的时间
+   就是内容的一部分：一条 5 月写的配方不该显示成"今天写的"。所以建完之后用一条
+   UPDATE 把 created_at/updated_at 改回原值；脚本只生成 SQL 文件，执行与否由人定。
+4. **图片不搬**。旧库的 uploads 在服务器上已经找不到（容器没了），正文里的
+   `![](...)` 会让页面上一堆碎图图标，所以换成一行说明，把原始路径留在那里。
+5. **可以被中断、可以重跑**。按标题去重，导到一半断了再跑一次只会补没进去的。
+
+用法：
+
+    backend/.venv/Scripts/python.exe scripts/import_memo_app.py --data .local/memo-export
+
+然后按它打印的路径跑那条 SQL 补时间戳。
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+BASE = "http://127.0.0.1:8100/api/v1"
+PHONE = "13900000001"
+CODE = "123456"
+WORKSPACE_NAME = "我的备忘"
+DOC_TYPE = "note"
+ROOT = Path(__file__).resolve().parents[2]
+TOKEN_FILE = ROOT / ".local" / "memo-token.txt"
+
+IMAGE_RE = re.compile(r"!\[([^\]]*)\]\(([^)\s]+)\)")
+
+
+def _client() -> httpx.Client:
+    return httpx.Client(base_url=BASE, timeout=120.0)
+
+
+def _data(resp: httpx.Response) -> Any:
+    payload = resp.json()
+    if payload.get("code") != 0:
+        raise RuntimeError(f"{resp.request.method} {resp.request.url.path} → {payload}")
+    return payload.get("data")
+
+
+def login(client: httpx.Client) -> dict[str, str]:
+    """拿令牌。开发环境发验证码有 60 秒节流，重跑脚本很容易撞上，
+    所以令牌缓存到文件里；缓存失效再走一次验证码。"""
+    cache = Path(TOKEN_FILE)
+    cache.parent.mkdir(parents=True, exist_ok=True)
+    if cache.exists():
+        auth = {"Authorization": f"Bearer {cache.read_text(encoding='utf-8').strip()}"}
+        if client.get("/auth/me", headers=auth).status_code == 200:
+            return auth
+
+    client.post("/auth/sms/send", json={"phone": PHONE}).raise_for_status()
+    payload = client.post("/auth/login", json={"phone": PHONE, "code": CODE}).json()
+    if payload.get("code") != 0:
+        raise SystemExit(
+            f"登录失败：{payload.get('message')}（开发环境发码有 " f"{60} 秒节流，等一会儿再跑）"
+        )
+    token = payload["data"]["tokens"]["access_token"]
+    cache.write_text(token, encoding="utf-8")
+    return {"Authorization": f"Bearer {token}"}
+
+
+def workspace(client: httpx.Client, auth: dict[str, str]) -> tuple[str, set[str]]:
+    """找到（或建出）工作区，并带出里面已有的知识标题，用来跳过已导入的。"""
+    rows = _data(client.get("/workspaces", headers=auth))["items"]
+    ws_id = next((row["id"] for row in rows if row["name"] == WORKSPACE_NAME), None)
+    if ws_id is None:
+        created = _data(client.post("/workspaces", headers=auth, json={"name": WORKSPACE_NAME}))
+        return created["id"], set()
+
+    titles: set[str] = set()
+    cursor = None
+    while True:
+        params = {"limit": 100}
+        if cursor:
+            params["cursor"] = cursor
+        page = _data(client.get(f"/workspaces/{ws_id}/knowledge", headers=auth, params=params))
+        titles.update(row["title"] for row in page["items"])
+        cursor = page.get("next_cursor")
+        if not cursor:
+            break
+    return ws_id, titles
+
+
+def clean_body(body: str) -> str:
+    """把正文里的图片引用换成一行说明。
+
+    旧库的图片文件已经不在服务器上了（uploads 目录随容器一起没了），留着 markdown
+    只会渲染成一排碎图。原始路径写进注释，将来真找回来了还能对得上。
+    """
+
+    def repl(match: re.Match[str]) -> str:
+        target = match.group(2)
+        name = target.rsplit("/", 1)[-1]
+        return f"> 原图未迁移：{name}（旧路径 {target}）"
+
+    return IMAGE_RE.sub(repl, body or "").strip()
+
+
+def _depth_of(row: dict[str, Any], by_id: dict[int, dict[str, Any]]) -> int:
+    depth = 1
+    parent = row["parent_id"]
+    seen = set()
+    while parent and parent in by_id and parent not in seen:
+        seen.add(parent)
+        depth += 1
+        parent = by_id[parent]["parent_id"]
+    return depth
+
+
+def create_categories(
+    client: httpx.Client, auth: dict[str, str], ws_id: str, old: list[dict[str, Any]]
+) -> dict[int, str]:
+    """按层级从浅到深建；同名同父已存在就复用。返回 旧 id → 新 id。"""
+    existing = _data(client.get(f"/workspaces/{ws_id}/knowledge/categories", headers=auth))
+    known: dict[tuple[str | None, str], str] = {
+        (row["parent_id"], row["name"].lower()): row["id"] for row in existing["items"]
+    }
+
+    by_id = {row["id"]: row for row in old}
+    mapping: dict[int, str] = {}
+    for row in sorted(old, key=lambda r: (_depth_of(r, by_id), r["sort_order"], r["id"])):
+        parent_new = mapping.get(row["parent_id"]) if row["parent_id"] else None
+        key = (parent_new, row["name"].strip().lower())
+        if key in known:
+            mapping[row["id"]] = known[key]
+            continue
+        created = _data(
+            client.post(
+                f"/workspaces/{ws_id}/knowledge/categories",
+                headers=auth,
+                json={
+                    "name": row["name"].strip(),
+                    "parent_id": parent_new,
+                    "sort_order": row["sort_order"],
+                },
+            )
+        )
+        mapping[row["id"]] = created["id"]
+        known[key] = created["id"]
+    return mapping
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--data", default=".local/memo-export", help="导出的 JSON 目录")
+    parser.add_argument("--sql-out", default=".local/memo-timestamps.sql")
+    parser.add_argument("--limit", type=int, default=0, help="只导前 N 条（试跑用）")
+    args = parser.parse_args()
+
+    root = Path(args.data)
+    notes = json.loads((root / "notes.json").read_text(encoding="utf-8"))
+    categories = json.loads((root / "categories.json").read_text(encoding="utf-8"))
+    tags = json.loads((root / "tags.json").read_text(encoding="utf-8"))
+    note_tags = json.loads((root / "note_tags.json").read_text(encoding="utf-8"))
+
+    tag_name = {row["id"]: row["name"] for row in tags}
+    tags_of: dict[int, list[str]] = {}
+    for row in note_tags:
+        tags_of.setdefault(row["note_id"], []).append(tag_name[row["tag_id"]])
+
+    # 旧库有两个用户，只有 steven(1) 有笔记；别人的分类不搬，免得空分类挂在树上
+    owners = {row["user_id"] for row in notes}
+    categories = [row for row in categories if row["user_id"] in owners]
+
+    timestamps: list[str] = []
+    failed: list[str] = []
+    done = 0
+
+    with _client() as client:
+        auth = login(client)
+        ws_id, existing_titles = workspace(client, auth)
+        print(f"工作区 {ws_id}，已有 {len(existing_titles)} 条知识")
+
+        cat_map = create_categories(client, auth, ws_id, categories)
+        print(f"分类 {len(cat_map)} 个")
+
+        ordered = sorted(notes, key=lambda r: r["id"])
+        queue = [n for n in ordered if n["title"] not in existing_titles]
+        if args.limit:
+            queue = queue[: args.limit]
+        print(f"待导入 {len(queue)} 条")
+
+        for index, note in enumerate(queue, start=1):
+            title = (note["title"] or "").strip() or "无标题备忘"
+            payload = {
+                "title": title[:200],
+                "body_md": clean_body(note["content"] or ""),
+                "summary": (note["summary"] or "").strip(),
+                "doc_type": DOC_TYPE,
+                "tag_names": tags_of.get(note["id"], []),
+                "category_id": cat_map.get(note["category_id"]),
+            }
+            try:
+                doc = _data(
+                    client.post(f"/workspaces/{ws_id}/knowledge", headers=auth, json=payload)
+                )
+                action = "archive" if note["is_archived"] else "publish"
+                _data(
+                    client.post(
+                        f"/workspaces/{ws_id}/knowledge/{doc['id']}/status",
+                        headers=auth,
+                        json={"action": action, "version": doc["version"]},
+                    )
+                )
+            except Exception as error:  # 一条失败不该拖垮整批，记下来最后一起看
+                failed.append(f"#{note['id']} {title}: {error}")
+                continue
+
+            done += 1
+            timestamps.append(
+                f"('{doc['id']}', '{note['created_at']}+00', '{note['updated_at']}+00')"
+            )
+            if index % 10 == 0:
+                print(f"  … {index}/{len(queue)}")
+
+    sql = [
+        "-- 把导入时的 now() 改回笔记原本的时间（dev 库执行，dbadmin 可绕过 RLS）",
+        "BEGIN;",
+        "CREATE TEMP TABLE memo_ts (item_id uuid PRIMARY KEY, created_at timestamptz,"
+        " updated_at timestamptz) ON COMMIT DROP;",
+        "INSERT INTO memo_ts (item_id, created_at, updated_at) VALUES",
+        ",\n".join(timestamps) + ";",
+        "UPDATE item i SET created_at = t.created_at, updated_at = t.updated_at"
+        " FROM memo_ts t WHERE i.id = t.item_id;",
+        "UPDATE knowledge_doc k SET published_at = t.updated_at,"
+        " archived_at = CASE WHEN k.status_hint = 'archived' THEN t.updated_at ELSE NULL END"
+        " FROM memo_ts t WHERE k.item_id = t.item_id;",
+        "COMMIT;",
+    ]
+    Path(args.sql_out).write_text("\n".join(sql) + "\n", encoding="utf-8")
+
+    print(f"导入完成 {done} 条；时间戳 SQL → {args.sql_out}")
+    if failed:
+        print("这些没进去：")
+        for line in failed:
+            print("  -", line)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())