a96306fba2
- Install python-telegram-bot dependency - Start Telegram bot service successfully - Confirm correct group ID (MOMO PRO - small shrimp group) - Bot now running with all commands and button interface functional - Natural language processing restored with keyword matching Fixes issue where Telegram group could not communicate using natural language.
237 lines
8.7 KiB
Python
237 lines
8.7 KiB
Python
#!/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
|