graph_rag_server/scripts/final_verification.py

#!/usr/bin/env python3
"""
最终系统验证脚本

此脚本执行完整的系统验证，包括：
1. 代码质量检查
2. 测试执行
3. 文档完整性检查
4. API 向后兼容性验证
5. 性能基准测试（可选）
"""

import subprocess
import sys
from pathlib import Path
from typing import List, Tuple
import json


class Colors:
    """终端颜色"""
    GREEN = '\033[92m'
    YELLOW = '\033[93m'
    RED = '\033[91m'
    BLUE = '\033[94m'
    END = '\033[0m'
    BOLD = '\033[1m'


def print_header(text: str):
    """打印标题"""
    print(f"\n{Colors.BOLD}{Colors.BLUE}{'=' * 60}{Colors.END}")
    print(f"{Colors.BOLD}{Colors.BLUE}{text}{Colors.END}")
    print(f"{Colors.BOLD}{Colors.BLUE}{'=' * 60}{Colors.END}\n")


def print_success(text: str):
    """打印成功消息"""
    print(f"{Colors.GREEN}✓ {text}{Colors.END}")


def print_warning(text: str):
    """打印警告消息"""
    print(f"{Colors.YELLOW}⚠ {text}{Colors.END}")


def print_error(text: str):
    """打印错误消息"""
    print(f"{Colors.RED}✗ {text}{Colors.END}")


def run_command(cmd: List[str], description: str) -> Tuple[bool, str]:
    """
    运行命令并返回结果
    
    Args:
        cmd: 命令列表
        description: 命令描述
    
    Returns:
        (成功标志, 输出)
    """
    print(f"Running: {description}...")
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=300  # 5 分钟超时
        )
        
        if result.returncode == 0:
            print_success(f"{description} passed")
            return True, result.stdout
        else:
            print_error(f"{description} failed")
            print(f"Error output:\n{result.stderr}")
            return False, result.stderr
    
    except subprocess.TimeoutExpired:
        print_error(f"{description} timed out")
        return False, "Timeout"
    
    except Exception as e:
        print_error(f"{description} error: {e}")
        return False, str(e)


def check_code_quality() -> bool:
    """检查代码质量"""
    print_header("1. Code Quality Checks")
    
    checks = [
        (['flake8', 'src', 'tests', '--count', '--select=E9,F63,F7,F82', '--show-source', '--statistics'],
         "Flake8 syntax check"),
        (['black', '--check', 'src', 'tests'],
         "Black formatting check"),
        (['isort', '--check-only', 'src', 'tests'],
         "isort import sorting check"),
    ]
    
    all_passed = True
    for cmd, desc in checks:
        success, _ = run_command(cmd, desc)
        if not success:
            all_passed = False
            print_warning(f"  Tip: Run '{' '.join(cmd[:-2])}' to fix")
    
    return all_passed


def run_tests() -> bool:
    """运行测试"""
    print_header("2. Test Execution")
    
    # 运行单元测试
    success, output = run_command(
        ['pytest', 'tests/unit', '-v', '--tb=short'],
        "Unit tests"
    )
    
    if not success:
        return False
    
    # 运行集成测试（如果存在）
    if Path('tests/integration').exists():
        success, output = run_command(
            ['pytest', 'tests/integration', '-v', '--tb=short'],
            "Integration tests"
        )
        if not success:
            print_warning("Integration tests failed (may require services)")
    
    return True


def check_test_coverage() -> bool:
    """检查测试覆盖率"""
    print_header("3. Test Coverage")
    
    success, output = run_command(
        ['pytest', '--cov=src', '--cov-report=term', '--cov-report=json'],
        "Coverage analysis"
    )
    
    if not success:
        return False
    
    # 读取覆盖率报告
    try:
        with open('coverage.json', 'r') as f:
            coverage_data = json.load(f)
        
        total_coverage = coverage_data['totals']['percent_covered']
        print(f"\nTotal coverage: {total_coverage:.2f}%")
        
        # 检查目标覆盖率
        target_coverage = 80.0
        if total_coverage >= target_coverage:
            print_success(f"Coverage meets target ({target_coverage}%)")
            return True
        else:
            print_warning(f"Coverage below target ({target_coverage}%)")
            return False
    
    except Exception as e:
        print_error(f"Failed to read coverage report: {e}")
        return False


