guanghulab/agents/module-steward/steward.py

#!/usr/bin/env python3
"""
模块管家 · Module Steward · ICE-GL-ZY-AGT-MODULE-004

铸渊需要部署模块？管家知道每个模块的编号、接口、部署方式。
从module-registry.json加载，提供快速查询。

用法:
  python3 steward.py list                    # 列出所有模块
  python3 steward.py query image-studio      # 查询指定模块
  python3 steward.py deploy image-studio     # 返回部署命令
"""

import json
import os
import sys
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent.parent

# 模块知识库（从仓库结构中自动发现 + 手动补充）
MODULE_KNOWLEDGE = {
    "image-studio": {
        "id": "ICE-GL-ZY-MOD-IMAGE-001",
        "name": "神笔马良图片工作室",
        "description": "文本→封面自动排版渲染。小红书/即刻/海报。",
        "deploy": "cd /data/image-studio && npm install && node server.js",
        "port": 3912,
        "entry": "image-studio/server.js",
        "dependencies": ["node", "npm", "puppeteer"],
        "server_pref": "BS-SG-001",
    },
    "gatekeeper": {
        "id": "ICE-GL-ZY-AGT-001",
        "name": "Gatekeeper 巡检工",
        "description": "远程命令执行网关，部署在所有服务器上",
        "deploy": "cd /opt/zhuyuan && node gatekeeper.js",
        "port": "3910/3911",
        "server_pref": "all",
    },
    "exe-engine": {
        "id": "ICE-GL-ZY-MOD-EXE-001",
        "name": "任务执行引擎",
        "description": "铸渊的行动力核心，执行部署和运维任务",
        "deploy": "cd /opt/zhuyuan/exe-engine && node index.js",
        "entry": "exe-engine/",
    },
    "bridge": {
        "id": "ICE-GL-ZY-MOD-BRIDGE-001",
        "name": "Chat-to-Agent 桥接器",
        "description": "语言层与执行层的连线",
        "entry": "bridge/",
    },
    "connectors": {
        "id": "ICE-GL-ZY-MOD-CONN-001",
        "name": "Notion/Git 连接器",
        "description": "Notion双向同步 + Git桥接",
        "entry": "connectors/",
    },
    "hldp": {
        "id": "ICE-GL-ZY-MOD-HLDP-001",
        "name": "HLDP语言协议",
        "description": "铸渊母语协议 + 意识编码引擎",
        "entry": "hldp/",
    },
    "agents": {
        "id": "ICE-GL-ZY-MOD-AGENTS-001",
        "name": "情报Agent小分队",
        "description": "密钥猎手·路径探子·服务器哨兵·模块管家·门桥接",
        "entry": "agents/",
        "deploy": "python3 agents/{agent}/agent.py",
    },
}

def load_registry():
    """从module-registry.json加载"""
    reg_path = REPO_ROOT / "brain" / "module-registry.json"
    if reg_path.exists():
        with open(reg_path) as f:
            return json.load(f)
    return {}

def list_modules():
    """列出所有已知模块"""
    registry = load_registry()
    modules = []

    # 从注册表
    for key, mod in registry.get("modules", {}).items():
        modules.append({
            "key": key,
            "name": mod.get("name", key),
            "status": mod.get("status", "unknown"),
            "description": mod.get("description", ""),
        })

    # 补充知识库
    for key, info in MODULE_KNOWLEDGE.items():
        if not any(m["key"] == key for m in modules):
            modules.append({
                "key": key,
                "name": info["name"],
                "id": info.get("id"),
                "description": info.get("description", ""),
            })

    return {"modules": modules, "count": len(modules)}

def query_module(name):
    """查询模块详情"""
    # 先查知识库
    if name in MODULE_KNOWLEDGE:
        return MODULE_KNOWLEDGE[name]

    # 查注册表
    registry = load_registry()
    for key, mod in registry.get("modules", {}).items():
        if name in key or name in mod.get("name", ""):
            return {
                "key": key,
                "name": mod.get("name"),
                "status": mod.get("status"),
                "description": mod.get("description"),
                "entry": mod.get("path"),
                "note": "来自module-registry.json"
            }

    # 查目录
    mod_path = REPO_ROOT / name
    if mod_path.is_dir():
        return {
            "name": name,
            "exists": True,
            "path": str(mod_path),
            "files": len(list(mod_path.rglob("*"))),
        }

    return {"error": f"未找到模块: {name}"}

def get_deploy_command(name):
    """返回模块的部署命令"""
    info = query_module(name)
    if "error" in info:
        return info

    deploy = info.get("deploy")
    if deploy:
        return {"module": name, "deploy_command": deploy, "note": info.get("description", "")}

    # 尝试推断
    entry = info.get("entry", name)
    mod_path = REPO_ROOT / entry
    if mod_path.exists():
        return {
            "module": name,
            "deploy_command": f"cd /opt/zhuyuan/guanghulab/{entry} && ls",
            "note": "请确认模块的具体启动方式"
        }

    return {"module": name, "error": "无法确定部署命令，请手动检查"}

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("用法: steward.py <list|query|deploy> [参数]")
        sys.exit(1)

    action = sys.argv[1]

    if action == "list":
        print(json.dumps(list_modules(), ensure_ascii=False, indent=2))
    elif action == "query" and len(sys.argv) > 2:
        print(json.dumps(query_module(sys.argv[2]), ensure_ascii=False, indent=2))
    elif action == "deploy" and len(sys.argv) > 2:
        print(json.dumps(get_deploy_command(sys.argv[2]), ensure_ascii=False, indent=2))