Jira to Github Project

README.md

Jira -> GitHub Project Migration

Simple script to migrate Jira issues (by JQL) into a GitHub repo and add them to a GitHub Project (Projects V2).

What it does

• Pulls Jira issues matching the JQL in migrate_jira_to_github.py
• Creates GitHub issues (by Jira key in title)
• Adds each issue to Project set in GITHUB_PROJECT_NUMBER
• Sets GitHub parent issue when the Jira Epic is mapped via PARENT_MAP

Requirements

• Python 3
• requests installed

Install dependency:

python3 -m pip install requests

Configuration

Create a .env in this repo (already supported by the script):

# Flags
DRY_RUN=1
LIST_JIRA_FIELDS=0
LIST_PROJECT_FIELDS=0
MAX_ISSUES=0

# Jira
JIRA_PROJECT_KEY=KAFF
JIRA_SPRINT_ID=your_sprint_id
JIRA_BASE_URL=https://your-domain.atlassian.net
JIRA_EMAIL=you@company.com
JIRA_STORY_POINTS_FIELD=customfield_XXXXX
JIRA_ACCEPTANCE_CRITERIA_FIELD=customfield_XXXXX
JIRA_API_TOKEN=your_jira_api_token

# GitHub
GITHUB_TOKEN=your_github_pat
GITHUB_OWNER=your-org-or-user
GITHUB_REPO=your-repo
GITHUB_PROJECT_NUMBER=your-project-number
GH_STORY_POINTS_FIELD_ID=PVTSSF_...
GH_SPRINT_FIELD_ID=PVTSSF_...
GH_SPRINT_VALUE=Iteration N
GH_DEFAULT_LABELS=from JIRA

# Comma-separated mappings (whitespace ignored)
# Jira assignee displayName -> GitHub username
# e.g. John Doe -> johndoe
ASSIGNEE_MAP="John Doe=johndoe, ..."

# Jira epic key -> GitHub parent issue number
# e.g. FCUK-2526 -> #1627
PARENT_MAP="FCUK-2526=1627, ..."

Helper modes

• LIST_JIRA_FIELDS=1 prints Jira fields matching Story point estimate.
• LIST_JIRA_FIELDS=all prints all Jira fields.
• LIST_PROJECT_FIELDS=1 prints GitHub Project field ids and iteration titles.
• DRY_RUN=1 prints actions without writing to GitHub.
• MAX_ISSUES=0 to fetch all jira issues.
• If you want to migrate in the JIRA backlog, set JIRA_SPRINT_ID = empty and remove value from GH_SPRINT_VALUE.

Run

python3 /Users/ardit/works/personal/jira-github/migrate_jira_to_github.py

Notes

• JQL is built from env: project = $JIRA_PROJECT_KEY AND sprint = $JIRA_SPRINT_ID ORDER BY created DESC

migrate_jira_to_github.py

#!/usr/bin/env python3
"""
Simple Jira -> GitHub Projects V2 migration helper.

- Pulls Jira issues from a JQL filter
- Creates (or reuses) GitHub issues in a repo
- Adds them to a GitHub Project (Projects V2)
- Optionally sets a project text field to a mapped parent issue

Requires: python3, requests
"""

from __future__ import annotations

import os
import sys
import textwrap
from typing import Dict, Iterable, List, Optional, Tuple

import requests

# Mappings are loaded from env (see README for format).
PARENT_MAP: Dict[str, int] = {}
ASSIGNEE_MAP: Dict[str, str] = {}

# ---- Config (.env + env) ----
def load_env_file(path: str) -> None:
    if not os.path.exists(path):
        return
    with open(path, "r", encoding="utf-8") as f:
        for raw in f:
            line = raw.strip()
            if not line or line.startswith("#"):
                continue
            if "=" not in line:
                continue
            key, val = line.split("=", 1)
            key = key.strip()
            val = val.strip().strip("'").strip('"')
            os.environ.setdefault(key, val)


load_env_file(os.path.join(os.path.dirname(__file__), ".env"))

