import requestsimport jsonimport timefrom datetime import datetime, timedeltafrom jira import JIRA
# 观测云配置，注意修改df_api_keybase_url = 'https://openapi.guance.com'channel_list_url = base_url + '/api/v1/channel/quick_list'issue_list_url = base_url + '/api/v1/issue/list'create_issue_reply_url = base_url + '/api/v1/issue/reply/create'df_api_key = 'vy2EV......fuTtn'
# JIRA配置，以下配置均为必填项,修改为自己的环境username = 'sutt'api_token = 'ATATT3xFfGF0eVvhZUkO0tTas8JnNYEsxGIJqWGinVyQL0ME......B6E'jira_server_url = 'https://***.net/'project_key = 'projectName'
#连接JIRAdef connect_to_jira(username, api_token, jira_server_url): try: jira_connection = JIRA(basic_auth=(username, api_token), server=jira_server_url) print("成功连接到JIRA!") return jira_connection except Exception as e: print(f"连接JIRA出错: {e}") return None
jira_instance = connect_to_jira(username, api_token, jira_server_url)
def sync_issues_from_guance_to_jira(): headers = { 'DF-API-KEY': df_api_key, 'Content-Type': 'application/json;charset=UTF-8' }
one_minute_ago = datetime.now() - timedelta(minutes=1) one_minute_ago_time = int(one_minute_ago.timestamp()) current_time = int(time.time())
response = requests.get(channel_list_url, headers=headers) if response.status_code == 200: channel_list = response.json()["content"] for channel in channel_list: if channel["name"] == "default": body = { 'channelUUID': channel["uuid"], 'startTime': one_minute_ago_time, 'endTime': current_time } issue_response = requests.post(issue_list_url, headers=headers, data=json.dumps(body)) print(issue_response.text) # 打印响应内容，帮助调试
if issue_response.status_code == 200: issue_lists = issue_response.json()['content'] for issue in issue_lists: issue_uuid = issue["uuid"] print(f"UUID from Guance: {issue_uuid}") # 打印 UUID 调试
issue_data = { 'project': {'key': project_key}, 'summary': issue["name"], 'description': issue["description"], 'issuetype': {'name': '缺陷'}, 'priority': {'name': 'Medium'}, 'labels': [issue_uuid] # 使用label来存储issue_id } created_issue = jira_instance.create_issue(fields=issue_data) print(f"Created JIRA issue: {created_issue.key}")
def create_issue_reply(issue_uuid, content): headers = { 'DF-API-KEY': df_api_key, 'Content-Type': 'application/json;charset=UTF-8' } body = { 'issueUUID': issue_uuid, 'content': content, 'extend': {} } response = requests.post(create_issue_reply_url, headers=headers, data=json.dumps(body)) if response.status_code == 200: print(f"Successfully created a reply for issueUUID: {issue_uuid}") else: print(f"Failed to create a reply for issueUUID: {issue_uuid}. Status code: {response.status_code}")
def sync_comments_from_jira_to_guance(): end_time = datetime.now() start_time = end_time - timedelta(minutes=1) start_time_str = start_time.strftime('%Y-%m-%d %H:%M') end_time_str = end_time.strftime('%Y-%m-%d %H:%M')
jql_str = f'project = {project_key} AND updated >= "{start_time_str}" AND updated <= "{end_time_str}"' recently_updated_issues = jira_instance.search_issues(jql_str)
has_updates = False for issue in recently_updated_issues: comments = jira_instance.comments(issue) new_comments = [comment for comment in comments if start_time_str <= comment.created.split('.')[0].replace("T", " ") <= end_time_str]
issue_labels = issue.fields.labels guance_issue_id = None for label in issue_labels: if label.startswith("issue"): guance_issue_id = label break
if guance_issue_id and new_comments: has_updates = True for comment in new_comments: create_issue_reply(guance_issue_id, comment.body)
if not has_updates: print("在过去的一分钟内没有更新的事务或者新的评论。")
@DFF.API('Create_JIRA_Issue_Reply2')def guance(): print("do start") sync_issues_from_guance_to_jira() sync_comments_from_jira_to_guance()
评论