290 lines
11 KiB
Python
Executable File
290 lines
11 KiB
Python
Executable File
#!/usr/bin/python3
|
|
'''
|
|
A utility to automatically create issues for new items in RSS feeds.
|
|
Its primary use is updating software packages when upstream dependencies
|
|
release new versions.
|
|
'''
|
|
|
|
import os
|
|
from typing import Any
|
|
import feedparser
|
|
import requests
|
|
|
|
|
|
CONFIG_DIR = os.getenv("XDG_CONFIG_DIR", os.getenv("HOME") + "/.config") + "/issue_generator/"
|
|
TIMEOUT = 5 # seconds
|
|
|
|
|
|
class IssuePoster:
|
|
'''Automated posting of issues to repositories.'''
|
|
|
|
def __init__(self, project: Any, token: str, assignee: Any, labels: Any):
|
|
'''Create a new IssuePoster.'''
|
|
self.project = project
|
|
self.token = token
|
|
self.assignee = assignee
|
|
self.labels = labels
|
|
|
|
def post_issue(self, name: str, version: str, link: str) -> bool:
|
|
'''Post a issue warning that version of software name has been published at link.'''
|
|
raise NotImplementedError()
|
|
|
|
def issue_exists(self, name: str, version: str) -> bool:
|
|
'''Checks whether the issue for the given name/version pair exists.'''
|
|
expected_title = IssuePoster.title(name, version)
|
|
get = requests.get(self.url,
|
|
headers = self.auth_headers,
|
|
timeout = TIMEOUT,
|
|
params = self.search_params(expected_title))
|
|
if not IssuePoster.check_req(get, name):
|
|
return False
|
|
data = get.json()
|
|
return type(data) == list and any(issue['title'] == expected_title for issue in data)
|
|
|
|
def search_params(self, expected_title: str) -> dict[str, Any]:
|
|
'''A parameter dictionary to query issues in different services.'''
|
|
raise NotImplementedError()
|
|
|
|
@staticmethod
|
|
def describe(name: str, version: str, link: str) -> str:
|
|
'''Generate a description for the issue.'''
|
|
return f"New release of {name}: [{version}]({link})" + \
|
|
"\n\nPlease update the corresponding package"
|
|
|
|
@staticmethod
|
|
def title(name: str, version: str) -> str:
|
|
'''Generate a title for the issue.'''
|
|
return f"Update {name} to {version}"
|
|
|
|
@staticmethod
|
|
def check_req(response: requests.Response, name: str) -> bool:
|
|
'''Check if the request succeeded.'''
|
|
if response.status_code // 100 != 2:
|
|
print(f"{response.status_code} when updating {name}, response reads:")
|
|
print(response.text)
|
|
return response.status_code // 100 == 2
|
|
|
|
|
|
class GitlabPoster(IssuePoster):
|
|
'''GitLab (MIST instance) issue poster'''
|
|
|
|
def __init__(self, project: int, token: str, instance: str = "https://gitlab.com",
|
|
assignee: int | None = None, labels: str = ""):
|
|
'''Create a Gitlab Issue Poster
|
|
- project: a project id
|
|
- token: API token for the specific project.
|
|
Go to Project > Settings > Access Tokens and create a new token
|
|
with Reporter role and `api` scope.
|
|
- instance: base url of the target instance, defaults to gitlab.com
|
|
- assignee: a user id, defaults to no user assigned
|
|
- labels: comma-separated list of labels, defaults to no label
|
|
'''
|
|
super().__init__(project, token, assignee, labels)
|
|
self.api_url = f"{instance}/api/v4"
|
|
self.url = f"{instance}/api/v4/projects/{project}/issues"
|
|
self.auth_headers = { 'PRIVATE-TOKEN': self.token }
|
|
|
|
def post_issue(self, name: str, version: str, link: str) -> bool:
|
|
payload = {
|
|
'description': IssuePoster.describe(name, version, link),
|
|
'title': IssuePoster.title(name, version),
|
|
'labels': self.labels,
|
|
}
|
|
if self.assignee:
|
|
payload['assignee_id'] = self.assignee
|
|
post = requests.post(self.url,
|
|
params = payload,
|
|
headers = self.auth_headers,
|
|
timeout = TIMEOUT)
|
|
return IssuePoster.check_req(post, name)
|
|
|
|
def search_params(self, expected_title: str) -> bool:
|
|
return { 'per_page': 1, 'search': expected_title, 'sort': 'asc', 'order_by': 'created_at' }
|
|
|
|
|
|
class ForgejoPoster(IssuePoster):
|
|
'''Forgejo (git.cgj.es instance) issue poster'''
|
|
|
|
def __init__(self, project: str, token: str, instance: str,
|
|
assignee: list[str] = None, labels: list[int] = None):
|
|
'''Create a new Forgejo Issue Poster
|
|
- project: a project path (string: owner/repo)
|
|
- token: access token (string)
|
|
Create by going to User Settings > Applications > Access Tokens.
|
|
Give it access to All and then give it read permissions on `issue`.
|
|
- instance: base url of the target instance
|
|
- assignee: a list of usernames
|
|
- labels: list of integer ids
|
|
You can check these ids by running self.list_labels(), or check
|
|
them at REPO_URL/labels (see open issues link).
|
|
'''
|
|
super().__init__(project, token, assignee, labels)
|
|
self.auth_headers = { 'Authorization': f"token {self.token}" }
|
|
self.api_url = f"{instance}/api/v1"
|
|
self.url = f"{instance}/api/v1/repos/{project}/issues"
|
|
|
|
def post_issue(self, name: str, version: str, link: str) -> bool:
|
|
payload = {
|
|
"assignees": self.assignee if self.assignee else [],
|
|
"body": IssuePoster.describe(name, version, link),
|
|
"title": IssuePoster.title(name, version),
|
|
"labels": self.labels if self.labels else [],
|
|
}
|
|
post = requests.post(self.url,
|
|
json = payload,
|
|
headers = self.auth_headers,
|
|
timeout = TIMEOUT)
|
|
return IssuePoster.check_req(post, name)
|
|
|
|
def search_params(self, expected_title: str) -> bool:
|
|
return { 'q': expected_title, 'state': 'all' }
|
|
|
|
def list_labels(self) -> dict[int, str] | None:
|
|
'''Lists the labels and their IDs for this repository.'''
|
|
get = requests.get(f"{self.api_url}/repos/{self.project}/labels",
|
|
headers = self.auth_headers,
|
|
timeout = TIMEOUT)
|
|
if get.status_code != 200:
|
|
print(f"Error {get.status_code} while reading labels, response was:\n{get.text}")
|
|
return None
|
|
res = {}
|
|
for label in get.json():
|
|
print(f"{label['id']}: {label['name']}")
|
|
res[label['id']] = label['name']
|
|
return res
|
|
|
|
|
|
class FeedReader:
|
|
'''Class to read an RSS/Atom feed and post an issue if a new version is found.'''
|
|
|
|
def __init__(self, name: str, url: str, target: IssuePoster):
|
|
'''Create a new feed reader'''
|
|
self.name = name
|
|
self.url = url
|
|
self.target = target
|
|
self.version_file = CONFIG_DIR + self.name
|
|
self.etag_file = CONFIG_DIR + self.name + ".etag"
|
|
self.beta_strings = [ "nightly", "beta", "alpha", "rc" ]
|
|
|
|
def first_item(self) -> dict[str, Any] | None | int:
|
|
'''Get the first item of the feed (newest)'''
|
|
if os.path.isfile(self.etag_file):
|
|
with open(self.etag_file, encoding="UTF-8") as file:
|
|
etag = file.readline()
|
|
else: etag = None
|
|
feed = feedparser.parse(self.url, etag=etag)
|
|
if feed.etag and feed.etag != etag:
|
|
with open(self.etag_file, mode='w', encoding="UTF-8") as file:
|
|
file.write(feed.etag)
|
|
if feed.status == 304:
|
|
return 304
|
|
if len(feed.entries) == 0:
|
|
return None
|
|
for entry in feed.entries:
|
|
skip = False
|
|
for beta in self.beta_strings:
|
|
if beta in self.entry_get_version(entry)[0]:
|
|
skip = True
|
|
break
|
|
if not skip:
|
|
return entry
|
|
return None
|
|
|
|
|
|
def read_feed(self) -> bool | None:
|
|
'''Read a feed and post an issue if a new item is found'''
|
|
entry = self.first_item()
|
|
if entry == 304:
|
|
return False
|
|
if not entry:
|
|
return None
|
|
version, _id = self.entry_get_version(entry)
|
|
return self.match_post_save(version, _id, entry.link)
|
|
|
|
def entry_get_version(self, entry: dict[str, Any]) -> tuple[str, str]:
|
|
'''Obtain the version of a given RSS entry.'''
|
|
raise NotImplementedError()
|
|
|
|
def entry_get_link(self, entry: dict[str, Any]) -> str:
|
|
'''Obtain the release link of a given RSS entry.'''
|
|
raise NotImplementedError()
|
|
|
|
def match_post_save(self, version: str, _id: str, link: str) -> bool:
|
|
'''Checks if a version is new, posts an issue and saves it as such.
|
|
If the version is not new, or the posting fails, the method stops and returns False.'''
|
|
# Match 1: with local file
|
|
try:
|
|
with open(self.version_file, encoding="utf-8") as file:
|
|
match = file.readline().strip("\n") == str(_id)
|
|
except FileNotFoundError:
|
|
match = False
|
|
if match:
|
|
return False
|
|
# Match 2: with repository issues
|
|
if not self.target.issue_exists(self.name, version):
|
|
# Post the issue
|
|
if not self.target.post_issue(self.name, version, link):
|
|
return False
|
|
# Save to disk
|
|
if not os.path.isdir(CONFIG_DIR):
|
|
os.makedirs(CONFIG_DIR)
|
|
with open(self.version_file, mode="w", encoding="utf-8") as file:
|
|
file.write(str(_id))
|
|
return True
|
|
|
|
|
|
class PIPYReader(FeedReader):
|
|
'''Reader specialized in the PIPY repository.'''
|
|
|
|
def __init__(self, name: str, package: str, target: IssuePoster):
|
|
'''Create a new PIPY reader for the given package.'''
|
|
super().__init__(name, f"https://pypi.org/rss/project/{package}/releases.xml", target)
|
|
|
|
def entry_get_version(self, entry: dict[str, Any]) -> tuple[str, str]:
|
|
return entry.title, entry.title
|
|
|
|
def entry_get_link(self, entry: dict[str, Any]) -> str:
|
|
return entry.link
|
|
|
|
|
|
class GithubTagReader(FeedReader):
|
|
'''Reader specialized in GitHub Tags Atom feed.'''
|
|
|
|
def __init__(self, name: str, project: str, target: IssuePoster):
|
|
'''Create a new GitHub Tags reader for the given project.'''
|
|
super().__init__(name, f"https://github.com/{project}/tags.atom", target)
|
|
|
|
def entry_get_version(self, entry: dict[str, Any]) -> tuple[str, str]:
|
|
return entry.title, str(entry.id)
|
|
|
|
def entry_get_link(self, entry: dict[str, Any]) -> str:
|
|
return entry.link
|
|
|
|
|
|
class GithubReader(GithubTagReader):
|
|
'''Reader specialized in GitHub releases' Atom feed.'''
|
|
|
|
def __init__(self, name: str, project: str, target: IssuePoster):
|
|
super().__init__(name, project, target)
|
|
self.url = f"https://github.com/{project}/releases.atom"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
from config import FEED_READERS
|
|
fail = 0
|
|
ok = 0
|
|
posted = 0
|
|
for reader in FEED_READERS:
|
|
res = reader.read_feed()
|
|
if res == None:
|
|
print(f"No versions of {reader.name} available.")
|
|
fail += 1
|
|
elif res == False:
|
|
ok += 1
|
|
elif res == True:
|
|
print(f"{reader.name} outdated, issue posted!")
|
|
posted += 1
|
|
else:
|
|
print(f"{reader.name} read with status {res}")
|
|
print(f"I've added {posted} new issues, {fail} items have no versions and {ok} items are up to date.")
|