openclaw-voc-skill/claude-code/claude-code-qiwe-assistant/scripts/build-catalog.py

#!/usr/bin/env python3
"""Build the enterprise WeChat endpoint catalog from markdown OpenAPI docs."""
import re, yaml, json, html, os, sys, glob

DOCS = sys.argv[1] if len(sys.argv) > 1 else os.environ.get('QIWE_DOCS_DIR')
OUT = sys.argv[2] if len(sys.argv) > 2 else os.path.join(os.path.dirname(__file__), '..', 'mcp', 'catalog', 'qiwe-endpoints.json')
if not DOCS:
    raise SystemExit('请通过第一个参数或 QIWE_DOCS_DIR 指定接口文档目录')

def strip_html(s):
    if not s: return ''
    s = re.sub(r'<[^>]+>', '', str(s))
    s = html.unescape(s)
    return re.sub(r'\s+', ' ', s).strip()

def simplify_props(schema, required=None, prefix=''):
    out = []
    props = (schema or {}).get('properties') or {}
    req = set((schema or {}).get('required') or [])
    for name, spec in props.items():
        name = str(name).strip()
        desc = strip_html(spec.get('description') or spec.get('title') or '')
        typ = spec.get('type', 'string')
        entry = {'name': prefix + name, 'in': 'body', 'type': typ, 'required': name in req}
        if desc: entry['desc'] = desc[:600]
        if typ == 'object' and spec.get('properties'):
            entry['children'] = simplify_props(spec)
        if typ == 'array' and isinstance(spec.get('items'), dict) and spec['items'].get('properties'):
            entry['children'] = simplify_props(spec['items'])
        out.append(entry)
    return out

def response_hint(post):
    try:
        schema = post['responses']['200']['content']['application/json']['schema']
    except Exception:
        return ''
    data = (schema.get('properties') or {}).get('data') or {}
    if data.get('type') == 'array':
        data = data.get('items') or {}
    fields = []
    for k, v in (data.get('properties') or {}).items():
        d = strip_html(v.get('description') or v.get('title') or '')
        fields.append(f"{str(k).strip()}{'（'+d[:120]+'）' if d else ''}")
    return ('data 字段：' + '；'.join(fields[:15])) if fields else ''

endpoints = []
skipped = []
for path in sorted(glob.glob(os.path.join(DOCS, '*.md'))):
    fname = os.path.basename(path)
    text = open(path, encoding='utf-8').read()
    m = re.search(r'```yaml\n(.*?)\n```', text, re.S)
    if not m:
        skipped.append(fname); continue
    try:
        spec = yaml.safe_load(m.group(1))
        paths = spec['paths']
        api_path = next(p for p in ('/api/qw/doApi', '/qiwe/api/qw/doApi', '/api/qw/doFileApi') if p in paths)
        post = paths[api_path]['post']
    except Exception:
        skipped.append(fname); continue
    is_file_api = 'doFileApi' in api_path
    content = post['requestBody']['content']
    content_type = 'application/json' if 'application/json' in content else next(iter(content))
    body = content[content_type].get('schema') or {}
    props = body.get('properties') or {}
    method_desc = (props.get('method') or {}).get('description', '')
    mm = re.search(r'(/[a-z]+/[A-Za-z0-9_]+)', strip_html(method_desc))
    if not mm:
        mm = re.search(r'(/[a-z]+/[A-Za-z0-9_]+)', method_desc)
    if not mm and (props.get('method') or {}).get('example'):
        mm = re.search(r'(/[a-z]+/[A-Za-z0-9_]+)', str(props['method']['example']))
    if not mm:
        mm = re.search(r'method[\'"：:\s]*[\'"]?(/[a-z]+/[A-Za-z0-9_]+)', text)
    if not mm:
        skipped.append(fname); continue
    method_path = mm.group(1)
    title = strip_html(post.get('summary') or fname[:-3])
    tags_raw = post.get('tags') or []
    category = tags_raw[0].split('/') if tags_raw else []
    params_schema = props.get('params')
    if params_schema:
        params = simplify_props(params_schema)
    else:
        flat = {'properties': {k: v for k, v in props.items() if k not in ('method', 'tokenId')},
                'required': [k for k in (body.get('required') or []) if k not in ('method', 'tokenId')]}
        params = simplify_props(flat)
    module = method_path.split('/')[1]
    ep = {
        'id': f"{module}.{method_path.split('/')[2]}",
        'module': module,
        'title': title,
        'summary': strip_html(re.sub(r'(?s)<details.*?</details>', '', post.get('description') or ''))[:300],
        'method': method_path,
        'category': [c for c in category if c],
        'docFile': fname,
        'tags': list({module, title, *[c for c in category if c]}),
        'params': params,
    }
    if is_file_api:
        ep['apiPath'] = '/api/qw/doFileApi'
        ep['contentType'] = content_type
        ep['flatParams'] = True
    hint = response_hint(post)
    if hint: ep['responseHint'] = hint[:800]
    endpoints.append(ep)

