workphone-sdk/sdk/scripts/gen_integration_manifest.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
四端可直调·模块化接口能力清单 生成器

从运行中的 SDK 实时拉取 skill-registry（211 动作 / 49 模块 / 9 平台），
生成模块化、可直读的接口能力清单 Markdown，供存客宝 / 触客宝 / AI数智员工 /
SuperAdmin 直接对接判断「哪些能力可直调、走什么通道、经哪个 BFF/SDK 路径」。

- 不改任何兄弟仓库代码；只输出工作手机侧文档。
- 真源：GET /api/v3/ai/brain/skill-registry（live）。
- 用法：python3 sdk/scripts/gen_integration_manifest.py [--base http://127.0.0.1:8899]

输出：开发文档/5、接口/01-规范与统一层/四端可直调接口能力清单.md
"""
from __future__ import annotations

import argparse
import datetime
import json
import os
import sys
import urllib.request

ROOT = os.path.normpath(os.path.join(os.path.dirname(__file__), "..", ".."))
OUT = os.path.join(ROOT, "开发文档", "5、接口", "01-规范与统一层", "四端可直调接口能力清单.md")

# 平台 → 默认执行通道 + 四端推荐入口（不改对方代码，全部 HTTP）
PLATFORM_CHANNEL = {
    "wechat":  ("Frida Hook（主） / u2（兜底）", "/api/v3/message/* friend/* group/* tag/* moments/* + hook/execute"),
    "douyin":  ("execute-script（AI Brain）", "/api/v3/ai/brain/execute-script (script=douyin)"),
    "xhs":     ("execute-script（AI Brain）", "/api/v3/ai/brain/execute-script (script=xhs)"),
    "xianyu":  ("execute-script（AI Brain）", "/api/v3/ai/brain/execute-script (script=xianyu)"),
    "soul":    ("execute-script（AI Brain）", "/api/v3/ai/brain/execute-script (script=soul)"),
    "system":  ("Agent 系统能力", "/api/v3/devices/{id}/* "),
    "hook":    ("Hook 统一执行", "/api/v3/hook/execute"),
    "ai_brain":("AI 中台编排", "/api/v3/ai/brain/* · /api/v3/devices/{id}/ai/*"),
    "anti_ban":("防封守护", "/api/v3/anti-ban/*"),
}

# 四端可直调矩阵（来自需求 §3.8 真源；纯文档约定，不改对方代码）
FOUR_END = [
    ("存客宝 H5 (:3100)",  "经存客宝 BFF /v1/workphone/* 或联调直连 :8899", "发消息/群发/加好友/朋友圈/标签/线索/状态/中台 Skill"),
    ("触客宝 (:3101)",     "必经存客宝 BFF（同域 JWT）",                    "message/send · messages/list · moments/post · agent/execute"),
    ("AI数智员工 (:3104)", "经存客宝 OpenPlatform / proxy（不直连 :8899）",   "设备状态/截图/终端绑定 + proxy→execute-script 多平台控机"),
    ("SuperAdmin (:3103)", "只读聚合 + 连接方案切换主控台",                  "GET /devices · status · connection/provider/switch"),
]


def fetch_registry(base: str) -> dict:
    url = base.rstrip("/") + "/api/v3/ai/brain/skill-registry"
    try:
        with urllib.request.urlopen(url, timeout=10) as r:
            return json.loads(r.read().decode())
    except Exception as e:  # 离线回退到缓存
        cache = os.path.join(ROOT, "sdk", "tmp", "manifest_20260530", "skill_registry.json")
        if os.path.exists(cache):
            sys.stderr.write(f"[warn] live 拉取失败({e})，用缓存 {cache}\n")
            return json.load(open(cache, encoding="utf-8"))
        raise


def render(reg: dict, base: str) -> str:
    data = reg["data"]
    summary = data.get("summary", {})
    skills = data.get("skills", {})
    now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M")

    L = []
    L.append("---")
    L.append("tags: [工作手机, 接口, 四端对接, 能力清单, 自动生成]")
    L.append("doc-type: 索引")
    L.append("layer: 5、接口/01-规范与统一层")
    L.append('parent: "[[5、接口/README|5、接口]]"')
    L.append("obsidian-color: \"#0277BD\"")
    L.append("---\n")
    L.append("# 四端可直调 · 模块化接口能力清单")
    L.append("")
    L.append(f"> **自动生成**（勿手改）：`python3 sdk/scripts/gen_integration_manifest.py`  ")
    L.append(f"> **真源**：`GET {base}/api/v3/ai/brain/skill-registry`（live） · **生成时间**：{now}  ")
    L.append("> **铁律**：四端**只调 HTTP**；工作手机**不改**存客宝/触客宝/AI数智员工/SuperAdmin 代码；缺口走 COORD 或 `sdk/proxy` 透传。")
    L.append("")
    L.append("## 〇、总览")
    L.append("")
    L.append("| 维度 | 数量 |")
    L.append("|:---|:---:|")
    L.append(f"| 平台（skill） | {summary.get('total_skills', len(skills))} |")
    L.append(f"| 功能模块 | {summary.get('total_modules', '—')} |")
    L.append(f"| 动作（action）总计 | {summary.get('total_actions', '—')} |")
    L.append(f"| 平台业务动作 | {summary.get('platform_actions', '—')} |")
    L.append("")
    L.append("**通道图例**：`Frida Hook`=设备本机 Frida RPC（微信主通道）· `execute-script`=AI Brain 多平台脚本 · `u2`=UI 自动化兜底 · `companion`=设备端无障碍/广播模块（资料/红包等需 APK 模块或测试号）。")
    L.append("")

    # 四端入口矩阵
    L.append("## 一、四端入口矩阵（不改对方代码）")
    L.append("")
    L.append("| 端 | 调用方式 | 典型可直调能力 |")
    L.append("|:---|:---|:---|")
    for end, how, caps in FOUR_END:
        L.append(f"| **{end}** | {how} | {caps} |")
    L.append("")
    L.append("> 连接方案切换（超管主控台）：`POST /api/v3/connection/provider/switch`（jiqing/aochuang/legacy/custom_*）· 列表 `GET /api/v3/connection/providers`。")
    L.append("")

    # 平台 → 模块 → 动作
    L.append("## 二、平台 × 模块 × 动作（实时）")
    L.append("")
    for plat, sk in skills.items():
        ch, entry = PLATFORM_CHANNEL.get(plat, ("—", "—"))
        name = sk.get("name", plat) if isinstance(sk, dict) else plat
        pkg = sk.get("package", "") if isinstance(sk, dict) else ""
        mods = sk.get("modules", {}) if isinstance(sk, dict) else {}
        total = sum(len(v) if isinstance(v, (list, dict)) else 0 for v in mods.values()) if isinstance(mods, dict) else 0
        L.append(f"### {name}  `{plat}`")
        L.append("")
        L.append(f"- **包名**：`{pkg or '—'}` · **动作数**：{total} · **默认通道**：{ch}")
        L.append(f"- **统一入口**：`{entry}`")
        L.append("")
        if isinstance(mods, dict) and mods:
            L.append("| 模块 | 动作数 | 动作（action） |")
            L.append("|:---|:---:|:---|")
            for mk, acts in mods.items():
                names = acts if isinstance(acts, list) else (acts.get("actions", []) if isinstance(acts, dict) else [])
                disp = " · ".join(f"`{a}`" for a in names) if names else "—"
                L.append(f"| {mk} | {len(names)} | {disp} |")
            L.append("")

    # 统一调用范式
    L.append("## 三、统一调用范式（可直接联调）")
    L.append("")
    L.append("```http")
    L.append("# 微信（Frida 主通道）")
    L.append("POST /api/v3/hook/execute")
    L.append('{"device_id":"<serial>","platform":"wechat","action":"send_message",')
    L.append(' "params":{"to_id":"文件传输助手","content":"hi"},"hook_only":1}')
    L.append("")
    L.append("# 多平台精确控机（抖音/小红书/闲鱼/Soul）")
    L.append("POST /api/v3/ai/brain/execute-script")
    L.append('{"device_id":"<serial>","script":"douyin","action":"send_message","params":{...}}')
    L.append("")
    L.append("# 未注册 BFF 的能力 → 经存客宝 BFF 透传（不新造接口）")
    L.append("POST /v1/workphone/sdk/proxy")
    L.append('{"method":"POST","path":"/api/v3/ai/brain/execute-script","body":{...}}')
    L.append("```")
    L.append("")
    L.append("## 四、关联文档")
    L.append("")
    L.append("- 四端开放接口与对接铁律：`5、接口/02-业务对接/四端开放接口汇总与对接铁律.md`")
    L.append("- 存客宝 BFF↔SDK 映射：`5、接口/02-业务对接/存客宝BFF与工作手机SDK映射表.md`")
    L.append("- 全量接口目录：`5、接口/01-规范与统一层/工作手机API全量接口目录.md`")
    L.append("- 需求真源：`1、需求/修改/工作手机_存客宝四端对接_20260529.md` §3.8 / §十")
    L.append("- 静态对齐校验：`python3 sdk/scripts/wechat_interface_audit.py`（0 缺失）")
    L.append("")
    return "\n".join(L)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--base", default="http://127.0.0.1:8899")
    args = ap.parse_args()
    reg = fetch_registry(args.base)
    md = render(reg, args.base)
    os.makedirs(os.path.dirname(OUT), exist_ok=True)
    with open(OUT, "w", encoding="utf-8") as f:
        f.write(md)
    print(f"[ok] 写入 {OUT}（{len(md)} 字）")


if __name__ == "__main__":
    main()