Linear → Jira migration script

linear_to_jira_migrate.py

#!/usr/bin/env python3
"""
Linear → Jira migration script.

Migrates Linear issues (title, description, comments, attachments, labels, status,
priority, due date, reporter/assignee) to Jira. Uses Jira REST API v2 for description
and comments (wiki markup with !filename! for embedded images); v3 for issue create/update.

Requirements: Python 3.10+, requests.

  pip install requests
  export LINEAR_API_KEY="lin_api_..."
  export JIRA_EMAIL="you@example.com"
  export JIRA_API_TOKEN="..."
  python linear_to_jira_migrate.py

Edit the config block below for LINEAR_TEAM_KEY, JIRA_BASE_URL, JIRA_PROJECT_KEY,
and custom field IDs. Duplicate detection: if a Jira issue already has the same
Linear ID (custom field), creation is skipped.
"""

import re
import os
import requests
from datetime import datetime
from typing import Any

# --- Config: set via env or edit defaults ---
LINEAR_API_KEY = os.environ.get("LINEAR_API_KEY", "")
LINEAR_TEAM_KEY = os.environ.get("LINEAR_TEAM_KEY", "YOUR_TEAM_KEY")
LINEAR_LABELS_FILTER = []  # e.g. ["ui bug"] to only import issues with these labels
LINEAR_ISSUE_URL_TEMPLATE = os.environ.get("LINEAR_ISSUE_URL_TEMPLATE", "https://linear.app/your-org/issue/{identifier}")

JIRA_BASE_URL = os.environ.get("JIRA_BASE_URL", "https://your-site.atlassian.net")
JIRA_EMAIL = os.environ.get("JIRA_EMAIL", "")
JIRA_API_TOKEN = os.environ.get("JIRA_API_TOKEN", "")
JIRA_PROJECT_KEY = os.environ.get("JIRA_PROJECT_KEY", "YOUR_PROJECT")
JIRA_CREATED_FIELD_ID = os.environ.get("JIRA_CREATED_FIELD_ID", "")  # optional, e.g. customfield_11268
JIRA_LINEAR_ID_FIELD_ID = os.environ.get("JIRA_LINEAR_ID_FIELD_ID", "") # Required for setting the Linear Issue ID custom field e.g. customfield_11269
JIRA_LINEAR_ID_FIELD_NAME = "Linear Issue ID" # Required for duplicate check
JIRA_ISSUE_TYPE = "Task"
JIRA_EPIC_ISSUE_TYPE = "Epic"
JIRA_SUMMARY_MAX_LENGTH = 255

# --- Jira auth (v2 and v3) ---
def _jira_headers():
    return {
        "Accept": "application/json",
        "Content-Type": "application/json",
        "Authorization": "Basic " + __import__("base64").b64encode(f"{JIRA_EMAIL}:{JIRA_API_TOKEN}".encode()).decode(),
    }


def _jira_account_id_for_email(email: str) -> str | None:
    """Resolve Jira accountId from email (cached)."""
    if not email:
        return None
    cache = getattr(_jira_account_id_for_email, "_cache", None)
    if cache is None:
        _jira_account_id_for_email._cache = cache = {}
    if email in cache:
        return cache[email]
    resp = requests.get(
        f"{JIRA_BASE_URL}/rest/api/3/user/search",
        headers=_jira_headers(),
        params={"query": email},
        timeout=30,
    )
    resp.raise_for_status()
    users = resp.json()
    for u in users:
        if (u.get("emailAddress") or "").lower() == email.lower():
            cache[email] = u["accountId"]
            return u["accountId"]
    cache[email] = None
    return None


def _jira_label_name(name: str) -> str:
    """Jira labels cannot contain spaces; use underscores."""
    return (name or "").replace(" ", "_")


def _normalize_summary(title: str) -> str:
    """Normalize title for Jira summary: strip and truncate to JIRA_SUMMARY_MAX_LENGTH."""
    s = (title or "Untitled").strip()
    if len(s) > JIRA_SUMMARY_MAX_LENGTH:
        s = s[:JIRA_SUMMARY_MAX_LENGTH]
    return s


