Helper for downloading targets from github

github_download

#!/usr/bin/env python3

import argparse
import json
import logging
import sys
import urllib.request
from pprint import pprint
from typing import Any, List, Optional, Dict

# Set up the main logger for the script
LOGGER = logging.getLogger("getver")

# Configure logging to output to stdout with a specific format
consolehandler = logging.StreamHandler(sys.stdout)
consolehandler.setFormatter(logging.Formatter("%(name)s %(levelname)s:  %(message)s"))
consolehandler.setLevel(logging.INFO)
LOGGER.addHandler(consolehandler)

# Custom exception for handling GitHub release-related errors
class GitHubReleaseError(Exception):
    """Custom exception for GitHub release errors."""
    pass

class GitHubAPI:
    """A class to interact with GitHub's API for fetching release information."""

    # URL format string for GitHub API releases endpoint
    _github_api_url_fmt: str = "https://api.github.com/repos/{owner}/{repo}/releases"
    _github_api_url: Optional[str] = None  # Cached API URL
    owner: Optional[str] = None  # Repository owner
    repo: Optional[str] = None   # Repository name
    _log = logging.getLogger("getver.GitHubAPI")  # Logger for this class

    def __init__(self, owner: str, repo: str, release: Optional[str] = None) -> None:
        """Initialize the GitHubAPI with owner, repo, and optional release tag."""
        self.release = release
        self.owner = owner
        self.repo = repo
        self._log.debug(f"Initialized GitHubAPI for {self.owner}/{self.repo}")

    def callapi(self, url: str) -> Any:
        """
        Fetch release information from the given URL using urllib.

        Args:
            url (str): The URL to fetch the release information from.

        Returns:
            Any: Parsed JSON response containing release information.

        Raises:
            GitHubReleaseError: If there is an error fetching the release.
        """
        self._log.debug(f"Fetching release from {url}")
        try:
            with urllib.request.urlopen(url) as response:
                if response.status != 200:
                    self._log.error(f"Error fetching release: {response.status} - {response.reason}")
                    raise GitHubReleaseError(f"Error fetching release: {response.status} - {response.reason}")
                retv = json.load(response)
                self._log.debug(f"Fetched release data")
                return retv

        except urllib.error.HTTPError as e:
            # Attempt to extract error message from the response
            try:
                error_message = json.load(e)
                message = error_message.get("message", "Unknown error")
            except Exception:
                message = e.reason
            self._log.error(f"HTTP Error: {e.code} - {message}")
            raise GitHubReleaseError(f"HTTP Error: {e.code} - {message}")

        except Exception as e:
            self._log.error(f"Error: {e}")
            raise GitHubReleaseError(f"Error: {e}")

    @property
    def github_api_url(self) -> str:
        """
        Return the formatted GitHub API URL for releases.

        Returns:
            str: The GitHub API URL for the repository's releases.
        """
        if self._github_api_url is None:
            self._github_api_url = self._github_api_url_fmt.format(owner=self.owner, repo=self.repo)
        return self._github_api_url

    def get_source_url(self, tag: str = "latest") -> str:
        """
        Return the source URL for a specific tag or the latest release.

        Args:
            tag (str): The release tag or "latest".

        Returns:
            str: The URL to fetch the release information.
        """
        if tag == "latest":
            url = f"{self.github_api_url}/latest"
        else:
            url = f"{self.github_api_url}/tags/{tag}"
        return url

    @property
    def releases(self) -> List[str]:
        """
        Fetch all release tags for the repository.

        Returns:
            List[str]: A list of release tag names.
        """
        retv: List[str] = []
        for row in self.callapi(self.github_api_url):
            if row.get("tag_name"):
                retv.append(row["tag_name"])
        return retv

    def is_valid_tag(self, tag: str) -> bool:
        """
        Check if the provided tag is a valid release tag.

        Args:
            tag (str): The release tag to check.

        Returns:
            bool: True if the tag exists, False otherwise.
        """
        return tag in self.releases

    def get_release(self) -> Any:
        """
        Fetch release information for a specific tag or the latest release.

        Returns:
            Any: Parsed JSON response for the release.
        """
        url = self.get_source_url()
        if self.release is not None:
            url = self.get_source_url(self.release)
        return self.callapi(url)

    def browser_download_urls(self) -> List[str]:
        """
        Return a list of download URLs for the release assets.

        Returns:
            List[str]: List of browser download URLs for the assets.
        """
        release_info = self.get_release()
        if "assets" not in release_info:
            return []
        return [asset["browser_download_url"] for asset in release_info["assets"]]

    @staticmethod
    def match_url(url: str, filter: str) -> bool:
        """
        Check if the URL matches the provided filter.

        Args:
            url (str): The URL to check.
            filter (str): The filter string.

        Returns:
            bool: True if the filter is in the URL's filename, False otherwise.
        """
        name = url.split("/")[-1]
        if filter in name:
            return True
        return False

    def download_urls(self, **kwargs) -> List[str]:
        """
        Return a list of download URLs filtered by the provided filters.

        Keyword Args:
            filter (List[str]): List of filter strings to include.
            ignore (List[str]): List of filter strings to exclude.

        Returns:
            List[str]: Filtered list of download URLs.
        """
        filters: List[str] = kwargs.get("filter", [])
        ignore: List[str] = kwargs.get("ignore", [])

        urls: List[str] = self.browser_download_urls()
        filtered_urls: List[str] = urls[:]

        # Apply include filters
        if filters:
            for filter in filters:
                tmplist: List[str] = []
                for url in filtered_urls:
                    if self.match_url(url, filter):
                        tmplist.append(url)
                filtered_urls = tmplist
        # Apply ignore filters
        if ignore:
            for ignore_filter in ignore:
                filtered_urls = [url for url in filtered_urls if not self.match_url(url, ignore_filter)]

        return filtered_urls

    def download_url(self, **kwargs) -> str:
        """
        Return a single download URL filtered by the provided filters.

        Keyword Args:
            filter (List[str]): List of filter strings to include.
            ignore (List[str]): List of filter strings to exclude.

        Returns:
            str: The single matching download URL.

        Raises:
            GitHubReleaseError: If no or multiple URLs are found.
        """
        urls = self.download_urls(**kwargs)
        if len(urls) == 0:
            raise GitHubReleaseError("No download URLs found matching the specified filters.")
        if len(urls) > 1:
            self._log.warning("Multiple download URLs found:")
            for url in urls:
                self._log.warning(f"- {url}")
            raise GitHubReleaseError("Multiple download URLs found. Please refine your filters.")
        return urls[0]

    def download(self, **kwargs) -> None:
        """
        Download the file from the filtered download URL.

        Keyword Args:
            outputfile (str): The output file path to save the downloaded file.
            filter (List[str]): List of filter strings to include.
            ignore (List[str]): List of filter strings to exclude.

        Raises:
            GitHubReleaseError: If the download fails.
        """
        url = self.download_url(**kwargs)
        self._log.debug(f"Downloading from {url}")
        outputfile: Optional[str] = kwargs.get("outputfile", None)
        if outputfile is None:
            self._log.warning("No output file specified, using 'downloaded_file' as default.")
            outputfile = url.split("/")[-1]

        try:
            with urllib.request.urlopen(url) as response:
                if response.status != 200:
                    raise GitHubReleaseError(f"Error downloading file: {response.status} - {response.reason}")

                with open(outputfile, "wb") as f:
                    f.write(response.read())
                    self._log.info(f"File downloaded successfully to {outputfile}")

        except Exception as e:
            raise GitHubReleaseError(f"Error downloading file: {e}")

