From bbd30ec85e55252f51a8b63f69e263b3dbcb4179 Mon Sep 17 00:00:00 2001 From: Steven Date: Sat, 31 Oct 2026 00:46:01 +0800 Subject: [PATCH] =?UTF-8?q?feat(knowledge):=20=E5=AF=BC=E5=85=A5=E8=84=9A?= =?UTF-8?q?=E6=9C=AC=E8=83=BD=E6=8C=87=E5=AE=9A=E7=8E=AF=E5=A2=83=E4=B8=8E?= =?UTF-8?q?=E8=B4=A6=E5=8F=B7=EF=BC=8C=E7=94=9F=E4=BA=A7=E4=B9=9F=E5=AF=BC?= =?UTF-8?q?=E4=BA=86=E4=B8=80=E4=BB=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 同一套备忘录要在 dev 和生产各导一次，所以把写死的东西参数化： --api 指定目标环境、--phone 指定记到谁名下、--workspace 指定工作区名。 令牌缓存按「环境 + 手机号」分文件（.local/memo-token--.txt）： 之前只按账号存，混着用会出现"刚登录完下一个请求就 401"——dev 的令牌拿不到生产的库。 生产（openclaw-01）实测：13605513683 名下「我的备忘」84 条（83 发布 + 1 归档）、 18 个分类，时间戳与 dev 一致。 --- backend/scripts/import_memo_app.py | 28 +++++++++++++++++++++------- 1 file changed, 21 insertions(+), 7 deletions(-) diff --git a/backend/scripts/import_memo_app.py b/backend/scripts/import_memo_app.py index 213d4ac..2f52ee8 100644 --- a/backend/scripts/import_memo_app.py +++ b/backend/scripts/import_memo_app.py @@ -41,11 +41,17 @@ 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 _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) @@ -58,9 +64,9 @@ def _data(resp: httpx.Response) -> Any: def login(client: httpx.Client) -> dict[str, str]: - """拿令牌。开发环境发验证码有 60 秒节流，重跑脚本很容易撞上， + """拿令牌。dev 发验证码有 60 秒节流，重跑脚本很容易撞上， 所以令牌缓存到文件里；缓存失效再走一次验证码。""" - cache = Path(TOKEN_FILE) + 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()}"} @@ -78,12 +84,12 @@ def login(client: httpx.Client) -> dict[str, str]: return {"Authorization": f"Bearer {token}"} -def workspace(client: httpx.Client, auth: dict[str, str]) -> tuple[str, set[str]]: +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"] == WORKSPACE_NAME), None) + 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": WORKSPACE_NAME})) + created = _data(client.post("/workspaces", headers=auth, json={"name": name})) return created["id"], set() titles: set[str] = set() @@ -160,12 +166,20 @@ def create_categories( 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")) @@ -187,7 +201,7 @@ def main() -> int: with _client() as client: auth = login(client) - ws_id, existing_titles = workspace(client, auth) + 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)