Skip to content

Instantly share code, notes, and snippets.

@Holzhaus
Created May 4, 2021 11:52
Show Gist options
  • Save Holzhaus/d3e5950b1ef9e35acb97e53592f34ee9 to your computer and use it in GitHub Desktop.
Save Holzhaus/d3e5950b1ef9e35acb97e53592f34ee9 to your computer and use it in GitHub Desktop.
#!/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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment