mirror of
https://github.com/movixcorp/MovixOpenSource.git
synced 2026-07-27 09:12:07 +00:00
413 lines
16 KiB
Python
413 lines
16 KiB
Python
import builtins
|
|
import json
|
|
import re
|
|
from os.path import join
|
|
from urllib.parse import unquote
|
|
|
|
import browser_cookie3
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
|
|
from utils.constants.macros import ERR_MSG, USER_ERROR, APP_ERROR
|
|
from utils.main_service import main_service
|
|
from utils.structs import BaseElement, CustomException, BaseService
|
|
from utils.tools.args import check_range
|
|
from utils.tools.cdm import get_pssh_from_cenc_pssh
|
|
from utils.tools.common import get_valid_filename, clean_url
|
|
|
|
|
|
class mtv_fi(BaseService):
|
|
DEMO_URLS = [
|
|
"https://www.mtv.fi/video/34b759d7d68e32d863b6/samu-sirkan-joulutervehdys-2024",
|
|
"https://www.mtv.fi/video/14866e70feb4bb53d04c/cliffhanger-kuilun-partaalla",
|
|
"https://www.mtv.fi/video/62eba76d8b449b3bf6d0/jakso-4-porvoon-mitalla",
|
|
"https://www.mtv.fi/lyhyet/da2fce85b18a2b7e6235/video-tytar-saara-muistelee-kultainen-venla-elamantyopalkinnon-saanutta-isaansa-heikki-silvennoista?first=129d8e9702d4f67430c3&playlist=2e5SUdVLjYGe3DWt6gGO39&isAutoplay=false",
|
|
"https://www.mtv.fi/hae/finland%20europe",
|
|
"https://www.mtv.fi/ohjelma/970e8c6279e5cf05de1a/hautalehto",
|
|
]
|
|
|
|
BASE_URL = "https://www.mtv.fi"
|
|
PLAY_URL = None
|
|
AUTH_URL = None
|
|
GRAPHQL_URL = None
|
|
|
|
BEARER_TOKEN = None
|
|
CLIENT_VERSION = '5.3.0'
|
|
PAGE_LIMIT = 50
|
|
VIDEO_URL = BASE_URL + "/{video_type}/{video_id}/{video_slug}"
|
|
|
|
@staticmethod
|
|
def test_service():
|
|
main_service.run_service(mtv_fi)
|
|
|
|
@staticmethod
|
|
def credentials_needed():
|
|
return {"FIREFOX_COOKIES": True}
|
|
|
|
@staticmethod
|
|
def set_api_endpoints():
|
|
response = requests.get(mtv_fi.BASE_URL).content.decode()
|
|
soup = BeautifulSoup(response, 'html5lib')
|
|
script = soup.find_all('script', {'type': 'application/json'})
|
|
script = [s for s in script if s.get("id", None) == "__NEXT_DATA__"][0]
|
|
script = json.loads(script.string)
|
|
|
|
response = script["runtimeConfig"]
|
|
mtv_fi.PLAY_URL = response["PLAYBACK_API"] + "/play/{vod_id}"
|
|
mtv_fi.AUTH_URL = response["AUTH_API"] + "/oauth/refresh"
|
|
mtv_fi.GRAPHQL_URL = response["GRAPHQL_URL"] + "/graphql"
|
|
|
|
@staticmethod
|
|
def set_bearer_token():
|
|
cookie_dict = {}
|
|
for c in browser_cookie3.firefox(domain_name='mtv.fi'):
|
|
cookie_dict[c.name] = c.value
|
|
|
|
refresh_token = cookie_dict.get("mtv-refresh-token", None)
|
|
if refresh_token in ["", None]:
|
|
raise CustomException(ERR_MSG.format(
|
|
type=USER_ERROR,
|
|
url=f'from {mtv_fi.__name__}',
|
|
reason='Need account for this service',
|
|
solution='Sign into your account using Firefox'
|
|
))
|
|
|
|
response = requests.post(mtv_fi.AUTH_URL, json={'refresh_token': refresh_token})
|
|
response = response.content.decode()
|
|
response = json.loads(response)
|
|
mtv_fi.BEARER_TOKEN = response["access_token"]
|
|
|
|
@staticmethod
|
|
def initialize_service():
|
|
if mtv_fi.PLAY_URL is None:
|
|
mtv_fi.set_api_endpoints()
|
|
if mtv_fi.BEARER_TOKEN is None:
|
|
mtv_fi.set_bearer_token()
|
|
return mtv_fi
|
|
|
|
@staticmethod
|
|
def get_keys(challenge, additional):
|
|
licence = requests.post(
|
|
additional["license_url"],
|
|
headers={'x-dt-auth-token': additional["license_token"]},
|
|
data=challenge
|
|
)
|
|
licence.raise_for_status()
|
|
return licence.content
|
|
|
|
@staticmethod
|
|
def get_video_data(source_element):
|
|
vod_id = None
|
|
for f in ["/video/", "/lyhyet/"]:
|
|
if f in source_element.url:
|
|
vod_id = re.search(fr"{f}([^/?]*)", source_element.url).group(1)
|
|
break
|
|
|
|
response = requests.get(
|
|
mtv_fi.PLAY_URL.format(vod_id=vod_id),
|
|
params={
|
|
'service': 'mtv', 'device': 'device',
|
|
'protocol': 'hls,dash', 'drm': 'widevine'
|
|
},
|
|
headers={'x-jwt': f'Bearer {mtv_fi.BEARER_TOKEN}'}
|
|
)
|
|
status_code = response.status_code
|
|
response = response.content.decode()
|
|
response = json.loads(response)
|
|
error_code = response.get("errorCode", "").lower()
|
|
|
|
if status_code in [403] or "missing_subscription" in error_code:
|
|
raise CustomException(ERR_MSG.format(
|
|
type=USER_ERROR,
|
|
url=source_element.url,
|
|
reason="Can't download paid content",
|
|
solution='Do not attempt to download it'
|
|
))
|
|
if status_code in [404] or "asset_not_found" in error_code:
|
|
raise CustomException(ERR_MSG.format(
|
|
type=f'{USER_ERROR}',
|
|
url=source_element.url,
|
|
reason="The content isn't available",
|
|
solution="Do not attempt to download it"
|
|
))
|
|
if status_code in [401] or "invalid_geo_location" in error_code or "proxy_blocked" in error_code:
|
|
raise CustomException(ERR_MSG.format(
|
|
type=f'{USER_ERROR}',
|
|
url=source_element.url,
|
|
reason="Need Finnish IP to access content",
|
|
solution="Use a VPN (a good one that is not detected)"
|
|
))
|
|
|
|
if source_element.element is None:
|
|
metadata = response["metadata"]
|
|
title = metadata.get("title", None)
|
|
if title in ["", None]:
|
|
title = metadata.get("seriesTitle", None)
|
|
if title in ["", None]:
|
|
title = source_element.url.split("/")[-1]
|
|
|
|
temp_title = ""
|
|
for f in ["seasonNumber", "episodeNumber"]:
|
|
if metadata.get(f, None) in ["", None]:
|
|
temp_title = ""
|
|
break
|
|
temp_title += " " + f[0] + str(metadata[f])
|
|
|
|
title = title + temp_title
|
|
source_element.element = get_valid_filename(title)
|
|
if source_element.collection is None:
|
|
source_element.collection = join(
|
|
str(builtins.CONFIG["DOWNLOAD_COMMANDS"]["OUTPUT_MEDIA_PATH"]),
|
|
mtv_fi.__name__
|
|
)
|
|
|
|
playback = response["playbackItem"]
|
|
manifest = playback["manifestUrl"]
|
|
license_url = None
|
|
license_token = None
|
|
pssh_value = None
|
|
|
|
try:
|
|
license_url = playback["license"]["castlabsServer"]
|
|
license_token = playback["license"]["castlabsToken"]
|
|
except:
|
|
pass
|
|
|
|
if license_url is not None:
|
|
manifest_content = requests.get(manifest).content.decode()
|
|
try:
|
|
pssh_value = re.search(r'base64,(AAAA[^"]+)"', manifest_content).group(1)
|
|
except:
|
|
pssh_value = None
|
|
|
|
if pssh_value is None:
|
|
try:
|
|
pssh_value = get_pssh_from_cenc_pssh(manifest_content)
|
|
except:
|
|
pssh_value = None
|
|
|
|
if pssh_value is None:
|
|
raise CustomException(ERR_MSG.format(
|
|
type=APP_ERROR,
|
|
url=source_element.url,
|
|
reason=f"Manifest format not supported: {manifest}",
|
|
solution=f"Extend the {mtv_fi.__name__} service"
|
|
))
|
|
|
|
return manifest, pssh_value, {
|
|
"license_url": license_url,
|
|
"license_token": license_token,
|
|
"FORCE_SHAKA": True
|
|
}
|
|
|
|
@staticmethod
|
|
def get_collection_elements(collection_url):
|
|
collection_url = clean_url(collection_url)
|
|
if "/lyhyet/" in collection_url:
|
|
return [BaseElement(url=collection_url)]
|
|
if "/video/" in collection_url:
|
|
return [BaseElement(url=collection_url)]
|
|
|
|
if "/hae/" in collection_url:
|
|
search_text = re.search(r"/hae/([^/?]*)", collection_url).group(1)
|
|
search_text = unquote(search_text)
|
|
collection_name = join(join(
|
|
str(builtins.CONFIG["DOWNLOAD_COMMANDS"]["OUTPUT_MEDIA_PATH"]),
|
|
mtv_fi.__name__
|
|
), get_valid_filename("Search clips " + search_text))
|
|
collection = []
|
|
|
|
offset = -mtv_fi.PAGE_LIMIT
|
|
clip_index = 0
|
|
while True:
|
|
offset += mtv_fi.PAGE_LIMIT
|
|
response = requests.get(
|
|
mtv_fi.GRAPHQL_URL,
|
|
params={
|
|
'operationName': 'ListSearch',
|
|
'variables': json.dumps({"input": {
|
|
"limit": mtv_fi.PAGE_LIMIT, "offset": offset,
|
|
"includeUpsell": True,
|
|
"query": search_text, "types": ["CLIP", "SHORT"]
|
|
}}),
|
|
'extensions': json.dumps({"persistedQuery": {
|
|
"version": 1,
|
|
"sha256Hash": "f3e6a09f1f4ccea89fb4d0c6af2e91fa898902db4b457223c509218d0ee25ca7"
|
|
}})
|
|
},
|
|
headers={
|
|
'client-name': 'client-name',
|
|
'client-version': mtv_fi.CLIENT_VERSION,
|
|
'authorization': f'Bearer {mtv_fi.BEARER_TOKEN}'
|
|
}
|
|
)
|
|
response = response.content.decode()
|
|
response = json.loads(response)
|
|
|
|
try:
|
|
clips = response["data"]["listSearch"]["items"]
|
|
except:
|
|
clips = []
|
|
if len(clips) == 0:
|
|
break
|
|
|
|
try:
|
|
has_next_page = response["data"]["listSearch"]["pageInfo"]["hasNextPage"]
|
|
assert type(has_next_page) is bool
|
|
except:
|
|
has_next_page = False
|
|
|
|
for clip in clips:
|
|
clip_index += 1
|
|
check = check_range(False, None, clip_index)
|
|
if check is True:
|
|
continue
|
|
elif check is False:
|
|
return collection
|
|
|
|
clip_url = mtv_fi.VIDEO_URL.format(
|
|
video_type="lyhyet",
|
|
video_id=clip["id"],
|
|
video_slug=clip.get("slug", "clip")
|
|
)
|
|
clip_title = clip.get("title", clip.get("slug", ""))
|
|
clip_title = get_valid_filename(f'Clip {clip_index} {clip_title}')
|
|
|
|
collection.append(BaseElement(
|
|
url=clip_url,
|
|
collection=collection_name,
|
|
element=clip_title
|
|
))
|
|
|
|
if not has_next_page:
|
|
break
|
|
|
|
return collection
|
|
|
|
if "/ohjelma/" in collection_url:
|
|
series_id = re.search(r"/ohjelma/([^/?]*)", collection_url).group(1)
|
|
collection = []
|
|
seasons_response = requests.get(
|
|
mtv_fi.GRAPHQL_URL,
|
|
params={
|
|
'operationName': 'ContentDetailsPage',
|
|
'variables': json.dumps({"mediaId": series_id}),
|
|
'extensions': json.dumps({"persistedQuery": {
|
|
"version": 1,
|
|
"sha256Hash": "5dac39a55f8ca745ab7e974260bc685c0de11349eb1df85023e90e3f6477f084"
|
|
}})
|
|
},
|
|
headers={
|
|
'client-name': 'client-name',
|
|
'client-version': mtv_fi.CLIENT_VERSION,
|
|
'authorization': f'Bearer {mtv_fi.BEARER_TOKEN}'
|
|
}
|
|
)
|
|
seasons_response = seasons_response.content.decode()
|
|
seasons_response = json.loads(seasons_response)
|
|
seasons_response = seasons_response["data"]["media"]
|
|
|
|
if seasons_response is None:
|
|
raise CustomException(ERR_MSG.format(
|
|
type=f'{USER_ERROR}',
|
|
url=collection_url,
|
|
reason="The content isn't available",
|
|
solution="Do not attempt to download it"
|
|
))
|
|
|
|
collection_name = seasons_response.get("title", seasons_response.get("slug", ""))
|
|
if collection_name in ["", None]:
|
|
collection_name = collection_url.split("/")[-1]
|
|
collection_name = join(join(
|
|
str(builtins.CONFIG["DOWNLOAD_COMMANDS"]["OUTPUT_MEDIA_PATH"]),
|
|
mtv_fi.__name__
|
|
), collection_name)
|
|
|
|
try:
|
|
seasons = seasons_response["allSeasonLinks"]
|
|
assert len(seasons) > 0
|
|
except:
|
|
seasons = []
|
|
|
|
season_index = 0
|
|
for season in seasons:
|
|
season_index += 1
|
|
check = check_range(True, season_index, None)
|
|
if check is True:
|
|
continue
|
|
elif check is False:
|
|
return collection
|
|
|
|
season_id = season["seasonId"]
|
|
season_title = season.get("title", "")
|
|
season_title = get_valid_filename(f'Season {season_index} {season_title}')
|
|
|
|
episode_offset = -mtv_fi.PAGE_LIMIT
|
|
episode_index = 0
|
|
while True:
|
|
episode_offset += mtv_fi.PAGE_LIMIT
|
|
episode_response = requests.get(
|
|
mtv_fi.GRAPHQL_URL,
|
|
params={
|
|
'operationName': 'SeasonEpisodes',
|
|
'variables': json.dumps({
|
|
"seasonId": season_id,
|
|
"input": {
|
|
"limit": mtv_fi.PAGE_LIMIT,
|
|
"offset": episode_offset
|
|
}
|
|
}),
|
|
'extensions': json.dumps({"persistedQuery": {
|
|
"version": 1,
|
|
"sha256Hash": "f705d5420f2d1cf2331f18059313bd93a6ccaaf18c4e96b38efee220ea116413"
|
|
}})
|
|
},
|
|
headers={
|
|
'client-name': 'client-name',
|
|
'client-version': mtv_fi.CLIENT_VERSION,
|
|
'authorization': f'Bearer {mtv_fi.BEARER_TOKEN}'
|
|
}
|
|
)
|
|
episode_response = episode_response.content.decode()
|
|
episode_response = json.loads(episode_response)
|
|
|
|
try:
|
|
episodes = episode_response["data"]["season"]["episodes"]["items"]
|
|
except:
|
|
episodes = []
|
|
if len(episodes) == 0:
|
|
break
|
|
|
|
try:
|
|
has_next_page = episode_response["data"]["season"]["episodes"]["pageInfo"]["hasNextPage"]
|
|
assert type(has_next_page) is bool
|
|
except:
|
|
has_next_page = False
|
|
|
|
for episode in episodes:
|
|
episode_index += 1
|
|
check = check_range(False, season_index, episode_index)
|
|
if check is True:
|
|
continue
|
|
elif check is False:
|
|
return collection
|
|
|
|
episode_url = mtv_fi.VIDEO_URL.format(
|
|
video_type="video",
|
|
video_id=episode["id"],
|
|
video_slug=episode.get("slug", "episode")
|
|
)
|
|
episode_title = episode.get("title", episode.get("slug", ""))
|
|
episode_title = get_valid_filename(f'Episode {episode_index} {episode_title}')
|
|
|
|
collection.append(BaseElement(
|
|
url=episode_url,
|
|
collection=join(collection_name, season_title),
|
|
element=episode_title
|
|
))
|
|
|
|
if not has_next_page:
|
|
break
|
|
|
|
return collection
|
|
return None
|