#!/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: if not os.path.isdir(CONFIG_DIR): os.mkdir(CONFIG_DIR) 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.")