gitea-mirror/FastAnime
1
0
Fork 0
mirror of https://github.com/Benexl/FastAnime.git synced 2025-12-25 04:15:19 -08:00
Code Issues Packages Projects Releases Wiki Activity

feat: only import types if type checking

Browse Source
This commit is contained in:
Benex254
2024-08-06 14:15:47 +03:00
parent 057596dfd9
commit 2e94e8494e
7 changed files with 107 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
fastanime/AnimeProvider.py
View File

@@ -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:

15
fastanime/cli/commands/anilist/completed.py
View File

@@ -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")

8
fastanime/cli/commands/anilist/dropped.py
View File

@@ -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")

7
fastanime/cli/commands/anilist/login.py
View File

@@ -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 = (

60
fastanime/cli/interfaces/anilist_interfaces.py
View File

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

30
fastanime/libs/anilist/api.py
View File

@@ -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:

43
fastanime/libs/anime_provider/allanime/api.py
View File

@@ -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 []