Migrate GitHub project between accounts

transporter.py

#
# Copyright (c) nexB Inc. and others. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
# See https://aboutcode.org for more information about nexB OSS projects.
#

from traceback import format_exc as traceback_format_exc
import requests

SOURCE_ACCOUNT_TYPE = "ORGANIZATION"
SOURCE_ACCOUNT_NAME = ""
SOURCE_PROJECT_NUMBER = 0

TARGET_ACCOUNT_TYPE = "ORGANIZATION"
TARGET_ACCOUNT_NAME = ""
TARGET_PROJECT_NUMBER = 0

GITHUB_TOKEN = ""

def graphql_query(query, variables=None):
    url = "https://api.github.com/graphql"
    headers = {
        "Authorization": f"Bearer {GITHUB_TOKEN}",
        "Accept": "application/vnd.github+json"
    }
    response = requests.post(url, headers=headers, json={"query": query, "variables": variables})
    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"Query failed: {response.status_code}. Response: {response.text}")

def fetch_project_node_id(account_type, account_name, project_number):
    query = f"""
    query {{
        {account_type.lower()}(login: "{account_name}") {{
            projectV2(number: {project_number}) {{
                id
            }}
        }}
    }}
    """
    data = graphql_query(query)
    return data["data"][account_type.lower()]["projectV2"]["id"]

def get_all_project_fields(project_id):
    """
    Fetch all custom fields from a project.
    Currently supports Single Select and (fallback) Text fields.
    """
    query = """
    query ($projectId: ID!) {
        node(id: $projectId) {
            ... on ProjectV2 {
                fields(first: 100) {
                    nodes {
                        __typename
                        ... on ProjectV2SingleSelectField {
                            id
                            name
                            options {
                                id
                                name
                            }
                        }
                        ... on ProjectV2Field {
                            id
                            name
                        }
                    }
                }
            }
        }
    }
    """
    data = graphql_query(query, {"projectId": project_id})
    # Debug output
    print("Response from get_all_project_fields:", data)
    if "errors" in data:
        raise Exception("GraphQL query errors: " + str(data["errors"]))
    if "data" not in data or data["data"].get("node") is None:
        raise Exception("No data returned in get_all_project_fields: " + str(data))
    fields_nodes = data["data"]["node"]["fields"]["nodes"]
    fields = {}
    for field in fields_nodes:
        field_name = field["name"]
        if field["__typename"] == "ProjectV2SingleSelectField":
            fields[field_name] = {
                "id": field["id"],
                "type": "SingleSelect",
                "options": {opt["name"]: opt["id"] for opt in field["options"]}
            }
        elif field["__typename"] == "ProjectV2Field":
            # Treat any generic field as a text field (or adjust based on a dataType if available)
            fields[field_name] = {
                "id": field["id"],
                "type": "Text"
            }
    return fields

def fetch_all_project_items(project_id):
    """
    Fetches all items from a project along with their field values.
    Note: The query now includes fragments for both single select and text field values.
    """
    all_items = []
    cursor = None
    has_next_page = True
    while has_next_page:
        query = """
        query($projectId: ID!, $cursor: String) {
            node(id: $projectId) {
                ... on ProjectV2 {
                    items(first: 100, after: $cursor) {
                        pageInfo {
                            hasNextPage
                            endCursor
                        }
                        nodes {
                            id
                            content {
                                ... on DraftIssue {
                                    title
                                    body
                                }
                                ... on Issue {
                                    title
                                    url
                                    id
                                }
                                ... on PullRequest {
                                    title
                                    url
                                    id
                                }
                            }
                            fieldValues(first: 100) {
                                nodes {
                                    __typename
                                    ... on ProjectV2ItemFieldSingleSelectValue {
                                        name
                                        field {
                                            ... on ProjectV2FieldCommon {
                                                name
                                            }
                                        }
                                    }
                                    ... on ProjectV2ItemFieldTextValue {
                                        text
                                        field {
                                            ... on ProjectV2FieldCommon {
                                                name
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
        """
        variables = {"projectId": project_id, "cursor": cursor}
        data = graphql_query(query, variables)
        items = data["data"]["node"]["items"]
        all_items.extend(items["nodes"])
        page_info = items["pageInfo"]
        has_next_page = page_info["hasNextPage"]
        cursor = page_info["endCursor"]
    return all_items

