mirror of
https://github.com/Benexl/FastAnime.git
synced 2025-12-31 23:15:51 -08:00
324 lines
12 KiB
Python
324 lines
12 KiB
Python
"""a module that handles the scraping of allanime
|
|
|
|
abstraction over allanime api
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from typing import TYPE_CHECKING
|
|
|
|
from requests.exceptions import Timeout
|
|
|
|
from ...anime_provider.base_provider import AnimeProvider
|
|
from ..utils import give_random_quality, one_digit_symmetric_xor
|
|
from .constants import ALLANIME_API_ENDPOINT, ALLANIME_BASE, ALLANIME_REFERER
|
|
from .gql_queries import ALLANIME_EPISODES_GQL, ALLANIME_SEARCH_GQL, ALLANIME_SHOW_GQL
|
|
|
|
if TYPE_CHECKING:
|
|
from typing import Iterator
|
|
|
|
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
|
|
#
|
|
# ** Based on ani-cli **
|
|
class AllAnimeAPI(AnimeProvider):
|
|
"""
|
|
Provides a fast and effective interface to AllAnime site.
|
|
"""
|
|
|
|
api_endpoint = ALLANIME_API_ENDPOINT
|
|
HEADERS = {
|
|
"Referer": ALLANIME_REFERER,
|
|
}
|
|
|
|
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,
|
|
params={
|
|
"variables": json.dumps(variables),
|
|
"query": query,
|
|
},
|
|
timeout=10,
|
|
)
|
|
if response.status_code == 200:
|
|
return response.json()["data"]
|
|
else:
|
|
logger.error("allanime(ERROR): ", response.text)
|
|
return {}
|
|
except Timeout:
|
|
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}")
|
|
return {}
|
|
|
|
def search_for_anime(
|
|
self,
|
|
user_query: str,
|
|
translation_type: str = "sub",
|
|
nsfw=True,
|
|
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
|
|
countryorigin = "all"
|
|
page = 1
|
|
variables = {
|
|
"search": search,
|
|
"limit": limit,
|
|
"page": page,
|
|
"translationtype": translationtype,
|
|
"countryorigin": countryorigin,
|
|
}
|
|
try:
|
|
search_results = self._fetch_gql(ALLANIME_SEARCH_GQL, variables)
|
|
page_info = search_results["shows"]["pageInfo"]
|
|
results = []
|
|
for result in search_results["shows"]["edges"]:
|
|
normalized_result = {
|
|
"id": result["_id"],
|
|
"title": result["name"],
|
|
"type": result["__typename"],
|
|
"availableEpisodes": result["availableEpisodes"],
|
|
}
|
|
results.append(normalized_result)
|
|
|
|
normalized_search_results = {
|
|
"pageInfo": page_info,
|
|
"results": results,
|
|
}
|
|
return normalized_search_results
|
|
|
|
except Exception as e:
|
|
logger.error(f"FA(AllAnime): {e}")
|
|
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)
|
|
id: str = anime["show"]["_id"]
|
|
title: str = anime["show"]["name"]
|
|
availableEpisodesDetail = anime["show"]["availableEpisodesDetail"]
|
|
type = anime.get("__typename")
|
|
normalized_anime = {
|
|
"id": id,
|
|
"title": title,
|
|
"availableEpisodesDetail": availableEpisodesDetail,
|
|
"type": type,
|
|
}
|
|
return normalized_anime
|
|
except Exception as e:
|
|
logger.error(f"AllAnime(get_anime): {e}")
|
|
return None
|
|
|
|
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,
|
|
"episodeString": episode_string,
|
|
}
|
|
try:
|
|
episode = self._fetch_gql(ALLANIME_EPISODES_GQL, variables)
|
|
return episode["episode"]
|
|
except Exception as e:
|
|
logger.error(f"FA(AllAnime): {e}")
|
|
return {}
|
|
|
|
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(
|
|
anime_id, episode_number, translation_type
|
|
)
|
|
if not allanime_episode:
|
|
return []
|
|
|
|
embeds = allanime_episode["sourceUrls"]
|
|
try:
|
|
for embed in embeds:
|
|
try:
|
|
# 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 (
|
|
# priorities based on death note
|
|
"Sak", # 7
|
|
"S-mp4", # 7.9
|
|
"Luf-mp4", # 7.7
|
|
"Default", # 8.5
|
|
"Yt-mp4", # 7.9
|
|
"Kir", # NA
|
|
# "Vid-mp4" # 4
|
|
# "Ok", # 3.5
|
|
# "Ss-Hls", # 5.5
|
|
# "Mp4", # 4
|
|
):
|
|
continue
|
|
url = embed.get("sourceUrl")
|
|
#
|
|
if not url:
|
|
continue
|
|
if url.startswith("--"):
|
|
url = url[2:]
|
|
url = one_digit_symmetric_xor(56, url)
|
|
|
|
if "tools.fast4speed.rsvp" in url:
|
|
yield {
|
|
"server": "Yt",
|
|
"episode_title": f'{anime["title"]}; Episode {episode_number}',
|
|
"headers": {"Referer": f"https://{ALLANIME_BASE}/"},
|
|
"subtitles": [],
|
|
"links": [
|
|
{
|
|
"link": url,
|
|
"quality": "1080",
|
|
}
|
|
],
|
|
} # pyright:ignore
|
|
continue
|
|
|
|
# get the stream url for an episode of the defined source names
|
|
embed_url = (
|
|
f"https://{ALLANIME_BASE}{url.replace('clock', 'clock.json')}"
|
|
)
|
|
resp = self.session.get(
|
|
embed_url,
|
|
timeout=10,
|
|
)
|
|
|
|
if resp.status_code == 200:
|
|
match embed["sourceName"]:
|
|
case "Luf-mp4":
|
|
logger.debug("allanime:Found streams from gogoanime")
|
|
yield {
|
|
"server": "gogoanime",
|
|
"headers": {},
|
|
"subtitles": [],
|
|
"episode_title": (
|
|
allanime_episode["notes"] or f'{anime["title"]}'
|
|
)
|
|
+ f"; Episode {episode_number}",
|
|
"links": give_random_quality(resp.json()["links"]),
|
|
} # pyright:ignore
|
|
case "Kir":
|
|
logger.debug("allanime:Found streams from wetransfer")
|
|
yield {
|
|
"server": "wetransfer",
|
|
"headers": {},
|
|
"subtitles": [],
|
|
"episode_title": (
|
|
allanime_episode["notes"] or f'{anime["title"]}'
|
|
)
|
|
+ f"; Episode {episode_number}",
|
|
"links": give_random_quality(resp.json()["links"]),
|
|
} # pyright:ignore
|
|
case "S-mp4":
|
|
logger.debug("allanime:Found streams from sharepoint")
|
|
yield {
|
|
"server": "sharepoint",
|
|
"headers": {},
|
|
"subtitles": [],
|
|
"episode_title": (
|
|
allanime_episode["notes"] or f'{anime["title"]}'
|
|
)
|
|
+ f"; Episode {episode_number}",
|
|
"links": give_random_quality(resp.json()["links"]),
|
|
} # pyright:ignore
|
|
case "Sak":
|
|
logger.debug("allanime:Found streams from dropbox")
|
|
yield {
|
|
"server": "dropbox",
|
|
"headers": {},
|
|
"subtitles": [],
|
|
"episode_title": (
|
|
allanime_episode["notes"] or f'{anime["title"]}'
|
|
)
|
|
+ f"; Episode {episode_number}",
|
|
"links": give_random_quality(resp.json()["links"]),
|
|
} # pyright:ignore
|
|
case "Default":
|
|
logger.debug("allanime:Found streams from wixmp")
|
|
yield {
|
|
"server": "wixmp",
|
|
"headers": {},
|
|
"subtitles": [],
|
|
"episode_title": (
|
|
allanime_episode["notes"] or f'{anime["title"]}'
|
|
)
|
|
+ f"; Episode {episode_number}",
|
|
"links": give_random_quality(resp.json()["links"]),
|
|
} # pyright:ignore
|
|
|
|
except Timeout:
|
|
logger.error(
|
|
"Timeout has been exceeded this could mean allanime is down or you have lost internet connection"
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"FA(Allanime): {e}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"FA(Allanime): {e}")
|
|
return []
|