def _find_jira_issue_by_linear_id(identifier: str) -> str | None:
    """Return Jira issue key if an issue already exists with this Linear identifier in JIRA_LINEAR_ID_FIELD_ID."""
    if not identifier or not JIRA_LINEAR_ID_FIELD_ID:
        return None
    # JQL: custom field equals identifier (escape double quotes in identifier for JQL)
    safe_val = (identifier or "").replace('\\', '\\\\').replace('"', '\\"')
    jql = f'project = {JIRA_PROJECT_KEY} AND "{JIRA_LINEAR_ID_FIELD_NAME}" ~ "{safe_val}"'
    resp = requests.post(
        f"{JIRA_BASE_URL}/rest/api/3/search/jql",
        headers=_jira_headers(),
        json={"jql": jql, "maxResults": 1, "fields": ["key"]},
        timeout=30,
    )
    resp.raise_for_status()
    issues = resp.json().get("issues", [])
    return issues[0]["key"] if issues else None


def _get_or_create_epic_for_linear_project(project_name: str) -> str:
    """Return Jira Epic key for this Linear project name. Creates Epic if none exists (cached per run)."""
    name = (project_name or "No project").strip()
    cache = getattr(_get_or_create_epic_for_linear_project, "_cache", None)
    if cache is None:
        _get_or_create_epic_for_linear_project._cache = cache = {}
    if name in cache:
        return cache[name]
    # Search for existing Epic with summary = project name
    summary = name
    safe_summary = summary.replace('\\', '\\\\').replace('"', '\\"')
    jql = f'project = {JIRA_PROJECT_KEY} AND issuetype = "{JIRA_EPIC_ISSUE_TYPE}" AND summary = "{safe_summary}"'
    resp = requests.post(
        f"{JIRA_BASE_URL}/rest/api/3/search/jql",
        headers=_jira_headers(),
        json={"jql": jql, "maxResults": 1, "fields": ["key"]},
        timeout=30,
    )
    resp.raise_for_status()
    issues = resp.json().get("issues", [])
    if issues:
        epic_key = issues[0]["key"]
        cache[name] = epic_key
        print(f"Using existing Epic {epic_key}: {summary!r}", flush=True)
        return epic_key
    # Create Epic
    create_resp = requests.post(
        f"{JIRA_BASE_URL}/rest/api/3/issue",
        headers=_jira_headers(),
        json={
            "fields": {
                "project": {"key": JIRA_PROJECT_KEY},
                "issuetype": {"name": JIRA_EPIC_ISSUE_TYPE},
                "summary": summary,
            }
        },
        timeout=30,
    )
    create_resp.raise_for_status()
    epic_key = create_resp.json()["key"]
    cache[name] = epic_key
    print(f"Created Epic {epic_key}: {summary!r}", flush=True)
    return epic_key


def _set_issue_parent(issue_key: str, parent_key: str) -> None:
    """Set the parent of an issue (e.g. link to Epic)."""
    resp = requests.put(
        f"{JIRA_BASE_URL}/rest/api/3/issue/{issue_key}",
        headers=_jira_headers(),
        json={"fields": {"parent": {"key": parent_key}}},
        timeout=30,
    )
    if resp.status_code not in (200, 204):
        raise RuntimeError(f"Failed to set parent on {issue_key}: {resp.status_code} {resp.text[:300]}")
    print(f"Linked {issue_key} to Epic {parent_key}", flush=True)


def _set_issue_duedate(issue_key: str, duedate_yyyymmdd: str) -> None:
    """Set the due date on an issue (YYYY-MM-DD)."""
    resp = requests.put(
        f"{JIRA_BASE_URL}/rest/api/3/issue/{issue_key}",
        headers=_jira_headers(),
        json={"fields": {"duedate": duedate_yyyymmdd}},
        timeout=30,
    )
    if resp.status_code not in (200, 204):
        print(f"  Warning: failed to set duedate on {issue_key}: {resp.status_code} {resp.text[:200]}", flush=True)


