forked from Lesin/reposync
108 lines
3.8 KiB
Python
108 lines
3.8 KiB
Python
import json
|
||
import urllib
|
||
import mysql.connector
|
||
import requests
|
||
|
||
def get_issue_details(owner, repo, issue_id, access_token):
|
||
headers = {
|
||
'Authorization': f'token {access_token}',
|
||
'Content-Type': 'application/json'
|
||
}
|
||
url = f'https://gitee.com/api/v5/repos/{owner}/{repo}/issues/{issue_id}'
|
||
|
||
response = requests.get(url, headers=headers)
|
||
|
||
if response.status_code == 200:
|
||
# print(response.json())
|
||
return response.json()
|
||
else:
|
||
print(f"Failed to retrieve issue {issue_id}. Status code: {response.status_code}")
|
||
return None
|
||
|
||
|
||
def get_issue_key_info(issue_details, conn, cursor):
|
||
|
||
if issue_details:
|
||
# print(f"Issue ID: {issue_details['number']}")
|
||
issue_Id = issue_details['number']
|
||
issue_Title = issue_details['title']
|
||
issue_Body = issue_details['body']
|
||
issue_State = issue_details['state']
|
||
issue_Created_At = issue_details['created_at']
|
||
issue_Priority = issue_details["priority"]
|
||
issue_start_date = issue_details['plan_started_at']
|
||
# print(issue_start_date)
|
||
try:
|
||
issue_start_date = issue_start_date.split('T')[0]
|
||
except:
|
||
issue_start_date = issue_details['plan_started_at']
|
||
issue_deadline = issue_details['deadline']
|
||
# print(issue_deadline)
|
||
try:
|
||
issue_deadline = issue_deadline.split('T')[0]
|
||
except:
|
||
issue_deadline = issue_details['deadline']
|
||
issue_branch = issue_details['branch']
|
||
|
||
cursor.execute(
|
||
"INSERT INTO gitee_issue (issue_id, issue_title, issue_body, issue_state, issue_created_at, issue_priority, issue_start_date, issue_deadline, issue_branch) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)",
|
||
(issue_Id, issue_Title, issue_Body, issue_State, issue_Created_At, issue_Priority, issue_start_date, issue_deadline, issue_branch))
|
||
conn.commit()
|
||
result = cursor.fetchall()
|
||
|
||
|
||
def get_gitee_issues(owner, repo, access_token, conn, cursor):
|
||
headers = {
|
||
'Authorization': f'token {access_token}',
|
||
'Content-Type': 'application/json'
|
||
}
|
||
url = f'https://gitee.com/api/v5/repos/{owner}/{repo}/issues'
|
||
# 发送POST请求 获取状态为open的issue
|
||
data = {
|
||
"access_token": access_token,
|
||
"owner": owner,
|
||
"repo": repo,
|
||
"state": 'all',
|
||
"sort": 'created',
|
||
"direction": 'asc'
|
||
}
|
||
# 将字典转换为查询字符串
|
||
query_string = urllib.parse.urlencode(data)
|
||
|
||
# 完整的请求 URL，包括查询字符串
|
||
url = f"{url}?{query_string}"
|
||
|
||
response = requests.get(url, headers=headers, json=data)
|
||
# print(response.text)
|
||
cursor.execute('DROP TABLE IF EXISTS gitee_issue')
|
||
# 创建一个表
|
||
cursor.execute('''CREATE TABLE IF NOT EXISTS gitee_issue
|
||
(id INT AUTO_INCREMENT PRIMARY KEY, issue_id VARCHAR(10), issue_title VARCHAR(100), issue_body VARCHAR(1000), issue_state VARCHAR(15), issue_created_at VARCHAR(100), issue_priority INT,issue_start_date VARCHAR(100), issue_deadline VARCHAR(100), issue_branch VARCHAR(100))''')
|
||
|
||
if response.status_code == 200:
|
||
issues = response.json()
|
||
for issue in issues:
|
||
# print(issue['number'])
|
||
issue_details = get_issue_details(owner, repo, issue['number'], access_token)
|
||
|
||
if issue_details:
|
||
get_issue_key_info(issue_details, conn, cursor)
|
||
print("获取gitee-issue成功")
|
||
|
||
# # 连接到数据库
|
||
# conn = mysql.connector.connect(
|
||
# host="localhost",
|
||
# user="root",
|
||
# password="251226X",
|
||
# database="reposyncer"
|
||
# )
|
||
#
|
||
# cursor = conn.cursor()
|
||
#
|
||
# # 你的Gitee Access Token
|
||
# access_token = '62b7d0577a13b75ff066047426df30cb'
|
||
# # 仓库的拥有者用户名和仓库名
|
||
# owner = 'wyf0625'
|
||
# repo = 'ob-reposyncer-gitee'
|
||
# get_gitee_issues(owner, repo, access_token, conn, cursor)
|