#!/usr/bin/env python3 # ================= MOMO 系統 - Code Review 工具類 ================= # 功能：提供Code Review相關的工具函數和錯誤處理 # 作者：AI Assistant # 版本：1.0 # ======================================================================= import os import sys import json import logging import traceback from datetime import datetime from pathlib import Path from typing import List, Dict, Any, Optional class CodeReviewLogger: """Code Review 專用日誌記錄器""" def __init__(self, project_root): self.project_root = Path(project_root) if not isinstance(project_root, Path) else project_root self.log_dir = self.project_root / "logs" self.log_file = self.log_dir / "code_review.log" # 確保日誌目錄存在 self.log_dir.mkdir(exist_ok=True) # 設置日誌記錄器 self.logger = logging.getLogger("code_review") self.logger.setLevel(logging.INFO) # 避免重複添加handler if not self.logger.handlers: # 檔案handler file_handler = logging.FileHandler(self.log_file, encoding='utf-8') file_handler.setLevel(logging.INFO) # 控制台handler console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) # 格式化器 formatter = logging.Formatter('[%(asctime)s] %(message)s', '%Y-%m-%d %H:%M:%S') file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) self.logger.addHandler(file_handler) self.logger.addHandler(console_handler) def info(self, message: str): """記錄資訊日誌""" self.logger.info(message) def error(self, message: str, exception: Optional[Exception] = None): """記錄錯誤日誌""" if exception: self.logger.error(f"{message}: {str(exception)}") self.logger.error(f"Traceback: {traceback.format_exc()}") else: self.logger.error(message) def warning(self, message: str): """記錄警告日誌""" self.logger.warning(message) class CodeReviewConfig: """Code Review 配置管理""" def __init__(self, project_root): self.project_root = Path(project_root) if not isinstance(project_root, Path) else project_root self.config_file = self.project_root / "config" / "code_review_config.json" self.default_config = self._get_default_config() self.config = self._load_config() def _get_default_config(self) -> Dict[str, Any]: """獲取預設配置""" return { "aider": { "path": "/Users/ooo/.local/bin/aider", "timeout": 600, "auto_yes": False, "no_git": True }, "review_types": { "basic": { "message": "請對這些檔案進行Code Review，檢查：\n1. 程式碼品質與最佳實踐\n2. 潛在的bug與安全問題\n3. 程式碼可讀性與維護性\n4. 效能優化建議" }, "security": { "message": "請重點檢查這些檔案的安全性：\n1. SQL注入風險\n2. XSS攻擊風險\n3. 認證與授權問題\n4. 敏感資料洩露風險\n5. 輸入驗證不足" }, "performance": { "message": "請分析這些檔案的效能問題：\n1. 演算法效率\n2. 資料庫查詢優化\n3. 記憶體使用\n4. 並發處理\n5. 快取策略" } }, "file_filters": { "allowed_extensions": [".py", ".js", ".ts", ".jsx", ".tsx", ".html", ".css"], "excluded_dirs": [".git", "__pycache__", "node_modules", "venv", "env", ".pytest_cache"], "max_file_size": 10485760 # 10MB }, "git": { "diff_filter": "ACM", # Added, Copied, Modified "use_staged": True } } def _load_config(self) -> Dict[str, Any]: """載入配置檔案""" if self.config_file.exists(): try: with open(self.config_file, 'r', encoding='utf-8') as f: user_config = json.load(f) # 合併預設配置和用戶配置 return self._merge_config(self.default_config, user_config) except Exception as e: print(f"⚠️ 載入配置檔案失敗，使用預設配置: {e}") return self.default_config else: # 創建預設配置檔案 self._save_config(self.default_config) return self.default_config def _merge_config(self, default: Dict, user: Dict) -> Dict: """合併配置""" result = default.copy() for key, value in user.items(): if key in result and isinstance(result[key], dict) and isinstance(value, dict): result[key] = self._merge_config(result[key], value) else: result[key] = value return result def _save_config(self, config: Dict[str, Any]): """保存配置檔案""" try: self.config_file.parent.mkdir(exist_ok=True) with open(self.config_file, 'w', encoding='utf-8') as f: json.dump(config, f, indent=2, ensure_ascii=False) except Exception as e: print(f"⚠️ 保存配置檔案失敗: {e}") def get(self, key_path: str, default=None): """獲取配置值""" keys = key_path.split('.') value = self.config for key in keys: if isinstance(value, dict) and key in value: value = value[key] else: return default return value class CodeReviewValidator: """Code Review 資料驗證器""" @staticmethod def validate_files(files: List[str], project_root: Path) -> List[str]: """驗證檔案列表""" valid_files = [] for file_path in files: abs_path = project_root / file_path if not Path(file_path).is_absolute() else Path(file_path) if not abs_path.exists(): print(f"⚠️ 檔案不存在: {file_path}") continue if not abs_path.is_file(): print(f"⚠️ 不是檔案: {file_path}") continue # 檢查檔案大小 if abs_path.stat().st_size > 10485760: # 10MB print(f"⚠️ 檔案過大: {file_path}") continue valid_files.append(str(abs_path.relative_to(project_root))) return valid_files @staticmethod def validate_review_type(review_type: str) -> bool: """驗證Review類型""" return review_type in ["basic", "security", "performance"] @staticmethod def sanitize_filename(filename: str) -> str: """清理檔案名稱""" # 移除危險字符 dangerous_chars = ['..', '/', '\\', ':', '*', '?', '"', '<', '>', '|'] for char in dangerous_chars: filename = filename.replace(char, '_') return filename class GitHelper: """Git 操作輔助工具""" def __init__(self, project_root: Path): self.project_root = project_root def get_changed_files(self, diff_filter: str = "ACM", use_staged: bool = True) -> List[str]: """獲取變更的檔案""" try: cmd = ["git"] if use_staged: cmd.extend(["diff", "--cached", "--name-only", "--diff-filter", diff_filter]) else: cmd.extend(["diff", "--name-only", "--diff-filter", diff_filter]) result = subprocess.run( cmd, cwd=self.project_root, capture_output=True, text=True, timeout=30 ) if result.returncode == 0: files = [f.strip() for f in result.stdout.split('\n') if f.strip()] return files else: print(f"❌ Git 命令失敗: {result.stderr}") return [] except subprocess.TimeoutExpired: print("❌ Git 命令超時") return [] except Exception as e: print(f"❌ Git 操作異常: {e}") return [] def is_git_repository(self) -> bool: """檢查是否為Git倉庫""" git_dir = self.project_root / ".git" return git_dir.exists() and git_dir.is_dir() # 導入subprocess用於GitHelper import subprocess