#!/usr/bin/env python3 # -*- coding: utf-8 -*- import os import requests import time import chardet # 用于自动检测文件编码 from dataclasses import dataclass import concurrent.futures from typing import List @dataclass class AnalysisResult: filepath: str issues: dict model_used: str elapsed: float class OllamaAnalyzer: def __init__(self, server_url, analysis_level="standard"): self.base_url = server_url.rstrip('/') self.session = requests.Session() self.session.timeout = 30 self.analysis_level = analysis_level self.rules = { "basic": ["security"], "standard": ["security", "code_smell", "io"], "strict": ["security", "code_smell", "io", "performance", "complexity"] } def detect_encoding(self, filepath: str) -> str: """自动检测文件编码""" with open(filepath, 'rb') as f: rawdata = f.read(1024) # 读取前1KB用于检测 result = chardet.detect(rawdata) return result['encoding'] or 'utf-8' def analyze(self, filepath: str) -> AnalysisResult: """执行代码分析并打印原始响应""" start_time = time.time() try: encoding = self.detect_encoding(filepath) with open(filepath, 'r', encoding=encoding) as f: code = f.read() response = self._send_analysis_request(code) # 打印原始响应 print(f"\n🔍 原始分析响应 ({filepath}):") print(json.dumps(response, indent=2, ensure_ascii=False)) print("-" * 50) return AnalysisResult( filepath=filepath, issues=self._parse_response(response), model_used="codellama:7b", elapsed=time.time() - start_time ) except Exception as e: return AnalysisResult( filepath=filepath, issues={"ERROR": [f"分析失败: {str(e)}"]}, model_used="N/A", elapsed=time.time() - start_time ) def _send_analysis_request(self, code: str) -> dict: """发送分析请求到Ollama""" payload = { "model": "codellama:7b", "prompt": f"Analyze this code for issues:\n```csharp\n{code[:5000]}\n```", "stream": False, "options": {"temperature": 0.2} } try: resp = self.session.post( f"{self.base_url}/api/generate", json=payload, timeout=45 ) resp.raise_for_status() return resp.json() except requests.exceptions.RequestException as e: return {"response": f"API请求失败: {str(e)}"} def _parse_response(self, response: dict) -> dict: """增强版响应解析器""" text = response.get('response', '').lower() issues = {} # 安全漏洞检测 if self.analysis_level in ["standard", "strict", "basic"]: security_terms = [ 'sql injection', 'xss', 'csrf', 'path traversal', 'insecure deserialization', 'buffer overflow', 'command injection' ] if any(term in text for term in security_terms): issues["SECURITY"] = ["发现潜在安全漏洞"] # 代码异味检测 if self.analysis_level in ["standard", "strict"]: code_smell_terms = [ 'unused variable', 'magic number', 'long method', 'duplicate code', 'god class', 'feature envy' ] if any(term in text for term in code_smell_terms): issues["CODE_SMELL"] = ["发现代码异味"] # 性能问题检测 if self.analysis_level == "strict": perf_terms = [ 'n+1 query', 'memory leak', 'inefficient loop', 'unbounded collection', 'excessive allocation' ] if any(term in text for term in perf_terms): issues["PERFORMANCE"] = ["发现潜在性能问题"] return issues if issues else {"INFO": ["未发现严重问题"]} def perform_analysis(server_url: str, filepaths: List[str]) -> None: """执行全自动分析流程""" class EnhancedAnalyzer(OllamaAnalyzer): def __init__(self, server_url): super().__init__(server_url, analysis_level="strict") self.success_count = 0 def analyze(self, filepath): result = super().analyze(filepath) if not any("ERROR" in k for k in result.issues): self.success_count += 1 return result print(f"\n🔧 正在初始化分析器 (服务器: {server_url})...") analyzer = EnhancedAnalyzer(server_url) with concurrent.futures.ThreadPoolExecutor() as executor: results = list(executor.map(analyzer.analyze, filepaths)) print("\n📊 分析完成!") print(f"✅ 成功分析文件数: {analyzer.success_count}/{len(filepaths)}") for result in results: print(f"\n📄 文件: {result.filepath}") for category, items in result.issues.items(): print(f" ⚠️ {category}: {', '.join(items)}") print(f" ⏱️ 耗时: {result.elapsed:.2f}秒") def create_test_files(): """创建包含多种编码的测试文件""" test_files = [] # UTF-8编码测试文件 utf8_file = "utf8_test.cs" with open(utf8_file, 'w', encoding='utf-8') as f: f.write(''' // UTF-8 测试文件 using System; class Program { static void Main() { Console.WriteLine("你好，世界!"); // 中文测试 var x = 100; // 魔法数字 } }''') test_files.append(utf8_file) # GBK编码测试文件 gbk_file = "gbk_test.cs" with open(gbk_file, 'w', encoding='gbk') as f: f.write(''' // GBK 测试文件 using System.Data.SqlClient; class Program { static void Main(string input) { var cmd = "SELECT * FROM Users WHERE Name='" + input + "'"; new SqlCommand(cmd).ExecuteReader(); // SQL注入测试 } }''') test_files.append(gbk_file) return test_files def main(): # 创建测试文件 test_files = create_test_files() # 使用扫描到的服务器 selected_server = "http://44.251.176.235:4711" perform_analysis(selected_server, test_files) # 清理测试文件 for filename in test_files: try: os.remove(filename) except Exception as e: print(f"⚠️ 清理文件 {filename} 失败: {str(e)}") if __name__ == "__main__": # 确保chardet包已安装 try: import chardet except ImportError: print("正在安装所需依赖...") os.system("pip install chardet requests") main()，返回python3 code_analyzer.py 🔧 正在初始化分析器 (服务器: http://44.251.176.235:4711)... 🔍 原始分析响应 (gbk_test.cs): 🔍 原始分析响应 (utf8_test.cs): 📊 分析完成! ✅ 成功分析文件数: 0/2 📄 文件: utf8_test.cs ⚠️ ERROR: 分析失败: name 'json' is not defined ⏱️ 耗时: 0.57秒 📄 文件: gbk_test.cs ⚠️ ERROR: 分析失败: name 'json' is not defined ⏱️ 耗时: 0.57秒，给出优化后的完整脚本