Files
DadaoTask/backend/scripts/import_memo_app.py
T
Steven bbd30ec85e
CI / backend (push) Canceled after 0s
feat(knowledge): 导入脚本能指定环境与账号,生产也导了一份
同一套备忘录要在 dev 和生产各导一次,所以把写死的东西参数化:
--api 指定目标环境、--phone 指定记到谁名下、--workspace 指定工作区名。

令牌缓存按「环境 + 手机号」分文件(.local/memo-token-<phone>-<host>.txt):
之前只按账号存,混着用会出现"刚登录完下一个请求就 401"——dev 的令牌拿不到生产的库。

生产(openclaw-01)实测:13605513683 名下「我的备忘」84 条(83 发布 + 1 归档)、
18 个分类,时间戳与 dev 一致。
2026-10-31 00:46:01 +08:00

276 lines
11 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""把旧 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())