# .gitlab-ci.yml - 基础流水线配置
stages:
  - build
  - test
  - deploy

variables:
  DOCKER_REGISTRY: registry.example.com
  AI_CACHE_DB: /data/cache-index.db

build:app:
  stage: build
  script:
    - |
      # 1. 检查 AI 缓存索引
      CACHE_RESULT=$(python3 ai_cache_checker.py \
        --project "$CI_PROJECT_PATH" \
        --commit "$CI_COMMIT_SHA" \
        --base-commit "$CI_COMMIT_BEFORE_SHA")
      
      if echo "$CACHE_RESULT" | grep -q "FULL_CACHE_HIT"; then
        echo "✅ AI 缓存命中，跳过构建"
        cp -r /cache/build ./dist
      else
        echo "🔨 开始构建..."
        docker build -t $DOCKER_REGISTRY/app:$CI_COMMIT_SHA .
        python3 ai_cache_updater.py \
          --project "$CI_PROJECT_PATH" \
          --commit "$CI_COMMIT_SHA" \
          --dependencies "$(cat requirements.txt)"
      fi
  cache:
    key: "${CI_COMMIT_REF_SLUG}"
    paths:
      - .cache/
  artifacts:
    paths:
      - dist/
    expire_in: 1 week

# ai_cache_checker.py - AI 缓存检查器
import json
import hashlib
import sqlite3
from datetime import datetime, timedelta

class AICacheChecker:
    def __init__(self, db_path="/data/cache-index.db"):
        self.db = sqlite3.connect(db_path)
        self._init_db()
    
    def _init_db(self):
        self.db.executescript("""
            CREATE TABLE IF NOT EXISTS build_cache (
                project TEXT,
                commit_hash TEXT,
                dependency_hash TEXT,
                build_artifacts_hash TEXT,
                cache_hit_rate REAL,
                last_build TIMESTAMP,
                PRIMARY KEY (project, dependency_hash)
            );
            CREATE INDEX IF NOT EXISTS idx_dep_hash ON build_cache(dependency_hash);
        """)
    
    def analyze_dependencies(self, current_deps: str, base_deps: str) -> dict:
        """
        使用 AI 分析依赖变更的影响范围
        
        传统方式：简单 diff 后全量重建
        AI 方式：理解依赖关系图，精确计算受影响模块
        """
        current_hash = hashlib.sha256(current_deps.encode()).hexdigest()
        base_hash = hashlib.sha256(base_deps.encode()).hexdigest()
        
        if current_hash == base_hash:
            return {"change_type": "none", "rebuild_modules": []}
        
        # 查询历史缓存命中率
        cursor = self.db.execute(
            """
            SELECT cache_hit_rate, COUNT(*) as build_count
            FROM build_cache
            WHERE dependency_hash = ?
            GROUP BY dependency_hash
            ORDER BY last_build DESC
            LIMIT 10
            """,
            (current_hash,)
        )
        
        results = cursor.fetchall()
        avg_hit_rate = sum(r[0] for r in results) / len(results) if results else 0.5
        
        # 判断是否值得缓存
        if avg_hit_rate > 0.8:
            return {"change_type": "cached", "confidence": avg_hit_rate}
        elif avg_hit_rate < 0.3:
            return {"change_type": "full_rebuild", "confidence": 1 - avg_hit_rate}
        else:
            return {
                "change_type": "partial_rebuild",
                "confidence": avg_hit_rate,
                "strategy": "incremental"
            }
    
    def check(self, project: str, commit: str, base_commit: str, deps: str) -> str:
        analysis = self.analyze_dependencies(deps, "")
        
        if analysis["change_type"] == "cached" and analysis["confidence"] > 0.8:
            return "FULL_CACHE_HIT"
        elif analysis["change_type"] == "full_rebuild":
            return "NO_CACHE"
        else:
            return "PARTIAL_CACHE"

# build_parallelizer.py - AI 构建并行化策略生成器
import networkx as nx

