From c8bf8f460908f198b62ba3cec47f89099e5a3a77 Mon Sep 17 00:00:00 2001 From: Sean Davis Date: Sun, 18 Sep 2022 08:18:06 -0400 Subject: Sync extended bug details from Launchpad --- .github/lpbugtracker.py | 295 ++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 250 insertions(+), 45 deletions(-) diff --git a/.github/lpbugtracker.py b/.github/lpbugtracker.py index 880bbce..4ccd6f0 100755 --- a/.github/lpbugtracker.py +++ b/.github/lpbugtracker.py @@ -10,6 +10,8 @@ Based on https://github.com/ubuntu/yaru/blob/master/.github/lpbugtracker.py import os import subprocess import logging +import json + from launchpadlib.launchpad import Launchpad log = logging.getLogger("lpbugtracker") @@ -24,6 +26,18 @@ LP_SOURCE_URL_NAME = "lightdm-gtk-greeter" HOME = os.path.expanduser("~") CACHEDIR = os.path.join(HOME, ".launchpadlib", "cache") +LP_OPEN_STATUS_LIST = ["New", + "Opinion", + "Confirmed", + "Triaged", + "In Progress", + "Fix Committed", + "Incomplete"] +LP_CLOSED_STATUS_LIST = ["Invalid", + "Won't Fix", + "Expired", + "Fix Released"] + def main(): lp_bugs = get_lp_bugs() @@ -34,15 +48,143 @@ def main(): for id in lp_bugs: if id in gh_bugs.keys(): + last_comment_id = get_gh_last_lp_comment(gh_bugs[id]["id"]) + add_comments(gh_bugs[id]["id"], last_comment_id, lp_bugs[id]["messages"]) + + gh_labels = parse_gh_labels(gh_bugs[id]["labels"]) if lp_bugs[id]["closed"] and gh_bugs[id]["status"] != "closed": - close_issue(gh_bugs[id]["id"], lp_bugs[id]["status"]) + close_issue(gh_bugs[id]["id"], gh_labels["labels"], lp_bugs[id]["status"]) + elif lp_bugs[id]["status"] != gh_labels["status"]: + update_issue(gh_bugs[id]["id"], gh_labels["labels"], lp_bugs[id]["status"]) elif not lp_bugs[id]["closed"] and lp_bugs[id]["status"] != "Incomplete": - create_issue(id, lp_bugs[id]["title"], lp_bugs[id]["link"]) + bug_id = create_issue(id, lp_bugs[id]["title"], lp_bugs[id]["link"], lp_bugs[id]["status"]) + add_comments(bug_id, -1, lp_bugs[id]["messages"]) def get_lp_bugs(): """Get a list of bugs from Launchpad""" + package = lp_get_package(LP_SOURCE_NAME) + open_bugs = lp_package_get_bugs(package, LP_OPEN_STATUS_LIST, True) + closed_bugs = lp_package_get_bugs(package, LP_CLOSED_STATUS_LIST, False) + + return {**open_bugs, **closed_bugs} + + +def get_gh_bugs(): + """Get the list of the LP bug already tracked in GitHub. + + Launchpad bugs tracked on GitHub have a title like + + "LP# " + + this function returns a list of the "LP#<id>" substring for each bug, + open or closed, found on the repository on GitHub. + """ + + output = subprocess.check_output( + ["hub", "issue", "--labels", "Launchpad", "--state", "all", "--format", "%I|%S|%L|%t%n"] + ) + bugs = {} + for line in output.decode().split("\n"): + issue = parse_gh_issue(line) + if issue is not None: + bugs[issue["lpid"]] = issue + return bugs + + +def create_issue(id, title, weblink, status): + """ Create a new Bug using HUB """ + print("creating:", id, title, weblink, status) + return gh_create_issue("LP#{} {}".format(id, title), + "Reported first on Launchpad at {}".format(weblink), + "Launchpad,%s" % status) + + +def update_issue(id, current_labels, status): + """ Update a Bug using HUB """ + print("updating:", id, status) + new_labels = ["Launchpad", status] + current_labels + gh_set_issue_labels(id, ",".join(new_labels)) + + +def close_issue(id, current_labels, status): + """ Close the Bug using HUB and leave a comment """ + print("closing:", id, status) + new_labels = ["Launchpad", status] + current_labels + gh_add_comment(id, "Issue closed on Launchpad with status: {}".format(status)) + gh_close_issue(id, ",".join(new_labels)) + + +def add_comments(issue_id, last_comment_id, comments): + for id in comments: + if id > last_comment_id: + print("adding comment:", issue_id, id) + gh_add_comment(issue_id, format_lp_comment(comments[id])) + + +def quote_str(string): + content = [] + for line in string.split("\n"): + content.append("> {}".format(line)) + return "\n".join(content) + + +def format_lp_comment(message): + output = "[LP#{}]({}): *{} ({}) wrote on {}:*\n\n{}".format(message["id"], + message["link"], + message["author"]["display_name"], + message["author"]["name"], + message["date"], + quote_str(message["content"])) + if len(message["attachments"]) > 0: + output += "\n\nAttachments:" + for attachment in message["attachments"]: + output += "\n- [{}]({})".format(attachment["title"], + attachment["link"]) + return output + + +def parse_gh_issue(issue): + if "LP#" in issue: + id, status, labels, lp = issue.strip().split("|", 3) + labels = labels.split(", ") + lpid, title = lp.split(" ", 1) + lpid = lpid[3:] + return {"id": id, "lpid": lpid, "status": status, "title": title, "labels": labels} + return None + + +def parse_gh_labels(labels): + result = { + "status": "Unknown", + "labels": [] + } + for label in labels: + if label == "Launchpad": + continue + elif label in LP_OPEN_STATUS_LIST + LP_CLOSED_STATUS_LIST: + result["status"] = label + else: + result["labels"].append(label) + return result + + +def get_gh_last_lp_comment(issue_id): + comments = gh_list_comments(issue_id) + last_comment_id = -1 + for comment in comments: + if comment["body"][0:4] == "[LP#": + comment_id = comment["body"].split("]")[0] + comment_id = comment_id[4:] + comment_id = int(comment_id) + if comment_id > last_comment_id: + last_comment_id = comment_id + return last_comment_id + + +# Launchpad API +def lp_get_package(source_name): lp = Launchpad.login_anonymously( "%s LP bug checker" % LP_SOURCE_NAME, "production", CACHEDIR, version="devel" ) @@ -50,82 +192,120 @@ def get_lp_bugs(): ubuntu = lp.distributions["ubuntu"] archive = ubuntu.main_archive - packages = archive.getPublishedSources(source_name=LP_SOURCE_NAME) + packages = archive.getPublishedSources(source_name=source_name) package = ubuntu.getSourcePackage(name=packages[0].source_package_name) - bug_tasks = package.searchTasks(status=["New", "Opinion", - "Invalid", "Won't Fix", - "Expired", "Confirmed", - "Triaged", "In Progress", - "Fix Committed", "Fix Released", - "Incomplete"]) + return package + + +def lp_package_get_bugs(package, status_list, get_messages = False): + """Get a list of bugs from Launchpad""" + + bug_tasks = package.searchTasks(status=status_list) bugs = {} for task in bug_tasks: + bug = lp_task_get_bug(task, get_messages) + if bug is not None: + bugs[bug["id"]] = bug + + return bugs + + +def lp_task_get_bug(task, get_messages = False): + try: id = str(task.bug.id) - title = task.title.split(": ")[1] + title = task.title.split(": ", 1)[1] status = task.status - closed = status in ["Invalid", "Won't Fix", "Expired", "Fix Released"] + closed = status in LP_CLOSED_STATUS_LIST link = "https://bugs.launchpad.net/ubuntu/+source/{}/+bug/{}".format(LP_SOURCE_URL_NAME, id) - bugs[id] = {"title": title, "link": link, "status": status, "closed": closed} + if get_messages: + messages = lp_bug_get_messages(task.bug) + else: + messages = {} + return {"id": id, "title": title, "link": link, "status": status, "closed": closed, "messages": messages} + except: + return None - return bugs +def lp_bug_get_messages(bug): + messages = {} + for message in bug.messages: + message_id = lp_message_get_id(message) + messages[message_id] = { + "id": str(message_id), + "link": message.web_link, + "content": message.content, + "date": lp_message_get_date_time(message), + "author": lp_message_get_author(message), + "attachments": lp_message_get_attachments(message) + } + return messages -def get_gh_bugs(): - """Get the list of the LP bug already tracked in GitHub. - Launchpad bugs tracked on GitHub have a title like +def lp_message_get_author(message): + return { + "name": message.owner.name, + "display_name": message.owner.display_name, + } - "LP#<id> <title>" - this function returns a list of the "LP#<id>" substring for each bug, - open or closed, found on the repository on GitHub. - """ +def lp_message_get_id(message): + return int(message.web_link.split("/")[-1]) - output = subprocess.check_output( - ["hub", "issue", "--labels", "Launchpad", "--state", "all", "--format", "%I %S %t%n"] - ) - bugs = {} - for line in output.decode().split("\n"): - if "LP#" in line: - id, status, lpid, title = line.strip().split(" ", 3) - lpid = lpid[3:] - bugs[lpid] = {"id": id, "status": status, "title": title} - return bugs +def lp_message_get_date_time(message): + dt = message.date_created + dt = dt.isoformat().split(".")[0] + dt = dt.split("T")[0] + return dt -def create_issue(id, title, weblink): - """ Create a new Bug using HUB """ - print("creating:", id, title, weblink) - subprocess.run( + +def lp_message_get_attachments(message): + attachments = [] + for attach in message.bug_attachments: + attachments.append({ + "link": attach.data_link, + "title": attach.title + }) + return attachments + + +# GitHub API +def gh_create_issue(summary, description, labels): + url = subprocess.check_output( [ "hub", "issue", "create", "--message", - "LP#{} {}".format(id, title), + summary, "--message", - "Reported first on Launchpad at {}".format(weblink), + description, "-l", - "Launchpad", + labels ] ) + url = url.decode("utf-8") + url = url.strip() + id = url.split("/")[-1] + return id -def close_issue(id, status): - """ Close the Bug using HUB and leave a comment """ - print("closing:", id, status) +def gh_set_issue_labels(id, labels): subprocess.run( [ "hub", - "api", - "repos/{}/{}/issues/{}/comments".format(GH_OWNER, GH_REPO, id), - "--field", - "body=Issue closed on Launchpad with status: {}".format(status) + "issue", + "update", + id, + "-l", + labels, ] ) + +def gh_close_issue(id, labels): subprocess.run( [ "hub", @@ -133,9 +313,34 @@ def close_issue(id, status): "update", id, "--state", - "closed" + "closed", + "-l", + labels, + ] + ) + + +def gh_add_comment(issue_id, comment): + subprocess.run( + [ + "hub", + "api", + "repos/{}/{}/issues/{}/comments".format(GH_OWNER, GH_REPO, issue_id), + "--field", + "body={}".format(comment) + ] + ) + + +def gh_list_comments(issue_id): + output = subprocess.check_output( + [ + "hub", + "api", + "repos/{}/{}/issues/{}/comments".format(GH_OWNER, GH_REPO, issue_id) ] ) + return json.loads(output) if __name__ == "__main__": -- cgit v1.2.1