kangshifu-board/scripts/audit-voc-skill-source.py

"""Inspect the installed VOC package as text/JSON only; no package code executes."""
import argparse
import collections
import datetime as dt
import hashlib
import json
from pathlib import Path
import re


FILES = ['package.json', 'SKILL.md', 'mcp/src/server.js',
         'mcp/src/core/api-catalog.js', 'mcp/src/core/budget-guard.js',
         'mcp/src/tools/voc-api-catalog-run.js', 'mcp/src/providers/voc-gateway.js',
         'mcp/src/providers/ecommerce-gateway.js', 'mcp/src/providers/overseas-gateway.js',
         'mcp/src/features/voc-competitor-map/competitor-map.js',
         'mcp/src/features/media/media-pipeline.js', 'mcp/catalog/voc-social-endpoints.json',
         'mcp/catalog/params/jd.json', 'mcp/catalog/params/taobao.json']


def inspect(root):
    root = Path(root).resolve()
    sources = {name: (root / name).read_bytes() for name in FILES}
    text = {name: value.decode('utf-8-sig') for name, value in sources.items()}
    package = json.loads(text['package.json'])
    if package.get('name') != '@vocmarket/voc-skill':
        raise ValueError('NOT_VOC_SKILL_PACKAGE')
    catalog = json.loads(text['mcp/catalog/voc-social-endpoints.json'])
    endpoints = catalog['endpoints']
    tools = re.findall(r"server\.registerTool\(\s*['\"]([^'\"]+)", text['mcp/src/server.js'])
    selected = []
    for endpoint in endpoints:
        if endpoint.get('platform') in ('jd', 'taobao'):
            store = json.loads(text['mcp/catalog/params/' + endpoint['platform'] + '.json'])
            selected.append({**{k: endpoint.get(k) for k in ('id','platform','title','method','proxyPath','channel',
                                                           'deprecated','responseHint','recommendedFor','freshness')},
                             'params': endpoint.get('params') or store.get(endpoint['id'], [])})
    root_skill_dirs = root / 'skills'
    subskills = sorted(p.parent.name for p in root_skill_dirs.glob('*/SKILL.md'))
    # A catalog match is discovery evidence only, never an online test result.
    rank_candidates = [{k:e.get(k) for k in ('id','platform','title','channel')} for e in endpoints
                       if e.get('platform') in ('jd','taobao') and
                       re.search(r'rank|ranking|榜单|排行|top.?100|top.?400',
                                 json.dumps(e, ensure_ascii=False), flags=re.I)]
    return {
        'schema': 'ksf.voc-skill-source-audit.v1', 'generatedAt': dt.datetime.now(dt.timezone.utc).isoformat(),
        'package': package['name'], 'version': package['version'], 'sourceRoot': str(root),
        'inspectionMode': 'static-text-and-json', 'packageCodeExecuted': False,
        'networkRequests': 0, 'collectionCalls': 0, 'credentialsRead': False,
        'files': [{'path': name, 'bytes': len(data), 'sha256': hashlib.sha256(data).hexdigest()}
                  for name, data in sources.items()],
        'toolCount': len(tools), 'tools': tools, 'subskills': subskills,
        'catalogEndpointCount': len(endpoints),
        'platformCounts': dict(collections.Counter(e['platform'] for e in endpoints)),
        'gatewayBaseUrls': {k:v.get('baseUrl') for k,v in catalog.get('gateway',{}).items()},
        'domesticEcommerceEndpoints': selected, 'jdTaobaoRankingCandidates': rank_candidates,
        'liveToolRegistrationVerified': False, 'accountPermissionVerified': False,
        'budgetVerified': False, 'liveCollectionVerified': False,
        'integration': {
            'discovery': 'voc_api_search → voc_api_doc（只读本地目录/参数）',
            'collection': 'voc_api_call；已登记京东/淘宝接口channel=ecommerce，自动路由',
            'cost': 'voc_cost_estimate / voc_billing_query / voc_usage_report；目录billing非人民币价',
            'storage': '返回结构化JSON、用量跟踪与本地媒体；不自动写入本项目研究库',
            'intake': '采集结果保留来源 → 九类batch映射校验 → 暂存 → 受限SSH dry/apply/status → 新代际回读',
            'limitation': '通用单接口有默认重试，不能据成本子技能文档声称每次调用都已硬限预算；试采须固定范围并设retries=0',
            'rankings': '当前本地京东/淘宝目录未据此证明完整Top100/400能力；需另核真实榜单端点',
            'analysis': '竞品图谱含通用候选及兜底示例；无真实证据不能当企业、店铺或商品事实',
        },
    }


def markdown(report):
    lines = ['# 本地VOC Skill源码能力核对', '',
             f"来源：{report['sourceRoot']}；{report['package']} {report['version']}。", '',
             f"本轮仅静态读取源码与目录：{report['toolCount']}个注册工具、{len(report['subskills'])}个子技能、{report['catalogEndpointCount']}条目录接口。未执行源码、未读凭据、未请求采集。", '',
             '## 可复用能力', '', *['- '+v for v in report['integration'].values()], '',
             '## 京东与淘宝/天猫的已登记操作', '',
             '|接口|通道|参数名（*为必填）|说明|', '|---|---|---|---|']
    for e in report['domesticEcommerceEndpoints']:
        params = '、'.join(p['name']+('*' if p.get('required') else '') for p in e['params'])
        note = ('已弃用；' if e.get('deprecated') else '') + (e.get('responseHint') or e['title'])
        lines.append('|'+e['id']+'|'+str(e['channel'])+'|'+params+'|'+note.replace('|','／').replace('\n',' ')+'|')
    lines += ['', '## 真实运行仍需核实', '',
              '- 目标Agent是否实际注册同版MCP工具、可用入口、账号权限与费用额度。源码存在不等于目标实例可调用。',
              '- 店铺ID/sellerId/userId是不同接口参数，不猜填、不互换；主店与平台稳定性待真实分操作实测。',
              '- 商品字段映射按真实响应核定；同平台不同详情版本的价格、销量和属性不能相互冒用。',
              '- 当前复用0.3.38，不自动升级、安装或触发付费接口。完整源文件指纹与参数见source-audit.json。', '']
    return '\n'.join(lines)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--source-root', required=True)
    parser.add_argument('--out', required=True)
    args = parser.parse_args()
    out = Path(args.out).resolve()
    if out.exists():
        parser.error('OUTPUT_DIRECTORY_ALREADY_EXISTS')
    root = Path(args.source_root).resolve()
    if out.is_relative_to(root):
        parser.error('OUTPUT_INSIDE_SOURCE_PACKAGE')
    report = inspect(root)
    out.mkdir(parents=True, mode=0o700)
    files = {'source-audit.json': json.dumps(report, ensure_ascii=False, indent=2)+'\n',
             'VOC能力核对.md': markdown(report)}
    for name, value in files.items():
        with (out/name).open('x',encoding='utf-8',newline='\n') as stream:
            stream.write(value)
        (out/name).chmod(0o600)
    with (out/'FILES.sha256').open('x',encoding='utf-8',newline='\n') as stream:
        for name in sorted(files):
            stream.write(hashlib.sha256((out/name).read_bytes()).hexdigest()+'  '+name+'\n')
    print(json.dumps({'ok': True,'version':report['version'],'tools':report['toolCount'],
                      'catalogEndpoints':report['catalogEndpointCount'],
                      'domesticEcommerceEndpoints':len(report['domesticEcommerceEndpoints']),
                      'report':str(out/'VOC能力核对.md'),'collectionCalls':0},ensure_ascii=False))


if __name__ == '__main__':
    main()