# 官方文档未公布 method 的接口，手工登记以保证清单全量可检索
MANUAL_ENDPOINTS = [
    {
        'id': 'cloud.bigFileDownloadAsync',
        'module': 'cloud',
        'title': '企微大文件异步下载',
        'summary': '文件大于 20M 使用此接口；下载地址为临时云资源（7–15 天不定期清理）；异步结果通过回调通知，根据 requestId 匹配。当前未登记 method 路由值，qiwe_api_call 暂不可调用。',
        'method': None,
        'category': ['API参考', '媒体与运营', '文件与媒体（下载/上传）'],
        'docFile': '企微大文件异步下载.md',
        'tags': ['cloud', '企微大文件异步下载', '媒体与运营', 'API参考', '大文件', '下载'],
        'params': [
            {'name': 'guid', 'in': 'body', 'type': 'string', 'required': True, 'desc': '设备唯一标识 guid'},
            {'name': 'filekey', 'in': 'body', 'type': 'string', 'required': True, 'desc': '当 fileId 以 * 开头时为空；否则不为空'},
            {'name': 'fileId', 'in': 'body', 'type': 'string', 'required': True, 'desc': '文件 Id'},
            {'name': 'fileSize', 'in': 'body', 'type': 'integer', 'required': True, 'desc': '文件大小'},
            {'name': 'fileType', 'in': 'body', 'type': 'integer', 'required': True, 'desc': '22 大视频 35 文件'},
            {'name': 'filename', 'in': 'body', 'type': 'string', 'required': True, 'desc': '文件名'}
        ],
        'responseHint': 'data 字段：requestId（请求 Id，用于在回调中匹配异步下载结果）'
    }
]
endpoints.extend(MANUAL_ENDPOINTS)

# de-dup ids
seen = {}
for ep in endpoints:
    if ep['id'] in seen:
        ep['id'] = ep['id'] + '.' + re.sub(r'\W+', '_', ep['title'])[:24]
    seen[ep['id']] = True

catalog = {
    'version': '0.2.0',
    'updatedAt': '2026-07-10',
    'source': '企业微信开放接口资料',
    'gateway': {
        'baseUrl': 'https://server.fmode.cn/api/qiwe',
        'endpoint': 'POST {baseUrl}/doApi',
        'auth': 'Authorization: Bearer <Fmode token>',
        'envelope': 'Fmode 网关请求信封：{"uid": "<本地稳定设备标识>", "method": "<接口 method 路径>", "params": {...}}；企业微信接口访问凭据和设备上下文由 Fmode 网关管理。',
        'note': '所有业务请求均通过 Fmode 网关转发的企业微信接口执行。清单未登记的普通 JSON 接口可用 qiwe_api_call 传 rawMethod + params；登录和设备接口必须使用 qiwe_login_* 专用工具。'
    },
    'modules': sorted({ep['module'] for ep in endpoints}),
    'endpoints': endpoints,
}
os.makedirs(os.path.dirname(OUT), exist_ok=True)
json.dump(catalog, open(OUT, 'w', encoding='utf-8'), ensure_ascii=False, indent=1)
print('endpoints:', len(endpoints))
print('modules:', catalog['modules'])
print('skipped:', skipped)