json2github.py

#!/usr/bin/env python3
"""
Alternative lp2gh JSON Importer for Mixxx issue migration

You can install this like this:

    $ python3 -m venv venv
    $ source venv/bin/activate
    $ pip install PyGithub

Then you can use it like this:

    $ python json2github.py --repo Holzhaus/mixxx-gh-issue-migration --token mytoken --output-file=mixxx_bugs.json mixxx_bugs.json mixxx_milestones.json
"""
import argparse
import binascii
import datetime
import github
import json
import logging
import random
import textwrap
import time

LAUNCHPAD_STATUS_MAP = {
    "Confirmed": None,
    "Fix Committed": None,
    "Fix Released": None,
    "Incomplete": "incomplete",
    "In Progress": None,
    "Invalid": "invalid",
    "New": None,
    "Triaged": None,
    "Won't Fix": "wontfix",
}

LAUNCHPAD_IMPORTANCE_MAP = {
    "Critical": "bug",
    "High": "bug",
    "Low": "bug",
    "Medium": "bug",
    "Undecided": None,
    "Wishlist": "wishlist",
}

LAUNCHPAD_USER_MAP = {
    "be.ing": "Be-ing",
    "bkgood": "bkgood",
    "crislacerda": "crislacerda",
    "daschuer": "daschuer",
    "default-kramer": "default-kramer",
    "ferranpujol": "ferranpujolcamins",
    "frank-breitling": "fkbreitl",
    "hacksdump": "hacksdump",
    "hile": "hile",
    "holthuis-jan": "Holzhaus",
    "iamcodemaker": "iamcodemaker",
    "joerg-ubuntu": "JoergAtGithub",
    "josepma": "JosepMaJAZ",
    "jus": "esbrandt",
    "kousu": "kousu",  # not sure, but first name seems to match
    "launchpad-net-poelzi": "poelzi",
    "max-linke": "kain88-de",
    "mevsme": "mevsme",
    "mxmilkiib": "mxmilkiib",
    "naught101": "naught101",
    "ninomp": "ninomp",
    "nopeppermint": "nopeppermint",
    "nschloe": "nschloe",
    "pasanen-tuukka": "illuusio",
    "pegasus-renegadetech": "Pegasus-RPG",
    "poelzi": "poelzi",
    "rawrr": "rawrr",
    "ronso0": "ronso0",
    "rryan": "rryan",
    "sblaisot": "sblaisot",
    "swiftb0y": "Swiftb0y",
    "toszlanyi": "toszlanyi",
    "uklotzde": "uklotzde",
    "vrince": "vrince",
    "xerus2000": "xeruf",
    "ywwg": "ywwwg",
    "zezic": "zezic",
}

GITHUB_ALLOWED_ASSIGNEES = {
    "Holzhaus",
    "asantoni",
    "Be-ing",
    "daschuer",
    "esbrandt",
    "Pegasus-RPG",
    "ronso0",
    "rryan",
    "sblaisot",
    "Swiftb0y",
    "uklotzde",
    "ywwg",
}

LAUNCHPAD_TAGS_TO_KEEP = {
    "aac",
    "accessibility",
    "analyzer",
    "autodj",
    "auxiliary",
    "beatgrid",
    "bpm",
    "broadcast",
    "browse",
    "build",
    "cloud",
    "cmake",
    "controllers",
    "coverart",
    "crash",
    "cue",
    "easy",
    "effects",
    "engine",
    "eq",
    "follower",
    "gui",
    "hackathon",
    "hid",
    "i18n",
    "installer",
    "itunes",
    "jack",
    "key",
    "keyboard",
    "library",
    "linux",
    "looping",
    "lyrics",
    "m4a",
    "macos",
    "manual",
    "metadata",
    "microphone",
    "midi",
    "mp3",
    "overview",
    "packaging",
    "passthrough",
    "performance",
    "playlist",
    "polish",
    "portaudio",
    "preferences",
    "qt5",
    "quantize",
    "raspberry",
    "recording",
    "regression",
    "rekordbox",
    "sampler",
    "scanner",
    "serato",
    "skin",
    "slip",
    "soundio",
    "soundsource",
    "standards",
    "sync",
    "tooltip",
    "touchscreen",
    "transport",
    "usability",
    "vinylcontrol",
    "waveform",
    "weekend",
    "windows",
}

