"""把旧 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] IMAGE_RE = re.compile(r"!\[([^\]]*)\]\(([^)\s]+)\)") def _token_file() -> Path: """令牌按「环境 + 手机号」分文件存：dev 和生产的令牌互不通用， 混着写会出现"刚登录完下一个请求又 401"这种很难看的事。""" host = BASE.split("//", 1)[-1].split("/", 1)[0] return ROOT / ".local" / f"memo-token-{PHONE}-{host.replace(':', '-')}.txt" 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]: """拿令牌。dev 发验证码有 60 秒节流，重跑脚本很容易撞上， 所以令牌缓存到文件里；缓存失效再走一次验证码。""" cache = _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], name: 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"] == name), None) if ws_id is None: created = _data(client.post("/workspaces", headers=auth, json={"name": 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: global BASE, PHONE 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 条（试跑用）") parser.add_argument("--api", default=BASE, help="目标环境的 API 根地址") parser.add_argument("--phone", default=PHONE, help="知识记到哪个手机号名下") parser.add_argument("--workspace", default=WORKSPACE_NAME, help="工作区名字，没有就新建") args = parser.parse_args() BASE = args.api.rstrip("/") PHONE = args.phone workspace_name = args.workspace 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, workspace_name) 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())