def main() -> None:
    """
    Main function to parse arguments and fetch GitHub release info.

    Handles command-line arguments for filtering, listing, and downloading releases.
    """
    # Set up argument parser for command-line options
    parser = argparse.ArgumentParser(description="Fetch GitHub release info.")
    parser.add_argument("-r", "--release", help="Fetch specific release by tag", dest="release")
    parser.add_argument(
        "-f",
        "--filter",
        dest="filter",
        action="append",
        default=[],
        metavar="FILTER",
        help="Filter output by name",
    )
    parser.add_argument(
        "-i",
        "--ignore",
        dest="ignore",
        action="append",
        default=[],
        metavar="IGNORE",
        help="Filter output output by name",
    )
    parser.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="Enable verbose output",
    )
    parser.add_argument(
        "-l",
        "--list",
        action="store_true",
        help="List releases",
    )
    parser.add_argument(
        "-o",
        "--outputfile",
        dest="outputfile",
        default=None,
        help="Output file to save the downloaded content",
    )
    parser.add_argument("owner", help="GitHub repo owner/org")
    parser.add_argument("repo", help="GitHub repository name")

    options = parser.parse_args()

    # Enable debug logging if verbose flag is set
    if options.verbose:
        LOGGER.setLevel(logging.DEBUG)
        LOGGER.debug("Verbose mode enabled")

    github_api = GitHubAPI(options.owner, options.repo, options.release)

    # Check if the specified release tag exists
    if options.release is not None and not github_api.is_valid_tag(options.release):
        raise GitHubReleaseError(f"Release {options.release not found in {options.owner}/{options.repo}")

    # List all releases if requested
    if options.list:
        LOGGER.info("Listing releases:")
        for release in github_api.releases:
            if not options.filter or any(f in release for f in options.filter):
                print(release)
        return

    # Download the release asset matching the filters
    github_api.download(filter=options.filter, ignore=options.ignore, outputfile=options.outputfile)

if __name__ == "__main__":
    # Entry point for the script; handle errors gracefully
    try:
        main()
    except GitHubReleaseError as e:
        LOGGER.error(f"GitHub Release Error: {e}")
        sys.exit(1)