JQL = ""
JIRA_PROJECT_KEY = os.getenv("JIRA_PROJECT_KEY", "")
JIRA_SPRINT_ID = os.getenv("JIRA_SPRINT_ID", "")
JIRA_BASE_URL = os.getenv("JIRA_BASE_URL", "").rstrip("/")
JIRA_EMAIL = os.getenv("JIRA_EMAIL", "")
JIRA_API_TOKEN = os.getenv("JIRA_API_TOKEN", "")

PROJECT_NUMBER = int(os.getenv("GITHUB_PROJECT_NUMBER", "0") or "0")
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN", "")
GITHUB_OWNER = os.getenv("GITHUB_OWNER", "")  # org or user
GITHUB_REPO = os.getenv("GITHUB_REPO", "")

# Optional: Jira story points field id (customfield_XXXX)
JIRA_STORY_POINTS_FIELD = os.getenv("JIRA_STORY_POINTS_FIELD", "")
# Optional: Jira acceptance criteria field id (customfield_XXXX)
JIRA_ACCEPTANCE_CRITERIA_FIELD = os.getenv("JIRA_ACCEPTANCE_CRITERIA_FIELD", "")

# Optional: Project field IDs for story points and sprint/iteration
GH_STORY_POINTS_FIELD_ID = os.getenv("GH_STORY_POINTS_FIELD_ID", "")
GH_SPRINT_FIELD_ID = os.getenv("GH_SPRINT_FIELD_ID", "")
GH_SPRINT_VALUE = os.getenv("GH_SPRINT_VALUE", "")
GH_DEFAULT_LABELS = os.getenv("GH_DEFAULT_LABELS", "from JIRA")
MAX_ISSUES = int(os.getenv("MAX_ISSUES", "0") or "0")

DRY_RUN = os.getenv("DRY_RUN", "0") == "1"
LIST_JIRA_FIELDS = os.getenv("LIST_JIRA_FIELDS", "0")
LIST_JIRA_FIELDS_ALL = LIST_JIRA_FIELDS.lower() in {"1", "true", "all"}
LIST_PROJECT_FIELDS = os.getenv("LIST_PROJECT_FIELDS", "0") == "1"


def parse_map(value: str) -> Dict[str, str]:
    result: Dict[str, str] = {}
    if not value:
        return result
    # Format: key=value, key2=value2
    parts = [p.strip() for p in value.split(",") if p.strip()]
    for part in parts:
        if "=" not in part:
            continue
        k, v = part.split("=", 1)
        k = k.strip()
        v = v.strip()
        if k:
            result[k] = v
    return result


def load_mappings() -> None:
    global PARENT_MAP, ASSIGNEE_MAP
    parent_raw = os.getenv("PARENT_MAP", "")
    assignee_raw = os.getenv("ASSIGNEE_MAP", "")
    parent_str = parse_map(parent_raw)
    PARENT_MAP = {}
    for k, v in parent_str.items():
        try:
            PARENT_MAP[k] = int(v)
        except ValueError:
            continue
    ASSIGNEE_MAP = parse_map(assignee_raw)

# ---- Helpers ----

def die(msg: str) -> None:
    print(msg, file=sys.stderr)
    sys.exit(1)


def require_env() -> None:
    missing = []
    if not JIRA_PROJECT_KEY:
        missing.append("JIRA_PROJECT_KEY")
    if not JIRA_SPRINT_ID:
        missing.append("JIRA_SPRINT_ID")
    if not PROJECT_NUMBER:
        missing.append("GITHUB_PROJECT_NUMBER")
    if not JIRA_BASE_URL:
        missing.append("JIRA_BASE_URL")
    if not JIRA_EMAIL:
        missing.append("JIRA_EMAIL")
    if not JIRA_API_TOKEN:
        missing.append("JIRA_API_TOKEN")
    if not GITHUB_TOKEN:
        missing.append("GITHUB_TOKEN")
    if not GITHUB_OWNER:
        missing.append("GITHUB_OWNER")
    if not GITHUB_REPO:
        missing.append("GITHUB_REPO")
    if missing:
        die("Missing required env vars: " + ", ".join(missing))


def jira_request(method: str, path: str, **kwargs) -> requests.Response:
    url = f"{JIRA_BASE_URL}{path}"
    auth = (JIRA_EMAIL, JIRA_API_TOKEN)
    headers = {"Accept": "application/json"}
    return requests.request(method, url, auth=auth, headers=headers, **kwargs)


