PrimeHub 市场发版火箭筒源码精讲

#!/usr/bin/env python3
"""PrimeHub 发布客户端（curl transport 版）

2026-10-19 实测：本机 python urllib 对 primehub SSL 握手会被服务端掐断
（UNEXPECTED_EOF_WHILE_READING），curl 正常 → 全链路改走 subprocess curl。

关键端点（均 https://primehub.primebot.com）：
  POST /app-api/content-server/file/upload                multipart 明文 + Bearer → fileUrl
  POST /app-api/content-server/skillCenter/publishSkill   SM4 加密体（本模块 call()）
  POST /app-api/search-server/portal/querySkillList       SM4 加密体（免登录可查）
    · 参数名是 PC 端约定：keywords / queryType / page / rows（pageNum/pageSize 会 9999）
  POST /app-api/content-server/skillCenter/queryPageByUserCode  我的发布列表

用法：
  python3 primehub_publish.py search '<keywords>'
  python3 primehub_publish.py publish <rpk路径> '<publishJson>'
"""
import json, uuid, time, subprocess, urllib.parse, sys, os
import argparse
from pathlib import Path

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from primehub_client import KEY, IV  # SM4 key/iv + unscramble
from extract_token import DEFAULT_TOKEN_FILE
from gmssl.sm4 import CryptSM4, SM4_ENCRYPT, SM4_DECRYPT

BASE = "https://primehub.primebot.com"
TOKEN_FILE = DEFAULT_TOKEN_FILE   # 提取器与发布器共用 workspace/.local，支持 PRIMEHUB_TOKEN_FILE


def call(path, params, token=None, ns="content-server"):
    params = dict(params); params["timestamp"] = int(time.time() * 1000)
    params["requestId"] = str(uuid.uuid4())
    inner = urllib.parse.quote(json.dumps(params, separators=(",", ":")), safe="")
    c = CryptSM4(); c.set_key(KEY, SM4_ENCRYPT)
    ct = c.crypt_cbc(IV, inner.encode()).hex()
    result = subprocess.run(
        ["curl", "-sS", "--max-time", "40", "-X", "POST",
         BASE + "/app-api/" + ns + "/" + path,
         "-H", "Content-Type: application/json", "-H", "request-t: true",
         "-H", "Origin: " + BASE, "-H", "Referer: " + BASE + "/",
         ] + (["-H", "Authorization: Bearer " + token] if token else [])
        + ["-d", json.dumps({"data": ct})],
        capture_output=True, text=True, encoding="utf-8")
    if result.returncode:
        raise RuntimeError("PrimeHub 网络请求失败，curl exit=%d" % result.returncode)
    d = json.loads(result.stdout)
    if isinstance(d.get("data"), str) and d.get("code") == 1000:
        c2 = CryptSM4(); c2.set_key(KEY, SM4_DECRYPT)
        d["data"] = json.loads(urllib.parse.unquote(
            c2.crypt_cbc(IV, bytes.fromhex(d["data"])).decode()))
    return d


def upload(rpk_path, token):
    boundary = "----" + uuid.uuid4().hex
    fn = os.path.basename(rpk_path)
    part = ("--" + boundary + "\r\nContent-Disposition: form-data; name=\"file\"; "
            "filename=\"" + fn + "\"\r\nContent-Type: application/octet-stream\r\n\r\n")
    body = part.encode() + open(rpk_path, "rb").read() + ("\r\n--" + boundary + "--\r\n").encode()
    out = subprocess.run(
        ["curl", "-s", "--max-time", "120", "-X", "POST",
         BASE + "/app-api/content-server/file/upload",
         "-H", "Content-Type: multipart/form-data; boundary=" + boundary,
         "-H", "Authorization: Bearer " + token, "-H", "request-t: true",
         "-H", "Origin: " + BASE, "-H", "Referer: " + BASE + "/",
         "--data-binary", "@-"], input=body, capture_output=True).stdout.decode()
    d = json.loads(out)
    if isinstance(d.get("data"), str):
        c = CryptSM4(); c.set_key(KEY, SM4_DECRYPT)
        d["data"] = json.loads(urllib.parse.unquote(
            c.crypt_cbc(IV, bytes.fromhex(d["data"])).decode()))
    return d["data"]["fileUrl"]


def my_skills(token):
    r = call("skillCenter/queryPageByUserCode", {"page": 1, "rows": 50}, token)
    if r.get("code") != 1000:
        raise RuntimeError("PrimeHub 我的技能查询失败：%s %s" % (r.get("code"), r.get("msg")))
    return (r.get("data") or {}).get("records") or []


def delete(skill_id, token):
    return call("skillCenter/deleteSkillById", {"id": skill_id}, token)


def search(keywords, token=None):
    r = call("portal/querySkillList", {"keywords": keywords, "queryType": 1, "page": 1, "rows": 30},
             token, ns="search-server")
    if r.get("code") != 1000:
        raise RuntimeError("PrimeHub 搜索失败：%s %s" % (r.get("code"), r.get("msg")))
    recs = (r.get("data") or {}).get("records") or []
    return [{"id": s.get("id"), "name": s.get("skillName"), "models": s.get("robotModels")} for s in recs]


def main():
    parser = argparse.ArgumentParser(description="PrimeHub 搜索、发布和管理")
    commands = parser.add_subparsers(dest="command", required=True)
    commands.add_parser("search").add_argument("keywords")
    commands.add_parser("my")
    commands.add_parser("delete").add_argument("skill_id", type=int)
    publish_parser = commands.add_parser("publish")
    publish_parser.add_argument("rpk")
    metadata = publish_parser.add_mutually_exclusive_group(required=True)
    metadata.add_argument("publish_json", nargs="?")
    metadata.add_argument("--metadata-file", type=Path)
    args = parser.parse_args()

    if args.command == "search":
        for s in search(args.keywords):
            print(s["id"], s["name"], s["models"])
        return
    token = Path(TOKEN_FILE).read_text(encoding="utf-8").strip()
    if args.command == "my":
        for s in my_skills(token):
            print(s.get("id"), s.get("skillName"))
    elif args.command == "delete":
        print(delete(args.skill_id, token))
    elif args.command == "publish":
        meta = json.loads(args.metadata_file.read_text(encoding="utf-8")
                          if args.metadata_file else args.publish_json)
        meta["packageUrl"] = upload(args.rpk, token)
        print(call("skillCenter/publishSkill", meta, token))


if __name__ == "__main__":
    main()