95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
import json
|
|
from django.views import View
|
|
from django.http import JsonResponse
|
|
|
|
from utils.git_config import GitLabProvider
|
|
from utils import constant
|
|
|
|
|
|
def load_project_config(
|
|
git_url, access_token, project_secret, openai_api_base, openai_key, llm_model
|
|
):
|
|
"""
|
|
加载项目配置
|
|
:param git_url: git服务器地址
|
|
:param access_token: 用户访问密钥
|
|
:param project_secret: 项目秘钥
|
|
:param openai_api_base: openai api base
|
|
:param openai_key: openai key
|
|
:param llm_model: llm model
|
|
:return:
|
|
"""
|
|
|
|
return {
|
|
"gitlab_url": git_url,
|
|
"access_token": access_token,
|
|
"secret": project_secret,
|
|
"openai_api_base": openai_api_base,
|
|
"openai_key": openai_key,
|
|
"llm_model": llm_model,
|
|
}
|
|
|
|
|
|
class WebHookView(View):
|
|
@staticmethod
|
|
def select_git_provider(git_type):
|
|
"""
|
|
根据git类型选择git服务提供商: 目前仅配置gitlab
|
|
:param git_type:
|
|
:return:
|
|
"""
|
|
if git_type == "gitlab":
|
|
return GitLabProvider()
|
|
else:
|
|
return GitLabProvider()
|
|
|
|
def post(self, request):
|
|
"""
|
|
处理Git服务器的Webhook事件
|
|
:param request:
|
|
:return:
|
|
"""
|
|
# 校验请求头
|
|
headers = request.headers
|
|
GIT_TYPE = constant.UA_TYPE.get(headers.get("User-Agent").split("/")[0])
|
|
|
|
if not GIT_TYPE:
|
|
return JsonResponse(status=400, data={"error": "Invalid Git Type"})
|
|
|
|
# 校验请求体
|
|
json_data = json.loads(request.body.decode('utf-8'))
|
|
if not json_data:
|
|
return JsonResponse(status=400, data={"error": "Invalid JSON"})
|
|
|
|
# 获取对应处理类
|
|
provider = self.select_git_provider(GIT_TYPE)
|
|
project_id = provider.get_project_id(request_json=json_data, git_type=GIT_TYPE)
|
|
if not project_id:
|
|
return JsonResponse(status=400, data={"error": "Missing project ID"})
|
|
|
|
# 获取项目配置
|
|
project_config = provider.get_project_config(project_id=project_id, git_type=GIT_TYPE)
|
|
|
|
# Token 校验
|
|
provider.check_secret(
|
|
request_headers=headers, project_secret=project_config.get("project_secret"), git_type=GIT_TYPE
|
|
)
|
|
|
|
provider.get_merge_request(
|
|
request_data=json_data,
|
|
git_url=project_config.get("git_url"),
|
|
access_token=project_config.get("access_token"),
|
|
api_base=project_config.get("api_base"),
|
|
api_key=project_config.get("api_key"),
|
|
llm_model=project_config.get("llm_model"),
|
|
project_commands=project_config.get("commands"),
|
|
)
|
|
|
|
# 记录请求日志: 目前仅记录合并日志
|
|
if json_data.get('object_kind') == 'merge_request':
|
|
provider.save_pr_agent_log(
|
|
request_data=json_data, project_id=project_config.get("project_id")
|
|
)
|
|
|
|
return JsonResponse(status=200, data={"status": "ignored"})
|