def github_request(method: str, path: str, **kwargs) -> requests.Response:
    url = f"https://api.github.com{path}"
    headers = {
        "Accept": "application/vnd.github+json",
        "Authorization": f"Bearer {GITHUB_TOKEN}",
    }
    return requests.request(method, url, headers=headers, **kwargs)


def github_graphql(query: str, variables: Dict, allow_errors: bool = False) -> Dict:
    url = "https://api.github.com/graphql"
    headers = {
        "Authorization": f"Bearer {GITHUB_TOKEN}",
        "Content-Type": "application/json",
    }
    resp = requests.post(url, json={"query": query, "variables": variables}, headers=headers)
    if resp.status_code >= 400:
        die(f"GitHub GraphQL error {resp.status_code}: {resp.text}")
    data = resp.json()
    if "errors" in data and not allow_errors:
        die(f"GitHub GraphQL errors: {data['errors']}")
    return data


def fetch_jira_issues(jql: str) -> List[Dict]:
    issues: List[Dict] = []
    start_at = 0
    max_results = 50
    while True:
        fields = ["summary", "description", "issuetype", "parent", "assignee"]
        if JIRA_STORY_POINTS_FIELD:
            fields.append(JIRA_STORY_POINTS_FIELD)
        if JIRA_ACCEPTANCE_CRITERIA_FIELD:
            fields.append(JIRA_ACCEPTANCE_CRITERIA_FIELD)
        params = {
            "jql": jql,
            "startAt": start_at,
            "maxResults": max_results,
            "fields": fields,
        }
        resp = jira_request("GET", "/rest/api/3/search/jql", params=params)
        if resp.status_code >= 400:
            die(f"Jira error {resp.status_code}: {resp.text}")
        payload = resp.json()
        issues.extend(payload.get("issues", []))
        if MAX_ISSUES and len(issues) >= MAX_ISSUES:
            issues = issues[:MAX_ISSUES]
            break
        if start_at + max_results >= payload.get("total", 0):
            break
        start_at += max_results
    return issues


def list_jira_fields() -> None:
    resp = jira_request("GET", "/rest/api/3/field")
    if resp.status_code >= 400:
        die(f"Jira error {resp.status_code}: {resp.text}")
    fields = resp.json()
    for f in fields:
        name = f.get("name", "")
        fid = f.get("id", "")
        if LIST_JIRA_FIELDS_ALL:
            print(f"Jira field: {name} -> {fid}")
            continue
        if "Story point estimate" in name:
            print(f"Jira field match: {name} -> {fid}")


def list_project_fields(project_id: str) -> None:
    q = """
    query($projectId:ID!) {
      node(id:$projectId) {
        ... on ProjectV2 {
          fields(first:100) {
            nodes {
              ... on ProjectV2Field {
                id
                name
              }
              ... on ProjectV2IterationField {
                id
                name
                configuration {
                  iterations { id title }
                }
              }
              ... on ProjectV2SingleSelectField {
                id
                name
                options { id name }
              }
            }
          }
        }
      }
    }
    """
    data = github_graphql(q, {"projectId": project_id})
    nodes = data.get("data", {}).get("node", {}).get("fields", {}).get("nodes", []) or []
    for n in nodes:
        fid = n.get("id")
        name = n.get("name")
        if n.get("configuration"):
            print(f"Field: {name} (iteration) -> {fid}")
            for it in n.get("configuration", {}).get("iterations", []) or []:
                print(f"  Iteration: {it.get('title')} -> {it.get('id')}")
        elif n.get("options"):
            print(f"Field: {name} (single-select) -> {fid}")
            for opt in n.get("options", []) or []:
                print(f"  Option: {opt.get('name')} -> {opt.get('id')}")
        else:
            print(f"Field: {name} -> {fid}")


def find_github_issue_by_key(key: str) -> Optional[Dict]:
    # Duplicate checks are disabled to avoid GitHub search rate limits.
    return None


