#!/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() @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.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(f"{self.api_url}/projects/{self.project}/issues", params = payload, headers = self.auth_headers, timeout = TIMEOUT) return IssuePoster.check_req(post, name) 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" 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(f"{self.api_url}/repos/{self.project}/issues", json = payload, headers = self.auth_headers, timeout = TIMEOUT) return IssuePoster.check_req(post, name) 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 def first_item(self) -> dict[str, Any] | None: '''Get the first item of the feed (newest)''' feed = feedparser.parse(self.url) if len(feed.entries) == 0: return None return feed.entries[0] def read_feed(self) -> bool | None: '''Read a feed and post an issue if a new item is found''' entry = self.first_item() 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.''' try: with open(CONFIG_DIR + self.name, encoding="utf-8") as file: match = file.readline().strip("\n") == str(_id) except FileNotFoundError: match = False if match: return False if not self.target.post_issue(self.name, version, link): return False if not os.path.isdir(CONFIG_DIR): os.makedirs(CONFIG_DIR) with open(CONFIG_DIR + self.name, 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 for reader in FEED_READERS: reader.read_feed()