def add_item_to_project(project_id, content_id):
    query = """
    mutation($projectId: ID!, $contentId: ID!) {
        addProjectV2ItemById(input: {projectId: $projectId, contentId: $contentId}) {
            item { id }
        }
    }
    """
    data = graphql_query(query, {"projectId": project_id, "contentId": content_id})
    return data["data"]["addProjectV2ItemById"]["item"]["id"]

def create_draft_issue_in_project(project_id, title, body):
    query = """
    mutation($projectId: ID!, $title: String!, $body: String!) {
        addProjectV2DraftIssue(input: {projectId: $projectId, title: $title, body: $body}) {
            projectItem { id }
        }
    }
    """
    data = graphql_query(query, {"projectId": project_id, "title": title, "body": body})
    return data["data"]["addProjectV2DraftIssue"]["projectItem"]["id"]

def update_item_field_value(project_id, item_id, field_id, field_type, value):
    """
    Update a custom field on an item.
    Builds the value input based on the field type.
    """
    query = """
    mutation($input: UpdateProjectV2ItemFieldValueInput!) {
        updateProjectV2ItemFieldValue(input: $input) {
            projectV2Item { id }
        }
    }
    """
    if field_type == "SingleSelect":
        value_input = {"singleSelectOptionId": value}
    elif field_type == "Text":
        value_input = {"text": value}
    else:
        # For other field types, adjust accordingly.
        value_input = value
    variables = {
        "input": {
            "projectId": project_id,
            "itemId": item_id,
            "fieldId": field_id,
            "value": value_input
        }
    }
    graphql_query(query, variables)

def handler():
    try:
        # Fetch project IDs
        source_project_id = fetch_project_node_id(SOURCE_ACCOUNT_TYPE, SOURCE_ACCOUNT_NAME, SOURCE_PROJECT_NUMBER)
        target_project_id = fetch_project_node_id(TARGET_ACCOUNT_TYPE, TARGET_ACCOUNT_NAME, TARGET_PROJECT_NUMBER)

        # Fetch target project's custom fields dynamically
        target_fields = get_all_project_fields(target_project_id)
        if target_fields:
            print("Target project custom fields detected. Migrating all custom field values...")
        else:
            print("No custom fields detected in target project.")

        # Migrate items from source to target
        items = fetch_all_project_items(source_project_id)
        for item in items:
            content = item.get("content")
            if not content:
                continue

            # Add item to target project
            if "id" in content:  # Issue or PR
                content_id = content["id"]
                new_item_id = add_item_to_project(target_project_id, content_id)
                print(f"Added item: {content['title'][:50]}...")
            else:  # Draft issue
                title = content.get("title", "Untitled Draft")
                body = content.get("body", "")
                new_item_id = create_draft_issue_in_project(target_project_id, title, body)
                print(f"Created draft issue: {title[:50]}...")

            # Migrate each custom field value if a matching field exists in the target project
            for fv in item["fieldValues"]["nodes"]:
                # Get the field name from the source item's field value
                field_name = fv.get("field", {}).get("name")
                if not field_name or field_name not in target_fields:
                    continue
                target_field = target_fields[field_name]
                if fv["__typename"] == "ProjectV2ItemFieldSingleSelectValue":
                    option_name = fv.get("name")
                    if option_name in target_field.get("options", {}):
                        option_id = target_field["options"][option_name]
                        update_item_field_value(target_project_id, new_item_id, target_field["id"], target_field["type"], option_id)
                        print(f"Updated '{field_name}' to: {option_name}")
                elif fv["__typename"] == "ProjectV2ItemFieldTextValue":
                    text_value = fv.get("text")
                    update_item_field_value(target_project_id, new_item_id, target_field["id"], target_field["type"], text_value)
                    print(f"Updated '{field_name}' with text value.")
                # Add more elif blocks here if you support additional field types.

        print("Migration completed. Visit project at:")
        print(f"https://github.com/orgs/{TARGET_ACCOUNT_NAME}/projects/{TARGET_PROJECT_NUMBER}")

    except Exception as e:
        print(f"Error: {e}\n{traceback_format_exc()}")

if __name__ == "__main__":
    handler()