def create_github_issue(key: str, summary: str, body: str) -> Dict:
    labels = [l.strip() for l in (GH_DEFAULT_LABELS or "").split(",") if l.strip()]
    payload = {"title": f"[{key}] {summary}", "body": body}
    if labels:
        payload["labels"] = labels
    if DRY_RUN:
        print(f"DRY_RUN create issue: {payload['title']}")
        # Fake response minimal fields
        return {"number": -1, "node_id": ""}
    resp = github_request("POST", f"/repos/{GITHUB_OWNER}/{GITHUB_REPO}/issues", json=payload)
    if resp.status_code >= 400:
        die(f"GitHub create issue error {resp.status_code}: {resp.text}")
    return resp.json()


def get_project_id(owner: str, number: int) -> str:
    # Try organization first, then user (avoid NOT_FOUND errors for the wrong type)
    org_q = """
    query($login:String!, $number:Int!) {
      organization(login:$login) { projectV2(number:$number){ id } }
    }
    """
    org_data = github_graphql(org_q, {"login": owner, "number": number}, allow_errors=True)
    org_errors = org_data.get("errors", [])
    if org_errors:
        for err in org_errors:
            if err.get("type") != "NOT_FOUND":
                die(f"GitHub GraphQL errors: {org_errors}")
    org = org_data.get("data", {}).get("organization")
    if org and org.get("projectV2"):
        return org["projectV2"]["id"]

    user_q = """
    query($login:String!, $number:Int!) {
      user(login:$login) { projectV2(number:$number){ id } }
    }
    """
    user_data = github_graphql(user_q, {"login": owner, "number": number}, allow_errors=True)
    user_errors = user_data.get("errors", [])
    if user_errors:
        for err in user_errors:
            if err.get("type") != "NOT_FOUND":
                die(f"GitHub GraphQL errors: {user_errors}")
    user = user_data.get("data", {}).get("user")
    if user and user.get("projectV2"):
        return user["projectV2"]["id"]
    die(f"Could not find project {number} under owner {owner}")


def add_issue_to_project(project_id: str, issue_node_id: str) -> str:
    q = """
    mutation($projectId:ID!, $contentId:ID!) {
      addProjectV2ItemById(input:{projectId:$projectId, contentId:$contentId}) {
        item { id }
      }
    }
    """
    if DRY_RUN:
        print("DRY_RUN add to project")
        return ""
    data = github_graphql(q, {"projectId": project_id, "contentId": issue_node_id})
    return data["data"]["addProjectV2ItemById"]["item"]["id"]


def set_number_field(project_id: str, item_id: str, field_id: str, value: float) -> None:
    q = """
    mutation($projectId:ID!, $itemId:ID!, $fieldId:ID!, $value:Float!) {
      updateProjectV2ItemFieldValue(
        input:{projectId:$projectId, itemId:$itemId, fieldId:$fieldId, value:{number:$value}}
      ) {
        projectV2Item { id }
      }
    }
    """
    if DRY_RUN:
        print(f"DRY_RUN set number field {field_id} to {value}")
        return
    github_graphql(q, {"projectId": project_id, "itemId": item_id, "fieldId": field_id, "value": value})


def get_iteration_id(project_id: str, field_id: str, title: str) -> str:
    if not title:
        return ""
    q = """
    query($projectId:ID!) {
      node(id:$projectId) {
        ... on ProjectV2 {
          fields(first:100) {
            nodes {
              ... on ProjectV2Field {
                id
                name
              }
              ... on ProjectV2IterationField {
                id
                name
                configuration {
                  iterations { id title }
                }
              }
              ... on ProjectV2SingleSelectField {
                id
                name
                options { id name }
              }
            }
          }
        }
      }
    }
    """
    data = github_graphql(q, {"projectId": project_id})
    nodes = data.get("data", {}).get("node", {}).get("fields", {}).get("nodes", []) or []
    field = None
    for n in nodes:
        if n.get("id") == field_id:
            field = n
            break
    if not field:
        die(f"Iteration field id not found in project: {field_id}")
    config = field.get("configuration", {})
    for it in config.get("iterations", []) or []:
        if it.get("title") == title:
            return it.get("id", "")
    # If title is provided but not found, keep empty to skip setting
    return ""