# --- Linear ---
LINEAR_GRAPHQL_URL = "https://api.linear.app/graphql"


def get_linear_issues() -> list[dict[str, Any]]:
    """Fetch issues from Linear (filtered by team key) with pagination."""
    issues: list[dict[str, Any]] = []
    cursor = None
    auth = {"Authorization": LINEAR_API_KEY}
    query = """
      query($first: Int!, $after: String, $filter: IssueFilter, $orderBy: PaginationOrderBy) {
        issues(first: $first, after: $after, filter: $filter, orderBy: $orderBy) {
          pageInfo { hasNextPage, endCursor }
          nodes {
            id title description identifier url
            createdAt dueDate
            creator { email }
            assignee { email }
            state { name }
            priority
            project { name }
            labels { nodes { name } }
            comments { nodes { body createdAt user { email name } } }
            attachments { nodes { id url title } }
          }
        }
      }
    """
    while True:
        variables: dict[str, Any] = {
            "first": 250,
            "filter": {"team": {"key": {"eq": LINEAR_TEAM_KEY}}},# oldest first so they get lowest Jira issue numbers
        }
        if cursor:
            variables["after"] = cursor
        resp = requests.post(
            LINEAR_GRAPHQL_URL,
            headers={"Content-Type": "application/json", **auth},
            json={"query": query, "variables": variables},
            timeout=60,
        )
        if resp.status_code == 401:
            raise SystemExit(
                "Linear API returned 401 Unauthorized. Check that LINEAR_API_KEY is a valid "
                "personal API key from Linear → Settings → Security → Personal API keys (no 'Bearer ' prefix)."
            )
        resp.raise_for_status()
        data = resp.json()
        if data.get("errors"):
            raise RuntimeError(data["errors"])
        node = data["data"]["issues"]
        issues.extend(node["nodes"])
        if not node["pageInfo"]["hasNextPage"]:
            break
        cursor = node["pageInfo"]["endCursor"]
    # Linear API returns newest first; sort oldest first so they get lowest Jira issue numbers
    issues.sort(key=lambda i: i.get("createdAt") or "")
    return issues


def _is_likely_file_attachment(url: str) -> bool:
    """True if URL looks like a downloadable file (e.g. Linear upload), not an external link."""
    return "uploads.linear.app" in url or url.rstrip("/").lower().endswith((".png", ".jpg", ".jpeg", ".gif", ".webp", ".pdf"))


def _jira_status_for_linear_state(linear_state_name: str | None) -> str | None:
    """Map Linear state name to Jira status name. Override in config if your project uses different names."""
    if not linear_state_name or not linear_state_name.strip():
        return None
    name = linear_state_name.strip()
    mapping = {
        "todo": "To Do",
    }
    return mapping.get(name.lower()) or name


def _jira_priority_for_linear(linear_priority: int | str | None) -> str | None:
    """Map Linear priority to Jira priority name. Linear: 0=none, 1=urgent, 2=high, 3=medium, 4=low (or label string)."""
    if linear_priority is None:
        return None
    if isinstance(linear_priority, str):
        mapping = {"urgent": "Highest", "high": "High", "medium": "Medium", "low": "Low", "none": None}
        return mapping.get(linear_priority.lower())
    mapping = {1: "Highest", 2: "High", 3: "Medium", 4: "Low"}
    return mapping.get(int(linear_priority))


def _format_linear_datetime(iso_str: str | None) -> str:
    """Format Linear ISO datetime for display (e.g. 'Jan 15, 2024 at 2:30 PM')."""
    if not iso_str or not iso_str.strip():
        return "unknown"
    try:
        s = iso_str.strip().replace("Z", "+00:00")
        dt = datetime.fromisoformat(s)
        return dt.strftime("%b %d, %Y at %I:%M %p")  # e.g. Jan 15, 2024 at 02:30 PM
    except (ValueError, TypeError):
        return iso_str[:19].replace("T", " ") if len(iso_str) >= 19 else iso_str


