mirror of
https://github.com/Benexl/FastAnime.git
synced 2025-12-12 15:50:01 -08:00
feat(animepahe): complete
This commit is contained in:
Benex254
@@ -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,
|
||||
}
|
||||
)
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user