def set_iteration_field(project_id: str, item_id: str, field_id: str, iteration_id: str) -> None:
    q = """
    mutation($projectId:ID!, $itemId:ID!, $fieldId:ID!, $iterationId:String!) {
      updateProjectV2ItemFieldValue(
        input:{projectId:$projectId, itemId:$itemId, fieldId:$fieldId, value:{iterationId:$iterationId}}
      ) {
        projectV2Item { id }
      }
    }
    """
    if DRY_RUN:
        print(f"DRY_RUN set iteration field {field_id} to {iteration_id}")
        return
    github_graphql(
        q, {"projectId": project_id, "itemId": item_id, "fieldId": field_id, "iterationId": iteration_id}
    )


def get_single_select_option_id(project_id: str, field_id: str, name: str) -> str:
    q = """
    query($projectId:ID!) {
      node(id:$projectId) {
        ... on ProjectV2 {
          fields(first:100) {
            nodes {
              ... on ProjectV2SingleSelectField {
                id
                name
                options { id name }
              }
            }
          }
        }
      }
    }
    """
    data = github_graphql(q, {"projectId": project_id})
    nodes = data.get("data", {}).get("node", {}).get("fields", {}).get("nodes", []) or []
    for n in nodes:
        if n.get("id") == field_id:
            for opt in n.get("options", []) or []:
                if opt.get("name") == name:
                    return opt.get("id", "")
            die(f"Option '{name}' not found for field {field_id}")
    die(f"Single-select field id not found in project: {field_id}")


def set_single_select_field(project_id: str, item_id: str, field_id: str, option_id: str) -> None:
    q = """
    mutation($projectId:ID!, $itemId:ID!, $fieldId:ID!, $optionId:String!) {
      updateProjectV2ItemFieldValue(
        input:{projectId:$projectId, itemId:$itemId, fieldId:$fieldId, value:{singleSelectOptionId:$optionId}}
      ) {
        projectV2Item { id }
      }
    }
    """
    if DRY_RUN:
        print(f"DRY_RUN set single-select field {field_id} to {option_id}")
        return
    github_graphql(q, {"projectId": project_id, "itemId": item_id, "fieldId": field_id, "optionId": option_id})


def set_parent_issue(parent_issue_node_id: str, sub_issue_node_id: str) -> None:
    q = """
    mutation($parentId:ID!, $subId:ID!) {
      addSubIssue(input:{issueId:$parentId, subIssueId:$subId, replaceParent:true}) {
        issue { id }
        subIssue { id }
      }
    }
    """
    if DRY_RUN:
        print("DRY_RUN set parent issue")
        return
    github_graphql(q, {"parentId": parent_issue_node_id, "subId": sub_issue_node_id})


def jira_issue_body(issue: Dict) -> str:
    key = issue.get("key")
    url = f"{JIRA_BASE_URL}/browse/{key}"
    fields = issue.get("fields", {})
    description = fields.get("description")
    acceptance = fields.get(JIRA_ACCEPTANCE_CRITERIA_FIELD) if JIRA_ACCEPTANCE_CRITERIA_FIELD else None
    desc_text = jira_doc_to_text(description)
    acc_text = jira_doc_to_text(acceptance)
    return textwrap.dedent(
        f"""
        Jira: {url}

        {desc_text}

        Acceptance Criteria:
        {acc_text}
        """
    ).strip()


def jira_doc_to_text(value: object) -> str:
    if isinstance(value, str):
        return value
    if not isinstance(value, dict):
        return ""
    # Minimal Atlassian Document Format (ADF) to plain text
    parts: List[str] = []

    def walk(node: Dict) -> None:
        ntype = node.get("type")
        if ntype == "inlineCard":
            url = (node.get("attrs") or {}).get("url")
            if url:
                parts.append(url)
            return
        if ntype == "text":
            text = node.get("text", "")
            marks = node.get("marks", []) or []
            link = None
            for m in marks:
                if m.get("type") == "link":
                    link = (m.get("attrs") or {}).get("href")
                    break
            if link:
                if text.strip() == link:
                    parts.append(link)
                else:
                    parts.append(f"{text} ({link})")
            else:
                parts.append(text)
            return
        if ntype == "hardBreak":
            parts.append("\n")
            return
        for child in node.get("content", []) or []:
            walk(child)
        if ntype in {"paragraph", "heading"}:
            parts.append("\n")

    walk(value)
    text = "".join(parts)
    # Normalize excessive blank lines
    lines = [ln.rstrip() for ln in text.splitlines()]
    return "\n".join([ln for ln in lines if ln.strip() or ln == ""])