def _is_pr_url(url: str) -> bool:
    """True if URL looks like a pull/merge request (GitHub PR, GitLab MR, Bitbucket PR). Excludes commits."""
    u = url.lower()
    if "/commit/" in u or "/commits/" in u:
        return False
    return "/pull/" in u or "/merge_requests" in u or "/pull-requests" in u


# --- Description: wiki markup string (for v2) with embedded images !filename! ---
def _markdown_to_wiki(text: str) -> str:
    """Rough markdown → Jira wiki markup."""
    if not text:
        return ""
    t = text
    t = re.sub(r"\*\*(.+?)\*\*", r"*\1*", t)
    t = re.sub(r"\*(.+?)\*", r"_\1_", t)
    t = re.sub(r"`([^`]+)`", r"{{{\1}}}", t)
    # [text](url) -> [text|url]; replace | in link text with fullwidth pipe so Jira's single | is text/url separator
    _PIPE_IN_LINK = "|"
    _PIPE_VISUAL = "\uFF5C"  # fullwidth vertical line, looks like | but not wiki separator
    def _wiki_link(match: re.Match[str]) -> str:
        link_text = match.group(1).replace(_PIPE_IN_LINK, _PIPE_VISUAL)
        return f"[{link_text}|{match.group(2)}]"
    t = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", _wiki_link, t)
    return t


def convert_description_to_wiki(
    raw: str,
    image_placeholders: list[tuple[str, str]],
) -> str:
    """
    Convert description to Jira wiki markup string.
    image_placeholders: list of (placeholder, ref) where ref is either the attachment content URL
    (preferred, so image loads from Jira) or filename for !ref! wiki embed.
    Images use width=800 so they don't stretch to full width but have a consistent max size.
    """
    if not raw:
        return ""
    t = _markdown_to_wiki(raw)
    for placeholder, ref in image_placeholders:
        t = t.replace(placeholder, f"!{ref}|width=800!")
    return t


def convert_comment_to_wiki(
    raw: str,
    image_placeholders: list[tuple[str, str]],
) -> str:
    """Same as description: wiki string with !filename! for images."""
    return convert_description_to_wiki(raw, image_placeholders)


# --- Jira: create issue (v3), upload attachments (v3), set description (v2 wiki) ---
def _jira_upload_headers() -> dict[str, str]:
    """Headers for attachment upload. Do NOT set Content-Type so requests sets multipart/form-data."""
    return {
        "Accept": "application/json",
        "X-Atlassian-Token": "no-check",
        "Authorization": "Basic "
        + __import__("base64").b64encode(f"{JIRA_EMAIL}:{JIRA_API_TOKEN}".encode()).decode(),
    }


def upload_to_jira(issue_key: str, file_path: str, filename: str | None = None) -> tuple[str, str]:
    """Upload file as attachment to issue. Returns (filename, content_url). Use content_url in wiki
    !url! so the image loads from Jira (avoids 'Preview unavailable' when !filename! doesn't resolve)."""
    with open(file_path, "rb") as f:
        data = f.read()
    name = filename or os.path.basename(file_path)
    resp = requests.post(
        f"{JIRA_BASE_URL}/rest/api/3/issue/{issue_key}/attachments",
        headers=_jira_upload_headers(),
        files={"file": (name, data)},
        timeout=60,
    )
    resp.raise_for_status()
    arr = resp.json()
    if not arr:
        raise RuntimeError(f"Jira attachment upload returned empty list for {name}")
    att = arr[0]
    content_url = att.get("content") or att.get("self") or ""
    return (att.get("filename", name), content_url)


