From d471ce607bd4ed4bf9967e367c6af8179d5f3ec5 Mon Sep 17 00:00:00 2001 From: cda2 Date: Tue, 11 Feb 2025 11:14:33 +0900 Subject: [PATCH] feat: Implement specified version installation feature --- pythonlib/camoufox/__main__.py | 22 +++-- pythonlib/camoufox/pkgman.py | 170 +++++++++++++++++++++++++++------ 2 files changed, 157 insertions(+), 35 deletions(-) diff --git a/pythonlib/camoufox/__main__.py b/pythonlib/camoufox/__main__.py index 8514c62..320195a 100644 --- a/pythonlib/camoufox/__main__.py +++ b/pythonlib/camoufox/__main__.py @@ -27,11 +27,11 @@ class CamoufoxUpdate(CamoufoxFetcher): Checks & updates Camoufox """ - def __init__(self) -> None: + def __init__(self, specified_version: Optional[str] = None) -> None: """ Initializes the CamoufoxUpdate class """ - super().__init__() + super().__init__(specified_version=specified_version) self.current_verstr: Optional[str] try: self.current_verstr = installed_verstr() @@ -53,7 +53,10 @@ class CamoufoxUpdate(CamoufoxFetcher): """ # Check if the version is the same as the latest available version if not self.is_updated_needed(): - rprint("Camoufox binaries up to date!", fg="green") + if not self.specified_version: + rprint("Camoufox binaries up to date!", fg="green") + else: + rprint("Target Camoufox binaries already installed!", fg="green") rprint(f"Current version: v{self.current_verstr}", fg="green") return @@ -77,13 +80,18 @@ def cli() -> None: @cli.command(name='fetch') @click.option( - '--browserforge', is_flag=True, help='Update browserforge\'s header and fingerprint definitions' + '--browserforge', is_flag=True, + help='Update browserforge\'s header and fingerprint definitions' ) -def fetch(browserforge=False) -> None: +@click.option( + '--version', type=str, default=None, + help='Download a specific release version instead of the latest' +) +def fetch(browserforge: bool, version: Optional[str]) -> None: """ - Fetch the latest version of Camoufox and optionally update Browserforge's database + Fetch the latest or specified version of Camoufox. """ - CamoufoxUpdate().update() + CamoufoxUpdate(specified_version=version).update() # Fetch the GeoIP database if ALLOW_GEOIP: download_mmdb() diff --git a/pythonlib/camoufox/pkgman.py b/pythonlib/camoufox/pkgman.py index b6f0c3f..fb82f35 100644 --- a/pythonlib/camoufox/pkgman.py +++ b/pythonlib/camoufox/pkgman.py @@ -9,7 +9,7 @@ from dataclasses import dataclass from functools import total_ordering from io import BufferedWriter, BytesIO from pathlib import Path -from typing import Any, Dict, List, Literal, Optional, Tuple, Union +from typing import Any, Callable, Dict, List, Literal, Optional, Tuple, Union, TypedDict from zipfile import ZipFile import click @@ -137,6 +137,44 @@ class Version: VERSION_MIN, VERSION_MAX = Version.build_minmax() +class GithubRelease(TypedDict): + id: int + noded_id: str + name: str + tag_name: str + author: Dict + target_commitish: str + draft: bool + prerelease: bool + created_at: str + published_at: str + assets: List[Dict] + url: str + assets_url: str + upload_url: str + html_url: str + tarball_url: str + zipball_url: str + body: str + reactions: Dict + + +class GithubAsset(TypedDict): + id: int + node_id: str + name: str + label: str + uploader: dict + browser_download_url: str + content_type: str + state: str + size: int + download_count: int + created_at: str + updated_at: str + browser_download_url: str + + class GitHubDownloader: """ Manages fetching and installing GitHub releases. @@ -146,17 +184,40 @@ class GitHubDownloader: self.github_repo = github_repo self.api_url = f"https://api.github.com/repos/{github_repo}/releases" - def check_asset(self, asset: Dict) -> Any: + def fetch_all_releases(self, per_page: int = 100) -> List[GithubRelease]: + """ + Internal function to iterate through GitHub release pages. + """ + releases_all = [] + page = 1 + while True: + url = f"{self.api_url}?page={page}&per_page={per_page}" + resp = requests.get(url, timeout=20) + resp.raise_for_status() + releases_page = resp.json() + if not releases_page: + break + releases_all.extend(releases_page) + page += 1 + return releases_all + + def _default_predicate(self, asset: GithubAsset) -> str: + return asset.get('browser_download_url') + + def check_asset( + self, + asset: Dict, + predicate: Optional[Callable[[GithubAsset], Optional[Tuple[Version, str]]]] = None + ) -> Optional[str]: """ Compare the asset to determine if it's the desired asset. - - Args: - asset: Asset information from GitHub API - - Returns: - Any: Data to be returned if this is the desired asset, or None/False if not + If predicate is provided, it is applied to the asset; otherwise, + the default predicate is used. """ - return asset.get('browser_download_url') + + if predicate is None: + predicate = self._default_predicate + return predicate(asset) def missing_asset_error(self) -> None: """ @@ -164,19 +225,24 @@ class GitHubDownloader: """ raise MissingRelease(f"Could not find a release asset in {self.github_repo}.") - def get_asset(self) -> Any: + def get_asset( + self, + predicate: Optional[Callable[[GithubAsset], Optional[str]]] = None + ) -> Any: """ Fetch the latest release from the GitHub API. - Gets the first asset that returns a truthy value from check_asset. + Iterates over all pages and returns the first asset for which + check_asset (with the predicate) returns a truthy value. """ - resp = requests.get(self.api_url, timeout=20) - resp.raise_for_status() - releases = resp.json() + if predicate is None: + predicate = self._default_predicate + # Search through releases for the first supported version + releases = self.fetch_all_releases() for release in releases: - for asset in release['assets']: - if data := self.check_asset(asset): + for asset in release.get('assets', []): + if data := self.check_asset(asset, predicate=predicate): return data self.missing_asset_error() @@ -187,26 +253,21 @@ class CamoufoxFetcher(GitHubDownloader): Handles fetching and installing the latest version of Camoufox. """ - def __init__(self) -> None: + def __init__(self, specified_version: Optional[str] = None) -> None: super().__init__("daijro/camoufox") - + self.specified_version = specified_version self.arch = self.get_platform_arch() self._version_obj: Optional[Version] = None self.pattern: re.Pattern = re.compile( rf'camoufox-(?P.+)-(?P.+)-{OS_NAME}\.{self.arch}\.zip' ) - self.fetch_latest() + if self.specified_version: + self.fetch_specific(self.specified_version) + else: + self.fetch_latest() - def check_asset(self, asset: Dict) -> Optional[Tuple[Version, str]]: - """ - Finds the latest release from a GitHub releases API response that - supports the Camoufox version constraints, the OS, and architecture. - - Returns: - Optional[Tuple[Version, str]]: The version and URL of a release - """ - # Search through releases for the first supported version + def _default_predicate(self, asset: Dict) -> Optional[Tuple[Version, str]]: match = self.pattern.match(asset['name']) if not match: return None @@ -219,6 +280,22 @@ class CamoufoxFetcher(GitHubDownloader): # Asset was found. Return data return version, asset['browser_download_url'] + def check_asset( + self, + asset: Dict, + predicate: Optional[Callable[[Dict], Optional[Tuple[Version, str]]]] = None + ) -> Optional[Tuple[Version, str]]: + """ + Finds the latest or specified release from a GitHub releases API response that + supports the Camoufox version constraints, the OS, and architecture. + + Returns: + Optional[Tuple[Version, str]]: The version and URL of a release + """ + + if checked_result := super().check_asset(asset, predicate): + return checked_result + def missing_asset_error(self) -> None: """ Raise a MissingRelease exception if no release is found. @@ -254,6 +331,15 @@ class CamoufoxFetcher(GitHubDownloader): return arch + def convert_asset_to_version(self, asset: GithubAsset) -> Version: + """ + Convert an github release asset info to a Version object. + """ + match = self.pattern.match(asset['name']) + if not match: + raise ValueError(f"Invalid asset name: {asset['name']}") + return Version(release=match['release'], version=match['version']) + def fetch_latest(self) -> None: """ Fetch the URL of the latest camoufox release for the current platform. @@ -268,6 +354,34 @@ class CamoufoxFetcher(GitHubDownloader): # Set the version and URL self._version_obj, self._url = release_data + def fetch_specific(self, version: str) -> None: + """ + Fetch the URL of a specific camoufox release for the current platform. + Sets the version, release, and url properties. + + Args: + version (str): The version to fetch + + Raises: + requests.RequestException: If there's an error fetching release data + ValueError: If no matching release is found for the current platform + """ + + def _find_specific_version_predicate(asset: Dict) -> Optional[tuple[Version, str]]: + try: + candidate_version = self.convert_asset_to_version(asset) + except ValueError: + return None + + if candidate_version.full_string == version: + return candidate_version, asset['browser_download_url'] + return None + + # get_asset will raise a MissingRelease exception if no release is found + specific_version, download_url = self.get_asset(_find_specific_version_predicate) + self._version_obj, self._url = specific_version, download_url + + @staticmethod def download_file(file: DownloadBuffer, url: str) -> DownloadBuffer: """