Migration - Redmine 2 Jira

Test.py

import requests
import json
from requests.auth import HTTPBasicAuth

# JIRA API 정보
JIRA_BASE_URL = "https://companyName.atlassian.net"
JIRA_API_URL = f"{JIRA_BASE_URL}/rest/api/2/issue"
EMAIL = "system@yourcompany.com"  # JIRA 계정 이메일
API_TOKEN = "ATATT3xFfGF04icyV0-GbWZqE0_UnEzlGx6S1yVCPy7grfDGl1_XIQPMTAcIWMkF9g89uWpPc0TFWfzbMY-GdB7geLWFnGj2HtkDJIXPd-1UEMPoXJAOskfG9FKWIorRmS00000000000000000000000000000000000000000000000000"

HEADERS = {
    "Content-Type": "application/json"
}

# JIRA 프로젝트 키
JIRA_PROJECT_KEY = "V1"

def create_jira_issue(issue_data):
    """JIRA API를 통해 단일 이슈를 생성합니다."""
    response = requests.post(
        JIRA_API_URL,
        headers=HEADERS,
        auth=HTTPBasicAuth(EMAIL, API_TOKEN),
        data=json.dumps(issue_data)
    )
    if response.status_code == 201:
        print("이슈가 성공적으로 생성되었습니다.")
    else:
        print(f"이슈 생성 실패: {response.status_code} - {response.text}")

def main():
    """JIRA에 이슈를 생성합니다."""
    issue_data = {
        "fields": {
            "project": {"key": JIRA_PROJECT_KEY},
            "summary": "예시 이슈 제목",
            "description": "이슈 설명을 입력하세요.",
            "issuetype": {"name": "Task"},  # 이슈 타입을 정확히 지정
        }
    }

    print("JIRA에 이슈 생성 중...")
    create_jira_issue(issue_data)

if __name__ == "__main__":
    main()

redmine.py

import requests
import json
from requests.auth import HTTPBasicAuth

# JIRA API 정보
JIRA_BASE_URL = "https://companyName.atlassian.net"
JIRA_API_URL = f"{JIRA_BASE_URL}/rest/api/2/issue/bulk"
JIRA_PRIORITY_URL = f"{JIRA_BASE_URL}/rest/api/2/priority"
EMAIL = "system@yourcompany.com"  # JIRA 계정 이메일
API_TOKEN = "ATATT3xFfGF04icyV0-GbWZqE0_UnEzlGx6S1yVCPy7grfDGl1_XIQPMTAcIWMkF9g89uWpPc0TFWfzbMY-GdB7geLWFnGj2HtkDJIXPd-1UEMPoXJAOskfG9FKWIorRmSDjIIFAccn42AxhVvOy000000000000000000000000000000000000"

HEADERS = {
    "Content-Type": "application/json"
}

# 기본 설정
#REDMINE_JSON_URL = "https://r.sharevaluecorp.com/projects/sharevalue_v1/issues.json"
REDMINE_JSON_URL = "https://r.sharevaluecorp.com/projects/sales/issues.json"


REDMINE_API_KEY = "2f6ca25af0fd2f8812aaf7cakeykeykeykeykeykeykeykeykeykey"
JIRA_PROJECT_KEY = "UKYT"
DEFAULT_PRIORITY = "Medium"  # 기본 우선순위
DEFAULT_DUE_DATE = "2024-01-01"
DEFAULT_ISSUE_TYPE = "Task"  # 기본 Issue Type
DEFAULT_ASSIGNEE = "jhryu@companyname.com"  # 담당자 기본값
ISSUE_TYPE_MAPPING = {
    "기능": "Task",
    "버그": "Bug",
    "지원": "Story",
    "기획": "Story"
}

def fetch_paginated_data(base_url, api_key):
    """Redmine API를 통해 페이징 데이터를 가져옵니다."""
    all_issues = []
    offset = 0
    limit = 100
    headers = {
        "Content-Type": "application/json",
        "X-Redmine-API-Key": api_key
    }

    while True:
        url = f"{base_url}?offset={offset}&limit={limit}"
        print(f"Fetching: {url}")
        response = requests.get(url, headers=headers)
        if response.status_code != 200:
            print(f"Error fetching data: {response.status_code} - {response.text}")
            break
        try:
            data = response.json()
            all_issues.extend(data["issues"])
        except json.JSONDecodeError as e:
            print(f"Error parsing JSON response: {e}")
            break

        if len(data["issues"]) < limit:
            break
        offset += limit

    return all_issues

