feat(knowledge): 导入脚本能指定环境与账号,生产也导了一份
CI / backend (push) Canceled after 0s

同一套备忘录要在 dev 和生产各导一次,所以把写死的东西参数化:
--api 指定目标环境、--phone 指定记到谁名下、--workspace 指定工作区名。

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

生产(openclaw-01)实测:13605513683 名下「我的备忘」84 条(83 发布 + 1 归档)、
18 个分类,时间戳与 dev 一致。
This commit is contained in:
Steven
2026-10-31 00:46:01 +08:00
parent 039e5b903c
commit bbd30ec85e
+21 -7
View File
@@ -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)