diff --git a/fastanime/AnimeProvider.py b/fastanime/AnimeProvider.py index d40fd16..22a79c7 100644 --- a/fastanime/AnimeProvider.py +++ b/fastanime/AnimeProvider.py @@ -4,11 +4,15 @@ """ import logging -from typing import Iterator +from typing import TYPE_CHECKING, Iterator -from .libs.anilist.anilist_data_schema import AnilistBaseMediaDataSchema from .libs.anime_provider import anime_sources -from .libs.anime_provider.types import Anime, SearchResults, Server + +if TYPE_CHECKING: + from typing import Union + + from .libs.anilist.anilist_data_schema import AnilistBaseMediaDataSchema + from .libs.anime_provider.types import Anime, SearchResults, Server logger = logging.getLogger(__name__) @@ -42,10 +46,10 @@ class AnimeProvider: self, user_query, translation_type, - anilist_obj: AnilistBaseMediaDataSchema | None = None, + anilist_obj: "Union[AnilistBaseMediaDataSchema ,None]" = None, nsfw=True, unknown=True, - ) -> SearchResults | None: + ) -> "SearchResults | None": """core abstraction over all providers search functionality Args: @@ -69,8 +73,10 @@ class AnimeProvider: return results def get_anime( - self, anime_id: str, anilist_obj: AnilistBaseMediaDataSchema | None = None - ) -> Anime | None: + self, + anime_id: str, + anilist_obj: "Union[AnilistBaseMediaDataSchema,None]" = None, + ) -> "Anime | None": """core abstraction over getting info of an anime from all providers Args: @@ -93,8 +99,8 @@ class AnimeProvider: anime, episode: str, translation_type: str, - anilist_obj: AnilistBaseMediaDataSchema | None = None, - ) -> Iterator[Server] | None: + anilist_obj: "Union[AnilistBaseMediaDataSchema,None]" = None, + ) -> Iterator["Server"] | None: """core abstractions for getting juicy streams from all providers Args: diff --git a/fastanime/cli/commands/anilist/completed.py b/fastanime/cli/commands/anilist/completed.py index 80e09ef..3ca0c88 100644 --- a/fastanime/cli/commands/anilist/completed.py +++ b/fastanime/cli/commands/anilist/completed.py @@ -1,15 +1,18 @@ +from typing import TYPE_CHECKING + import click -from fastanime.cli.config import Config -from fastanime.cli.interfaces import anilist_interfaces -from fastanime.cli.utils.tools import QueryDict, exit_app - -from ....anilist import AniList +if TYPE_CHECKING: + from fastanime.cli.config import Config @click.command(help="View anime you completed") @click.pass_obj -def completed(config: Config): +def completed(config: "Config"): + from ....anilist import AniList + from ...interfaces import anilist_interfaces + from ...utils.tools import QueryDict, exit_app + if not config.user: print("Not authenticated") print("Please run: fastanime anilist loggin") diff --git a/fastanime/cli/commands/anilist/dropped.py b/fastanime/cli/commands/anilist/dropped.py index 0844916..17125b6 100644 --- a/fastanime/cli/commands/anilist/dropped.py +++ b/fastanime/cli/commands/anilist/dropped.py @@ -1,15 +1,19 @@ +from typing import TYPE_CHECKING + import click -from fastanime.cli.config import Config from fastanime.cli.interfaces import anilist_interfaces from fastanime.cli.utils.tools import QueryDict, exit_app from ....anilist import AniList +if TYPE_CHECKING: + from fastanime.cli.config import Config + @click.command(help="View anime you dropped") @click.pass_obj -def dropped(config: Config): +def dropped(config: "Config"): if not config.user: print("Not authenticated") print("Please run: fastanime anilist loggin") diff --git a/fastanime/cli/commands/anilist/login.py b/fastanime/cli/commands/anilist/login.py index e331841..da5d568 100644 --- a/fastanime/cli/commands/anilist/login.py +++ b/fastanime/cli/commands/anilist/login.py @@ -1,18 +1,21 @@ import webbrowser +from typing import TYPE_CHECKING import click from rich import print from rich.prompt import Confirm, Prompt from ....anilist import AniList -from ...config import Config from ...utils.tools import exit_app +if TYPE_CHECKING: + from ...config import Config + @click.command(help="Login to your anilist account") @click.option("--status", "-s", help="Whether you are logged in or not", is_flag=True) @click.pass_obj -def login(config: Config, status): +def login(config: "Config", status): if status: is_logged_in = True if config.user else False message = ( diff --git a/fastanime/cli/interfaces/anilist_interfaces.py b/fastanime/cli/interfaces/anilist_interfaces.py index 9bc7a5a..24fc5ad 100644 --- a/fastanime/cli/interfaces/anilist_interfaces.py +++ b/fastanime/cli/interfaces/anilist_interfaces.py @@ -3,6 +3,7 @@ from __future__ import annotations import os import random from datetime import datetime +from typing import TYPE_CHECKING from InquirerPy import inquirer from InquirerPy.validator import EmptyInputValidator @@ -12,18 +13,20 @@ from rich.prompt import Confirm, Prompt from ...anilist import AniList from ...constants import USER_CONFIG_PATH -from ...libs.anilist.anilist_data_schema import AnilistBaseMediaDataSchema -from ...libs.anime_provider.types import Anime, SearchResult, Server from ...libs.fzf import fzf from ...libs.rofi import Rofi from ...Utility.data import anime_normalizer from ...Utility.utils import anime_title_percentage_match, sanitize_filename -from ..config import Config from ..utils.mpv import mpv from ..utils.tools import QueryDict, exit_app from ..utils.utils import clear, fuzzy_inquirer from .utils import aniskip +if TYPE_CHECKING: + from ...libs.anilist.anilist_data_schema import AnilistBaseMediaDataSchema + from ...libs.anime_provider.types import Anime, SearchResult, Server + from ..config import Config + def calculate_time_delta(start_time, end_time): time_format = "%H:%M:%S" @@ -38,7 +41,7 @@ def calculate_time_delta(start_time, end_time): return delta -def player_controls(config: Config, anilist_config: QueryDict): +def player_controls(config: "Config", anilist_config: QueryDict): # user config config.translation_type.lower() @@ -56,7 +59,7 @@ def player_controls(config: Config, anilist_config: QueryDict): fetch_streams(config, anilist_config) def _replay(): - selected_server: Server = anilist_config.current_server + selected_server: "Server" = anilist_config.current_server print( "[bold magenta]Now Replaying:[/]", anime_title, @@ -233,7 +236,7 @@ def player_controls(config: Config, anilist_config: QueryDict): options[action]() -def fetch_streams(config: Config, anilist_config: QueryDict): +def fetch_streams(config: "Config", anilist_config: QueryDict): # user config quality: int = config.quality @@ -241,7 +244,7 @@ def fetch_streams(config: Config, anilist_config: QueryDict): episode_number: str = anilist_config.episode_number anime_title: str = anilist_config.anime_title anime_id: int = anilist_config.anime_id - anime: Anime = anilist_config.anime + anime: "Anime" = anilist_config.anime translation_type = config.translation_type anime_provider = config.anime_provider @@ -375,7 +378,7 @@ def fetch_streams(config: Config, anilist_config: QueryDict): player_controls(config, anilist_config) -def fetch_episode(config: Config, anilist_config: QueryDict): +def fetch_episode(config: "Config", anilist_config: QueryDict): # user config translation_type: str = config.translation_type.lower() continue_from_history: bool = config.continue_from_history @@ -384,9 +387,9 @@ def fetch_episode(config: Config, anilist_config: QueryDict): anime_title: str = anilist_config.anime_title # internal config - anime: Anime = anilist_config.anime - _anime: SearchResult = anilist_config._anime - selected_anime_anilist: AnilistBaseMediaDataSchema = ( + anime: "Anime" = anilist_config.anime + _anime: "SearchResult" = anilist_config._anime + selected_anime_anilist: "AnilistBaseMediaDataSchema" = ( anilist_config.selected_anime_anilist ) # prompt for episode number @@ -437,7 +440,7 @@ def fetch_episode(config: Config, anilist_config: QueryDict): def fetch_anime_episode(config, anilist_config: QueryDict): - selected_anime: SearchResult = anilist_config._anime + selected_anime: "SearchResult" = anilist_config._anime anime_provider = config.anime_provider with Progress() as progress: progress.add_task("Fetching Anime Info...", total=None) @@ -459,14 +462,14 @@ def fetch_anime_episode(config, anilist_config: QueryDict): fetch_episode(config, anilist_config) -def provide_anime(config: Config, anilist_config: QueryDict): +def provide_anime(config: "Config", anilist_config: QueryDict): # user config translation_type = config.translation_type.lower() # internal config selected_anime_title = anilist_config.selected_anime_title - anime_data: AnilistBaseMediaDataSchema = anilist_config.selected_anime_anilist + anime_data: "AnilistBaseMediaDataSchema" = anilist_config.selected_anime_anilist anime_provider = config.anime_provider # search and get the requested title from provider @@ -529,12 +532,12 @@ def provide_anime(config: Config, anilist_config: QueryDict): def anilist_options(config, anilist_config: QueryDict): - selected_anime: AnilistBaseMediaDataSchema = anilist_config.selected_anime_anilist + selected_anime: "AnilistBaseMediaDataSchema" = anilist_config.selected_anime_anilist selected_anime_title: str = anilist_config.selected_anime_title progress = (selected_anime["mediaListEntry"] or {"progress": 0}).get("progress", 0) episodes_total = selected_anime["episodes"] or "Inf" - def _watch_trailer(config: Config, anilist_config: QueryDict): + def _watch_trailer(config: "Config", anilist_config: QueryDict): if trailer := selected_anime.get("trailer"): trailer_url = "https://youtube.com/watch?v=" + trailer["id"] print("[bold magenta]Watching Trailer of:[/]", selected_anime_title) @@ -552,7 +555,7 @@ def anilist_options(config, anilist_config: QueryDict): exit(0) anilist_options(config, anilist_config) - def _add_to_list(config: Config, anilist_config: QueryDict): + def _add_to_list(config: "Config", anilist_config: QueryDict): # config.update_anime_list(anilist_config.anime_id) anime_lists = { "Watching": "CURRENT", @@ -589,7 +592,7 @@ def anilist_options(config, anilist_config: QueryDict): input("Enter to continue...") anilist_options(config, anilist_config) - def _score_anime(config: Config, anilist_config: QueryDict): + def _score_anime(config: "Config", anilist_config: QueryDict): if config.use_rofi: score = Rofi.ask("Enter Score", is_int=True) score = max(100, min(0, score)) @@ -612,7 +615,7 @@ def anilist_options(config, anilist_config: QueryDict): input("Enter to continue...") anilist_options(config, anilist_config) - def _remove_from_list(config: Config, anilist_config: QueryDict): + def _remove_from_list(config: "Config", anilist_config: QueryDict): if Confirm.ask( f"Are you sure you want to procede, the folowing action will permanently remove {selected_anime_title} from your list and your progress will be erased", default=False, @@ -630,7 +633,7 @@ def anilist_options(config, anilist_config: QueryDict): input("Enter to continue...") anilist_options(config, anilist_config) - def _change_translation_type(config: Config, anilist_config: QueryDict): + def _change_translation_type(config: "Config", anilist_config: QueryDict): # prompt for new translation type options = ["Sub", "Dub"] if config.use_fzf: @@ -733,12 +736,12 @@ def anilist_options(config, anilist_config: QueryDict): options[action](config, anilist_config) -def select_anime(config: Config, anilist_config: QueryDict): +def select_anime(config: "Config", anilist_config: QueryDict): search_results = anilist_config.data["data"]["Page"]["media"] anime_data = {} for anime in search_results: - anime: AnilistBaseMediaDataSchema + anime: "AnilistBaseMediaDataSchema" progress = (anime["mediaListEntry"] or {"progress": 0}).get("progress", 0) episodes_total = anime["episodes"] or "Inf" title = str( @@ -746,7 +749,11 @@ def select_anime(config: Config, anilist_config: QueryDict): ) title = sanitize_filename(f"{title} ({progress} of {episodes_total})") # Check if the anime is currently airing and has new/unwatched episodes - if anime["status"] == "RELEASING" and anime["nextAiringEpisode"] and progress > 0: + if ( + anime["status"] == "RELEASING" + and anime["nextAiringEpisode"] + and progress > 0 + ): last_aired_episode = anime["nextAiringEpisode"]["episode"] - 1 if last_aired_episode - progress > 0: title += f" 🔹{last_aired_episode - progress} new episode(s)🔹" @@ -791,7 +798,7 @@ def select_anime(config: Config, anilist_config: QueryDict): anilist(config, anilist_config) return - selected_anime: AnilistBaseMediaDataSchema = anime_data[selected_anime_title] + selected_anime: "AnilistBaseMediaDataSchema" = anime_data[selected_anime_title] anilist_config.selected_anime_anilist = selected_anime anilist_config.selected_anime_title = ( selected_anime["title"]["romaji"] or selected_anime["title"]["english"] @@ -801,7 +808,7 @@ def select_anime(config: Config, anilist_config: QueryDict): anilist_options(config, anilist_config) -def handle_animelist(anilist_config, config: Config, list_type: str): +def handle_animelist(anilist_config, config: "Config", list_type: str): if not config.user: if not config.use_rofi: print("You haven't logged in please run: fastanime anilist login") @@ -854,7 +861,7 @@ def handle_animelist(anilist_config, config: Config, list_type: str): return anime_list -def anilist(config: Config, anilist_config: QueryDict): +def anilist(config: "Config", anilist_config: QueryDict): def _anilist_search(): if config.use_rofi: search_term = str(Rofi.ask("Search for")) @@ -873,6 +880,7 @@ def anilist(config: Config, anilist_config: QueryDict): watch_history = list(map(int, config.watch_history.keys())) return AniList.search(id_in=watch_history, sort="TRENDING_DESC") + # NOTE: Will probably be depracated def _anime_list(): anime_list = config.anime_list return AniList.search(id_in=anime_list) diff --git a/fastanime/libs/anilist/api.py b/fastanime/libs/anilist/api.py index e57bcbf..4b12d7a 100644 --- a/fastanime/libs/anilist/api.py +++ b/fastanime/libs/anilist/api.py @@ -3,17 +3,10 @@ This is the core module availing all the abstractions of the anilist api """ import logging +from typing import TYPE_CHECKING import requests -from .anilist_data_schema import ( - AnilistDataSchema, - AnilistMediaLists, - AnilistMediaListStatus, - AnilistNotifications, - AnilistUser, - AnilistUserData, -) from .queries_graphql import ( airing_schedule_query, anime_characters_query, @@ -35,6 +28,15 @@ from .queries_graphql import ( upcoming_anime_query, ) +if TYPE_CHECKING: + from .anilist_data_schema import ( + AnilistDataSchema, + AnilistMediaLists, + AnilistMediaListStatus, + AnilistNotifications, + AnilistUser, + AnilistUserData, + ) logger = logging.getLogger(__name__) ANILIST_ENDPOINT = "https://graphql.anilist.co" @@ -81,7 +83,7 @@ class AniListApi: def get_notification( self, - ) -> tuple[bool, AnilistNotifications] | tuple[bool, None]: + ) -> tuple[bool, "AnilistNotifications"] | tuple[bool, None]: """get the top five latest notifications for anime thats airing Returns: @@ -89,7 +91,7 @@ class AniListApi: """ return self._make_authenticated_request(notification_query) - def update_login_info(self, user: AnilistUser, token: str): + def update_login_info(self, user: "AnilistUser", token: str): """method used to login a user enabling authenticated requests Args: @@ -101,7 +103,7 @@ class AniListApi: self.session.headers.update(self.headers) self.user_id = user["id"] - def get_logged_in_user(self) -> tuple[bool, AnilistUserData] | tuple[bool, None]: + def get_logged_in_user(self) -> tuple[bool, "AnilistUserData"] | tuple[bool, None]: """get the details of the user who is currently logged in Returns: @@ -124,8 +126,8 @@ class AniListApi: return self._make_authenticated_request(media_list_mutation, variables) def get_anime_list( - self, status: AnilistMediaListStatus - ) -> tuple[bool, AnilistMediaLists] | tuple[bool, None]: + self, status: "AnilistMediaListStatus" + ) -> tuple[bool, "AnilistMediaLists"] | tuple[bool, None]: """gets an anime list from your media list given the list status Args: @@ -225,7 +227,7 @@ class AniListApi: def get_data( self, query: str, variables: dict = {} - ) -> tuple[bool, AnilistDataSchema]: + ) -> tuple[bool, "AnilistDataSchema"]: """the abstraction over all none authenticated requests and that returns data of a similar type Args: diff --git a/fastanime/libs/anime_provider/allanime/api.py b/fastanime/libs/anime_provider/allanime/api.py index 014a80e..f470e85 100644 --- a/fastanime/libs/anime_provider/allanime/api.py +++ b/fastanime/libs/anime_provider/allanime/api.py @@ -5,12 +5,10 @@ abstraction over allanime api import json import logging -from typing import Iterator +from typing import TYPE_CHECKING, Iterator 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 ( @@ -22,7 +20,10 @@ from .constants import ( from .gql_queries import ALLANIME_EPISODES_GQL, ALLANIME_SEARCH_GQL, ALLANIME_SHOW_GQL from .normalizer import normalize_anime, normalize_search_results -Logger = logging.getLogger(__name__) +if TYPE_CHECKING: + from ....libs.anime_provider.allanime.types import AllAnimeEpisode + from ....libs.anime_provider.types import Anime, Server +logger = logging.getLogger(__name__) # TODO: create tests for the api @@ -58,15 +59,15 @@ class AllAnimeAPI(AnimeProvider): if response.status_code == 200: return response.json()["data"] else: - Logger.error("allanime(ERROR): ", response.text) + logger.error("allanime(ERROR): ", response.text) return {} except Timeout: - Logger.error( + logger.error( "allanime(Error):Timeout exceeded this could mean allanime is down or you have lost internet connection" ) return {} except Exception as e: - Logger.error(f"allanime:Error: {e}") + logger.error(f"allanime:Error: {e}") return {} def search_for_anime( @@ -105,7 +106,7 @@ class AllAnimeAPI(AnimeProvider): search_results = self._fetch_gql(ALLANIME_SEARCH_GQL, variables) return normalize_search_results(search_results) # pyright:ignore except Exception as e: - Logger.error(f"FA(AllAnime): {e}") + logger.error(f"FA(AllAnime): {e}") return {} def get_anime(self, allanime_show_id: str): @@ -122,12 +123,12 @@ class AllAnimeAPI(AnimeProvider): anime = self._fetch_gql(ALLANIME_SHOW_GQL, variables) return normalize_anime(anime["show"]) except Exception as e: - Logger.error(f"FA(AllAnime): {e}") + logger.error(f"FA(AllAnime): {e}") return None def _get_anime_episode( self, allanime_show_id: str, episode_string: str, translation_type: str = "sub" - ) -> AllAnimeEpisode | dict: + ) -> "AllAnimeEpisode | dict": """get the episode details and sources info Args: @@ -147,12 +148,12 @@ class AllAnimeAPI(AnimeProvider): episode = self._fetch_gql(ALLANIME_EPISODES_GQL, variables) return episode["episode"] except Exception as e: - Logger.error(f"FA(AllAnime): {e}") + logger.error(f"FA(AllAnime): {e}") return {} def get_episode_streams( - self, anime: Anime, episode_number: str, translation_type="sub" - ) -> Iterator[Server] | None: + self, anime: "Anime", episode_number: str, translation_type="sub" + ) -> Iterator["Server"] | None: """get the streams of an episode Args: @@ -205,7 +206,7 @@ class AllAnimeAPI(AnimeProvider): if resp.status_code == 200: match embed["sourceName"]: case "Luf-mp4": - Logger.debug("allanime:Found streams from gogoanime") + logger.debug("allanime:Found streams from gogoanime") yield { "server": "gogoanime", "episode_title": ( @@ -215,7 +216,7 @@ class AllAnimeAPI(AnimeProvider): "links": resp.json()["links"], } # pyright:ignore case "Kir": - Logger.debug("allanime:Found streams from wetransfer") + logger.debug("allanime:Found streams from wetransfer") yield { "server": "wetransfer", "episode_title": ( @@ -225,7 +226,7 @@ class AllAnimeAPI(AnimeProvider): "links": resp.json()["links"], } # pyright:ignore case "S-mp4": - Logger.debug("allanime:Found streams from sharepoint") + logger.debug("allanime:Found streams from sharepoint") yield { "server": "sharepoint", "episode_title": ( @@ -235,7 +236,7 @@ class AllAnimeAPI(AnimeProvider): "links": resp.json()["links"], } # pyright:ignore case "Sak": - Logger.debug("allanime:Found streams from dropbox") + logger.debug("allanime:Found streams from dropbox") yield { "server": "dropbox", "episode_title": ( @@ -245,7 +246,7 @@ class AllAnimeAPI(AnimeProvider): "links": resp.json()["links"], } # pyright:ignore case "Default": - Logger.debug("allanime:Found streams from wixmp") + logger.debug("allanime:Found streams from wixmp") yield { "server": "wixmp", "episode_title": ( @@ -255,15 +256,15 @@ class AllAnimeAPI(AnimeProvider): "links": resp.json()["links"], } # pyright:ignore except Timeout: - Logger.error( + logger.error( "Timeout has been exceeded this could mean allanime is down or you have lost internet connection" ) return [] except Exception as e: - Logger.error(f"FA(Allanime): {e}") + logger.error(f"FA(Allanime): {e}") return [] except Exception as e: - Logger.error(f"FA(Allanime): {e}") + logger.error(f"FA(Allanime): {e}") return []