def fetch_jira_priorities():
    """JIRA에서 허용되는 Priority 목록을 가져옵니다."""
    response = requests.get(JIRA_PRIORITY_URL, headers=HEADERS, auth=HTTPBasicAuth(EMAIL, API_TOKEN))
    if response.status_code == 200:
        priorities = {item['name']: item['id'] for item in response.json()}
        print(f"Available priorities in JIRA: {priorities}")
        return priorities
    else:
        print(f"Failed to fetch priorities from JIRA: {response.status_code} - {response.text}")
        return {}

def transform_to_jira_issues(issues, jira_priorities):
    """Redmine 이슈 데이터를 JIRA 이슈 포맷으로 변환합니다."""
    jira_issues = []
    unsupported_trackers = set()

    for issue in issues:
        issue_type = ISSUE_TYPE_MAPPING.get(issue["tracker"]["name"], DEFAULT_ISSUE_TYPE)
        priority = issue.get("priority", {}).get("name", DEFAULT_PRIORITY)
        priority = jira_priorities.get(priority, jira_priorities.get(DEFAULT_PRIORITY))  # 유효한 Priority로 변환
        duedate = issue.get("due_date", DEFAULT_DUE_DATE)

        if not issue.get("due_date"):
            print(f"Due Date is missing for issue: {issue['subject']}. Using default due date: {DEFAULT_DUE_DATE}")

        if not priority:
            print(f"Priority '{priority}' is not valid for JIRA. Using default priority '{DEFAULT_PRIORITY}'.")
            priority = jira_priorities.get(DEFAULT_PRIORITY)

        fields = {
            "project": {"key": JIRA_PROJECT_KEY},
            "summary": issue["subject"],
            "description": issue.get("description", "No description provided."),
            "issuetype": {"name": issue_type},  # 기본 이슈 유형 설정
            "priority": {"id": priority},  # Priority ID 사용
            "assignee": {"name": DEFAULT_ASSIGNEE},  # 담당자 고정
            "duedate": duedate
        }

        jira_issue = {"fields": fields}
        jira_issues.append(jira_issue)

    if unsupported_trackers:
        print(f"Unsupported tracker types: {', '.join(unsupported_trackers)}")

    return jira_issues

def upload_to_jira(payload):
    """JIRA API를 통해 이슈를 업로드합니다."""
    response = requests.post(JIRA_API_URL, headers=HEADERS, auth=HTTPBasicAuth(EMAIL, API_TOKEN), data=json.dumps(payload))
    if response.status_code == 201:
        print("이슈가 성공적으로 업로드되었습니다.")
    else:
        print(f"이슈 업로드 실패: {response.status_code} - {response.text}")

def chunk_list(data, chunk_size):
    """데이터를 특정 크기로 나눕니다."""
    for i in range(0, len(data), chunk_size):
        yield data[i:i + chunk_size]

def main():
    """Redmine 데이터를 가져와 JIRA에 업로드합니다."""
    print("Fetching available priorities from JIRA...")
    jira_priorities = fetch_jira_priorities()
    if not jira_priorities:
        print("Failed to fetch priorities. Terminating.")
        return

    print("Fetching issues from Redmine...")
    redmine_issues = fetch_paginated_data(REDMINE_JSON_URL, REDMINE_API_KEY)

    if not redmine_issues:
        print("No issues fetched from Redmine.")
        return

    print(f"Fetched {len(redmine_issues)} issues from Redmine.")

    print("Transforming issues to JIRA format...")
    jira_issues = transform_to_jira_issues(redmine_issues, jira_priorities)

    if not jira_issues:
        print("No valid JIRA issues to upload. Terminating.")
        return

    print("Uploading issues to JIRA...")
    for chunk in chunk_list(jira_issues, 50):
        jira_payload = {"issueUpdates": chunk}
        upload_to_jira(jira_payload)

if __name__ == "__main__":
    main()