208 lines
7.4 KiB
Python
208 lines
7.4 KiB
Python
import ollama
|
||
from mcp.server.fastmcp import FastMCP
|
||
import os
|
||
import json
|
||
import logging
|
||
import time
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
|
||
_CONFIG = None
|
||
mcp = FastMCP("code_parser")
|
||
|
||
def get_config():
|
||
"""获取配置并缓存"""
|
||
global _CONFIG
|
||
if _CONFIG is None:
|
||
try:
|
||
|
||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
config_path = os.path.join(script_dir, '..', 'config.json')
|
||
with open(config_path, 'r', encoding='utf-8') as f:
|
||
_CONFIG = json.load(f)
|
||
logging.info("配置加载成功")
|
||
except Exception as e:
|
||
logging.error(f"配置文件加载失败: {str(e)}")
|
||
raise
|
||
return _CONFIG
|
||
|
||
def llm_parse_code(code_path, code_content):
|
||
"""使用OLLAMA处理代码注释翻译和添加"""
|
||
config = get_config()
|
||
|
||
client = ollama.Client(host=config['OLLAMA_URL'])
|
||
prompt = f'''
|
||
请严格遵循以下要求处理代码：
|
||
1. 仅添加中文注释，不要修改任何原始代码逻辑、格式和变量名
|
||
2. 对于已有的英文注释，请将其准确地翻译成中文
|
||
3. 请保留原代码的格式和结构
|
||
4. 最后输出完整且可运行的代码（不要使用markdown格式）
|
||
|
||
文件路径：{code_path}
|
||
代码内容：
|
||
{code_content}
|
||
'''
|
||
|
||
max_retries = 3
|
||
for attempt in range(max_retries):
|
||
try:
|
||
response = ""
|
||
chat_stream = client.generate(
|
||
model=config['OLLAMA_MODEL'],
|
||
prompt=prompt,
|
||
stream=True,
|
||
think=False
|
||
)
|
||
|
||
for chunk in chat_stream:
|
||
content = chunk.get('response', '')
|
||
response += content
|
||
if response.strip():
|
||
return response
|
||
logging.warning(f"LLM响应可能无效，重试中... ({attempt+1}/{max_retries})")
|
||
|
||
except Exception as e:
|
||
logging.error(f"LLM请求失败: {str(e)}，尝试重新连接...")
|
||
time.sleep(2 ** attempt)
|
||
|
||
logging.error("LLM处理失败，返回原始代码")
|
||
return code_content
|
||
|
||
def detect_and_read_file(file_path):
|
||
"""尝试多种编码方式读取文件内容"""
|
||
encodings = ['utf-8', 'gbk', 'latin-1', 'cp1252', 'iso-8859-1']
|
||
|
||
for encoding in encodings:
|
||
try:
|
||
with open(file_path, 'r', encoding=encoding) as f:
|
||
content = f.read()
|
||
return content
|
||
except UnicodeDecodeError:
|
||
continue
|
||
|
||
|
||
try:
|
||
with open(file_path, 'rb') as f:
|
||
raw_data = f.read()
|
||
return raw_data.decode('utf-8', errors='ignore')
|
||
except Exception as e:
|
||
logging.error(f"无法读取文件 {file_path}: {str(e)}")
|
||
return None
|
||
|
||
@mcp.tool()
|
||
def parse_code(project_path, excluded_items_path=None, save_path=None):
|
||
"""为指定目录下的代码文件添加中文注释。
|
||
|
||
Args:
|
||
project_path: 项目根目录路径
|
||
excluded_items_path: 需要排除的文件或目录列表(可选)
|
||
save_path: 解析后文件保存路径(可选，默认为项目根目录下的 'parsed_code' 目录)
|
||
"""
|
||
|
||
project_path = os.path.abspath(project_path)
|
||
|
||
|
||
if save_path is None:
|
||
save_path = os.path.join(project_path, 'parsed_code')
|
||
else:
|
||
save_path = os.path.abspath(save_path)
|
||
|
||
|
||
os.makedirs(save_path, exist_ok=True)
|
||
abs_save_path = os.path.abspath(save_path)
|
||
|
||
excluded_abs_paths = set()
|
||
if excluded_items_path is not None:
|
||
if isinstance(excluded_items_path, str):
|
||
|
||
with open(excluded_items_path, 'r', encoding='utf-8') as f:
|
||
excluded_items = [line.strip() for line in f.readlines()]
|
||
|
||
for item in excluded_items:
|
||
abs_item = os.path.abspath(os.path.join(project_path, item))
|
||
excluded_abs_paths.add(abs_item)
|
||
|
||
|
||
code_extensions = [
|
||
'.py', '.js', '.jsx', '.java', '.c', '.cpp', '.h', '.hpp',
|
||
'.cs', '.go', '.rs', '.ts', '.tsx', '.html', '.css', '.scss',
|
||
'.php', '.rb', '.swift', '.kt', '.m', '.sql', '.sh', '.bat'
|
||
]
|
||
|
||
files_to_process = []
|
||
for root, dirs, files in os.walk(project_path):
|
||
root_abs = os.path.abspath(root)
|
||
|
||
|
||
if root_abs.startswith(abs_save_path + os.sep) or root_abs == abs_save_path:
|
||
continue
|
||
dirs[:] = [d for d in dirs if os.path.join(root_abs, d) not in excluded_abs_paths]
|
||
for file in files:
|
||
file_path = os.path.join(root_abs, file)
|
||
if file_path in excluded_abs_paths:
|
||
continue
|
||
_, ext = os.path.splitext(file)
|
||
if ext.lower() not in code_extensions:
|
||
continue
|
||
relative_path = os.path.relpath(root_abs, project_path)
|
||
save_dir = os.path.join(save_path, relative_path)
|
||
target_path = os.path.join(save_dir, file)
|
||
if os.path.exists(target_path):
|
||
logging.info(f"跳过已处理文件: {file_path}")
|
||
continue
|
||
|
||
files_to_process.append((file_path, save_dir, target_path))
|
||
|
||
if not files_to_process:
|
||
return "没有找到需要处理的代码文件"
|
||
|
||
logging.info(f"发现 {len(files_to_process)} 个文件需要处理")
|
||
def process_file(file_data):
|
||
file_path, save_dir, target_path = file_data
|
||
try:
|
||
|
||
code_content = detect_and_read_file(file_path)
|
||
if code_content is None:
|
||
logging.warning(f"无法读取文件 {file_path}，跳过处理")
|
||
return
|
||
|
||
MAX_LINES = 800
|
||
if code_content.count('\n') > MAX_LINES:
|
||
logging.warning(f"文件过大({file_path}，{code_content.count('\n')}行)，跳过处理")
|
||
return
|
||
|
||
logging.info(f"处理文件: {file_path}")
|
||
|
||
relative_file_path = os.path.relpath(file_path, project_path)
|
||
parsed_code = llm_parse_code(relative_file_path, code_content)
|
||
|
||
|
||
os.makedirs(save_dir, exist_ok=True)
|
||
with open(target_path, 'w', encoding='utf-8') as out_file:
|
||
out_file.write(parsed_code)
|
||
|
||
except Exception as e:
|
||
logging.error(f"处理文件失败 {file_path}: {str(e)}")
|
||
|
||
|
||
max_workers = min(os.cpu_count() or 1, 4)
|
||
logging.info(f"使用线程池处理，最大线程数: {max_workers}")
|
||
|
||
processed_count = 0
|
||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||
futures = {executor.submit(process_file, file_data): file_data for file_data in files_to_process}
|
||
for future in as_completed(futures):
|
||
file_data = futures[future]
|
||
try:
|
||
future.result()
|
||
processed_count += 1
|
||
except Exception as e:
|
||
logging.error(f"处理文件 {file_data[0]} 出现异常: {str(e)}")
|
||
|
||
return f"代码注释添加完成: 处理了 {processed_count}/{len(files_to_process)} 个文件，保存路径: {save_path}"
|
||
|
||
|
||
if __name__ == "__main__":
|
||
mcp.run(transport='stdio')
|