From c4255fc74897ca78175a0b3f2c01cb6b4288dece Mon Sep 17 00:00:00 2001 From: Benex254 Date: Sun, 11 Aug 2024 11:20:08 +0300 Subject: [PATCH] feat(animepahe): complete --- .../libs/anime_provider/animepahe/api.py | 161 ++++++++++++++---- .../libs/anime_provider/animepahe/types.py | 61 +++++++ 2 files changed, 187 insertions(+), 35 deletions(-) diff --git a/fastanime/libs/anime_provider/animepahe/api.py b/fastanime/libs/anime_provider/animepahe/api.py index 95aba3d..993c8ff 100644 --- a/fastanime/libs/anime_provider/animepahe/api.py +++ b/fastanime/libs/anime_provider/animepahe/api.py @@ -1,6 +1,8 @@ +import logging import re import shutil import subprocess +from typing import TYPE_CHECKING from yt_dlp.utils import ( extract_attributes, @@ -17,72 +19,146 @@ from .constants import ( SERVER_HEADERS, ) +if TYPE_CHECKING: + from ..types import Anime + from .types import AnimePaheAnimePage, AnimePaheSearchPage, AnimeSearchResult JUICY_STREAM_REGEX = re.compile(r"source='(.*)';") +logger = logging.getLogger(__name__) # TODO: hack this to completion class AnimePaheApi(AnimeProvider): - def search_for_anime(self, user_query, *args): + search_page: "AnimePaheSearchPage" + anime: "AnimePaheAnimePage" + + def search_for_anime(self, user_query: str, *args): try: url = f"{ANIMEPAHE_ENDPOINT}m=search&q={user_query}" headers = {**REQUEST_HEADERS} response = self.session.get(url, headers=headers) if not response.status_code == 200: return - data = response.json() + data: "AnimePaheSearchPage" = response.json() + self.search_page = data return { - "pageInfo": {"total": data["total"]}, + "pageInfo": { + "total": data["total"], + "perPage": data["per_page"], + "currentPage": data["current_page"], + }, "results": [ { + "availableEpisodes": list(range(result["episodes"])), "id": result["session"], "title": result["title"], - "availableEpisodes": result["episodes"], "type": result["type"], + "year": result["year"], + "score": result["score"], + "status": result["status"], + "season": result["season"], + "poster": result["poster"], } for result in data["data"] ], } except Exception as e: - print(e) - input() + logger.error(f"AnimePahe(search): {e}") + return {} def get_anime(self, session_id: str, *args): - url = f"{ANIMEPAHE_ENDPOINT}m=release&id={session_id}&sort=episode_asc&page=1" - response = self.session.get(url, headers=REQUEST_HEADERS) - if not response.status_code == 200: - return - data = response.json() - self.current = data - episodes = list(map(str, range(data["total"]))) - return { - "id": session_id, - "title": "none", - "availableEpisodesDetail": { - "sub": episodes, - "dub": episodes, - "raw": episodes, - }, - } + try: + anime_result: "AnimeSearchResult" = [ + anime + for anime in self.search_page["data"] + if anime["session"] == session_id + ][0] + url = ( + f"{ANIMEPAHE_ENDPOINT}m=release&id={session_id}&sort=episode_asc&page=1" + ) + response = self.session.get(url, headers=REQUEST_HEADERS) + if not response.status_code == 200: + return {} + data: "AnimePaheAnimePage" = response.json() + self.anime = data + episodes = list(map(str, range(data["total"]))) + title = "" + return { + "id": session_id, + "title": anime_result["title"], + "year": anime_result["year"], + "season": anime_result["season"], + "poster": anime_result["poster"], + "score": anime_result["score"], + "availableEpisodesDetail": { + "sub": episodes, + "dub": episodes, + "raw": episodes, + }, + "episodesInfo": [ + { + "title": episode["title"] or f"{title};{episode['episode']}", + "episode": episode["episode"], + "id": episode["session"], + "translation_type": episode["audio"], + "duration": episode["duration"], + "poster": episode["snapshot"], + } + for episode in data["data"] + ], + } + except Exception as e: + logger.error(f"AnimePahe(anime): {e}") + return {} + + def get_episode_streams( + self, anime: "Anime", episode_number: str, translation_type, *args + ): + # extract episode details from memory + episode = [ + episode + for episode in self.anime["data"] + if float(episode["episode"]) == float(episode_number) + ] + if not episode: + logger.error(f"AnimePahe(streams): episode {episode_number} doesn't exist") + raise Exception("Episode not found") + episode = episode[0] - def get_episode_streams(self, anime, episode, *args): - episode_id = self.current["data"][int(episode)]["session"] anime_id = anime["id"] - url = f"{ANIMEPAHE_BASE}/play/{anime_id}/{episode_id}" - # response = requests.get(url, headers=REQUEST_HEADERS) + # fetch the episode page + url = f"{ANIMEPAHE_BASE}/play/{anime_id}/{episode['session']}" response = self.session.get(url, headers=REQUEST_HEADERS) - # print(clean_html(response.text)) + # get the element containing links to juicy streams c = get_element_by_id("resolutionMenu", response.text) resolutionMenuItems = get_elements_html_by_class("dropdown-item", c) + # convert the elements containing embed links to a neat dict containing: + # data-src + # data-audio + # data-resolution res_dicts = [extract_attributes(item) for item in resolutionMenuItems] - streams = {"server": "kwik", "links": [], "episode_title": f"{episode}"} + # get the episode title + episode_title = ( + episode["title"] or f"{anime['title']}; Episode {episode['episode']}" + ) + # get all links + streams = {"server": "kwik", "links": [], "episode_title": episode_title} for res_dict in res_dicts: # get embed url embed_url = res_dict["data-src"] + data_audio = "dub" if res_dict["data-audio"] == "eng" else "sub" + # filter streams by translation_type + if data_audio != translation_type: + continue + if not embed_url: - return + logger.warn( + "AnimePahe: embed url not found please report to the developers" + ) + raise Exception("Episode not found") + # get embed page embed_response = self.session.get(embed_url, headers=SERVER_HEADERS) embed = embed_response.text # search for the encoded js @@ -95,28 +171,43 @@ class AnimePaheApi(AnimeProvider): encoded_js = content break if not encoded_js: - return - # execute the encoded js with node for now or maybe forever + logger.warn( + "AnimePahe: Encoded js not found please report to the developers" + ) + raise Exception("Episode not found") + # execute the encoded js with node for now or maybe forever in odrder to get a more workable info NODE = shutil.which("node") if not NODE: - return + logger.warn( + "AnimePahe: animepahe currently requires node js to extract them juicy streams" + ) + raise Exception("Episode not found") result = subprocess.run( [NODE, "-e", encoded_js], text=True, capture_output=True, ) + # decoded js evaluted_js = result.stderr if not evaluted_js: - return + logger.warn( + "AnimePahe: could not decode encoded js using node please report to developers" + ) + raise Exception("Episode not found") # get that juicy stream match = JUICY_STREAM_REGEX.search(evaluted_js) if not match: - return + logger.warn( + "AnimePahe: could not find the juicy stream please report to developers" + ) + raise Exception("Episode not found") + # get the actual hls stream link juicy_stream = match.group(1) + # add the link streams["links"].append( { "quality": res_dict["data-resolution"], - "audio_language": res_dict["data-audio"], + "translation_type": data_audio, "link": juicy_stream, } ) diff --git a/fastanime/libs/anime_provider/animepahe/types.py b/fastanime/libs/anime_provider/animepahe/types.py index e69de29..0bfa245 100644 --- a/fastanime/libs/anime_provider/animepahe/types.py +++ b/fastanime/libs/anime_provider/animepahe/types.py @@ -0,0 +1,61 @@ +from typing import Literal, TypedDict + + +class AnimeSearchResult(TypedDict): + id: int + title: str + type: str + episodes: int + status: str + season: str + year: int + score: int + poster: str + session: str + + +class AnimePaheSearchPage(TypedDict): + total: int + per_page: int + current_page: int + last_page: int + _from: int + to: int + data: list[AnimeSearchResult] + + +class Episode(TypedDict): + id: int + anime_id: int + episode: int + episode2: int + edition: str + title: str + snapshot: str # episode image + disc: str + audio: Literal["eng", "jpn"] + duration: str # time 00:00:00 + session: str + filler: int + created_at: str + + +class AnimePaheAnimePage(TypedDict): + total: int + per_page: int + current_page: int + last_page: int + next_page_url: str | None + prev_page_url: str | None + _from: int + to: int + data: list[Episode] + + +class Server: + type: str + data_src = "https://kwik.si/e/PImJ0u7Y3M0G" + data_fansub: str + data_resolution: Literal["360", "720", "1080"] + data_audio: Literal["eng", "jpn"] + data_av1: str