From 1ddb82e63df21461f66ddcf12cdbe15f29ef2d54 Mon Sep 17 00:00:00 2001 From: Benex254 Date: Tue, 6 Aug 2024 13:24:52 +0300 Subject: [PATCH] update docs and move allanime utils to be accessible by all providers --- fastanime/libs/anime_provider/allanime/api.py | 87 ++++++++++++++++--- .../anime_provider/{allanime => }/utils.py | 34 ++------ 2 files changed, 85 insertions(+), 36 deletions(-) rename fastanime/libs/anime_provider/{allanime => }/utils.py (55%) diff --git a/fastanime/libs/anime_provider/allanime/api.py b/fastanime/libs/anime_provider/allanime/api.py index a0c9e38..014a80e 100644 --- a/fastanime/libs/anime_provider/allanime/api.py +++ b/fastanime/libs/anime_provider/allanime/api.py @@ -1,3 +1,8 @@ +"""a module that handles the scraping of allanime + +abstraction over allanime api +""" + import json import logging from typing import Iterator @@ -7,6 +12,7 @@ from requests.exceptions import Timeout from ....libs.anime_provider.allanime.types import AllAnimeEpisode from ....libs.anime_provider.types import Anime, Server from ...anime_provider.base_provider import AnimeProvider +from ..utils import decode_hex_string from .constants import ( ALLANIME_API_ENDPOINT, ALLANIME_BASE, @@ -15,7 +21,6 @@ from .constants import ( ) from .gql_queries import ALLANIME_EPISODES_GQL, ALLANIME_SEARCH_GQL, ALLANIME_SHOW_GQL from .normalizer import normalize_anime, normalize_search_results -from .utils import decode_hex_string Logger = logging.getLogger(__name__) @@ -31,6 +36,15 @@ class AllAnimeAPI(AnimeProvider): api_endpoint = ALLANIME_API_ENDPOINT def _fetch_gql(self, query: str, variables: dict): + """main abstraction over all requests to the allanime api + + Args: + query: [TODO:description] + variables: [TODO:description] + + Returns: + [TODO:return] + """ try: response = self.session.get( self.api_endpoint, @@ -63,6 +77,18 @@ class AllAnimeAPI(AnimeProvider): unknown=True, **kwargs, ): + """search for an anime title using allanime provider + + Args: + nsfw ([TODO:parameter]): [TODO:description] + unknown ([TODO:parameter]): [TODO:description] + user_query: [TODO:description] + translation_type: [TODO:description] + **kwargs: [TODO:args] + + Returns: + [TODO:return] + """ search = {"allowAdult": nsfw, "allowUnknown": unknown, "query": user_query} limit = 40 translationtype = translation_type @@ -83,6 +109,14 @@ class AllAnimeAPI(AnimeProvider): return {} def get_anime(self, allanime_show_id: str): + """get an anime details given its id + + Args: + allanime_show_id: [TODO:description] + + Returns: + [TODO:return] + """ variables = {"showId": allanime_show_id} try: anime = self._fetch_gql(ALLANIME_SHOW_GQL, variables) @@ -91,9 +125,19 @@ class AllAnimeAPI(AnimeProvider): Logger.error(f"FA(AllAnime): {e}") return None - def get_anime_episode( + def _get_anime_episode( self, allanime_show_id: str, episode_string: str, translation_type: str = "sub" ) -> AllAnimeEpisode | dict: + """get the episode details and sources info + + Args: + allanime_show_id: [TODO:description] + episode_string: [TODO:description] + translation_type: [TODO:description] + + Returns: + [TODO:return] + """ variables = { "showId": allanime_show_id, "translationType": translation_type, @@ -101,7 +145,7 @@ class AllAnimeAPI(AnimeProvider): } try: episode = self._fetch_gql(ALLANIME_EPISODES_GQL, variables) - return episode["episode"] # pyright: ignore + return episode["episode"] except Exception as e: Logger.error(f"FA(AllAnime): {e}") return {} @@ -109,18 +153,29 @@ class AllAnimeAPI(AnimeProvider): def get_episode_streams( self, anime: Anime, episode_number: str, translation_type="sub" ) -> Iterator[Server] | None: + """get the streams of an episode + + Args: + translation_type ([TODO:parameter]): [TODO:description] + anime: [TODO:description] + episode_number: [TODO:description] + + Yields: + [TODO:description] + """ anime_id = anime["id"] - allanime_episode = self.get_anime_episode( + allanime_episode = self._get_anime_episode( anime_id, episode_number, translation_type ) if not allanime_episode: - return {} + return [] embeds = allanime_episode["sourceUrls"] try: for embed in embeds: try: - # filter the working streams + # filter the working streams no need to get all since the others are mostly hsl + # TODO: should i just get all the servers and handle the hsl?? if embed.get("sourceName", "") not in ( "Sak", "Kir", @@ -218,7 +273,7 @@ if __name__ == "__main__": import subprocess import sys - from .utils import run_fzf + from InquirerPy import inquirer, validator anime = input("Enter the anime name: ") translation = input("Enter the translation type: ") @@ -232,7 +287,11 @@ if __name__ == "__main__": search_results = search_results["results"] options = {show["title"]: show for show in search_results} - anime = run_fzf(options.keys()) + anime = inquirer.fuzzy( + "Enter the anime title", + list(options.keys()), + validate=validator.EmptyInputValidator(), + ).execute() if anime is None: print("No anime was selected") sys.exit(1) @@ -248,7 +307,11 @@ if __name__ == "__main__": stream_link = True while stream_link != "quit": print("select episode") - episode = run_fzf(availableEpisodesDetail[translation.strip()]) + episode = inquirer.fuzzy( + "Choose an episode", + availableEpisodesDetail[translation.strip()], + validate=validator.EmptyInputValidator(), + ).execute() if episode is None: print("No episode was selected") sys.exit(1) @@ -267,7 +330,11 @@ if __name__ == "__main__": for server in episode_streams: stream_links.extend([link["link"] for link in server["links"]]) stream_links.append("back") - stream_link = run_fzf(stream_links) + stream_link = inquirer.fuzzy( + "Choose a link to stream", + stream_links, + validate=validator.EmptyInputValidator(), + ).execute() if stream_link == "quit": print("Have a nice day") sys.exit() diff --git a/fastanime/libs/anime_provider/allanime/utils.py b/fastanime/libs/anime_provider/utils.py similarity index 55% rename from fastanime/libs/anime_provider/allanime/utils.py rename to fastanime/libs/anime_provider/utils.py index 0f6419b..e98b1d5 100644 --- a/fastanime/libs/anime_provider/allanime/utils.py +++ b/fastanime/libs/anime_provider/utils.py @@ -1,5 +1,4 @@ import re -import subprocess # Dictionary to map hex values to characters hex_to_char = { @@ -36,6 +35,14 @@ hex_to_char = { def decode_hex_string(hex_string): + """some of the sources encrypt the urls into hex codes this function decrypts the urls + + Args: + hex_string ([TODO:parameter]): [TODO:description] + + Returns: + [TODO:return] + """ # Split the hex string into pairs of characters hex_pairs = re.findall("..", hex_string) @@ -43,28 +50,3 @@ def decode_hex_string(hex_string): decoded_chars = [hex_to_char.get(pair.lower(), pair) for pair in hex_pairs] return "".join(decoded_chars) - - -def run_fzf(options): - """ - Run fzf with a list of options and return the selected option. - """ - # Join the list of options into a single string with newlines - options_str = "\n".join(options) - - # Run fzf as a subprocess - result = subprocess.run( - ["fzf"], - input=options_str, - text=True, - stdout=subprocess.PIPE, - ) - - # Check if fzf was successful - if result.returncode == 0: - # Return the selected option - return result.stdout.strip() - else: - # Handle the case where fzf fails or is canceled - print("fzf was canceled or failed") - return None