update docs and move allanime utils to be accessible by all providers

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()

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