109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
import json
|
||
from django.views import View
|
||
from django.http import JsonResponse
|
||
|
||
from utils.git_config import GitLabProvider, GiteaProvider
|
||
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()
|
||
# TODO: 暂时官方没适配gitea
|
||
# elif git_type == "gitea":
|
||
# return GiteaProvider()
|
||
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 GIT_TYPE == "gitlab":
|
||
if json_data.get('object_kind') == 'merge_request':
|
||
labels = json_data["labels"]
|
||
if labels and labels[0].get("title"):
|
||
if json_data.get("object_attributes").get("detailed_merge_status") == "mergeable":
|
||
if json_data.get("object_attributes").get("action") == "update":
|
||
provider.save_pr_agent_log(
|
||
request_data=json_data, project_id=project_config.get("project_id")
|
||
)
|
||
# TODO：暂时官方没适配gitea
|
||
# elif GIT_TYPE == "gitea":
|
||
# if json_data.get("action") == "opened":
|
||
# provider.save_pr_agent_log(
|
||
# request_data=json_data, project_id=project_config.get("project_id")
|
||
# )
|
||
|
||
return JsonResponse(status=200, data={"status": "ignored"})
|