From eb4228b698902015788437b607add973f5d25291 Mon Sep 17 00:00:00 2001 From: Carlos Galindo Date: Wed, 6 Sep 2023 23:33:19 +0200 Subject: [PATCH] initial commit --- README.md | 32 +++++++ issue_generator.py | 223 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 255 insertions(+) create mode 100644 README.md create mode 100755 issue_generator.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..4f806d1 --- /dev/null +++ b/README.md @@ -0,0 +1,32 @@ +# Updates issue generator + +A project to turn RSS update feeds into new issues in the projects that depend on them. + +## Usage + +Create a file called `config.py`, which should contain a list `FEED\_READERS`. The +list should contain one or more feed readers (`issue_generator.FeedReader`), which +in turn contain an issue poster (`issue_generator.IssuePoster`). + +The reader will check up on the RSS feed and compare against a local folder with the +last version of the software it saw (stored at `~/.config/issue_generator` or +`$XDG_CONFIG_DIR/issue_generator` by default). Then, if the version differs (it doesn't +sort them, just compare equality), it opens an issue with the corresponding issue poster. + +See the documentation of each feed reader and issue poster with `import issue_generator`, +`help(issue_generator)` to see how each parameter works. + +## Example configuration file + +Example: +```python +from issue_generator import GithubReader, GitlabPoster + +FEED_READERS = [ + # Post an issue on Gitlab.com for new releases of Gitea + GithubReader(name = "gitea", + project = "gitea-go/gitea", + target = GitlabPoster(project = 65536, # your project id here + token = "qwerty-secret-token")) +] +``` diff --git a/issue_generator.py b/issue_generator.py new file mode 100755 index 0000000..317542b --- /dev/null +++ b/issue_generator.py @@ -0,0 +1,223 @@ +#!/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()