issue_generator/issue_generator.py

#!/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
    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.")