def _transition_issue_to_status(issue_key: str, target_status_name: str) -> None:
    """Move issue to target status via Jira transitions API. Works when any state can transition to any other."""
    target = target_status_name.strip().lower()
    resp = requests.get(
        f"{JIRA_BASE_URL}/rest/api/3/issue/{issue_key}/transitions",
        headers=_jira_headers(),
        timeout=30,
    )
    if resp.status_code != 200:
        return
    data = resp.json()
    for t in data.get("transitions") or []:
        to_status = (t.get("to") or {}).get("name") or ""
        if to_status.lower() == target:
            post_resp = requests.post(
                f"{JIRA_BASE_URL}/rest/api/3/issue/{issue_key}/transitions",
                headers=_jira_headers(),
                json={"transition": {"id": t["id"]}},
                timeout=30,
            )
            if post_resp.status_code in (200, 204):
                return
            return


def update_jira_description_v2(issue_key: str, description_wiki: str) -> None:
    """Set issue description via REST API v2 with wiki markup. v2 accepts plain wiki (e.g. !filename! for
    embedded images); v3 would require ADF and does not render !filename! as inline images."""
    resp = requests.put(
        f"{JIRA_BASE_URL}/rest/api/2/issue/{issue_key}",
        headers=_jira_headers(),
        json={"fields": {"description": description_wiki}},
        timeout=30,
    )
    resp.raise_for_status()


def add_jira_comment_v2(issue_key: str, body_wiki: str) -> None:
    """Add comment using v2 API so wiki markup (e.g. !filename!) embeds images."""
    resp = requests.post(
        f"{JIRA_BASE_URL}/rest/api/2/issue/{issue_key}/comment",
        headers=_jira_headers(),
        json={"body": body_wiki},
        timeout=30,
    )
    resp.raise_for_status()


# --- Main migration flow ---
def process_description_and_attachments(
    description: str | None,
    attachments: list[dict],
    issue_key: str,
    linear_identifier: str,
    linear_url_template: str,
) -> str:
    """
    Download any file attachments, upload to Jira, build wiki description with !filename! refs.
    Returns the final wiki markup string for the description.
    """
    import tempfile
    image_placeholders: list[tuple[str, str]] = []
    prefix_parts = []
    if linear_identifier and linear_url_template:
        url = linear_url_template.rstrip("/").replace("{identifier}", linear_identifier)
        prefix_parts.append(f"Migrated from Linear [{linear_identifier}|{url}]")
    elif linear_identifier:
        prefix_parts.append(f"Migrated from Linear {linear_identifier}")
    text = description or ""
    placeholder_idx = [0]

    def repl(m: re.Match) -> str:
        url = m.group(1)
        if not _is_likely_file_attachment(url):
            return m.group(0)
        ph = f"<<<IMG{placeholder_idx[0]}>>>"
        placeholder_idx[0] += 1
        try:
            r = requests.get(url, timeout=30, headers={"Authorization": LINEAR_API_KEY})
            r.raise_for_status()
            ext = ".png"
            if "content-type" in r.headers:
                ct = r.headers["content-type"].lower()
                if "jpeg" in ct or "jpg" in ct:
                    ext = ".jpg"
                elif "gif" in ct:
                    ext = ".gif"
                elif "webp" in ct:
                    ext = ".webp"
                elif "pdf" in ct:
                    ext = ".pdf"
            with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp:
                tmp.write(r.content)
                tmp_path = tmp.name
            try:
                upload_name = f"image_{placeholder_idx[0] - 1}{ext}"
                stored_name, content_url = upload_to_jira(issue_key, tmp_path, upload_name)
                # Use content URL for embed so image loads from Jira (avoids "Preview unavailable")
                ref = content_url if content_url else stored_name
                image_placeholders.append((ph, ref))
            finally:
                try:
                    os.unlink(tmp_path)
                except Exception:
                    pass
        except Exception as e:
            raise RuntimeError(f"Failed to download/upload image from {url}: {e}") from e
        return ph

    text = re.sub(r"!\[[^\]]*\]\(([^)]+)\)", repl, text)
    text = re.sub(r"<img[^>]+src=[\"']([^\"']+)[\"'][^>]*>", repl, text)
    if prefix_parts:
        text = "\n\n".join(["\n".join(prefix_parts), text]) if text else "\n".join(prefix_parts)

    # Append Linear link attachments: only PRs (no commits or other links)
    link_lines = []
    for att in attachments or []:
        url = (att.get("url") or "").strip()
        title = (att.get("title") or url or "Link").strip()
        if not url:
            continue
        if _is_likely_file_attachment(url):
            continue
        if not _is_pr_url(url):
            continue  # only link PRs, not commits or other URLs
        link_lines.append(f"* [{title}|{url}]")
    if link_lines:
        text = text.rstrip()
        text += "\n\nh3. Linear PR Links\n\n" + "\n".join(link_lines)

    return convert_description_to_wiki(text, image_placeholders)