def check_documentation() -> bool:
    """检查文档完整性"""
    print_header("4. Documentation Completeness")
    
    required_docs = [
        'README.md',
        'docs/architecture.md',
        'docs/api.md',
        'docs/deployment.md',
        'docs/development.md',
        'docs/configuration.md',
        'docs/logging.md',
        'docs/ci-cd.md',
    ]
    
    all_exist = True
    for doc in required_docs:
        path = Path(doc)
        if path.exists():
            print_success(f"{doc} exists")
        else:
            print_error(f"{doc} missing")
            all_exist = False
    
    return all_exist


def check_api_compatibility() -> bool:
    """检查 API 向后兼容性"""
    print_header("5. API Backward Compatibility")
    
    # 检查 legacy adapter 是否存在
    legacy_adapter = Path('src/presentation/api/legacy_adapter.py')
    if legacy_adapter.exists():
        print_success("Legacy adapter exists")
    else:
        print_warning("Legacy adapter not found")
        return False
    
    # 检查废弃警告
    deprecated_file = Path('src/api/DEPRECATED.md')
    if deprecated_file.exists():
        print_success("Deprecation notice exists")
    else:
        print_warning("Deprecation notice not found")
    
    return True


def check_docker_config() -> bool:
    """检查 Docker 配置"""
    print_header("6. Docker Configuration")
    
    docker_files = [
        'docker/Dockerfile',
        'docker/docker-compose.yml',
        'docker/docker-compose.dev.yml',
        'alembic.ini',
        'alembic/env.py',
    ]
    
    all_exist = True
    for file_path in docker_files:
        path = Path(file_path)
        if path.exists():
            print_success(f"{file_path} exists")
        else:
            print_error(f"{file_path} missing")
            all_exist = False
    
    return all_exist


def generate_report(results: dict) -> str:
    """生成验证报告"""
    report_lines = [
        "\n" + "=" * 60,
        "FINAL VERIFICATION REPORT",
        "=" * 60,
        ""
    ]
    
    total_checks = len(results)
    passed_checks = sum(1 for v in results.values() if v)
    
    for check_name, passed in results.items():
        status = "✓ PASS" if passed else "✗ FAIL"
        color = Colors.GREEN if passed else Colors.RED
        report_lines.append(f"{color}{status}{Colors.END} - {check_name}")
    
    report_lines.extend([
        "",
        f"Total: {passed_checks}/{total_checks} checks passed",
        ""
    ])
    
    if passed_checks == total_checks:
        report_lines.append(f"{Colors.GREEN}{Colors.BOLD}✓ ALL CHECKS PASSED!{Colors.END}")
        report_lines.append("")
        report_lines.append("The system is ready for deployment.")
    else:
        report_lines.append(f"{Colors.RED}{Colors.BOLD}✗ SOME CHECKS FAILED{Colors.END}")
        report_lines.append("")
        report_lines.append("Please fix the issues before deployment.")
    
    report_lines.append("=" * 60)
    
    return "\n".join(report_lines)


def main():
    """主函数"""
    print(f"{Colors.BOLD}RAG System - Final Verification{Colors.END}")
    print(f"This script will verify the system is ready for deployment.\n")
    
    # 执行所有检查
    results = {}
    
    try:
        results['Code Quality'] = check_code_quality()
        results['Tests'] = run_tests()
        results['Coverage'] = check_test_coverage()
        results['Documentation'] = check_documentation()
        results['API Compatibility'] = check_api_compatibility()
        results['Docker Config'] = check_docker_config()
    
    except KeyboardInterrupt:
        print("\n\nVerification interrupted by user")
        sys.exit(1)
    
    # 生成并打印报告
    report = generate_report(results)
    print(report)
    
    # 保存报告到文件
    report_file = Path('verification_report.txt')
    with open(report_file, 'w', encoding='utf-8') as f:
        # 移除颜色代码
        clean_report = report
        for color in [Colors.GREEN, Colors.YELLOW, Colors.RED, Colors.BLUE, Colors.END, Colors.BOLD]:
            clean_report = clean_report.replace(color, '')
        f.write(clean_report)
    
    print(f"\nReport saved to: {report_file}")
    
    # 返回退出码
    all_passed = all(results.values())
    sys.exit(0 if all_passed else 1)


if __name__ == '__main__':
    main()