Gitnexus-python/api/aiRouter.py

import os, json, time, asyncio
from base_config import ai_key, path, file_summary_app_id, commit_summary_app_id, filter_code_files_app_id, analysis_results_app_id
from fastapi import APIRouter, BackgroundTasks
from pathlib import Path
from pydantic import BaseModel
from git import Repo

from http import HTTPStatus
from dashscope import Application

from models.aiModels import Scan_Tasks, Commit_Summary_Tasks, File_Summary_Tasks

from mcp_.client import MCPClient

airouter = APIRouter()
class RequestCommit(BaseModel):
    task_id: str
    uuid: str
    repo_url: str

class RequestScan(BaseModel):
    task_id: str
    uuid: str
    repo_url: str

class RequestFile(BaseModel):
    task_id: str
    uuid: str
    repo_url: str
    file_path: str

class RequestChat(BaseModel):
    uuid: str
    message: str

def generate_repo_path(uuid, repo_url):
    repo_name = repo_url.split("/")[-1].replace(".git", "")
    base_path = os.path.join(path, uuid)
    return os.path.join(base_path, repo_name), repo_name

async def file_summary(content):
    response = Application.call(
        api_key=ai_key,
        app_id=file_summary_app_id,
        prompt=content)
    if response.status_code == HTTPStatus.OK:
        try:
            json_data = json.loads(response.output.text)
            print(json_data)
        except json.JSONDecodeError:
            print("返回内容不是有效的 JSON 格式！")
            print(response.output.text)
            json_data = {"summary": []}
    else:
        print(f"请求失败: {response.message}")
        json_data = {"summary": []}
    return json_data

async def commit_summary(content):
    response = Application.call(
        api_key=ai_key,
        app_id=commit_summary_app_id,
        prompt=content)
    if response.status_code == HTTPStatus.OK:
        try:
            json_data = json.loads(response.output.text)
            print(json_data)
        except json.JSONDecodeError:
            print("返回内容不是有效的 JSON 格式！")
            print(response.output.text)
            json_data = {"null": []}
    else:
        print(f"请求失败: {response.message}")
        json_data = {"null": []}
    return json_data

def filter_code_files(prompt):
    response = Application.call(
        api_key=ai_key,
        app_id=filter_code_files_app_id,
        prompt=prompt)
    if response.status_code == HTTPStatus.OK:
        try:
            json_data = json.loads(response.output.text)
            print(json_data)
        except json.JSONDecodeError:
            print("返回内容不是有效的 JSON 格式！")
            print(response.output.text)
            json_data={"files":[]}
    else:
        print(f"请求失败: {response.message}")
        json_data = {"files": []}
    return json_data

def analysis_results(local_path,path):
    prompt=""
    file_path=os.path.join(local_path,path)
    with open(file_path, 'r',encoding="utf8") as f:
        for line_num, line in enumerate(f, start=1):
            prompt+=f"{line_num}\t{line}"
    response = Application.call(
        api_key=ai_key,
        app_id=analysis_results_app_id,
        prompt=prompt)
    if response.status_code == HTTPStatus.OK:
        try:
            json_data = json.loads(response.output.text)
        except json.JSONDecodeError:
            print("返回内容不是有效的 JSON 格式！")
            print(response.output.text)
            json_data={"summary":None}

    else:
        print(f"请求失败: {response.message}")
        json_data = {"summary":None}
    json_data["path"]=path

    return json_data


async def get_filtered_files(folder_path):
    base_path = Path(folder_path).resolve()
    if not base_path.is_dir():
        raise ValueError("无效的目录路径")
    file_list = []
    for root, dirs, files in os.walk(base_path):
        dirs[:] = [d for d in dirs if not d.startswith('.')]
        files = [f for f in files if not f.startswith('.')]
        for file in files:
            abs_path = Path(root) / file
            rel_path = abs_path.relative_to(base_path)
            file_list.append(str(rel_path))
    return file_list
async def process_batch1(batch_files):
    """多线程处理单个文件批次的函数"""
    try:
        js = filter_code_files(str(batch_files))
        return js["files"]
    except Exception as e:
        print(f"处理批次时出错: {e}")
        return []