def _process_comment_body_wiki(body: str, issue_key: str) -> str:
    """Convert comment body to wiki; optionally upload inline images to issue and use !filename!."""
    image_placeholders: list[tuple[str, str]] = []
    placeholder_idx = [0]

    def repl(m: re.Match) -> str:
        url = m.group(1)
        if not _is_likely_file_attachment(url):
            return m.group(0)
        ph = f"<<<IMG{placeholder_idx[0]}>>>"
        placeholder_idx[0] += 1
        try:
            r = requests.get(url, timeout=30, headers={"Authorization": LINEAR_API_KEY})
            r.raise_for_status()
            ext = ".png"
            if "content-type" in r.headers:
                ct = r.headers["content-type"].lower()
                if "jpeg" in ct or "jpg" in ct:
                    ext = ".jpg"
                elif "gif" in ct:
                    ext = ".gif"
                elif "webp" in ct:
                    ext = ".webp"
            import tempfile
            with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp:
                tmp.write(r.content)
                tmp_path = tmp.name
            try:
                upload_name = f"comment_img_{placeholder_idx[0] - 1}{ext}"
                stored_name, content_url = upload_to_jira(issue_key, tmp_path, upload_name)
                ref = content_url if content_url else stored_name
                image_placeholders.append((ph, ref))
            finally:
                try:
                    os.unlink(tmp_path)
                except Exception:
                    pass
        except Exception as e:
            raise RuntimeError(f"Failed to download/upload comment image from {url}: {e}") from e
        return ph

    text = re.sub(r"!\[[^\]]*\]\(([^)]+)\)", repl, body)
    text = re.sub(r"<img[^>]+src=[\"']([^\"']+)[\"'][^>]*>", repl, text)
    return convert_comment_to_wiki(text, image_placeholders)