def migrate() -> None:
    require_env()
    global JQL
    JQL = f"project = {JIRA_PROJECT_KEY} AND sprint = {JIRA_SPRINT_ID} ORDER BY created DESC"
    load_mappings()
    # Sanity check GitHub repo access (even in DRY_RUN)
    resp = github_request("GET", f"/repos/{GITHUB_OWNER}/{GITHUB_REPO}")
    if resp.status_code >= 400:
        die(f"GitHub repo access error {resp.status_code}: {resp.text}")
    if LIST_JIRA_FIELDS_ALL or LIST_JIRA_FIELDS == "1":
        list_jira_fields()
        return
    if LIST_PROJECT_FIELDS:
        project_id = get_project_id(GITHUB_OWNER, PROJECT_NUMBER)
        list_project_fields(project_id)
        return
    print(f"Fetching Jira issues for JQL: {JQL}")
    issues = fetch_jira_issues(JQL)
    print(f"Found {len(issues)} Jira issues")

    project_id = get_project_id(GITHUB_OWNER, PROJECT_NUMBER)
    sprint_iteration_id = ""
    if GH_SPRINT_FIELD_ID:
        sprint_iteration_id = get_iteration_id(project_id, GH_SPRINT_FIELD_ID, GH_SPRINT_VALUE)

    for issue in issues:
        key = issue.get("key")
        fields = issue.get("fields", {})
        summary = fields.get("summary", "")
        parent_key = None
        if fields.get("parent"):
            parent_key = fields["parent"].get("key")
        assignee = fields.get("assignee") or {}
        assignee_name = assignee.get("displayName")
        assignee_id = assignee.get("accountId")
        mapped_assignee = ASSIGNEE_MAP.get(assignee_name) if assignee_name else ""
        story_points = None
        if JIRA_STORY_POINTS_FIELD:
            story_points = fields.get(JIRA_STORY_POINTS_FIELD)

        body = jira_issue_body(issue)
        created = create_github_issue(key, summary, body)
        gh_number = created.get("number")
        gh_node_id = created.get("node_id", "")
        print(f"Created GitHub issue #{gh_number} for {key}")

        if DRY_RUN:
            # Avoid API calls when dry-running
            if parent_key:
                mapped_parent = PARENT_MAP.get(parent_key)
                if mapped_parent:
                    print(f"Parent mapping: {parent_key} -> #{mapped_parent}")
                else:
                    print(f"Parent mapping: {parent_key} -> (unmapped)")
            if assignee_name or assignee_id:
                if mapped_assignee:
                    print(f"Assignee mapping: {assignee_name} -> {mapped_assignee}")
                else:
                    print(f"Assignee mapping: {assignee_name} -> (unmapped)")
            if story_points is not None:
                print(f"Story points: {story_points}")
            print("Description:")
            print(body)
            continue

        if not gh_node_id:
            # fetch node_id for existing issue
            resp = github_request("GET", f"/repos/{GITHUB_OWNER}/{GITHUB_REPO}/issues/{gh_number}")
            if resp.status_code >= 400:
                die(f"GitHub issue fetch error {resp.status_code}: {resp.text}")
            gh_node_id = resp.json().get("node_id", "")

        item_id = add_issue_to_project(project_id, gh_node_id)

        if parent_key and parent_key in PARENT_MAP:
            parent_issue_number = PARENT_MAP[parent_key]
            resp = github_request(
                "GET", f"/repos/{GITHUB_OWNER}/{GITHUB_REPO}/issues/{parent_issue_number}"
            )
            if resp.status_code >= 400:
                die(f"GitHub parent issue fetch error {resp.status_code}: {resp.text}")
            parent_node_id = resp.json().get("node_id", "")
            if parent_node_id:
                set_parent_issue(parent_node_id, gh_node_id)

        if GH_STORY_POINTS_FIELD_ID and story_points is not None:
            set_number_field(project_id, item_id, GH_STORY_POINTS_FIELD_ID, float(story_points))

        if GH_SPRINT_FIELD_ID and sprint_iteration_id:
            set_iteration_field(project_id, item_id, GH_SPRINT_FIELD_ID, sprint_iteration_id)

    print("Done")


if __name__ == "__main__":
    migrate()