ai-execution-platform/scripts/verify_setup.py

#!/usr/bin/env python3
"""
工具链完整性验证脚本
验证所有组件是否正确安装和配置
"""

import os
import sys
import json
import importlib
import subprocess
from pathlib import Path
from typing import Dict, List, Tuple, Any

class SetupVerifier:
    def __init__(self):
        self.base_dir = Path(__file__).parent.parent
        self.issues = []
        self.warnings = []

    def log_info(self, message: str):
        print(f"ℹ️  {message}")

    def log_success(self, message: str):
        print(f"✅ {message}")

    def log_warning(self, message: str):
        print(f"⚠️  {message}")
        self.warnings.append(message)

    def log_error(self, message: str):
        print(f"❌ {message}")
        self.issues.append(message)

    def check_python_version(self) -> bool:
        """检查Python版本"""
        self.log_info("检查Python版本...")

        version = sys.version_info
        if version.major == 3 and version.minor >= 11:
            self.log_success(f"Python版本: {version.major}.{version.minor}.{version.micro}")
            return True
        else:
            self.log_error(f"Python版本过低: {version.major}.{version.minor}.{version.micro} (需要3.11+)")
            return False

    def check_required_packages(self) -> bool:
        """检查必需的Python包"""
        self.log_info("检查必需的Python包...")

        required_packages = [
            'fastapi',
            'sqlalchemy',
            'alembic',
            'redis',
            'pydantic',
            'click',
            'pyyaml',
            'jinja2',
            'pytest',
            'black',
            'ruff',
            'mypy',
            'bandit'
        ]

        all_installed = True
        for package in required_packages:
            try:
                importlib.import_module(package.replace('-', '_'))
                self.log_success(f"包 {package} 已安装")
            except ImportError:
                self.log_error(f"包 {package} 未安装")
                all_installed = False

        return all_installed

    def check_directory_structure(self) -> bool:
        """检查目录结构"""
        self.log_info("检查目录结构...")

        required_dirs = [
            'tools',
            'tools/requirements-engine',
            'tools/task-generator',
            'tools/execution-engine',
            'tools/validators',
            'tools/pipeline',
            'tools/web',
            'utils',
            'prompts',
            'docs',
            'docs/process',
            'docs/engine',
            'requirements',
            'requirements/raw',
            'requirements/parsed',
            'development',
            'development/tasks',
            'reports',
            'results',
            'tests',
            'config'
        ]

        all_exist = True
        for dir_path in required_dirs:
            full_path = self.base_dir / dir_path
            if full_path.exists():
                self.log_success(f"目录 {dir_path} 存在")
            else:
                self.log_error(f"目录 {dir_path} 不存在")
                all_exist = False
                # 尝试创建目录
                try:
                    full_path.mkdir(parents=True, exist_ok=True)
                    self.log_warning(f"已创建目录: {dir_path}")
                except Exception as e:
                    self.log_error(f"无法创建目录 {dir_path}: {e}")

        return all_exist

    def check_core_files(self) -> bool:
        """检查核心文件"""
        self.log_info("检查核心文件...")

        required_files = [
            'tools/requirements-engine/requirement_parser.py',
            'tools/task-generator/task_decomposer.py',
            'tools/execution-engine/claude_executor.py',
            'tools/validators/process_gatekeeper.py',
            'tools/pipeline/pipeline_orchestrator.py',
            'utils/generate_execution_contract.py',
            'prompts/design-mode-enhanced.md',
            'prompts/development-mode-enhanced.md',
            'prompts/review-mode-enhanced.md',
            'docs/process/requirement-to-execution-pipeline.md',
            'docs/engine/claude-execution-engine-blueprint.md'
        ]

        all_exist = True
        for file_path in required_files:
            full_path = self.base_dir / file_path
            if full_path.exists() and full_path.is_file():
                self.log_success(f"文件 {file_path} 存在")

                # 检查文件是否可执行（对于.py文件）
                if file_path.endswith('.py'):
                    if os.access(full_path, os.X_OK):
                        self.log_success(f"文件 {file_path} 可执行")
                    else:
                        self.log_warning(f"文件 {file_path} 不可执行，正在修复...")
                        try:
                            os.chmod(full_path, 0o755)
                            self.log_success(f"文件 {file_path} 权限已修复")
                        except Exception as e:
                            self.log_error(f"无法修复文件权限 {file_path}: {e}")
            else:
                self.log_error(f"文件 {file_path} 不存在")
                all_exist = False

        return all_exist

    def check_config_files(self) -> bool:
        """检查配置文件"""
        self.log_info("检查配置文件...")

        config_files = [
            '.env.example',
            'requirements.txt',
            'pyproject.toml',
            'README.md'
        ]

        all_exist = True
        for file_path in config_files:
            full_path = self.base_dir / file_path
            if full_path.exists():
                self.log_success(f"配置文件 {file_path} 存在")
            else:
                self.log_warning(f"配置文件 {file_path} 不存在")
                all_exist = False

        return all_exist

    def check_external_services(self) -> bool:
        """检查外部服务连接"""
        self.log_info("检查外部服务连接...")

        # 检查Redis连接
        try:
            import redis
            r = redis.Redis(host='localhost', port=6379, db=0, socket_connect_timeout=2)
            r.ping()
            self.log_success("Redis连接正常")
            redis_ok = True
        except Exception as e:
            self.log_warning(f"Redis连接失败: {e}")
            redis_ok = False

        # 检查PostgreSQL连接（如果配置了）
        try:
            import sqlalchemy
            # 这里可以添加数据库连接检查
            self.log_info("PostgreSQL连接检查跳过（需要配置）")
            postgres_ok = True
        except Exception as e:
            self.log_warning(f"PostgreSQL检查失败: {e}")
            postgres_ok = False

        return redis_ok and postgres_ok

    def check_tool_functionality(self) -> bool:
        """检查工具功能"""
        self.log_info("检查工具功能...")

        tools_status = {}

        # 检查需求解析器
        try:
            result = subprocess.run([
                sys.executable,
                str(self.base_dir / 'tools/requirements-engine/requirement_parser.py'),
                '--help'
            ], capture_output=True, text=True, timeout=5)

            if result.returncode == 0:
                self.log_success("需求解析器功能正常")
                tools_status['requirement_parser'] = True
            else:
                self.log_error("需求解析器功能异常")
                tools_status['requirement_parser'] = False
        except Exception as e:
            self.log_error(f"需求解析器检查失败: {e}")
            tools_status['requirement_parser'] = False

        # 检查任务分解器
        try:
            result = subprocess.run([
                sys.executable,
                str(self.base_dir / 'tools/task-generator/task_decomposer.py'),
                '--help'
            ], capture_output=True, text=True, timeout=5)

            if result.returncode == 0:
                self.log_success("任务分解器功能正常")
                tools_status['task_decomposer'] = True
            else:
                self.log_error("任务分解器功能异常")
                tools_status['task_decomposer'] = False
        except Exception as e:
            self.log_error(f"任务分解器检查失败: {e}")
            tools_status['task_decomposer'] = False

        # 检查流水线编排器
        try:
            result = subprocess.run([
                sys.executable,
                str(self.base_dir / 'tools/pipeline/pipeline_orchestrator.py'),
                '--help'
            ], capture_output=True, text=True, timeout=5)

            if result.returncode == 0:
                self.log_success("流水线编排器功能正常")
                tools_status['pipeline_orchestrator'] = True
            else:
                self.log_error("流水线编排器功能异常")
                tools_status['pipeline_orchestrator'] = False
        except Exception as e:
            self.log_error(f"流水线编排器检查失败: {e}")
            tools_status['pipeline_orchestrator'] = False

        return all(tools_status.values())

    def run_syntax_check(self) -> bool:
        """运行语法检查"""
        self.log_info("运行Python语法检查...")

        python_files = list(self.base_dir.rglob("*.py"))
        syntax_errors = []

        for py_file in python_files:
            if '__pycache__' in str(py_file):
                continue

            try:
                with open(py_file, 'r', encoding='utf-8') as f:
                    compile(f.read(), str(py_file), 'exec')
                self.log_success(f"语法检查通过: {py_file.relative_to(self.base_dir)}")
            except SyntaxError as e:
                syntax_errors.append(f"{py_file.relative_to(self.base_dir)}: {e}")
                self.log_error(f"语法错误: {py_file.relative_to(self.base_dir)}: {e}")

        return len(syntax_errors) == 0

    def generate_report(self) -> Dict[str, Any]:
        """生成验证报告"""
        self.log_info("生成验证报告...")

        total_checks = 0
        passed_checks = 0

        # 执行所有检查
        checks = [
            ("Python版本", self.check_python_version),
            ("必需包", self.check_required_packages),
            ("目录结构", self.check_directory_structure),
            ("核心文件", self.check_core_files),
            ("配置文件", self.check_config_files),
            ("外部服务", self.check_external_services),
            ("工具功能", self.check_tool_functionality),
            ("语法检查", self.run_syntax_check)
        ]

        results = {}
        for check_name, check_func in checks:
            total_checks += 1
            try:
                result = check_func()
                results[check_name] = result
                if result:
                    passed_checks += 1
            except Exception as e:
                self.log_error(f"检查 {check_name} 时发生异常: {e}")
                results[check_name] = False

        # 生成报告
        report = {
            "timestamp": "2025-01-29T10:30:00Z",
            "summary": {
                "total_checks": total_checks,
                "passed_checks": passed_checks,
                "failed_checks": total_checks - passed_checks,
                "success_rate": f"{(passed_checks/total_checks*100):.1f}%" if total_checks > 0 else "0%"
            },
            "detailed_results": results,
            "issues": self.issues,
            "warnings": self.warnings,
            "recommendations": self._generate_recommendations()
        }

        return report

    def _generate_recommendations(self) -> List[str]:
        """生成改进建议"""
        recommendations = []

        if self.issues:
            recommendations.append("存在严重问题，需要立即解决后才能正常使用")

        if len(self.warnings) > 3:
            recommendations.append("建议修复警告项以获得更好的使用体验")

        # 检查常见的改进建议
        recommendations.extend([
            "定期运行 `python scripts/verify_setup.py` 检查系统状态",
            "确保Redis和PostgreSQL服务正常运行",
            "定期更新依赖包版本",
            "查看docs/目录获取详细使用指南"
        ])

        return recommendations

    def save_report(self, report: Dict[str, Any]) -> None:
        """保存验证报告"""
        report_path = self.base_dir / "reports" / "setup-verification-report.json"
        report_path.parent.mkdir(exist_ok=True)

        with open(report_path, 'w', encoding='utf-8') as f:
            json.dump(report, f, ensure_ascii=False, indent=2)

        self.log_success(f"验证报告已保存: {report_path}")

