From 039e5b903c45020e6eb9d7591f9c179ed3289e0a Mon Sep 17 00:00:00 2001 From: Steven Date: Sat, 31 Oct 2026 00:22:08 +0800 Subject: [PATCH] =?UTF-8?q?feat(knowledge):=20=E6=97=A7=20memos=20?= =?UTF-8?q?=E5=A4=87=E5=BF=98=E6=90=AC=E8=BF=9B=E7=9F=A5=E8=AF=86=E5=BA=93?= =?UTF-8?q?=E7=9A=84=E5=AF=BC=E5=85=A5=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 一次性的搬迁脚本：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 个分类、语义检索可命中。 --- backend/scripts/import_memo_app.py | 261 +++++++++++++++++++++++++++++ 1 file changed, 261 insertions(+) create mode 100644 backend/scripts/import_memo_app.py diff --git a/backend/scripts/import_memo_app.py b/backend/scripts/import_memo_app.py new file mode 100644 index 0000000..213d4ac --- /dev/null +++ b/backend/scripts/import_memo_app.py @@ -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())