97 lines
4.1 KiB
Python
97 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
||
"""IMA 上传单篇 Markdown 知识笔记到 daily 知识库。
|
||
用法: python3 ima_upload_one.py "<绝对路径/summary.md>" "<完整文章标题>"
|
||
依赖环境变量: IMA_OPENAPI_CLIENTID / IMA_OPENAPI_APIKEY / IMA_DAILY_KNOWLEDGE_BASE_ID
|
||
流程: preflight -> check_repeated_names -> create_media -> cos-upload -> add_knowledge
|
||
说明: 用 Python 而非 bash 编排，避免中文文件名/引号转义问题。
|
||
退出码 2 = 文件名重复（需与用户确认保留双方或取消），非 0 均为失败。
|
||
"""
|
||
import json, os, subprocess, sys
|
||
|
||
SKILL_DIR = "/root/.hermes/skills/openclaw-imports/ima-skill"
|
||
IMA_API = os.path.join(SKILL_DIR, "ima_api.cjs")
|
||
COS_UPLOAD = os.path.join(SKILL_DIR, "knowledge-base/scripts/cos-upload.cjs")
|
||
PREFLIGHT = os.path.join(SKILL_DIR, "knowledge-base/scripts/preflight-check.cjs")
|
||
|
||
def run_node(script, args):
|
||
r = subprocess.run(["node", script] + args, capture_output=True, text=True, timeout=120)
|
||
if r.returncode != 0:
|
||
raise RuntimeError(f"{script} exit={r.returncode} stderr={r.stderr[:500]}")
|
||
return json.loads(r.stdout)
|
||
|
||
def ima_api(api_path, body):
|
||
r = subprocess.run(["node", IMA_API, api_path, json.dumps(body, ensure_ascii=False)],
|
||
capture_output=True, text=True, timeout=120)
|
||
if r.returncode != 0:
|
||
raise RuntimeError(f"ima_api {api_path} exit={r.returncode} stderr={r.stderr[:500]}")
|
||
resp = json.loads(r.stdout)
|
||
if resp.get("code") != 0:
|
||
raise RuntimeError(f"ima_api {api_path} code={resp.get('code')} msg={resp.get('msg')}")
|
||
return resp.get("data", {})
|
||
|
||
def main():
|
||
kb_id = os.environ["IMA_DAILY_KNOWLEDGE_BASE_ID"]
|
||
file_path = sys.argv[1]
|
||
title = sys.argv[2] # 完整文章标题（不含 .md）
|
||
|
||
pf = run_node(PREFLIGHT, ["--file", file_path])
|
||
if not pf.get("pass"):
|
||
raise RuntimeError(f"preflight failed: {pf}")
|
||
file_name = pf["file_name"]; media_type = pf["media_type"]
|
||
content_type = pf["content_type"]; file_size = pf["file_size"]; file_ext = pf["file_ext"]
|
||
print(f"[preflight] ok file={file_name} ext={file_ext} size={file_size} media_type={media_type}")
|
||
|
||
dup = ima_api("openapi/wiki/v1/check_repeated_names", {
|
||
"params": [{"name": file_name, "media_type": media_type}],
|
||
"knowledge_base_id": kb_id
|
||
})
|
||
is_rep = dup.get("results", [{}])[0].get("is_repeated", False) if dup.get("results") else False
|
||
if is_rep:
|
||
print(f"[check_repeated_names] REPEATED: {file_name} — 需要处理")
|
||
sys.exit(2)
|
||
print("[check_repeated_names] no duplicate")
|
||
|
||
cm = ima_api("openapi/wiki/v1/create_media", {
|
||
"file_name": file_name,
|
||
"file_size": file_size,
|
||
"content_type": content_type,
|
||
"knowledge_base_id": kb_id,
|
||
"file_ext": file_ext
|
||
})
|
||
media_id = cm["media_id"]
|
||
cos = cm["cos_credential"]
|
||
print(f"[create_media] media_id={media_id} cos_bucket={cos.get('bucket_name')} cos_key={cos.get('cos_key','')[:40]}")
|
||
|
||
r = subprocess.run(["node", COS_UPLOAD,
|
||
"--file", file_path,
|
||
"--secret-id", cos["secret_id"],
|
||
"--secret-key", cos["secret_key"],
|
||
"--token", cos["token"],
|
||
"--bucket", cos["bucket_name"],
|
||
"--region", cos["region"],
|
||
"--cos-key", cos["cos_key"],
|
||
"--content-type", content_type,
|
||
"--start-time", str(cos.get("start_time", "")),
|
||
"--expired-time", str(cos.get("expired_time", "")),
|
||
"--timeout", "300000"
|
||
], capture_output=True, text=True, timeout=360)
|
||
if r.returncode != 0:
|
||
raise RuntimeError(f"cos-upload exit={r.returncode} stderr={r.stderr[:800]}")
|
||
print(f"[cos-upload] ok rc=0 stdout={r.stdout.strip()[:200]}")
|
||
|
||
ak = ima_api("openapi/wiki/v1/add_knowledge", {
|
||
"media_type": media_type,
|
||
"media_id": media_id,
|
||
"title": title,
|
||
"knowledge_base_id": kb_id,
|
||
"file_info": {
|
||
"cos_key": cos["cos_key"],
|
||
"file_size": file_size,
|
||
"file_name": file_name
|
||
}
|
||
})
|
||
print(f"[add_knowledge] ok media_id={ak.get('media_id') or media_id}")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|