#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Git pre-commit 钩子 - 自动化规范检查 安装方法: Windows: copy scripts/pre-commit-hook.py .git/hooks/pre-commit Linux/Mac: cp scripts/pre-commit-hook.py .git/hooks/pre-commit 功能: - 提交前自动检查代码规范 - 阻塞性问题阻止提交 - 可配置跳过检查的文件 """ import os import re import subprocess import sys from pathlib import Path from typing import List, Tuple # ANSI 颜色 RED = '\033[91m' GREEN = '\033[92m' YELLOW = '\033[93m' BLUE = '\033[94m' CYAN = '\033[96m' END = '\033[0m' BOLD = '\033[1m' def colored(text: str, color: str) -> str: return f"{color}{text}{END}" def print_info(msg: str): print(f"{CYAN}ℹ️{END} {msg}") def print_success(msg: str): print(f"{GREEN}✅{END} {msg}") def print_warning(msg: str): print(f"{YELLOW}⚠️{END} {msg}") def print_error(msg: str): print(f"{RED}❌{END} {msg}") # 检查规则 RULES = [ { "id": "DATE001", "name": "禁止使用Date类型", "pattern": r'\bjava\.util\.Date\b', "severity": "BLOCKER", "message": "必须使用java.time.LocalDateTime" }, { "id": "SQL001", "name": "禁止SELECT *", "pattern": r'SELECT\s+\*\s+FROM', "severity": "BLOCKER", "message": "必须明确列出查询字段" }, { "id": "SQL002", "name": "禁止使用$拼接", "pattern": r'\$\{[^}]+\}', "severity": "BLOCKER", "message": "必须使用#{}而非${}" }, { "id": "DI001", "name": "禁止字段注入", "pattern": r'@Autowired\s+private\s+', "severity": "BLOCKER", "message": "必须使用构造器注入" }, { "id": "EXEC001", "name": "禁止Executors", "pattern": r'Executors\.(newFixedThreadPool|newCachedThreadPool)', "severity": "BLOCKER", "message": "必须使用ThreadPoolExecutor" }, { "id": "LOG001", "name": "禁止+拼接日志", "pattern": r'log\.(info|warn|error)\s*\([^"\']*\+[^)]+\)', "severity": "MAJOR", "message": "必须使用占位符" }, { "id": "SEC001", "name": "禁止硬编码密码", "pattern": r'password\s*=\s*["\'][^"\']{8,}["\']', "severity": "BLOCKER", "message": "禁止硬编码密码" }, { "id": "SYS001", "name": "禁止System.out", "pattern": r'System\.(out|err)\.(print|println)', "severity": "MAJOR", "message": "必须使用日志框架" }, ] def get_staged_files() -> List[str]: """获取暂存的文件""" try: result = subprocess.run( ['git', 'diff', '--cached', '--name-only', '--diff-filter=ACM'], capture_output=True, text=True, check=True ) return [f.strip() for f in result.stdout.strip().split('\n') if f.strip()] except subprocess.CalledProcessError: return [] def get_file_diff(file_path: str) -> List[Tuple[int, str]]: """获取文件的变更内容""" try: result = subprocess.run( ['git', 'diff', '--cached', file_path], capture_output=True, text=True, check=True ) lines = [] for line in result.stdout.split('\n'): if line.startswith('+') and not line.startswith('+++'): line_num = len([l for l in lines if l[1].startswith('+')]) lines.append((line_num, line[1:])) elif line.startswith(' ') and not line.startswith('---'): lines.append((len([l for l in lines if not l[1].startswith('@@')]), line)) return lines except subprocess.CalledProcessError: return [] # JWT相关文件允许使用Date类型（JWT库要求） JWT_ALLOW_DATE_FILES = [ 'JwtUtil.java', 'AuthService.java', 'TokenService.java', 'JwtService.java' ] def check_file(file_path: str, content: str) -> List[dict]: """检查文件内容""" violations = [] lines = content.split('\n') # 检查是否是JWT相关文件 is_jwt_file = any(jwt_file in file_path for jwt_file in JWT_ALLOW_DATE_FILES) for line_no, line in enumerate(lines, 1): # 跳过注释 if line.strip().startswith('//') or line.strip().startswith('/*'): continue for rule in RULES: # JWT相关文件允许使用Date类型 if is_jwt_file and rule['id'] == 'DATE001': continue if re.search(rule['pattern'], line, re.IGNORECASE): violations.append({ 'file': file_path, 'line': line_no, 'rule': rule['id'], 'name': rule['name'], 'severity': rule['severity'], 'message': rule['message'], 'code': line.strip()[:60] }) return violations def main(): print() print("=" * 60) print(f"{BOLD}🔍 执行提交前规范检查{END}") print("=" * 60) print() staged_files = get_staged_files() if not staged_files: print_warning("没有暂存的文件，跳过检查") sys.exit(0) # 只检查Java文件 java_files = [f for f in staged_files if f.endswith('.java')] if not java_files: print_info("没有暂存Java文件，跳过检查") sys.exit(0) print_info(f"检查 {len(java_files)} 个Java文件...") all_violations = [] blocker_count = 0 for file_path in java_files: # 获取暂存的文件内容 try: result = subprocess.run( ['git', 'show', f':{file_path}'], capture_output=True, text=True, check=True ) content = result.stdout except subprocess.CalledProcessError: continue violations = check_file(file_path, content) all_violations.extend(violations) if any(v['severity'] == 'BLOCKER' for v in violations): blocker_count += 1 if not all_violations: print_success("所有检查通过！") sys.exit(0) # 按严重性分组 blockers = [v for v in all_violations if v['severity'] == 'BLOCKER'] majors = [v for v in all_violations if v['severity'] == 'MAJOR'] if blockers: print() print_error(f"发现 {len(blockers)} 个阻塞性问题，必须修复后才能提交！") print() for v in blockers: print(f" {RED}✗{END} [{v['rule']}] {v['file']}:{v['line']}") print(f" {CYAN}{v['code']}{END}") print(f" {YELLOW}建议: {v['message']}{END}") print() if majors: print() print_warning(f"发现 {len(majors)} 个重要问题（建议修复）:") for v in majors[:5]: # 只显示前5个 print(f" {YELLOW}⚠{END} [{v['rule']}] {v['file']}:{v['line']}") print(f" {CYAN}{v['code']}{END}") if len(majors) > 5: print(f" ... 还有 {len(majors) - 5} 个问题") print() print("-" * 60) if blockers: print(f"{RED}❌ 提交被阻止，请修复阻塞性问题后重试{END}") print() sys.exit(1) else: print(f"{YELLOW}⚠️ 存在 {len(majors)} 个建议修复的问题{END}") print(f"{GREEN}✅ 可以提交（但建议先修复以上问题）{END}") print() sys.exit(0) if __name__ == '__main__': main()