Merge pull request 'Updated Crunchyroll service.' (#1) from ToonsHub/devine:master into master

Reviewed-on: TPD94/devine#1
master
TPD94 2024-03-06 17:34:55 +00:00
commit 096ba1932a
1 changed files with 82 additions and 47 deletions

View File

@ -1,6 +1,7 @@
import base64
import uuid
import math
import time
import datetime
import logging
from abc import ABCMeta, abstractmethod
@ -20,7 +21,7 @@ from devine.core.console import console
from devine.core.constants import AnyTrack
from devine.core.credential import Credential
from devine.core.titles import Title_T, Titles_T
from devine.core.tracks import Chapters, Tracks
from devine.core.tracks import Video, Audio, Subtitle, Chapter, Chapters, Tracks
from devine.core.utilities import get_ip_info
from devine.core.manifests import HLS, DASH
@ -29,15 +30,15 @@ class CR(Service):
Service code for Crunchyroll
Written by TPD94
Authorization: Cookies (Free and Paid Titles)
Security: FHD@L3
Authorization: None (Free) | Cookies (Free and Paid Titles)
Security: L3 FHD
"""
# Static method, this method belongs to the class
@staticmethod
# The command name, must much the service tag (and by extension the service folder)
@click.command(name="CR", short_help="https://crucnhyroll.com", help=__doc__)
@click.command(name="CR", short_help="https://crunchyroll.com", help=__doc__)
# Using series ID for crunchyroll
@click.argument("title", type=str)
@ -52,64 +53,77 @@ class CR(Service):
# Pass the series_id argument to self so it's accessable across all methods
self.title = title
self.no_login = False
self.token = None
self.token_expiry = 0
# Overriding the constructor
super().__init__(ctx)
# Define function to get a session for service
@staticmethod
def get_session() -> requests.Session:
# Start a session
session = requests.Session()
# Set the headers
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:123.0) Gecko/20100101 Firefox/123.0',
'Accept': 'application/json, text/plain, */*',
'Accept-Language': 'en-US,en;q=0.5',
}
# Update the headers for the session
session.headers.update(headers)
# return the session
return session
# Defining an authinticate function
def authenticate(self, cookies: Optional[CookieJar], credential: Optional[Credential] = None):
# Check for cached token
if self.token_expiry > time.time():
return self.token
# Login session
login_session = requests.Session()
# Load the cookies for login session
login_session.cookies.get(cookies)
# Check whether cookies are available
if not cookies:
self.no_login = True
if cookies:
# Load the cookies for login session
login_session.cookies.get(cookies)
# Add cookies
login_session.cookies.update(cookies)
# Add cookies
login_session.cookies.update(cookies)
# Load the cookies for self sessions later
self.session.cookies.get(cookies)
# Load the cookies for self sessions later
self.session.cookies.get(cookies)
# Add cookies for self sessions later
self.session.cookies.update(cookies)
# Add cookies for self sessions later
self.session.cookies.update(cookies)
# Set headers for the token request
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0',
# Setting to Firefox
'Authorization': 'Basic bm9haWhkZXZtXzZpeWcwYThsMHE6', # Seems to be the same across all browsers on PC
'ETP-Anonymous-ID': f'{uuid.uuid4()}', # Device ID, can be a randomized UUID
'Origin': 'https://www.crunchyroll.com', # Crunchyroll origin
'Referer': 'https://www.crunchyroll.com/', # Crunchyroll referer
}
# If cookies are not available, log in anonymously
if self.no_login:
# Set headers for the token request
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0',
# Setting to Firefox
"Authorization":"Basic Y3Jfd2ViOg==", # Seems to be the same across all browsers on PC
'ETP-Anonymous-ID': f'{uuid.uuid4()}', # Device ID, can be a randomized UUID
'Origin': 'https://www.crunchyroll.com', # Crunchyroll origin
'Referer': 'https://www.crunchyroll.com/', # Crunchyroll referer
}
# Set data for the token request
data = {
'grant_type': 'client_id', # Value as per browser for anonymous login
}
else:
# Set headers for the token request
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0',
# Setting to Firefox
'Authorization': 'Basic bm9haWhkZXZtXzZpeWcwYThsMHE6', # Seems to be the same across all browsers on PC
'ETP-Anonymous-ID': f'{uuid.uuid4()}', # Device ID, can be a randomized UUID
'Origin': 'https://www.crunchyroll.com', # Crunchyroll origin
'Referer': 'https://www.crunchyroll.com/', # Crunchyroll referer
}
# Set data for the token request
data = {
'device_id': f'{uuid.uuid4()}', # Device ID, can be randomized UUID
'device_type': 'Firefox on Windows', # Setting to FireFox
'grant_type': 'etp_rt_cookie', # Seems some sort of refresh token
}
# Set data for the token request
data = {
'device_id': f'{uuid.uuid4()}', # Device ID, can be randomized UUID
'device_type': 'Firefox on Windows', # Setting to FireFox
'grant_type': 'etp_rt_cookie', # Not sure what that is
}
# Send a post request to the auth login
response = login_session.post(url='https://www.crunchyroll.com/auth/v1/token', data=data, headers=headers)
@ -117,6 +131,10 @@ class CR(Service):
# Retrieve the token from the response
token = f"Bearer {response.json()['access_token']}"
# Cache token
self.token = token
self.token_expiry = time.time() + response.json()["expires_in"]
# Return the token
return token
@ -181,6 +199,11 @@ class CR(Service):
# Set a class for each episode
episode_class = Episode(id_=episode_id, title=episode_season_title, season=episode_season, number=episode_number, name=episode_name, year=episode_year, service=self.__class__)
# Check whether you can stream the title or not
if not episode_metadata.json()['data'][episode].get('streams_link'):
logging.getLogger("CR").info(f"Unable to stream ID: {episode_id}")
continue
# Append the stream link
episode_class.data = episode_metadata.json()['data'][episode]['streams_link']
@ -206,9 +229,21 @@ class CR(Service):
# Get the MPD URL
mpd_url = mpd_info.json()['data'][0]['drm_adaptive_dash']['']['url']
mpd_lang = mpd_info.json()['meta']['audio_locale']
# Grab the tracks from the MPD
tracks = DASH.from_url(url=mpd_url).to_tracks(language="en")
tracks = DASH.from_url(url=mpd_url).to_tracks(language=mpd_lang)
# Get subtitles
for _, sub in mpd_info.json()['meta']['subtitles'].items():
tracks.add(
Subtitle(
url=sub['url'],
codec=Subtitle.Codec.from_mime(sub['format']),
language=sub['locale'],
forced=(sub['locale'] == mpd_lang), # If audio language matches subtitle language, it's Forced.
)
)
# Return the tracks
return tracks