인시던트와 JIRA 양방향 연동 모범 사례¶

import requests
import json
import time
from datetime import datetime, timedelta
from jira import JIRA

# Guance 설정, df_api_key 수정 필요
base_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'

#JIRA 연결
def 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"Guance의 UUID: {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"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"issueUUID에 대한 댓글 생성 성공: {issue_uuid}")
    else:
        print(f"issueUUID에 대한 댓글 생성 실패: {issue_uuid}. 상태 코드: {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("지난 1분 동안 업데이트된 이슈나 새로운 댓글이 없습니다.")

@DFF.API('Create_JIRA_Issue_Reply2')
def guance():
    print("시작")
    sync_issues_from_guance_to_jira()
    sync_comments_from_jira_to_guance()