Sync issues from Launchpad to GitHub
This commit is contained in:
parent
252e706e49
commit
02c9117fb8
|
@ -0,0 +1,10 @@
|
|||
# Set update schedule for GitHub Actions
|
||||
|
||||
version: 2
|
||||
updates:
|
||||
|
||||
- package-ecosystem: "github-actions"
|
||||
directory: "/"
|
||||
schedule:
|
||||
# Check for updates to GitHub Actions every weekday
|
||||
interval: "daily"
|
|
@ -0,0 +1,347 @@
|
|||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
# vi: set ft=python :
|
||||
"""
|
||||
Launchpad Bug Tracker uses launchpadlib to get the bugs.
|
||||
|
||||
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")
|
||||
log.setLevel(logging.DEBUG)
|
||||
|
||||
GH_OWNER = "bluesabre"
|
||||
GH_REPO = "mugshot"
|
||||
|
||||
LP_SOURCE_NAME = "mugshot"
|
||||
LP_SOURCE_URL_NAME = "mugshot"
|
||||
|
||||
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()
|
||||
if len(lp_bugs) == 0:
|
||||
return
|
||||
|
||||
gh_bugs = get_gh_bugs()
|
||||
|
||||
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"], 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":
|
||||
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#<id> <title>"
|
||||
|
||||
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"
|
||||
)
|
||||
|
||||
ubuntu = lp.distributions["ubuntu"]
|
||||
archive = ubuntu.main_archive
|
||||
|
||||
packages = archive.getPublishedSources(source_name=source_name)
|
||||
package = ubuntu.getSourcePackage(name=packages[0].source_package_name)
|
||||
|
||||
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)[1]
|
||||
status = task.status
|
||||
closed = status in LP_CLOSED_STATUS_LIST
|
||||
link = "https://bugs.launchpad.net/ubuntu/+source/{}/+bug/{}".format(LP_SOURCE_URL_NAME, id)
|
||||
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
|
||||
|
||||
|
||||
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 lp_message_get_author(message):
|
||||
return {
|
||||
"name": message.owner.name,
|
||||
"display_name": message.owner.display_name,
|
||||
}
|
||||
|
||||
|
||||
def lp_message_get_id(message):
|
||||
return int(message.web_link.split("/")[-1])
|
||||
|
||||
|
||||
def lp_message_get_date_time(message):
|
||||
dt = message.date_created
|
||||
dt = dt.isoformat().split(".")[0]
|
||||
dt = dt.split("T")[0]
|
||||
return dt
|
||||
|
||||
|
||||
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",
|
||||
summary,
|
||||
"--message",
|
||||
description,
|
||||
"-l",
|
||||
labels
|
||||
]
|
||||
)
|
||||
url = url.decode("utf-8")
|
||||
url = url.strip()
|
||||
id = url.split("/")[-1]
|
||||
return id
|
||||
|
||||
|
||||
def gh_set_issue_labels(id, labels):
|
||||
subprocess.run(
|
||||
[
|
||||
"hub",
|
||||
"issue",
|
||||
"update",
|
||||
id,
|
||||
"-l",
|
||||
labels,
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def gh_close_issue(id, labels):
|
||||
subprocess.run(
|
||||
[
|
||||
"hub",
|
||||
"issue",
|
||||
"update",
|
||||
id,
|
||||
"--state",
|
||||
"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__":
|
||||
main()
|
|
@ -0,0 +1,34 @@
|
|||
name: Sync Launchpad issues to GitHub
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- master
|
||||
paths:
|
||||
- '.github/workflows/track-lp-issues.yaml'
|
||||
- '.github/lpbugtracker.py'
|
||||
schedule:
|
||||
- cron: '0 1 * * *'
|
||||
|
||||
jobs:
|
||||
add-lp-issues:
|
||||
name: Sync Launchpad issues to GitHub bug tracker
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
steps:
|
||||
# Checkout code
|
||||
- uses: actions/checkout@v3.0.2
|
||||
- name: Install Python 3
|
||||
uses: actions/setup-python@v4.2.0
|
||||
with:
|
||||
python-version: 3.6
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
pip install launchpadlib
|
||||
- name: Mirror GitHub bugs from Launchpad
|
||||
id: getlpbugs
|
||||
run: |
|
||||
python .github/lpbugtracker.py
|
||||
|
||||
env:
|
||||
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
|
Loading…
Reference in New Issue