async def get_code_files(path):
    file_list = []
    files = await get_filtered_files(path)
    print(files)
    print(f"找到 {len(files)} 个文件")

    # 将文件列表分块（每500个一组）
    chunks = [files[i * 500: (i + 1) * 500]
              for i in range(0, len(files) // 500 + 1)]
    # 提交所有批次任务
    tasks = [process_batch1(chunk) for chunk in chunks]
    futures = await asyncio.gather(*tasks, return_exceptions=True)
    # 实时获取已完成任务的结果
    for future in futures[0]:
        if isinstance(future, Exception):
            print(f"处理出错: {future}")
        else:
            file_list.append(future)

    return file_list
async def process_batch2(local_path,path):
    """多线程处理单个文件批次的函数"""
    try:
        # print(local_path, path)
        js = analysis_results(local_path,path)
        return js
    except Exception as e:
        print(11111)
        print(f"处理批次时出错: {e}")
        return {"summary":None}


async def analysis(local_path, task_id):
    file_list = await get_code_files(local_path)
    all_extensions = [
         "adoc", "asm", "awk", "bas", "bat", "bib", "c", "cbl", "cls", "clj",
        "cljc", "cljs", "cmd", "conf", "cpp", "cr", "cs", "css", "cxx", "dart",
        "dockerfile", "edn", "el", "env", "erl", "ex", "exs", "f", "f90", "f95",
        "fs", "fsscript", "fsi", "fsx", "g4", "gd", "gql", "graphql", "groovy",
        "gsh", "gvy", "h", "hbs", "hcl", "hh", "hl", "hpp", "hrl", "hs", "htm",
        "html", "hx", "ini", "jad", "jade", "java", "jl", "js", "json", "json5",
        "jsx", "kt", "kts", "less", "lfe", "lgt", "lhs", "log", "ltx", "lua",
        "m", "mjs", "ml", "mli", "mm", "nim", "nims", "nlogo", "pas", "php",
        "pl", "plantuml", "pro", "ps1", "pug", "puml", "py", "qml", "r", "rb",
        "re", "rei", "res", "resi", "rkt", "rs", "rst", "s", "sass", "scala",
        "scm", "scss", "sed", "sh", "sol", "sql", "ss", "st", "squeak", "swift",
        "tcl", "tex", "tf", "tfvars", "toml", "ts", "tsx", "txt", "v", "vb",
        "vbs", "vh", "vhd", "vhdl", "vim", "vue", "xml", "yaml", "yang", "yml"]
    file_list = [file for file in file_list if file.split(".")[-1]  in all_extensions]
    results = []
    tasks = [process_batch2(local_path, file) for file in file_list]
    batch_results = await asyncio.gather(*tasks, return_exceptions=True)

    for result in batch_results:
        if isinstance(result, Exception):
            print(f"处理出错: {result}")
            await Scan_Tasks.filter(id=task_id).update(state=3, result={"results": results},
                                                 scan_end_time=int(time.time() * 1000))
        else:
            results.append(result)
    await Scan_Tasks.filter(id=task_id).update(state=2, result={"results": results},
                                               scan_end_time=int(time.time() * 1000))
    print("扫描完成")

async def commit_task(content,task_id):
    commit_result=await asyncio.gather(commit_summary(content), return_exceptions=True)
    if isinstance(commit_result, Exception):
        print(f"提交出错: {commit_result}")
    else:
        print("提交成功")
        await Commit_Summary_Tasks.filter(id=task_id).update(result=commit_result[0],end_time=int(time.time() * 1000))

async def file_task(file_path,task_id):
    with open(file_path, 'r', encoding="utf8") as f:
        content = f.read()
    file_result = await asyncio.gather(file_summary(content), return_exceptions=True)
    if isinstance(file_result, Exception):
        print(f"提交出错: {file_result}")
    else:
        print("提交成功")
        await File_Summary_Tasks.filter(id=task_id).update(result=file_result[0], end_time=int(time.time() * 1000))




@airouter.post("/scan")
async def scan(request: RequestScan, background_tasks: BackgroundTasks):
    local_path, repo_name = generate_repo_path(request.uuid, request.repo_url)
    print(f"开始扫描仓库: {repo_name}")
    await Scan_Tasks.filter(id=request.task_id).update(state=1, scan_start_time=int(time.time() * 1000))
    background_tasks.add_task(analysis, local_path, request.task_id)
    return {"code": 200, "msg": "添加扫描任务成功"}

@airouter.post("/summaryCommit")
async def summaryCommit(request: RequestCommit, background_tasks: BackgroundTasks):
    local_path, repo_name = generate_repo_path(request.uuid, request.repo_url)
    repo_commit=await Commit_Summary_Tasks.get(id=request.task_id)
    repo_commit_hash=repo_commit.repo_hash
    print(f"开始提交仓库: {repo_name}")
    await Commit_Summary_Tasks.filter(id=request.task_id).update(start_time=int(time.time() * 1000))
    commit_content = Repo(local_path).git.diff(f"{repo_commit_hash}^", repo_commit_hash)
    background_tasks.add_task(commit_task,commit_content, request.task_id)
    return {"code": 200, "msg": "添加提交任务成功"}

@airouter.post("/summaryFile")
async def summaryFile(request: RequestFile,background_tasks: BackgroundTasks):
    await File_Summary_Tasks.filter(id=request.task_id).update(start_time=int(time.time() * 1000))
    background_tasks.add_task(file_task, request.file_path, request.task_id)
    return {"code": 200, "msg": "添加提交任务成功"}

@airouter.post("/chat")
async def chat(request: RequestChat):
    client = MCPClient(request.uuid)
    await client.connect_to_server()
    response = await client.process_query(request.message)
    await client.cleanup()
    return {"code": 200, "msg": response}