def main():
    """主函数"""
    print("🔍 LJWX-Stock AI执行平台 - 工具链完整性验证")
    print("=" * 60)

    verifier = SetupVerifier()

    try:
        # 生成验证报告
        report = verifier.generate_report()

        # 保存报告
        verifier.save_report(report)

        # 显示总结
        print("\n" + "=" * 60)
        print("📊 验证总结")
        print("=" * 60)

        summary = report["summary"]
        print(f"总检查项: {summary['total_checks']}")
        print(f"通过项: {summary['passed_checks']}")
        print(f"失败项: {summary['failed_checks']}")
        print(f"成功率: {summary['success_rate']}")

        if report["issues"]:
            print(f"\n❌ 发现 {len(report['issues'])} 个问题:")
            for issue in report["issues"]:
                print(f"   - {issue}")

        if report["warnings"]:
            print(f"\n⚠️  发现 {len(report['warnings'])} 个警告:")
            for warning in report["warnings"]:
                print(f"   - {warning}")

        if report["recommendations"]:
            print(f"\n💡 建议:")
            for rec in report["recommendations"]:
                print(f"   - {rec}")

        # 设置退出码
        if report["issues"]:
            sys.exit(1)
        else:
            print("\n🎉 工具链验证完成！")
            sys.exit(0)

    except KeyboardInterrupt:
        print("\n\n⏹️  验证被用户中断")
        sys.exit(130)
    except Exception as e:
        print(f"\n\n💥 验证过程中发生异常: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()