#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ services/code_review_pipeline_service.py Post-Deploy AI Agent Code Review Pipeline 觸發時機:CD 健康檢查通過後,由 Gitea Action webhook 呼叫 Pipeline: Step 1 system 讀取變更檔案內容 Step 2 Hermes 程式碼掃描(bugs / security / performance) Step 3 OpenClaw 架構品質評估(Gemini 2.5 Flash) Step 4 ElephantAlpha 決策協調(severity 判定 + auto-fix 裁量) Step 5 NemoTron 行動派遣(action_plans 寫入 + AiderHeal 觸發) 結果輸出: - ai_insights(type='code_review_result') - action_plans(type='code_review_fix') - Telegram 告警(啟動 / 完成 / 錯誤) - 前端 /code-review/ 即時 polling 狀態 """ import json import logging import os import re import threading from datetime import datetime from typing import Any, Dict, List, Optional from database.manager import get_session from sqlalchemy import text logger = logging.getLogger(__name__) # ── Pipeline 全域狀態(供前端 polling)───────────────────────────────────── _current_pipeline: Dict[str, Any] = {} _pipeline_lock = threading.Lock() GEMINI_API_KEY = os.getenv("GEMINI_API_KEY", "") REVIEW_MODEL = os.getenv("OPENCLAW_MODEL", "gemini-2.5-flash-preview-05-20") INTERNAL_TOKEN = os.getenv("INTERNAL_WEBHOOK_TOKEN", "") # ═══════════════════════════════════════════════════════════════════════════════ # Pipeline Class # ═══════════════════════════════════════════════════════════════════════════════ class CodeReviewPipeline: """ 5-Step post-deploy code review pipeline. Call pipeline.run() inside a daemon thread. """ def __init__(self, commit_sha: str, changed_files: List[str], branch: str = "main", deploy_type: str = "sync"): self.commit_sha = commit_sha self.branch = branch self.deploy_type = deploy_type self.started_at = datetime.now() self.pipeline_id = f"cr_{commit_sha[:8]}_{self.started_at.strftime('%Y%m%d_%H%M%S')}" # 只 review Python + YAML 檔(跳過靜態資源) self.changed_files = [ f for f in changed_files if f.endswith(('.py', '.yaml', '.yml', '.json')) and not f.startswith(('node_modules/', '.git/')) ] self.state: Dict[str, Any] = { "pipeline_id": self.pipeline_id, "commit_sha": commit_sha, "branch": branch, "changed_files": self.changed_files, "status": "running", "current_step": 0, "total_steps": 5, "steps": [], "findings": [], "severity_summary": {"critical": 0, "high": 0, "medium": 0, "low": 0}, "openclaw_report": "", "ea_decision": {}, "auto_fix_triggered": False, "started_at": self.started_at.isoformat(), "completed_at": None, "message": "", } self._sync_global() # ── State helpers ───────────────────────────────────────────────────────── def _sync_global(self): global _current_pipeline with _pipeline_lock: _current_pipeline = dict(self.state) def _step_start(self, num: int, name: str, agent: str): self.state["current_step"] = num self.state["steps"].append({ "step": num, "name": name, "agent": agent, "status": "running", "started_at": datetime.now().isoformat(), "completed_at": None, "summary": "", }) self._sync_global() logger.info("[CodeReview] ▶ Step %d/5 %s (%s)", num, name, agent) def _step_done(self, num: int, summary: str, ok: bool = True): for s in self.state["steps"]: if s["step"] == num: s["status"] = "ok" if ok else "error" s["completed_at"] = datetime.now().isoformat() s["summary"] = summary[:300] self._sync_global() logger.info("[CodeReview] %s Step %d — %s", "✓" if ok else "✗", num, summary[:100]) def _finish(self, status: str, message: str): self.state["status"] = status self.state["completed_at"] = datetime.now().isoformat() self.state["message"] = message self._sync_global() # ── Main pipeline ───────────────────────────────────────────────────────── def run(self): """Execute full pipeline. Designed to run in daemon thread.""" try: self._notify_start() # Step 1 ─ read files self._step_start(1, "讀取變更檔案", "system") file_contents = self._read_changed_files() if not file_contents: self._step_done(1, "無有效 Python/YAML 變更,跳過 Review") self._finish("skipped", "無有效變更檔案") return self._step_done(1, f"讀取 {len(file_contents)} 個檔案") # Step 2 ─ Hermes scan self._step_start(2, "Hermes 程式碼掃描", "Hermes") findings = self._hermes_scan(file_contents) cnt = self.state["severity_summary"] self._step_done(2, f"CRITICAL={cnt['critical']} HIGH={cnt['high']} MEDIUM={cnt['medium']} LOW={cnt['low']}") # Step 3 ─ OpenClaw assessment self._step_start(3, "OpenClaw 架構品質評估", "OpenClaw") openclaw_report = self._openclaw_assess(file_contents, findings) self.state["openclaw_report"] = openclaw_report self._step_done(3, openclaw_report[:120] if openclaw_report else "(Gemini 未回應)") # Step 4 ─ ElephantAlpha orchestration self._step_start(4, "ElephantAlpha 決策協調", "ElephantAlpha") ea = self._ea_orchestrate(findings, openclaw_report) self.state["ea_decision"] = ea self._step_done(4, f"優先度={ea.get('priority','?')} auto_fix={ea.get('auto_fix',False)}") # Step 5 ─ NemoTron dispatch self._step_start(5, "NemoTron 行動派遣", "NemoTron") dispatch = self._nemotron_dispatch(ea, findings) self._step_done(5, f"寫入 {dispatch['actions']} 筆 action_plans,自動修復={'是' if dispatch['auto_fix'] else '否'}") # Persist + notify self._save_to_db(findings, openclaw_report, ea) self._notify_complete(findings, openclaw_report, ea) self._finish("completed", "Pipeline 執行完成") except Exception as e: logger.error("[CodeReview] Pipeline 例外: %s", e, exc_info=True) self._finish("error", str(e)[:200]) self._notify_error(str(e)) # ── Step 1:讀取檔案 ─────────────────────────────────────────────────────── def _read_changed_files(self) -> Dict[str, str]: project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) contents: Dict[str, str] = {} for rel_path in self.changed_files[:8]: # 最多 8 個檔案 abs_path = os.path.join(project_root, rel_path) try: with open(abs_path, encoding="utf-8", errors="ignore") as fh: raw = fh.read() contents[rel_path] = raw[:6000] + ("\n... (截斷)" if len(raw) > 6000 else "") except OSError: logger.debug("[CodeReview] 無法讀取 %s(部署路徑不同?)", rel_path) return contents # ── Step 2:Hermes 掃描 ─────────────────────────────────────────────────── def _hermes_scan(self, files: Dict[str, str]) -> List[Dict]: try: from services.ai_provider import ai_provider_service # 最多送 4 個檔案給 Hermes,避免 context overflow files_text = "\n\n".join( f"### {name}\n```python\n{content}\n```" for name, content in list(files.items())[:4] ) prompt = f"""你是資深程式碼審查工程師,請掃描以下程式碼並列出所有問題。 {files_text} 輸出格式:純 JSON 陣列,每項包含以下欄位: - severity: "CRITICAL" | "HIGH" | "MEDIUM" | "LOW" - type: "bug" | "security" | "performance" | "maintainability" - file: 檔案名稱 - line_hint: 約略行號或函式名稱 - description: 問題說明(繁體中文,精簡一句) - suggestion: 修復建議(繁體中文,精簡一句) 只輸出 JSON 陣列,不含其他文字。無問題時輸出 []""" resp = ai_provider_service.generate( prompt=prompt, provider="ollama", model="hermes3:latest", temperature=0.1, timeout=120, ) if not (resp and resp.success): logger.warning("[CodeReview] Hermes 未回應: %s", getattr(resp, "error", "")) return [] match = re.search(r"\[.*\]", resp.content, re.DOTALL) if not match: return [] findings = json.loads(match.group()) # 更新 severity 計數 for f in findings: sev = f.get("severity", "LOW").lower() if sev in self.state["severity_summary"]: self.state["severity_summary"][sev] += 1 self.state["findings"] = findings self._sync_global() return findings except Exception as e: logger.warning("[CodeReview] Hermes 掃描失敗: %s", e) return [] # ── Step 3:OpenClaw 評估 ────────────────────────────────────────────────── def _openclaw_assess(self, files: Dict[str, str], findings: List[Dict]) -> str: if not GEMINI_API_KEY: logger.warning("[CodeReview] GEMINI_API_KEY 未設定,跳過 OpenClaw") return "" try: import google.generativeai as genai genai.configure(api_key=GEMINI_API_KEY) model = genai.GenerativeModel( model_name=REVIEW_MODEL, generation_config=genai.types.GenerationConfig( temperature=0.3, max_output_tokens=1500, ), system_instruction=( "你是 OpenClaw 程式碼品質戰略分析師,以技術主管視角評估部署後程式碼。" "語言:繁體中文。風格:精準、數據導向、可執行建議。" ), ) sev = self.state["severity_summary"] findings_json = json.dumps(findings[:8], ensure_ascii=False, indent=2) files_list = "\n".join(f"- {k} ({len(v)} 字元)" for k, v in list(files.items())[:5]) prompt = f"""【部署資訊】Commit {self.commit_sha[:8]} @ {self.branch} 【變更檔案】 {files_list} 【Hermes 掃描摘要】CRITICAL={sev['critical']} HIGH={sev['high']} MEDIUM={sev['medium']} LOW={sev['low']} 【Hermes 詳細問題】 {findings_json} 請產出程式碼品質評估(使用 HTML 標題,150字以內): 🔍 整體風險等級(CRITICAL / HIGH / MEDIUM / LOW,一句理由) ⚠️ 最需關注問題(TOP 2,具體說明) 💡 架構優化方向(1 條長期建議) ✅ 本次部署亮點(值得肯定的地方)""" resp = model.generate_content(prompt, request_options={"timeout": 90}) return resp.text or "" except Exception as e: logger.warning("[CodeReview] OpenClaw 評估失敗: %s", e) return "" # ── Step 4:ElephantAlpha 決策 ───────────────────────────────────────────── def _ea_orchestrate(self, findings: List[Dict], openclaw_report: str) -> Dict: sev = self.state["severity_summary"] critical_n = sev["critical"] high_n = sev["high"] # 嘗試呼叫 ElephantAlpha 做精細判斷 try: from services.elephant_service import elephant_service top3 = json.dumps( [f for f in findings if f.get("severity") in ("CRITICAL", "HIGH")][:3], ensure_ascii=False, ) prompt = f"""你是 Elephant Alpha,負責協調 Code Review 後的修復決策。 【部署】commit={self.commit_sha[:8]} branch={self.branch} 【問題統計】CRITICAL={critical_n} HIGH={high_n} MEDIUM={sev['medium']} LOW={sev['low']} 【Top 問題】{top3} 【OpenClaw評估摘要】{openclaw_report[:300]} 請以 JSON 回答(不含其他文字): {{ "priority": "critical|high|medium|low", "auto_fix": true|false, "reasoning": "決策理由(繁體中文,一句話,需含具體數字)", "fix_files": ["需自動修復的檔案(最多3個,只填 CRITICAL/HIGH 問題的檔案)"], "human_review_needed": true|false }} 規則: - CRITICAL ≥ 1 → priority=critical, auto_fix=true - HIGH ≥ 3 → priority=high, auto_fix=true - HIGH 1-2 → priority=high, auto_fix=false, human_review_needed=true - 其餘 → priority=medium|low, auto_fix=false""" resp = elephant_service.generate( prompt=prompt, json_mode=True, temperature=0.1, timeout=60, ) if resp.success: return json.loads(resp.content) except Exception as e: logger.warning("[CodeReview] ElephantAlpha 決策失敗,回退規則: %s", e) # 規則 fallback auto_fix = critical_n > 0 or high_n >= 3 priority = ( "critical" if critical_n > 0 else "high" if high_n > 0 else "medium" if sev["medium"] > 0 else "low" ) fix_files = list({ f.get("file", "") for f in findings if f.get("severity") in ("CRITICAL", "HIGH") and f.get("file") })[:3] return { "priority": priority, "auto_fix": auto_fix, "reasoning": f"規則判斷:CRITICAL={critical_n} HIGH={high_n},{'觸發自動修復' if auto_fix else '需人工審查'}", "fix_files": fix_files, "human_review_needed": not auto_fix and (critical_n + high_n) > 0, } # ── Step 5:NemoTron 派遣 ────────────────────────────────────────────────── def _nemotron_dispatch(self, ea: Dict, findings: List[Dict]) -> Dict: auto_fix = ea.get("auto_fix", False) fix_files = ea.get("fix_files", []) priority_map = {"critical": 1, "high": 2, "medium": 3, "low": 4} priority_num = priority_map.get(ea.get("priority", "low"), 4) actions_created = 0 session = get_session() try: # 每個需修復的檔案建立一筆 action_plan for fpath in fix_files[:3]: related = [f for f in findings if f.get("file") == fpath][:3] desc = f"Code Review 修復:{fpath}|{', '.join(f.get('description','')[:40] for f in related)}" session.execute(text(""" INSERT INTO action_plans (action_type, description, status, priority, metadata_json, created_at) VALUES ('code_review_fix', :desc, :status, :priority, :meta, NOW()) """), { "desc": desc[:500], "status": "auto_pending" if auto_fix else "pending_review", "priority": priority_num, "meta": json.dumps({ "pipeline_id": self.pipeline_id, "commit_sha": self.commit_sha, "file": fpath, "auto_fix": auto_fix, "ea_priority": ea.get("priority"), "findings": related, }, ensure_ascii=False), }) actions_created += 1 session.commit() except Exception as e: logger.warning("[CodeReview] action_plans 寫入失敗: %s", e) session.rollback() finally: session.close() # 觸發 AiderHeal(非阻塞) if auto_fix and fix_files: self.state["auto_fix_triggered"] = True self._sync_global() self._trigger_aider_heal(findings, fix_files) return {"actions": actions_created, "auto_fix": auto_fix} def _trigger_aider_heal(self, findings: List[Dict], fix_files: List[str]): """非阻塞觸發 AiderHeal 自動修復""" def _heal_worker(): try: from services.aider_heal_executor import execute_code_fix for fpath in fix_files[:2]: # 最多同時修 2 個檔案 related = [f for f in findings if f.get("file") == fpath] if not related: continue worst = sorted(related, key=lambda x: {"CRITICAL":0,"HIGH":1,"MEDIUM":2,"LOW":3}.get(x.get("severity","LOW"),3))[0] result = execute_code_fix( error_type=f"code_review_{worst.get('type','bug')}", error_message=worst.get("description", "Code Review 發現問題"), target_file=fpath, context={"suggestion": worst.get("suggestion", ""), "pipeline_id": self.pipeline_id}, ) logger.info("[CodeReview] AiderHeal %s → %s", fpath, result.get("message", "")) except Exception as e: logger.error("[CodeReview] AiderHeal 觸發失敗: %s", e) t = threading.Thread(target=_heal_worker, daemon=True, name=f"aider-heal-{self.commit_sha[:8]}") t.start() # ── DB 持久化 ────────────────────────────────────────────────────────────── def _save_to_db(self, findings: List[Dict], openclaw_report: str, ea: Dict): session = get_session() try: session.execute(text(""" INSERT INTO ai_insights (insight_type, content, confidence, created_by, status, metadata_json, period, created_at) VALUES ('code_review_result', :content, :conf, 'code_review_pipeline', 'active', :meta, :period, NOW()) """), { "content": json.dumps({ "findings": findings, "openclaw_report": openclaw_report, "ea_decision": ea, "severity_summary": self.state["severity_summary"], }, ensure_ascii=False)[:8000], "conf": 0.90, "meta": json.dumps({ "pipeline_id": self.pipeline_id, "commit_sha": self.commit_sha, "branch": self.branch, "changed_files": self.changed_files, "auto_fix_triggered": self.state["auto_fix_triggered"], }, ensure_ascii=False), "period": datetime.now().strftime("%Y-%m-%d"), }) session.commit() logger.info("[CodeReview] ai_insights 寫入成功 pipeline=%s", self.pipeline_id) except Exception as e: logger.error("[CodeReview] DB 寫入失敗: %s", e) session.rollback() finally: session.close() # ── Telegram 通知 ───────────────────────────────────────────────────────── def _notify_start(self): try: from services.telegram_templates import _send_telegram_raw files_list = "\n".join(f" • {f}" for f in self.changed_files[:5]) if len(self.changed_files) > 5: files_list += f"\n (+{len(self.changed_files)-5} 個)" _send_telegram_raw( f"🔍 Code Review 啟動\n" f"══════════════════════════\n" f"📦 Commit {self.commit_sha[:8]} 🌿 {self.branch}\n" f"📝 變更檔案:\n{files_list}\n" f"══════════════════════════\n" f"🤖 Hermes → OpenClaw → Elephant Alpha → NemoTron\n" f"📊 即時進度:https://mo.wooo.work/code-review/" ) except Exception as e: logger.warning("[CodeReview] 啟動通知失敗: %s", e) def _notify_complete(self, findings: List[Dict], openclaw_report: str, ea: Dict): try: from services.telegram_templates import _send_telegram_raw sev = self.state["severity_summary"] priority = ea.get("priority", "medium") auto_fix = ea.get("auto_fix", False) icon = {"critical": "🔴", "high": "🟠", "medium": "🟡", "low": "🟢"}.get(priority, "🟡") top_issues = [f for f in findings if f.get("severity") in ("CRITICAL", "HIGH")][:3] issues_lines = "\n".join( f" {'🔴' if f['severity']=='CRITICAL' else '🟠'} [{f['severity']}] {f.get('description','')} — {f.get('file','')}" for f in top_issues ) or " ✅ 無高風險問題" msg = ( f"{icon} Code Review 完成 · {self.commit_sha[:8]}\n" f"══════════════════════════\n" f"🔴 CRITICAL {sev['critical']} " f"🟠 HIGH {sev['high']} " f"🟡 MEDIUM {sev['medium']} " f"🟢 LOW {sev['low']}\n" f"══════════════════════════\n" f"⚠️ 主要問題\n{issues_lines}\n" ) if openclaw_report: msg += f"\n{openclaw_report[:400]}\n" fix_status = "🔧 已觸發自動修復(AiderHeal)" if auto_fix else ( "👁 需人工審查" if ea.get("human_review_needed") else "✅ 無需修復動作" ) msg += ( f"══════════════════════════\n" f"🤖 Elephant Alpha:{priority.upper()} {fix_status}\n" f"📊 完整報告:https://mo.wooo.work/code-review/" ) _send_telegram_raw(msg) except Exception as e: logger.warning("[CodeReview] 完成通知失敗: %s", e) def _notify_error(self, error: str): try: from services.telegram_templates import _send_telegram_raw _send_telegram_raw( f"🚨 Code Review Pipeline 失敗\n" f"Commit:{self.commit_sha[:8]} @ {self.branch}\n" f"錯誤:{error[:200]}\n" f"📊 查看:https://mo.wooo.work/code-review/" ) except Exception: pass # ═══════════════════════════════════════════════════════════════════════════════ # 公開 API # ═══════════════════════════════════════════════════════════════════════════════ def trigger_post_deploy_review( commit_sha: str, changed_files: List[str], branch: str = "main", deploy_type: str = "sync", ) -> str: """ 啟動 Pipeline(後台 daemon thread)。 回傳 pipeline_id。由 routes/code_review_routes.py 的 webhook 端點呼叫。 """ pipeline = CodeReviewPipeline(commit_sha, changed_files, branch, deploy_type) t = threading.Thread( target=pipeline.run, daemon=True, name=f"code-review-{commit_sha[:8]}", ) t.start() logger.info("[CodeReview] 已派發 pipeline=%s files=%d", pipeline.pipeline_id, len(pipeline.changed_files)) return pipeline.pipeline_id def get_current_state() -> Dict[str, Any]: """前端 polling 用:取得目前 pipeline 即時狀態""" with _pipeline_lock: return dict(_current_pipeline) def get_history(limit: int = 20) -> List[Dict]: """取得 ai_insights 中歷史 code_review_result 記錄""" session = get_session() try: rows = session.execute(text(""" SELECT id, content, confidence, metadata_json, created_at, status FROM ai_insights WHERE insight_type = 'code_review_result' ORDER BY created_at DESC LIMIT :lim """), {"lim": limit}).fetchall() results = [] for r in rows: meta, content = {}, {} try: meta = json.loads(r[3]) if r[3] else {} content = json.loads(r[1]) if r[1] else {} except Exception: pass sev = content.get("severity_summary", {}) results.append({ "id": r[0], "pipeline_id": meta.get("pipeline_id", ""), "commit_sha": meta.get("commit_sha", "")[:8], "branch": meta.get("branch", ""), "changed_files": meta.get("changed_files", []), "severity_summary": sev, "total_issues": sum(sev.values()), "auto_fix": meta.get("auto_fix_triggered", False), "ea_decision": content.get("ea_decision", {}), "created_at": r[4].isoformat() if r[4] else "", "status": r[5] or "active", }) return results except Exception as e: logger.warning("[CodeReview] 歷史讀取失敗: %s", e) return [] finally: session.close() def verify_internal_token(request_token: str) -> bool: """驗證 CD webhook 來源 token。未設定 env 時直接放行(dev 環境)""" if not INTERNAL_TOKEN: return True return request_token == INTERNAL_TOKEN