run.py

#!/usr/bin/env python3
"""
Four-Agent Harness v6.4 — 入口脚本

v6.0 简化：
- 移除 adaptive_plan（LLM 元决策）
- 移除任务澄清循环（多余的 LLM 调用）
- 移除事件 JSONL 写入（MonitorTerminal 已废弃）
- 直接调用 Orchestrator.run(requirement)

v6.2 新增：
- 自动加载 .env 文件
- 输出目录管理：所有生成的项目放在 output/ 目录下

v6.4 新增：
- 酷炫终端监视器
- 自动启动/关闭监视器
- 会话日志生成
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import platform
import re
import sys
from datetime import datetime

# 确保项目根目录在 sys.path 中
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))


def load_env_file(env_path: str = ".env") -> None:
    """加载 .env 文件到环境变量。"""
    if not os.path.exists(env_path):
        return
    
    with open(env_path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            if "=" in line:
                key, value = line.split("=", 1)
                key = key.strip()
                value = value.strip()
                if key and not os.environ.get(key):
                    os.environ[key] = value


def generate_project_name(requirement: str) -> str:
    """根据需求生成项目名称。"""
    requirement = requirement.lower()
    
    game_keywords = {
        "贪吃蛇": "snake-game",
        "俄罗斯方块": "tetris",
        "五子棋": "gomoku",
        "赛车": "racing",
        "飞机": "aircraft",
        "打砖块": "breakout",
        "游戏": "game",
    }
    
    other_keywords = {
        "计算器": "calculator",
        "记事本": "notepad",
        "时钟": "clock",
        "报时": "time-announcer",
        "hello": "hello-world",
        "web": "web-app",
        "api": "api-server",
        "博客": "blog",
        "聊天": "chat",
        "待办": "todo",
        "文件": "file-manager",
        "图片": "image-tool",
        "音乐": "music-player",
        "视频": "video-player",
    }
    
    for key, name in game_keywords.items():
        if key in requirement:
            return name
    
    for key, name in other_keywords.items():
        if key in requirement:
            return name
    
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    return f"project_{timestamp}"


def main():
    # 自动加载 .env 文件
    script_dir = os.path.dirname(os.path.abspath(__file__))
    load_env_file(os.path.join(script_dir, ".env"))
    
    parser = argparse.ArgumentParser(
        description="Four-Agent Harness v6.0 — 流式验证架构",
    )
    parser.add_argument("requirement", nargs="?", help="自然语言需求描述")
    parser.add_argument("--name", "-n", default=None, help="项目名称（默认根据需求自动生成）")
    parser.add_argument("--project-dir", "-d", default=None, help="项目目录（默认 output/<项目名>）")
    parser.add_argument("--config", "-c", default=None, help="配置文件路径（JSON）")
    parser.add_argument("--verbose", "-v", action="store_true", help="详细日志")
    parser.add_argument("--list", "-l", action="store_true", help="列出 output 目录下的所有项目")
    parser.add_argument("--no-monitor", action="store_true", help="禁用终端监视器")
    parser.add_argument("--clarify", action="store_true", help="启用需求澄清模式，通过多轮提问明确需求")
    args = parser.parse_args()

    # 列出已有项目
    if args.list:
        output_dir = os.path.join(script_dir, "output")
        if os.path.exists(output_dir):
            projects = sorted(os.listdir(output_dir))
            if projects:
                print("\n已生成的项目:")
                for p in projects:
                    project_path = os.path.join(output_dir, p)
                    if os.path.isdir(project_path):
                        print(f"  - {p}")
            else:
                print("\n暂无已生成的项目")
        else:
            print("\noutput 目录不存在")
        sys.exit(0)
    
    # 检查是否提供了需求
    if not args.requirement:
        parser.error("请提供需求描述，或使用 --list 查看已有项目")

    # 配置日志
    level = logging.DEBUG if args.verbose else logging.INFO
    logging.basicConfig(
        level=level,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        datefmt="%H:%M:%S",
    )

    # 加载配置（默认加载项目目录下的 config.json）
    config = None
    config_path = args.config
    if config_path is None:
        # 自动查找配置文件：先项目目录，再脚本所在目录
        for candidate in [
            os.path.join(script_dir, "config.json"),
        ]:
            if os.path.exists(candidate):
                config_path = candidate
                break

    if config_path and os.path.exists(config_path):
        with open(config_path, "r", encoding="utf-8") as f:
            config = json.load(f)
        logging.info("已加载配置: %s", config_path)
    else:
        logging.warning("未找到配置文件，使用默认配置")

    # 导入并运行
    from orchestrator import Orchestrator, DEFAULT_CONFIG

    # 需求澄清模式
    requirement = args.requirement
    if args.clarify:
        try:
            from requirements_clarifier import clarify_requirement
            llm_config = (config or DEFAULT_CONFIG).get("llm", {})
            clarified = clarify_requirement(
                requirement,
                llm_config=llm_config,
                output_dir=os.path.join(script_dir, "output"),
            )
            requirement = clarified["requirement"]
            print("\n" + "=" * 60)
            print("使用澄清后的需求继续执行...")
            print("=" * 60)
        except ImportError as e:
            logging.warning(f"需求澄清模块导入失败: {e}，使用原始需求")
        except KeyboardInterrupt:
            print("\n\n用户取消澄清，使用原始需求继续执行...")
            requirement = args.requirement

    # 确定项目目录
    if args.project_dir:
        project_dir = os.path.abspath(args.project_dir)
    else:
        # 自动生成项目名和目录
        project_name = args.name or generate_project_name(requirement)
        output_dir = os.path.join(script_dir, "output")
        os.makedirs(output_dir, exist_ok=True)
        project_dir = os.path.join(output_dir, project_name)
        
        # 如果目录已存在，添加序号
        if os.path.exists(project_dir):
            counter = 1
            while os.path.exists(f"{project_dir}_{counter}"):
                counter += 1
            project_dir = f"{project_dir}_{counter}"

    os.makedirs(project_dir, exist_ok=True)

    # 预检查：API Key 是否存在
    llm_config = (config or DEFAULT_CONFIG).get("llm", {})
    api_key_env = llm_config.get("api_key_env", "DEEPSEEK_API_KEY")
    api_key = os.environ.get(api_key_env, "")

    if not api_key:
        print(f"\n❌ 错误: 环境变量 {api_key_env} 未设置", file=sys.stderr)
        print(f"\n请先设置 API Key:", file=sys.stderr)
        if platform.system() == "Windows":
            print(f"  set {api_key_env}=your-api-key-here", file=sys.stderr)
        else:
            print(f"  export {api_key_env}=your-api-key-here", file=sys.stderr)
        print(f"\n或者通过 --config 指定其他 Provider:", file=sys.stderr)
        print(f"  python run.py \"你的需求\" --config path/to/config.json", file=sys.stderr)
        print(f"\nconfig.json 中可用的 Provider:", file=sys.stderr)
        print(f"  deepseek  → 环境变量 DEEPSEEK_API_KEY", file=sys.stderr)
        print(f"  openai    → 环境变量 OPENAI_API_KEY", file=sys.stderr)
        print(f"  anthropic → 环境变量 ANTHROPIC_API_KEY", file=sys.stderr)
        print(f"  gemini    → 环境变量 GOOGLE_API_KEY", file=sys.stderr)
        sys.exit(1)

    logging.info("=" * 60)
    logging.info("Four-Agent Harness v6.4 — Stream 架构 + 终端监视器")
    logging.info("=" * 60)
    logging.info("需求: %s", requirement)
    logging.info("项目目录: %s", project_dir)
    logging.info("Provider: %s / %s", llm_config.get("provider", "?"), llm_config.get("model", "?"))
    logging.info("终端监视器: %s", "禁用" if args.no_monitor else "启用")
    logging.info("=" * 60)

    monitor_bridge = None
    external_monitor = None
    if not args.no_monitor:
        try:
            from monitor_bridge import create_monitor_bridge
            token_budget = (config or DEFAULT_CONFIG).get("task_token_budget", 500000)
            monitor_bridge = create_monitor_bridge(
                project_root=project_dir,
                token_budget=token_budget,
                enable_terminal=True,
            )
            monitor_bridge.start("task-" + datetime.now().strftime("%Y%m%d%H%M%S"), requirement)
            external_monitor = monitor_bridge.monitor
        except Exception as e:
            logging.warning(f"终端监视器启动失败: {e}")
            monitor_bridge = None

    orch = Orchestrator(project_root=project_dir, config=config, monitor=external_monitor)
    result = orch.run(requirement)

    if monitor_bridge:
        monitor_bridge.stop(result.get("status", "completed"))

    # 输出结果
    print("\n" + "=" * 60)
    print("执行结果")
    print("=" * 60)
    print(json.dumps(result, indent=2, ensure_ascii=False))

    # 返回退出码
    if result.get("status") == "completed":
        sys.exit(0)
    else:
        sys.exit(1)


if __name__ == "__main__":
    main()