diff --git a/app/controllers/Releases.py b/app/controllers/Releases.py index e2bc7ed..ce9bfcd 100644 --- a/app/controllers/Releases.py +++ b/app/controllers/Releases.py @@ -7,49 +7,53 @@ from app.utils.HTTPXClient import HTTPXClient class Releases: - + """Implements the methods required to get the latest releases and patches from revanced repositories.""" - + uvloop.install() - + httpx_client = HTTPXClient.create() - + async def __get_release(self, repository: str) -> list: - # Get assets from latest release in a given repository. - # - # Args: - # repository (str): Github's standard username/repository notation - # - # Returns: - # dict: dictionary of filename and download url - + """Get assets from latest release in a given repository. + + Args: + repository (str): Github's standard username/repository notation + + Returns: + dict: dictionary of filename and download url + """ + assets: list = [] response = await self.httpx_client.get(f"https://api.github.com/repos/{repository}/releases/latest") - + if response.status_code == 200: release_assets: dict = response.json()['assets'] release_version: str = response.json()['tag_name'] release_tarball: str = response.json()['tarball_url'] release_timestamp: str = response.json()['published_at'] - - if release_assets: - for asset in release_assets: - assets.append({ 'repository': repository, - 'version': release_version, - 'timestamp': asset['updated_at'], - 'name': asset['name'], - 'size': asset['size'], - 'browser_download_url': asset['browser_download_url'], - 'content_type': asset['content_type'] - }) - else: - assets.append({ 'repository': repository, - 'version': release_version, - 'timestamp': release_timestamp, - 'name': f"{repository.split('/')[1]}-{release_version}.tar.gz", - 'browser_download_url': release_tarball, - 'content_type': 'application/gzip' - }) + + async def get_asset_data(asset: dict) -> dict: + return {'repository': repository, + 'version': release_version, + 'timestamp': asset['updated_at'], + 'name': asset['name'], + 'size': asset['size'], + 'browser_download_url': asset['browser_download_url'], + 'content_type': asset['content_type'] + } + + no_release_assets_data: dict = {'repository': repository, + 'version': release_version, + 'timestamp': release_timestamp, + 'name': f"{repository.split('/')[1]}-{release_version}.tar.gz", + 'browser_download_url': release_tarball, + 'content_type': 'application/gzip' + } + + assets = [map(get_asset_data, release_assets) + if release_assets else no_release_assets_data] + return assets async def get_latest_releases(self, repositories: list) -> dict: @@ -61,58 +65,60 @@ class Releases: Returns: dict: A dictionary containing assets from each repository """ - + releases: dict[str, list] = {} releases['tools'] = [] - + results: list = await asyncio.gather(*[self.__get_release(repository) for repository in repositories]) - - for result in results: - for asset in result: - releases['tools'].append(asset) - + + releases['tools'] = [asset for result in results for asset in result] + return releases - + async def __get_patches_json(self) -> dict: - # Get revanced-patches repository's README.md. - # - # Returns: - # dict: JSON content - # - + """Get revanced-patches repository's README.md. + + Returns: + dict: JSON content + """ + response = await self.httpx_client.get(f"https://api.github.com/repos/revanced/revanced-patches/contents/patches.json") - content = orjson.loads(b64decode(response.json()['content']).decode('utf-8')) - + content = orjson.loads( + b64decode(response.json()['content']).decode('utf-8')) + return content - + async def get_patches_json(self) -> dict: """Get patches.json from revanced-patches repository. - + Returns: dict: Patches available for a given app """ + patches: dict = await self.__get_patches_json() - + return patches async def __get_contributors(self, repository: str) -> list: - # Get contributors from a given repository. - # - # Args: - # repository (str): Github's standard username/repository notation - # - # Returns: - # list: a list of dictionaries containing the repository's contributors - + """Get contributors from a given repository. + + Args: + repository (str): Github's standard username/repository notation + + Returns: + list: a list of dictionaries containing the repository's contributors + """ + keep: set = {'login', 'avatar_url', 'html_url'} - + response = await self.httpx_client.get(f"https://api.github.com/repos/{repository}/contributors") - + + # Looping over each contributor, filtering each contributor so that + # keyfilter() returns a dictionary with only the key-value pairs that are in the "keep" set. contributors: list = [keyfilter(lambda k: k in keep, contributor) for contributor in response.json()] - - + return contributors - + async def get_contributors(self, repositories: list) -> dict: """Runs get_contributors() asynchronously for each repository. @@ -122,22 +128,23 @@ class Releases: Returns: dict: A dictionary containing the contributors from each repository """ - + contributors: dict[str, list] - + contributors = {} contributors['repositories'] = [] - - revanced_repositories = [repository for repository in repositories if 'revanced' in repository] - + + revanced_repositories = [ + repository for repository in repositories if 'revanced' in repository] + results: list[dict] = await asyncio.gather(*[self.__get_contributors(repository) for repository in revanced_repositories]) - + for key, value in zip(revanced_repositories, results): - data = { 'name': key, 'contributors': value } + data = {'name': key, 'contributors': value} contributors['repositories'].append(data) - + return contributors - + async def get_commits(self, org: str, repository: str, path: str) -> dict: """Get commit history from a given repository. @@ -154,28 +161,28 @@ class Releases: Returns: dict: a dictionary containing the repository's latest commits """ - + payload: dict = {} payload["repository"] = f"{org}/{repository}" payload["path"] = path payload["commits"] = [] - + if org == 'revanced' or org == 'vancedapp': _releases = await self.httpx_client.get( f"https://api.github.com/repos/{org}/{repository}/releases?per_page=2" - ) - + ) + releases = _releases.json() - + since = releases[1]['created_at'] until = releases[0]['created_at'] - + _response = await self.httpx_client.get( f"https://api.github.com/repos/{org}/{repository}/commits?path={path}&since={since}&until={until}" - ) - + ) + response = _response.json() - + for commit in response: data: dict[str, str] = {} data["sha"] = commit["sha"] @@ -183,7 +190,7 @@ class Releases: data["message"] = commit["commit"]["message"] data["html_url"] = commit["html_url"] payload['commits'].append(data) - + return payload else: raise Exception("Invalid organization.")