xiaowei-system/scripts/npc-dispatch.py

#!/usr/bin/env python3
"""
CNB CodeBuddy NPC 看板调度器
通过 Issue @npc/CodeBuddy 触发云端编码

用法:
    # 创建仓库
    python3 npc-dispatch.py --create-repo muchen-org/test-repo

    # 触发任务（自动建 Issue + @CodeBuddy 启动）
    python3 npc-dispatch.py "实现一个 HTTP 服务器" --repo muchen-org/test-repo

    # 带验收标准
    python3 npc-dispatch.py "写 REST API" --repo muchen-org/test --criteria "有单元测试" "支持 POST/GET"
"""

import argparse
import json
import os
import sys
import urllib.error
import urllib.request
from pathlib import Path

API = "https://api.cnb.cool"


def load_token():
    """从 ~/.hermes/.env 读取 CNB_TOKEN"""
    env_path = Path.home() / ".hermes" / ".env"
    try:
        with open(env_path) as f:
            for line in f:
                line = line.strip()
                if line.startswith("CNB_TOKEN="):
                    return line.split("=", 1)[1].strip()
    except FileNotFoundError:
        pass
    return os.environ.get("CNB_TOKEN", "")


def api_call(method, path, data=None):
    """CNB API 统一调用"""
    token = load_token()
    if not token:
        print("❌ CNB_TOKEN 未配置", file=sys.stderr)
        sys.exit(1)

    url = f"{API}{path}"
    headers = {
        "Authorization": f"Bearer {token}",
        "accept": "application/json",
        "Content-Type": "application/json",
    }
    body = json.dumps(data).encode("utf-8") if data else None
    req = urllib.request.Request(url, data=body, headers=headers, method=method)

    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            content = resp.read().decode()
            # 空响应（201 Created）直接返回 None
            if not content.strip():
                return {"status": resp.status, "empty": True}
            return json.loads(content)
    except urllib.error.HTTPError as e:
        error_body = e.read().decode()
        print(f"❌ HTTP {e.code}: {error_body}", file=sys.stderr)
        sys.exit(1)


def create_repo(org, name, description="", visibility="public"):
    """创建仓库"""
    result = api_call("POST", f"/{org}/-/repos", {
        "name": name,
        "description": description,
        "visibility": visibility,
    })
    # 201 Created 返回空 body，用 path/name 构造结果
    if result.get("empty"):
        return {"path": f"{org}/{name}", "name": name, "visibility": visibility}
    return result


def create_issue_with_npc(repo, task_title, task_body, criteria=None, work_mode=True):
    """创建 Issue 并触发 NPC"""
    # 构造 body
    body_parts = [f"## 任务描述\n{task_body}\n"]
    if criteria:
        body_parts.append("## 验收标准\n")
        for i, c in enumerate(criteria, 1):
            body_parts.append(f"{i}. {c}")
    body = "\n".join(body_parts)

    # 创建 Issue（带 work_mode: true）
    issue = api_call("POST", f"/{repo}/-/issues", {
        "title": f"@npc/CodeBuddy 替我上班：{task_title}",
        "body": body,
        "labels": ["enhancement"],
        "priority": "P2",
        "work_mode": work_mode,
    })
    print(f"✅ Issue 已创建: #{issue.get('number')} - {issue.get('title')}")

    # 立即发评论再次触发（保险起见）
    issue_number = issue.get("iid") or issue.get("number")
    comment = api_call("POST", f"/{repo}/-/issues/{issue_number}/comments", {
        "body": "@npc/CodeBuddy 替我上班，请开始执行这个任务。",
        "work_mode": True,
    })
    print(f"✅ 触发评论已发: NPC 约 3 分钟内开始响应")

    return {
        "issue_number": issue_number,
        "title": issue.get("title"),
        "url": issue.get("html_url"),
        "state": issue.get("state"),
    }


def dispatch_to_npc(task_desc, repo="", criteria=None):
    """主入口：分派任务给 NPC"""
    if not repo:
        repo = os.environ.get("CNB_DEFAULT_REPO", "muchen-org/auto-npc")

    # 确保仓库存在
    repo_info = None
    try:
        repo_info = api_call("GET", f"/{repo}")
    except SystemExit:
        print(f"📦 仓库 {repo} 不存在，尝试创建...")
        if not repo or "/" not in repo:
            print("❌ 需要有效的 org/repo 格式", file=sys.stderr)
            sys.exit(1)
        org, name = repo.split("/", 1)
        create_repo(org, name, description="Auto-created by 小唯看板")
        repo_info = api_call("GET", f"/{repo}")

    # 提取简短标题
    short_title = task_desc[:60] + ("..." if len(task_desc) > 60 else "")
    return create_issue_with_npc(repo, short_title, task_desc, criteria)


def main():
    parser = argparse.ArgumentParser(description="CNB CodeBuddy NPC 看板调度器")
    parser.add_argument("task", nargs="?", help="任务描述")
    parser.add_argument("--repo", help="目标仓库 (org/name)")
    parser.add_argument("--criteria", nargs="+", help="验收标准（多项）")
    parser.add_argument("--create-repo", metavar="ORG/NAME", help="创建仓库")
    parser.add_argument("--no-work-mode", action="store_true", help="禁用 work_mode（仅编码不能 push）")

    args = parser.parse_args()

    if args.create_repo:
        org, name = args.create_repo.split("/", 1)
        result = create_repo(org, name)
        print(f"✅ 仓库已创建: {result.get('path') or args.create_repo}")
        return

    if not args.task:
        parser.print_help()
        sys.exit(1)

    result = dispatch_to_npc(args.task, args.repo, args.criteria)
    print(json.dumps(result, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()