def generate_build_graph(requirements: list, package_info: dict) -> nx.DiGraph:
    """
    构建模块依赖图
    
    示例：
    项目结构:
      - frontend (依赖: shared-utils, api-client)
      - backend (依赖: shared-utils, db-driver)
      - shared-utils (无依赖)
      - api-client (依赖: shared-utils)
      - db-driver (依赖: shared-utils)
    
    依赖图:
      shared-utils → frontend
                   → api-client → frontend
                   → backend
                   → db-driver → backend
    """
    graph = nx.DiGraph()
    
    for pkg in requirements:
        deps = package_info.get(pkg, {}).get("depends", [])
        graph.add_node(pkg)
        for dep in deps:
            graph.add_edge(dep, pkg)
    
    return graph

def optimal_parallel_strategy(graph: nx.DiGraph) -> list:
    """
    使用拓扑排序生成最优并行构建批次
    
    返回: [[独立模块], [第二批], ...]
    """
    levels = []
    remaining = set(graph.nodes())
    
    while remaining:
        # 找出所有入度为 0 的节点（可以并行构建）
        ready = []
        for node in remaining:
            dependents_in_remaining = [
                m for m in remaining 
                if node in graph.predecessors(m)
            ]
            if not dependents_in_remaining:
                ready.append(node)
        
        if not ready:
            break  # 避免循环依赖
        
        levels.append(sorted(ready))
        remaining -= set(ready)
    
    return levels

# 使用示例
if __name__ == "__main__":
    requirements = [
        "frontend", "backend", "shared-utils", 
        "api-client", "db-driver"
    ]
    package_info = {
        "frontend": {"depends": ["shared-utils", "api-client"]},
        "backend": {"depends": ["shared-utils", "db-driver"]},
        "shared-utils": {"depends": []},
        "api-client": {"depends": ["shared-utils"]},
        "db-driver": {"depends": ["shared-utils"]},
    }
    
    graph = generate_build_graph(requirements, package_info)
    batches = optimal_parallel_strategy(graph)
    
    print("📊 推荐构建批次:")
    for i, batch in enumerate(batches, 1):
        print(f"   第{i}批 (可并行): {', '.join(batch)}")

# test_selector.py - AI 测试选择器
import difflib
import re
from collections import defaultdict

class TestImpactAnalyzer:
    def __init__(self):
        # 建立文件路径到测试文件的映射
        self.file_to_tests = {
            "src/models/user.py": ["tests/test_user_model.py"],
            "src/api/auth.py": ["tests/test_auth.py", "tests/test_api.py"],
            "src/utils/cache.py": ["tests/test_cache.py"],
            "src/services/payment.py": ["tests/test_payment.py"],
            "docker-compose.yml": ["tests/test_deploy.py"],
        }
    
    def analyze_changes(self, old_content: str, new_content: str, 
                        changed_files: list) -> dict:
        """
        分析代码变更，确定需要运行的测试
        
        使用语义分析 + 规则匹配的方式，比传统的 glob 模式更精准
        """
        affected_tests = set()
        
        for file_path in changed_files:
            # 直接映射
            if file_path in self.file_to_tests:
                affected_tests.update(self.file_to_tests[file_path])
            
            # 分析变更内容
            if file_path.endswith(".py"):
                changes = self._analyze_python_changes(old_content, new_content)
                if changes["type"] == "api_change":
                    # API 变更影响所有测试
                    affected_tests.update(["tests/test_api.py"])
                elif changes["type"] == "model_change":
                    # 模型变更影响相关测试
                    affected_tests.update(["tests/test_user_model.py"])
        
        return {
            "affected_tests": list(affected_tests),
            "test_count": len(affected_tests),
            "original_total": 150,  # 总测试数
            "reduction_ratio": round(1 - len(affected_tests)/150, 2)
        }
    
    def _analyze_python_changes(self, old: str, new: str) -> dict:
        """分析 Python 代码变更类型"""
        diff = difflib.unified_diff(old.splitlines(), new.splitlines())
        added_lines = []
        removed_lines = []
        
        for line in diff:
            if line.startswith("+") and not line.startswith("+++"):
                added_lines.append(line[1:])
            elif line.startswith("-") and not line.startswith("---"):
                removed_lines.append(line[1:])
        
        # 检测是否为 API 变更
        api_patterns = [r"def\s+\w+\(", r"@router\.", r"@app\."]
        model_patterns = [r"class\s+\w+\(BaseModel\)", r"class\s+\w+\(SQLModel\)"]
        
        has_api_change = any(
            re.search(p, "\n".join(added_lines)) 
            for p in api_patterns
        )
        has_model_change = any(
            re.search(p, "\n".join(added_lines))
            for p in model_patterns
        )
        
        if has_api_change:
            return {"type": "api_change"}
        elif has_model_change:
            return {"type": "model_change"}
        return {"type": "minor"}

