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(cli): normalize allanime api output

Browse Source
This commit is contained in:
Benex254
2024-08-05 09:47:01 +03:00
parent ae5e20505a
commit 88388dd182
6 changed files with 159 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -10,7 +10,7 @@ from ... import APP_CACHE_DIR, USER_CONFIG_PATH
from ...libs.anilist.anilist import AniList
from ...libs.anilist.anilist_data_schema import AnilistBaseMediaDataSchema
from ...libs.anime_provider.allanime.api import anime_provider
from ...libs.anime_provider.allanime.types import AllAnimeEpisode, AllAnimeShow
from ...libs.anime_provider.types import Anime, SearchResult, Server
from ...libs.fzf import fzf
from ...Utility import anilist_data_helper
from ...Utility.data import anime_normalizer
@@ -105,7 +105,7 @@ def player_controls(config: Config, anilist_config: QueryDict):
translation_type: str = config.translation_type.lower()
# internal config
_anime: AllAnimeShow = anilist_config._anime
anime: Anime = anilist_config.anime
current_episode: str = anilist_config.episode_number
episodes: list = sorted(anilist_config.episodes, key=float)
links: list = anilist_config.current_stream_links
@@ -119,6 +119,7 @@ def player_controls(config: Config, anilist_config: QueryDict):
fetch_streams(config, anilist_config)
def _replay():
selected_server: Server = anilist_config.selected_server
print(
"[bold magenta]Now Replaying:[/]",
anime_title,
@@ -126,7 +127,7 @@ def player_controls(config: Config, anilist_config: QueryDict):
current_episode,
)
mpv(current_link)
mpv(current_link, selected_server["episode_title"])
clear()
player_controls(config, anilist_config)
@@ -135,7 +136,7 @@ def player_controls(config: Config, anilist_config: QueryDict):
if next_episode >= len(episodes):
next_episode = len(episodes) - 1
episode = anime_provider.get_anime_episode(
_anime["_id"], episodes[next_episode], translation_type
anime["id"], episodes[next_episode], translation_type
)
# update internal config
@@ -160,7 +161,7 @@ def player_controls(config: Config, anilist_config: QueryDict):
if prev_episode <= 0:
prev_episode = 0
episode = anime_provider.get_anime_episode(
_anime["_id"], episodes[prev_episode], config.translation_type.lower()
anime["id"], episodes[prev_episode], config.translation_type.lower()
)
# update internal config
@@ -225,15 +226,23 @@ def fetch_streams(config: Config, anilist_config: QueryDict):
quality: int = config.quality
# internal config
episode: AllAnimeEpisode = anilist_config.episode
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
translation_type = config.translation_type
# get streams for episode from provider
episode_streams = anime_provider.get_episode_streams(episode)
episode_streams = anime_provider.get_episode_streams(
anime, episode_number, translation_type
)
if not episode_streams:
print("Failed to fetch :cry:")
input("Enter to retry...")
return fetch_streams(config, anilist_config)
episode_streams = {
episode_stream[0]: episode_stream[1] for episode_stream in episode_streams
episode_stream["server"]: episode_stream for episode_stream in episode_streams
}
# prompt for preferred server
@@ -277,7 +286,7 @@ def fetch_streams(config: Config, anilist_config: QueryDict):
episode_number,
)
mpv(stream_link)
mpv(stream_link, selected_server["episode_title"])
# update_watch_history
config.update_watch_history(anime_id, str(int(episode_number) + 1))
@@ -296,8 +305,8 @@ def fetch_episode(config: Config, anilist_config: QueryDict):
anime_id: int = anilist_config.anime_id
# internal config
anime = anilist_config.anime
_anime: AllAnimeShow = anilist_config._anime
anime: Anime = anilist_config.anime
_anime: SearchResult = anilist_config._anime
# prompt for episode number
episodes = anime["availableEpisodesDetail"][translation_type]
@@ -318,7 +327,7 @@ def fetch_episode(config: Config, anilist_config: QueryDict):
# get the episode info from provider
episode = anime_provider.get_anime_episode(
_anime["_id"], episode_number, translation_type
_anime["id"], episode_number, translation_type
)
# update internal config
@@ -332,8 +341,8 @@ def fetch_episode(config: Config, anilist_config: QueryDict):
def fetch_anime_episode(config, anilist_config: QueryDict):
selected_anime: AllAnimeShow = anilist_config._anime
anilist_config.anime = anime_provider.get_anime(selected_anime["_id"])
selected_anime: SearchResult = anilist_config._anime
anilist_config.anime = anime_provider.get_anime(selected_anime["id"])
fetch_episode(config, anilist_config)
@@ -350,9 +359,7 @@ def provide_anime(config: Config, anilist_config: QueryDict):
selected_anime_title, translation_type
)
search_results = {
anime["name"]: anime for anime in search_results["shows"]["edges"]
}
search_results = {anime["title"]: anime for anime in search_results["results"]}
_title = None
if _title := next(
(
@@ -386,7 +393,7 @@ def anilist_options(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)
mpv(trailer_url)
mpv(trailer_url, selected_anime_title)
anilist_options(config, anilist_config)
def _add_to_list(config: Config, anilist_config: QueryDict):

4
fastanime/cli/utils/mpv.py
View File

@@ -2,9 +2,9 @@ import shutil
import subprocess
def mpv(link, *custom_args):
def mpv(link, title, *custom_args):
MPV = shutil.which("mpv")
if not MPV:
print("mpv not found")
return
subprocess.run([MPV, *custom_args, link])
subprocess.run([MPV, *custom_args, f"--title={title}", link])

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

@@ -7,13 +7,8 @@ from requests.exceptions import Timeout
from rich import print
from rich.progress import Progress
from ....libs.anime_provider.allanime.types import (
AllAnimeEpisode,
AllAnimeSearchResults,
AllAnimeShow,
AllAnimeStreams,
Server,
)
from ....libs.anime_provider.allanime.types import AllAnimeEpisode
from ....libs.anime_provider.types import Anime, Server
from .constants import (
ALLANIME_API_ENDPOINT,
ALLANIME_BASE,
@@ -21,6 +16,7 @@ from .constants import (
USER_AGENT,
)
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__)
@@ -60,7 +56,7 @@ class AllAnimeAPI:
def search_for_anime(
self, user_query: str, translation_type: str = "sub", nsfw=True, unknown=True
) -> AllAnimeSearchResults:
):
search = {"allowAdult": nsfw, "allowUnknown": unknown, "query": user_query}
limit = 40
translationtype = translation_type
@@ -77,14 +73,14 @@ class AllAnimeAPI:
progress.add_task("[cyan]searching..", start=False, total=None)
search_results = self._fetch_gql(ALLANIME_SEARCH_GQL, variables)
return search_results # pyright:ignore
return normalize_search_results(search_results) # pyright:ignore
def get_anime(self, allanime_show_id: str) -> AllAnimeShow:
def get_anime(self, allanime_show_id: str):
variables = {"showId": allanime_show_id}
with Progress() as progress:
progress.add_task("[cyan]fetching anime..", start=False, total=None)
anime = self._fetch_gql(ALLANIME_SHOW_GQL, variables)
return anime["show"] # pyright:ignore
return normalize_anime(anime["show"])
def get_anime_episode(
self, allanime_show_id: str, episode_string: str, translation_type: str = "sub"
@@ -97,74 +93,100 @@ class AllAnimeAPI:
with Progress() as progress:
progress.add_task("[cyan]fetching episode..", start=False, total=None)
episode = self._fetch_gql(ALLANIME_EPISODES_GQL, variables)
return episode # pyright: ignore
return episode["episode"] # pyright: ignore
def get_episode_streams(self, allanime_episode_embeds_data) -> (
def get_episode_streams(
self, anime: Anime, episode_number: str, translation_type="sub"
) -> (
Generator[
tuple[Server, AllAnimeStreams],
tuple[Server, AllAnimeStreams],
tuple[Server, AllAnimeStreams],
Server,
Server,
Server,
]
| dict
| None
):
if (
not allanime_episode_embeds_data
or allanime_episode_embeds_data.get("episode") is None
):
anime_id = anime["id"]
allanime_episode = self.get_anime_episode(
anime_id, episode_number, translation_type
)
if not allanime_episode:
return {}
embeds = allanime_episode_embeds_data["episode"]["sourceUrls"]
embeds = allanime_episode["sourceUrls"]
with Progress() as progress:
progress.add_task("[cyan]fetching streams..", start=False, total=None)
for embed in embeds:
# filter the working streams
if embed.get("sourceName", "") not in (
"Sak",
"Kir",
"S-mp4",
"Luf-mp4",
):
continue
url = embed.get("sourceUrl")
try:
# filter the working streams
if embed.get("sourceName", "") not in (
"Sak",
"Kir",
"S-mp4",
"Luf-mp4",
):
continue
url = embed.get("sourceUrl")
if not url:
continue
if url.startswith("--"):
url = url[2:]
if not url:
continue
if url.startswith("--"):
url = url[2:]
# get the stream url for an episode of the defined source names
parsed_url = decode_hex_string(url)
embed_url = (
f"https://{ALLANIME_BASE}{parsed_url.replace('clock','clock.json')}"
)
resp = requests.get(
embed_url,
headers={
"Referer": ALLANIME_REFERER,
"User-Agent": USER_AGENT,
},
timeout=10,
)
if resp.status_code == 200:
match embed["sourceName"]:
case "Luf-mp4":
Logger.debug("allanime:Found streams from gogoanime")
print("[yellow]GogoAnime Fetched")
yield "gogoanime", resp.json()
case "Kir":
Logger.debug("allanime:Found streams from wetransfer")
print("[yellow]WeTransfer Fetched")
yield "wetransfer", resp.json()
case "S-mp4":
Logger.debug("allanime:Found streams from sharepoint")
print("[yellow]Sharepoint Fetched")
yield "sharepoint", resp.json()
case "Sak":
Logger.debug("allanime:Found streams from dropbox")
print("[yellow]Dropbox Fetched")
yield "dropbox", resp.json()
else:
return {}
# get the stream url for an episode of the defined source names
parsed_url = decode_hex_string(url)
embed_url = f"https://{ALLANIME_BASE}{parsed_url.replace('clock','clock.json')}"
resp = requests.get(
embed_url,
headers={
"Referer": ALLANIME_REFERER,
"User-Agent": USER_AGENT,
},
timeout=10,
)
if resp.status_code == 200:
match embed["sourceName"]:
case "Luf-mp4":
Logger.debug("allanime:Found streams from gogoanime")
print("[yellow]GogoAnime Fetched")
yield {
"server": "gogoanime",
"episode_title": allanime_episode["notes"]
or f"{anime["title"]} Episode:{episode_number}",
"links": resp.json()["links"],
} # pyright:ignore
case "Kir":
Logger.debug("allanime:Found streams from wetransfer")
print("[yellow]WeTransfer Fetched")
yield {
"server": "wetransfer",
"episode_title": allanime_episode["notes"]
or f"{anime["title"]} Episode:{episode_number}",
"links": resp.json()["links"],
} # pyright:ignore
case "S-mp4":
Logger.debug("allanime:Found streams from sharepoint")
print("[yellow]Sharepoint Fetched")
yield {
"server": "sharepoint",
"episode_title": allanime_episode["notes"]
or f"{anime["title"]} Episode:{episode_number}",
"links": resp.json()["links"],
} # pyright:ignore
case "Sak":
Logger.debug("allanime:Found streams from dropbox")
print("[yellow]Dropbox Fetched")
yield {
"server": "dropbox",
"episode_title": allanime_episode["notes"]
or f"{anime["title"]} Episode:{episode_number}",
"links": resp.json()["links"],
} # pyright:ignore
except Timeout:
print(
"Timeout has been exceeded :cry: this could mean allanime is down or your internet connection is poor"
)
except Exception as e:
print("Sth went wrong :confused:", e)
anime_provider = AllAnimeAPI()
@@ -187,53 +209,48 @@ if __name__ == "__main__":
if not search_results:
raise Exception("No results found")
search_results = search_results["shows"]["edges"]
options = [show["name"] for show in search_results]
anime = run_fzf(options)
search_results = search_results["results"]
options = {show["title"]: show for show in search_results}
anime = run_fzf(options.keys())
if anime is None:
print("No anime was selected")
sys.exit(1)
anime_result = list(filter(lambda x: x["name"] == anime, search_results))[0]
anime_data = anime_provider.get_anime(anime_result["_id"])
anime_result = options[anime]
anime_data = anime_provider.get_anime(anime_result["id"])
if anime_data is None:
raise Exception("Anime not found")
availableEpisodesDetail = anime_data["availableEpisodesDetail"]
if not availableEpisodesDetail.get(translation.strip()):
raise Exception("No episodes found")
print("select episode")
stream_link = True
while stream_link != "quit":
print("select episode")
episode = run_fzf(availableEpisodesDetail[translation.strip()])
if episode is None:
print("No episode was selected")
sys.exit(1)
episode_data = anime_provider.get_anime_episode(
anime_result["_id"], episode, translation.strip()
episode_streams_ = anime_provider.get_episode_streams(
anime_data, episode, translation.strip()
)
if episode_data is None:
if episode_streams_ is None:
raise Exception("Episode not found")
episode_streams = anime_provider.get_episode_streams(episode_data)
if not episode_streams:
raise Exception("No streams found")
episode_streams = list(episode_streams)
episode_streams = list(episode_streams_)
stream_links = []
for server in episode_streams:
# FIXME:
stream_links = [
*stream_links,
*[stream["link"] for stream in server[1]["links"]],
]
stream_links = stream_link = run_fzf([*stream_links, "quit"])
stream_links.extend([link["link"] for link in server["links"]])
stream_links.append("back")
stream_link = run_fzf(stream_links)
if stream_link == "quit":
print("Have a nice day")
sys.exit()
if not stream_link:
raise Exception("No stream was selected")
subprocess.run(["mpv", stream_link])
title = episode_streams[0].get(
"episode_title", "%s: Episode %s" % (anime_data["title"], episode)
)
subprocess.run(["mpv", f"--title={title}", stream_link])

8
fastanime/libs/anime_provider/allanime/normalizer.py
View File

@@ -1,5 +1,5 @@
from ..types import Anime, EpisodesDetail, SearchResults
from .types import AllAnimeSearchResults, AllAnimeShow
from .types import AllAnimeEpisode, AllAnimeSearchResults, AllAnimeShow
def normalize_search_results(search_results: AllAnimeSearchResults) -> SearchResults:
@@ -26,7 +26,7 @@ def normalize_anime(anime: AllAnimeShow) -> Anime:
id: str = anime["_id"]
title: str = anime["name"]
availableEpisodesDetail: EpisodesDetail = anime["availableEpisodesDetail"]
type: str = anime["__typename"]
type = anime.get("__typename")
normalized_anime: Anime = {
"id": id,
"title": title,
@@ -34,3 +34,7 @@ def normalize_anime(anime: AllAnimeShow) -> Anime:
"type": type,
}
return normalized_anime
def normalize_episode(episode: AllAnimeEpisode):
pass

2
fastanime/libs/anime_provider/allanime/types.py
View File

@@ -22,7 +22,7 @@ class AllAnimeSearchResult(TypedDict):
_id: str
name: str
availableEpisodes: list[str]
__typename: str
__typename: str | None
class AllAnimeShows(TypedDict):

18
fastanime/libs/anime_provider/types.py
View File

@@ -29,4 +29,20 @@ class Anime(TypedDict):
id: str
title: str
availableEpisodesDetail: EpisodesDetail
type: str
type: str | None
class EpisodeStream(TypedDict):
resolution: str
link: str
hls: bool | None
mp4: bool
priority: int
headers: dict
fromCache: str
class Server(TypedDict):
server: str
episode_title: str | None
links: list