class LaunchpadImporter:
    def __init__(self, token, repo, milestonedata):
        self.logger = logging.getLogger(__name__)
        self.gh = github.Github(login_or_token=token)
        self.repo = self.gh.get_repo(repo)
        self.gh_labels = {
            label.name: label
            for label in self.repo.get_labels()
        }
        self.gh_milestones = {
            milestone.title: milestone
            for milestone in self.repo.get_milestones(state="all")
        }
        self.lp_milestones = {x["name"]: x for x in milestonedata}

    def get_user(self, username):
        try:
            username = "@" + LAUNCHPAD_USER_MAP[username]
        except KeyError:
            pass
        return username

    def get_label(self, name):
        try:
            label = self.gh_labels[name]
        except KeyError:
            label = self.import_label(name)
        return label

    def format_body(self, issuedata):
        header = [
            f"Reported by: {self.get_user(issuedata['owner'])}",
            f"Date: {issuedata['date_created']}",
            f"Status: {issuedata['status']}",
            f"Importance: {issuedata['importance']}",
            f"Launchpad Issue: [lp{issuedata['id']}]({issuedata['lp_url']})",
        ]
        if issuedata["tags"]:
            header.append("Tags: %s" % ", ".join(issuedata["tags"]))
        body = textwrap.indent(issuedata["description"], prefix="    ")
        return "\n".join(header) + "\n\n" + body

    def format_comment(self, commentdata):
        header = [
            f"Created by: {self.get_user(commentdata['owner'])}",
            f"Date: {commentdata['date_created']}",
        ]
        body = textwrap.indent(commentdata["content"], prefix="    ")
        return "\n".join(header) + "\n\n" + body

    def name_to_milestone(self, milestone_name):
        try:
            milestone = self.gh_milestones[milestone_name]
        except KeyError:
            milestonedata = self.lp_milestones.get(milestone_name, {
                "active": True,
                "date_targeted": None,
                "name": milestone_name,
                "summary": "",
            })
            milestone = self.import_milestone(milestonedata)
        return milestone

    def handle_ratelimit(self, func):
        abuse_timeout = 30
        while True:
            try:
                return func()
            except github.RateLimitExceededException:
                rate_limit_resettime = datetime.datetime.fromtimestamp(
                    self.gh.rate_limiting_resettime
                )
                self.logger.warning(
                    "Rate limit exceeded (%d left/%d total), waiting until %r",
                    *self.gh.rate_limiting, rate_limit_resettime,
                )
                seconds_to_wait = (
                    rate_limit_resettime - datetime.datetime.now()
                ).total_seconds()
                if seconds_to_wait <= 0:
                    self.logger.warning(
                        "Failed to detect wait time, assuming 10 seconds...",
                        seconds_to_wait)
                    seconds_to_wait = 10
                self.logger.warning("Sleeping for %d seconds", seconds_to_wait)
                time.sleep(seconds_to_wait)
            except github.GithubException as e:
                if e.status == 403 and "abuse" in e.data.get("message", ""):
                    self.logger.warning(
                        "Triggered abuse detection, sleeping %d seconds...",
                        abuse_timeout)
                    time.sleep(abuse_timeout)
                    abuse_timeout *= 2
                else:
                    raise
            else:
                break

    def import_milestone(self, milestonedata):
        state = "open" if milestonedata["active"] else "closed"
        due_on = github.GithubObject.NotSet
        if milestonedata["date_targeted"]:
            due_on = datetime.datetime.strptime(
                milestonedata["date_targeted"],
                "%Y-%m-%dT%H:%M:%SZ"
            )
        description = github.GithubObject.NotSet
        if milestonedata["summary"]:
            description = milestonedata["summary"]
        milestone = self.handle_ratelimit(lambda: self.repo.create_milestone(
            milestonedata["name"], state, description, due_on
        ))
        self.logger.info("Created milestone: %r", milestone)
        self.gh_milestones[milestone.title] = milestone
        return milestone

    def status_to_label(self, status):
        label_name = LAUNCHPAD_STATUS_MAP.get(status)
        if label_name is None:
            return None
        return self.get_label(label_name)

    def importance_to_label(self, importance):
        label_name = LAUNCHPAD_IMPORTANCE_MAP.get(importance)
        if label_name is None:
            return None
        return self.get_label(label_name)

    def import_label(self, label_name):
        color = binascii.hexlify(random.randbytes(3)).decode()
        label = self.handle_ratelimit(
            lambda: self.repo.create_label(label_name, color))
        self.logger.info("Created label: %r", label)
        self.gh_labels[label_name] = label
        return label

    def name_to_assignee(self, name):
        username = LAUNCHPAD_USER_MAP.get(name)
        if username and username in GITHUB_ALLOWED_ASSIGNEES:
            return username
        else:
            return github.GithubObject.NotSet

    def import_issue(self, issuedata):
        milestone = github.GithubObject.NotSet
        if issuedata["milestone"]:
            milestone = self.name_to_milestone(issuedata["milestone"])
        labels = []
        status_label = self.status_to_label(issuedata["status"])
        if status_label:
            labels.append(status_label)
        importance_label = self.importance_to_label(issuedata["importance"])
        if importance_label:
            labels.append(importance_label)
        if issuedata["duplicate_of"]:
            labels.append(self.get_label("duplicate"))
        if issuedata["security_related"]:
            labels.append(self.get_label("security"))
        if issuedata["tags"]:
            labels.extend([self.get_label(tag) for tag in issuedata["tags"]
                           if tag in LAUNCHPAD_TAGS_TO_KEEP])
        assignee = self.name_to_assignee(issuedata["assignee"])
        issue = self.handle_ratelimit(lambda: self.repo.create_issue(
            title=issuedata["title"],
            body=self.format_body(issuedata),
            milestone=milestone,
            assignee=assignee,
            labels=labels
        ))
        self.logger.info("Created issue: %r", issue)
        return issue

    def import_issuecomment(self, issue, commentdata):
        comment = self.handle_ratelimit(
            lambda: issue.create_comment(self.format_comment(commentdata)))
        self.logger.info("Created issue comment: %r", comment)
        return comment

    def run_import(self, lp_issues, lp_milestones):
        for issuedata in sorted(
                lp_issues.values(),
                key=lambda x: (x["date_created"], x["id"])):
            gh_issue_number = issuedata.get("gh_issue_number")
            if gh_issue_number:
                issue = self.handle_ratelimit(
                    lambda: self.repo.get_issue(gh_issue_number))
            else:
                issue = self.import_issue(issuedata)
                lp_issues[issuedata["id"]]["gh_issue_number"] = issue.number

            comments_imported = issuedata.get("gh_comments_imported", 0)
            lp_issues[issuedata["id"]][
                "gh_comments_imported"] = comments_imported

            comments = issuedata["comments"][comments_imported:]
            for comment in comments:
                comment = self.import_issuecomment(issue, comment)
                lp_issues[issuedata["id"]]["gh_comments_imported"] += 1

            if issuedata.get("gh_status_comment_imported", False):
                continue

            if issuedata["status"] in (
                    "Fix Released", "Fix Committed", "Invalid", "Won't Fix"):
                comment = f'Issue closed with status "{issuedata["status"]}".'
                self.handle_ratelimit(lambda: issue.create_comment(comment))
                if issue.state != "closed":
                    self.handle_ratelimit(lambda: issue.edit(state="closed"))
            lp_issues[issuedata["id"]]["gh_status_comment_imported"] = True

        for issuedata in sorted(
                lp_issues.values(),
                key=lambda x: (x["date_created"], x["id"])):
            if issuedata.get("gh_duplicate_comment_imported", False):
                continue

            if issuedata["duplicate_of"] is None:
                issue_number = lp_issues[issuedata["id"]]["gh_issue_number"]
                comment = f'Marked as duplicate of issue #{issue_number}.'
                self.handle_ratelimit(lambda: issue.create_comment(comment))
                if issue.state != "closed":
                    self.handle_ratelimit(lambda: issue.edit(state="closed"))
            lp_issues[issuedata["id"]]["gh_duplicate_comment_imported"] = True


def main(argv=None):
    parser = argparse.ArgumentParser()
    parser.add_argument("--repo", required=True)
    parser.add_argument("--token", required=True)
    parser.add_argument("--output-file")
    parser.add_argument("bugs_file", type=argparse.FileType("r"))
    parser.add_argument("milestone_file", type=argparse.FileType("r"))
    args = parser.parse_args(argv)

    logging.basicConfig(level=logging.INFO)

    lp_milestones = json.load(args.milestone_file)
    lp_issues = {x["id"]: x for x in json.load(args.bugs_file)}
    importer = LaunchpadImporter(args.token, args.repo, lp_milestones)
    try:
        importer.run_import(lp_issues, lp_milestones)
    finally:
        if args.output_file:
            with open(args.output_file, mode="w") as f:
                json.dump(list(lp_issues.values()), f)


if __name__ == "__main__":
    main()