def migrate_one(issue: dict[str, Any], linear_url_template: str) -> str | None:
    """Migrate one Linear issue to Jira. Returns Jira issue key or None on skip."""
    title = issue.get("title") or "Untitled"
    identifier = issue.get("identifier") or ""
    description = issue.get("description") or ""
    created_at = issue.get("createdAt")
    due_date = (issue.get("dueDate") or "")[:10] if issue.get("dueDate") else None  # YYYY-MM-DD for Jira
    creator = issue.get("creator") or {}
    assignee = issue.get("assignee") or {}
    attachments = (issue.get("attachments") or {}).get("nodes") or []
    comments = (issue.get("comments") or {}).get("nodes") or []

    # Check for existing Jira issue by Linear ID to avoid creating duplicates
    existing_key = _find_jira_issue_by_linear_id(identifier) if (identifier and JIRA_LINEAR_ID_FIELD_ID) else None
    if existing_key:
        print(f"Skipping creation: existing issue {existing_key} (Linear {identifier}: {title!r})", flush=True)
        return existing_key

    # Minimal description for create (required by some projects); overwritten with wiki via v2 below
    summary_for_jira = _normalize_summary(title)
    fields: dict[str, Any] = {
        "project": {"key": JIRA_PROJECT_KEY},
        "summary": summary_for_jira,
        "issuetype": {"name": JIRA_ISSUE_TYPE},
        "description": {
            "type": "doc",
            "version": 1,
            "content": [{"type": "paragraph", "content": []}],
        },
    }
    creator_id = _jira_account_id_for_email(creator.get("email")) if creator.get("email") else None
    if creator_id:
        fields["reporter"] = {"accountId": creator_id}
    assignee_id = _jira_account_id_for_email(assignee.get("email")) if assignee.get("email") else None
    if assignee_id:
        fields["assignee"] = {"accountId": assignee_id}
    linear_priority = issue.get("priority")
    if isinstance(linear_priority, dict) and "label" in linear_priority:
        linear_priority = linear_priority.get("label")
    jira_priority = _jira_priority_for_linear(linear_priority)
    if jira_priority:
        fields["priority"] = {"name": jira_priority}
    labels = (issue.get("labels") or {}).get("nodes") or []
    if labels:
        fields["labels"] = [_jira_label_name(n.get("name")) for n in labels if n.get("name")]
    # Status and custom fields omitted on create (may not be on create screen); set via update after

    resp = requests.post(
        f"{JIRA_BASE_URL}/rest/api/3/issue",
        headers=_jira_headers(),
        json={"fields": fields},
        timeout=30,
    )
    if resp.status_code == 400:
        try:
            err = resp.json()
            msg = err.get("errorMessages", [])
            if isinstance(msg, list):
                msg = "; ".join(str(m) for m in msg)
            details = err.get("errors", {})
            if details:
                msg = f"{msg} " if msg else ""
                msg += "errors: " + str(details)
            raise RuntimeError(f"Jira 400: {msg or resp.text}")
        except (ValueError, KeyError):
            raise RuntimeError(f"Jira 400: {resp.text}")
    resp.raise_for_status()
    issue_key = resp.json()["key"]

    # Set custom fields (created date, Linear id) and duedate in a separate PUT so they aren't blocked by status failing
    custom_updates: dict[str, Any] = {}
    if JIRA_CREATED_FIELD_ID and created_at:
        custom_updates[JIRA_CREATED_FIELD_ID] = created_at[:10]  # YYYY-MM-DD for Jira Date Picker
    if JIRA_LINEAR_ID_FIELD_ID and identifier:
        custom_updates[JIRA_LINEAR_ID_FIELD_ID] = identifier
    if due_date:
        custom_updates["duedate"] = due_date
    if custom_updates:
        patch_resp = requests.put(
            f"{JIRA_BASE_URL}/rest/api/3/issue/{issue_key}",
            headers=_jira_headers(),
            json={"fields": custom_updates},
            timeout=30,
        )
        if patch_resp.status_code not in (200, 204):
            print(f"  Warning: custom fields update failed for {issue_key}: {patch_resp.status_code} {patch_resp.text[:200]}")

    # Set status via transitions API (Jira does not allow setting status via PUT)
    state = (issue.get("state") or {}).get("name")
    jira_status = _jira_status_for_linear_state(state)
    if jira_status:
        _transition_issue_to_status(issue_key, jira_status)

    description_wiki = process_description_and_attachments(
        description, attachments, issue_key, identifier, linear_url_template
    )
    update_jira_description_v2(issue_key, description_wiki)

    for c in sorted(comments, key=lambda x: x.get("createdAt") or ""):
        body = c.get("body") or ""
        body_wiki = _process_comment_body_wiki(body, issue_key)
        posted_at = _format_linear_datetime(c.get("createdAt"))
        user = c.get("user") or {}
        author = (user.get("name") or user.get("email") or "Unknown").strip()
        body_wiki = ("_Linear comment by *" + author + "* from " + posted_at + ":_\n\n" + body_wiki.strip()).strip()
        if body_wiki:
            add_jira_comment_v2(issue_key, body_wiki)

    project_name = (issue.get("project") or {}).get("name") or "No project"
    epic_key = _get_or_create_epic_for_linear_project(project_name)
    _set_issue_parent(issue_key, epic_key)

    print(f"Created {issue_key}: {title!r}", flush=True)
    return issue_key