# test_diagnoser.py - AI 测试失败诊断器
import subprocess
import json

class TestDiagnoser:
    def __init__(self, llm_endpoint="http://localhost:11434"):
        self.llm_url = llm_endpoint
    
    def diagnose_failure(self, test_name: str, error_output: str, 
                         test_code: str, recent_commits: list) -> dict:
        """
        诊断测试失败的根本原因
        
        传统方式: 返回原始错误堆栈
        AI 方式: 分析错误上下文，给出可操作的修复建议
        """
        prompt = f"""你是一个资深 QA 工程师。以下测试失败了，请分析原因并给出修复建议。

## 测试名称
{test_name}

## 错误输出

# auto_rollback.py - 智能回滚决策器
import time

class AutoRollbackEngine:
    def __init__(self):
        self.metrics_history = []
        self.alert_thresholds = {
            "error_rate": 0.05,       # 错误率超过 5%
            "latency_p99": 2000,      # P99 延迟超过 2 秒
            "cpu_usage": 90,          # CPU 超过 90%
            "memory_usage": 85,       # 内存超过 85%
        }
    
    def monitor_and_decide(self, deployment_id: str, 
                           current_metrics: dict) -> dict:
        """
        实时监控部署后的指标，决定是否回滚
        
        不只是看单一指标，而是综合分析多个维度的变化趋势
        """
        self.metrics_history.append({
            "timestamp": time.time(),
            "metrics": current_metrics,
            "deployment": deployment_id
        })
        
        # 计算各指标的异常程度
        anomalies = {}
        for metric, threshold in self.alert_thresholds.items():
            if metric in current_metrics:
                value = current_metrics[metric]
                if isinstance(threshold, float):
                    deviation = abs(value - threshold) / threshold
                else:
                    deviation = abs(value - threshold) / threshold * 100
                
                anomalies[metric] = {
                    "value": value,
                    "threshold": threshold,
                    "deviation_pct": round(deviation * 100, 1)
                }
        
        # 综合判断
        critical_count = sum(
            1 for a in anomalies.values() 
            if a["deviation_pct"] > 50
        )
        
        if critical_count >= 2:
            return {
                "action": "immediate_rollback",
                "reason": f"检测到 {critical_count} 个关键指标异常",
                "anomalies": anomalies,
                "confidence": 0.95
            }
        elif critical_count == 1:
            return {
                "action": "monitor_closely",
                "reason": "单个关键指标异常，继续观察",
                "anomalies": anomalies,
                "check_interval_seconds": 30
            }
        elif len(anomalies) > 0:
            return {
                "action": "proceed",
                "reason": "轻微指标波动，在正常范围内",
                "anomalies": anomalies
            }
        else:
            return {
                "action": "proceed",
                "reason": "所有指标正常"
            }

# ai_alert_aggregator.py 集成到 GitLab CI
stages:
  - build
  - test
  - notify

notify:smart:
  stage: test
  script:
    - |
      if [ "$PIPELINE_STATUS" = "failed" ]; then
        # 收集所有失败的 job 信息
        FAILED_JOBS=$(gitlab-ci-jobs list --failed --json)
        
        # 调用 AI 聚合分析
        AGGREGATED=$(python3 ai_alert_aggregator.py \
          --jobs "$FAILED_JOBS" \
          --pipeline-id "$CI_PIPELINE_ID")
        
        # 发送聚合后的通知
        curl -X POST "$SLACK_WEBHOOK" \
          -H "Content-Type: application/json" \
          -d "{\"text\": \"$AGGREGATED\"}"
      fi

# ai_alert_aggregator.py - AI 告警聚合器
class AlertAggregator:
    def __init__(self):
        self.failure_patterns = {
            "build_failure": {
                "keywords": ["compile error", "syntax error", "missing dependency"],
                "severity": "high"
            },
            "test_failure": {
                "keywords": ["assertion failed", "timeout", "connection refused"],
                "severity": "medium"
            },
            "deploy_failure": {
                "keywords": ["port already in use", "image pull failed", "permission denied"],
                "severity": "critical"
            }
        }
    
    def aggregate(self, failed_jobs: list) -> str:
        """
        将分散的失败告警聚合为一条有意义的报告
        """
        categories = defaultdict(list)
        
        for job in failed_jobs:
            category = self._categorize(job)
            categories[category].append(job)
        
        # 生成聚合报告
        report_parts = ["🚨 CI/CD 流水线失败报告\n"]
        
        for category, jobs in categories.items():
            if len(jobs) == 1:
                report_parts.append(f"• {category}: {jobs[0]['name']}")
            else:
                report_parts.append(
                    f"• {category} ({len(jobs)} 个任务失败): "
                    f"{', '.join(j['name'] for j in jobs)}"
                )
        
        # AI 总结
        summary_prompt = (
            f"以下 CI/CD 流水线失败，请给出简短总结和建议:\n"
            + "\n".join(report_parts)
        )
        
        return self._get_ai_summary(summary_prompt)
    
    def _categorize(self, job: dict) -> str:
        name = job.get("name", "").lower()
        output = job.get("output", "").lower()
        
        for pattern, config in self.failure_patterns.items():
            for keyword in config["keywords"]:
                if keyword in output or keyword in name:
                    return pattern
        
        return "unknown_failure"
    
    def _get_ai_summary(self, prompt: str) -> str:
        """调用本地 LLM 生成总结"""
        # 简化版：实际生产中调用 Ollama
        return "建议优先检查构建阶段的编译错误，这通常会导致后续所有测试和部署失败。"

# .gitlab-ci.yml
stages:
  - validate
  - build
  - test
  - security
  - deploy

variables:
  AI_MODEL: qwen2.5:7b-instruct
  AI_ENDPOINT: http://localhost:11434
  CACHE_DIR: .cache/build

# 阶段 1: AI 验证
validate:code:
  stage: validate
  script:
    - python3 ai_code_validator.py --diff HEAD~1 HEAD
  only:
    - merge_requests

# 阶段 2: AI 优化的构建
build:app:
  stage: build
  script:
    - |
      # AI 缓存检查
      CACHE_STATUS=$(python3 ai_cache_checker.py \
        --deps "$(cat requirements.txt)")
      
      if [ "$CACHE_STATUS" = "FULL_CACHE_HIT" ]; then
        echo "使用 AI 缓存，跳过构建"
        cp -r $CACHE_DIR/dist ./dist
      else
        echo "开始构建..."
        docker build -t app:$CI_COMMIT_SHA .
        mkdir -p $CACHE_DIR/dist
        cp -r dist/* $CACHE_DIR/dist/
      fi
  cache:
    key: ${CI_COMMIT_REF_SLUG}
    paths:
      - .cache/
  artifacts:
    paths:
      - dist/

# 阶段 3: AI 优化的测试
test:unit:
  stage: test
  script:
    - |
      # AI 测试选择
      AFFECTED_TESTS=$(python3 test_selector.py \
        --changed-files $(git diff --name-only HEAD~1 HEAD))
      
      # 只运行受影响的测试
      pytest $AFFECTED_TESTS -v
      
      # 如果全部通过，记录成功
      if [ $? -eq 0 ]; then
        python3 test_success_logger.py \
          --pipeline $CI_PIPELINE_ID
      fi

# 阶段 4: AI 安全扫描
security:scan:
  stage: security
  script:
    - trivy image --severity HIGH,CRITICAL app:$CI_COMMIT_SHA
    - python3 ai_security_reviewer.py --scan-results ./trivy-report.json

# 阶段 5: AI 智能部署
deploy:staging:
  stage: deploy
  script:
    - |
      # 风险评估
      RISK=$(python3 deploy_risk_assessor.py \
        --change-log "$CHANGELOG" \
        --diff-summary "$(git diff --stat HEAD~1)")
      
      echo "📊 部署风险评估: $RISK"
      
      # 根据风险等级选择部署策略
      STRATEGY=$(echo "$RISK" | jq -r '.recommended_strategy')
      
      case "$STRATEGY" in
        direct)
          kubectl rollout restart deployment/app
          ;;
        canary)
          kubectl set image deployment/app-canary \
            app=app:$CI_COMMIT_SHA
          sleep 60
          # 监控金丝雀实例
          python3 canary_monitor.py --deployment app-canary
          ;;
        blue_green)
          kubectl apply -f deploy/blue-green.yaml
          kubectl patch service/app \
            -p '{"spec":{"selector":{"version":"green"}}}'
          ;;
      esac
      
      # 部署后监控
      python3 auto_rollback.py \
        --deployment-id $CI_PIPELINE_ID \
        --metrics-endpoint http://prometheus:9090
  environment:
    name: staging
  when: manual

deploy:production:
  stage: deploy
  script:
    - |
      # 生产部署需要人工确认 + AI 风险评估
      RISK=$(python3 deploy_risk_assessor.py \
        --change-log "$CHANGELOG" \
        --diff-summary "$(git diff --stat HEAD~1)" \
        --previous-success-rate 0.95)
      
      echo "📊 生产部署风险评估: $RISK"
      
      # 高风险部署必须人工二次确认
      if [ "$(echo "$RISK" | jq -r '.risk_level')" = "high" ]; then
        echo "⚠️ 高风险部署，需要额外确认"
        # 发送 Slack 通知要求确认
      fi
      
      # 使用蓝绿部署确保零停机
      kubectl apply -f deploy/blue-green-prod.yaml
      kubectl rollout status deployment/app-green
      
      # 部署后持续监控 30 分钟
      timeout 1800 python3 deploy_health_monitor.py \
        --deployment app-green
  environment:
    name: production
  when: manual
  rules:
    - if: $CI_COMMIT_BRANCH == "main"

#!/bin/bash
# setup_ai_engine.sh - 设置本地 AI 推理引擎

# 安装 Ollama
curl -fsSL https://ollama.com/install.sh | sh

# 拉取适合推理的模型
ollama pull qwen2.5:7b-instruct

# 创建 AI 工作目录
mkdir -p ~/ai-cicd/{cache,models,policies}

# 初始化缓存数据库
python3 -c "
import sqlite3
conn = sqlite3.connect('~/ai-cicd/cache/build-index.db')
conn.executescript('''
    CREATE TABLE build_cache (
        project TEXT, commit_hash TEXT,
        dependency_hash TEXT, build_time REAL,
        cache_hit BOOLEAN, timestamp TIMESTAMP
    );
    CREATE INDEX idx_project ON build_cache(project);
''')
print('✅ AI 引擎初始化完成')
"

echo "🎉 AI-CI/CD 引擎已就绪!"
echo "   模型: qwen2.5:7b-instruct"
echo "   缓存: ~/ai-cicd/cache/"
echo "   端口: 11434"