99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
from requests import request
|
|
|
|
|
|
class ProjectApi:
|
|
"""
|
|
GitLink平台相关接口
|
|
"""
|
|
|
|
def __init__(self, host, project, access_token=None):
|
|
"""
|
|
:param host: 平台域名
|
|
:param project: 项目拥有者/项目标识
|
|
:param access_token: 授权token
|
|
"""
|
|
self.host = host
|
|
self.project = project
|
|
self.headers = {
|
|
"Content-Type": "application/json",
|
|
}
|
|
if access_token:
|
|
self.headers["Authorization"] = f"Bearer {access_token}"
|
|
|
|
def get_commit_info(self):
|
|
"""
|
|
获取仓库的commit列表
|
|
:return:
|
|
"""
|
|
try:
|
|
url = f"{self.host}/api/{self.project}/webhooks/4110/tasks.json?page=1&limit=10"
|
|
response = request(method="GET", url=url, headers=self.headers)
|
|
if response.status_code == 200:
|
|
return response
|
|
except Exception as e:
|
|
raise e
|
|
|
|
def create_version(self, req_data):
|
|
"""
|
|
给仓库新建发行版
|
|
:param req_data:
|
|
:return:
|
|
"""
|
|
try:
|
|
url = f"{self.host}/api/{self.project}/releases.json"
|
|
payload = {
|
|
"tag_name": req_data.get('tag_name'),
|
|
"name": f"{req_data.get('name')}",
|
|
"body": f"{req_data.get('body')}",
|
|
"target_commitish": req_data.get('target_commitish')
|
|
}
|
|
response = request(method="POST", url=url, json=payload, headers=self.headers)
|
|
if response.status_code == 200 and response.json()["message"] == "发布成功":
|
|
return response.json()
|
|
else:
|
|
return response.text
|
|
except Exception as e:
|
|
raise e
|
|
|
|
def get_issue_by_title(self, keyword):
|
|
"""
|
|
根据Issue标题搜索获取issue
|
|
:return:
|
|
"""
|
|
try:
|
|
url = f"{self.host}/api/v1/{self.project}/issues?participant_category=all&category=all&limit=15&page=1&keyword={keyword}"
|
|
response = request(method="GET", url=url, headers=self.headers, timeout=10)
|
|
if response.status_code == 200 and response.json()["total_count"] >= 0:
|
|
return response.json()
|
|
else:
|
|
return response
|
|
except Exception as e:
|
|
raise e
|
|
|
|
def get_issue_detail(self, issue_id):
|
|
"""
|
|
获取易修详情页数据
|
|
"""
|
|
try:
|
|
url = self.host + f"/api/v1/{self.project}/issues/{issue_id}"
|
|
method = "GET"
|
|
response = request(url=url, method=method, headers=self.headers, timeout=10)
|
|
return response.json()
|
|
except Exception as e:
|
|
raise e
|
|
|
|
def new_issue_journals(self, issue_id, payload):
|
|
"""
|
|
新建疑修评论
|
|
"""
|
|
try:
|
|
url = self.host + f"/api/v1/{self.project}/issues/{issue_id}/journals"
|
|
method = "POST"
|
|
response = request(url=url, method=method, headers=self.headers, json=payload, timeout=5)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
return response
|
|
except Exception as e:
|
|
raise e
|