def main() -> None:
    print("Starting Linear → Jira migration...", flush=True)
    if not LINEAR_API_KEY:
        print("Error: LINEAR_API_KEY is not set. Set it in the environment or edit the script.")
        print("  Get a key from Linear: Settings → Security → Personal API keys")
        raise SystemExit(1)
    if not JIRA_BASE_URL or not JIRA_EMAIL or not JIRA_API_TOKEN or not JIRA_PROJECT_KEY:
        print("Error: Jira config missing. Set JIRA_BASE_URL, JIRA_EMAIL, JIRA_API_TOKEN, JIRA_PROJECT_KEY")
        raise SystemExit(1)
    linear_url_template = LINEAR_ISSUE_URL_TEMPLATE or ""
    print(f"Fetching issues from Linear (team {LINEAR_TEAM_KEY})...", flush=True)
    issues = get_linear_issues()
    print(f"Fetched {len(issues)} issues from Linear.", flush=True)
    if LINEAR_LABELS_FILTER:
        label_set = set(LINEAR_LABELS_FILTER)
        issues = [
            i
            for i in issues
            if label_set & {n.get("name") for n in (i.get("labels") or {}).get("nodes") or [] if n.get("name")}
        ]
        print(f"Filtered to {len(issues)} issues with at least one of: {LINEAR_LABELS_FILTER}", flush=True)
    if not issues:
        print("No issues to migrate. Check LINEAR_TEAM_KEY if you expected issues.")
        return
    for issue in issues:
        try:
            key = migrate_one(issue, linear_url_template)
        except Exception as e:
            print(f"Skip/Error {issue.get('identifier')}: {e}", flush=True)
    print("Done.", flush=True)


if __name__ == "__main__":
    main()

linear_to_jira_migrate_README.md

Linear → Jira migration script

Migrates Linear issues to Jira Cloud: title, description, comments, attachments, labels, status, priority, due date, reporter/assignee. Creates Epics per Linear project and links issues to them.

Requirements

• Python 3.10+
• requests

pip install requests

Setup

1. Linear: Create a Personal API key (Settings → Security → Personal API keys). No Bearer prefix.

2. Jira: Create an API token for your Atlassian account.

3. Jira custom fields (optional but recommended):

   • A custom field to store the Linear issue ID (e.g. DES-123) so the script can skip duplicates. Set JIRA_LINEAR_ID_FIELD_ID to its ID (e.g. customfield_11269) and JIRA_LINEAR_ID_FIELD_NAME to its name (e.g. "Linear Issue ID").
   • Optionally a custom field for Linear created date; set JIRA_CREATED_FIELD_ID.

Configuration

Set via environment or edit the config block at the top of the script:

Variable	Description
LINEAR_API_KEY	Linear personal API key (required)
LINEAR_TEAM_KEY	Linear team key to import (e.g. DES)
LINEAR_LABELS_FILTER	Optional list of label names; only issues with at least one are imported
JIRA_BASE_URL	e.g. https://your-site.atlassian.net
JIRA_EMAIL	Jira user email
JIRA_API_TOKEN	Jira API token
JIRA_PROJECT_KEY	Jira project key (e.g. MYPROJ)
JIRA_LINEAR_ID_FIELD_ID	Custom field ID for Linear issue ID (for duplicate detection)
JIRA_LINEAR_ID_FIELD_NAME	Custom field name for Linear issue ID (for duplicate detection)
JIRA_CREATED_FIELD_ID	Optional custom field ID for created date

Usage

export LINEAR_API_KEY="lin_api_..."
export JIRA_EMAIL="you@example.com"
export JIRA_API_TOKEN="..."
# Optionally: LINEAR_TEAM_KEY, JIRA_BASE_URL, JIRA_PROJECT_KEY, JIRA_LINEAR_ID_FIELD_ID, etc.

python linear_to_jira_migrate.py

• Duplicate detection: If a Jira issue already has the same Linear ID (in the custom field), the script skips creation.
• Labels: Jira labels cannot contain spaces; the script replaces spaces with underscores.
• Images: Inline images in descriptions/comments are downloaded from Linear and uploaded to Jira, then embedded with wiki markup.

Gist

This script is also available as a GitHub Gist for easy sharing and forking.

linear_to_jira_migrate_requirements.txt

requests>=2.28.0
python-dotenv>=1.0.0