mirror of
https://github.com/movixcorp/MovixOpenSource.git
synced 2026-07-27 05:02:08 +00:00
321 lines
12 KiB
Python
321 lines
12 KiB
Python
import builtins
|
|
import json
|
|
import re
|
|
from os.path import join
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
import browser_cookie3
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
from slugify import slugify
|
|
|
|
from utils.constants.macros import USER_ERROR, ERR_MSG, APP_ERROR
|
|
from utils.main_service import main_service
|
|
from utils.structs import BaseElement, BaseService, CustomException
|
|
from utils.tools.args import check_range
|
|
from utils.tools.common import get_valid_filename, get_ext_from_url, update_url_params
|
|
|
|
|
|
class midnightpulp_com(BaseService):
|
|
DEMO_URLS = [
|
|
"https://www.midnightpulp.com/details/HIGHLANDER/Highlander",
|
|
"https://www.midnightpulp.com/details/1000000047253/Jekyll",
|
|
'https://www.midnightpulp.com/livetv?q=MidnightPulp',
|
|
"https://www.midnightpulp.com/livetv?q=cinedigm-bloody-disgusting",
|
|
"https://www.midnightpulp.com/watch/NNVG1260/The-Hunters?vod_type=",
|
|
"https://www.midnightpulp.com/watch/1000000029854/Quantum-Cowboys?vod_type=",
|
|
"https://www.midnightpulp.com/watch/1000000022496/Fighting?vod_type=trailer",
|
|
]
|
|
|
|
BASE_URL = "https://www.midnightpulp.com"
|
|
VIDEO_URL = BASE_URL + "/watch/{video_id}/{video_title}"
|
|
|
|
LOGIN_COOKIES = None
|
|
AUTH_TOKEN = None
|
|
|
|
@staticmethod
|
|
def test_service():
|
|
main_service.run_service(midnightpulp_com)
|
|
|
|
@staticmethod
|
|
def is_content_livestream(content, additional):
|
|
return "/livetv?" in content
|
|
|
|
@staticmethod
|
|
def credentials_needed():
|
|
return {
|
|
"OPTIONAL": True,
|
|
"FIREFOX_COOKIES": True
|
|
}
|
|
|
|
@staticmethod
|
|
def get_login_cookies():
|
|
cookie_dict = {}
|
|
try:
|
|
for c in browser_cookie3.firefox(domain_name='midnightpulp.com'):
|
|
cookie_dict[c.name] = c.value
|
|
except browser_cookie3.BrowserCookieError:
|
|
pass
|
|
|
|
try:
|
|
assert len(cookie_dict.keys()) > 0
|
|
return cookie_dict
|
|
except:
|
|
return {}
|
|
|
|
@staticmethod
|
|
def get_auth_token():
|
|
response = requests.get(midnightpulp_com.BASE_URL).content.decode()
|
|
response = re.findall(r'<script[^<>]*src="([^"]*app-[^"]*.js)"[^<>]*>', response)
|
|
response = response[0]
|
|
if not response.startswith("http"):
|
|
response = f'{midnightpulp_com.BASE_URL}{response}'
|
|
|
|
response = requests.get(response).content.decode()
|
|
response = re.findall(r'Authorization:"Basic ([^"]+)"', response)
|
|
return response[0]
|
|
|
|
@staticmethod
|
|
def initialize_service():
|
|
if midnightpulp_com.LOGIN_COOKIES is None:
|
|
midnightpulp_com.LOGIN_COOKIES = midnightpulp_com.get_login_cookies()
|
|
if midnightpulp_com.AUTH_TOKEN is None:
|
|
midnightpulp_com.AUTH_TOKEN = midnightpulp_com.get_auth_token()
|
|
return midnightpulp_com
|
|
|
|
@staticmethod
|
|
def get_keys(challenge, additional):
|
|
licence = requests.post(
|
|
additional["license_url"],
|
|
data=challenge
|
|
)
|
|
licence.raise_for_status()
|
|
return licence.content
|
|
|
|
@staticmethod
|
|
def get_video_data(source_element):
|
|
if source_element.collection is None:
|
|
source_element.collection = join(
|
|
str(builtins.CONFIG["DOWNLOAD_COMMANDS"]["OUTPUT_MEDIA_PATH"]),
|
|
midnightpulp_com.__name__
|
|
)
|
|
|
|
response = requests.get(
|
|
source_element.url,
|
|
cookies=midnightpulp_com.LOGIN_COOKIES
|
|
).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)
|
|
|
|
if "/livetv?" in source_element.url:
|
|
params_dict = parse_qs(urlparse(source_element.url).query)
|
|
channel_query = params_dict["q"][0]
|
|
|
|
live_channel = script["props"]["pageProps"]["liveChannelListURL"]
|
|
live_feeds = []
|
|
for _ in range(0, 2):
|
|
response = json.loads(requests.get(
|
|
live_channel,
|
|
cookies=midnightpulp_com.LOGIN_COOKIES,
|
|
headers={"Authorization": f"Basic {midnightpulp_com.AUTH_TOKEN}"}
|
|
).content.decode())
|
|
|
|
live_feeds = response.get("liveFeeds", [])
|
|
if len(live_feeds) > 0:
|
|
break
|
|
if "jloc=us" in live_channel:
|
|
break
|
|
live_channel = update_url_params(live_channel, {"jloc": "us"})
|
|
|
|
live_feeds = list(filter(lambda lf: lf.get("item_id", None) == channel_query, live_feeds))
|
|
if len(live_feeds) == 0:
|
|
raise CustomException(ERR_MSG.format(
|
|
type=f'{USER_ERROR}',
|
|
url=source_element.url,
|
|
reason="The content wasn't found or you need a VPN",
|
|
solution="Do not attempt to download it or use a VPN that lets you see the livestream in your browser"
|
|
))
|
|
|
|
live_feed = live_feeds[0]
|
|
if source_element.element is None:
|
|
title = live_feed.get("title", channel_query)
|
|
source_element.element = get_valid_filename(f"Live_{title}")
|
|
|
|
manifest = live_feed["content"]["hls_url"]
|
|
return manifest, None, {}
|
|
|
|
details = script["props"]["pageProps"]["idetails"]
|
|
err_code = details.get("err_code", None)
|
|
err_msg = details.get("err_msg", "").lower()
|
|
if err_code == 1002 or "access" in err_msg:
|
|
raise CustomException(ERR_MSG.format(
|
|
type=USER_ERROR,
|
|
url=source_element.url,
|
|
reason='Need account for this content, it is paid or it is not available',
|
|
solution="Sign into your account using Firefox or don't attempt to download it"
|
|
))
|
|
if err_code == 1200 or ("available" in err_msg and "location" in err_msg):
|
|
raise CustomException(ERR_MSG.format(
|
|
type=f'{USER_ERROR}',
|
|
url=source_element.url,
|
|
reason="Need American IP to access content",
|
|
solution="Use a VPN"
|
|
))
|
|
|
|
is_trailer = details.get("trailer", False)
|
|
if not is_trailer:
|
|
is_trailer = "trailer" in script.get("query", {}).get("vod_type", "").lower()
|
|
if not is_trailer:
|
|
is_trailer = "vod_type=trailer" in source_element.url
|
|
|
|
if source_element.element is None:
|
|
title = details.get("title", None)
|
|
if title is None:
|
|
title = details.get("details", {}).get("title", None)
|
|
if title is None:
|
|
title = script.get("query", {}).get("play-asset-title", None)
|
|
if title is None:
|
|
title = source_element.url.split("/watch/")[1]
|
|
|
|
if is_trailer:
|
|
title = f"Trailer_{title}"
|
|
source_element.element = get_valid_filename(title)
|
|
|
|
has_drm = details.get("is_drm_encrypted", False) is True
|
|
additional = {}
|
|
|
|
if has_drm:
|
|
drm_info = details["drm_info"]
|
|
manifest = drm_info.get("wv_dash_url", None)
|
|
if manifest is None:
|
|
manifest = drm_info["wv_hls_url"]
|
|
|
|
license_url = drm_info.get("wv_drm_server", None)
|
|
if license_url is None:
|
|
license_url = drm_info["drm_server"]
|
|
additional["license_url"] = license_url
|
|
|
|
else:
|
|
manifest = details.get("url", None)
|
|
if manifest is None and is_trailer:
|
|
manifest = details["details"]["media_trailer_url"]
|
|
|
|
subtitles = []
|
|
srt_url = details.get("subtitle_url", None)
|
|
if srt_url in [None, ""]:
|
|
srt_url = details.get("cc_url_vtt", None)
|
|
|
|
if srt_url not in [None, ""]:
|
|
srt_ext = get_ext_from_url(srt_url)
|
|
subtitles.append((False, BaseElement(
|
|
url=srt_url,
|
|
collection=join(source_element.collection, source_element.element),
|
|
element=f'subtitle{srt_ext}'
|
|
)))
|
|
|
|
additional["SUBTITLES"] = subtitles
|
|
try:
|
|
if not has_drm:
|
|
raise
|
|
|
|
pssh_value = str(min(re.findall(
|
|
r'<[^<>]*cenc:pssh[^<>]*>(.*?)</[^<>]*cenc:pssh[^<>]*>',
|
|
requests.get(manifest).content.decode()
|
|
), key=len))
|
|
except:
|
|
pssh_value = None
|
|
|
|
if has_drm and pssh_value is None:
|
|
raise CustomException(ERR_MSG.format(
|
|
type=APP_ERROR,
|
|
url=source_element.url,
|
|
reason=f"Manifest format not supported: {manifest}. Can't extract pssh",
|
|
solution=f"Extend the {midnightpulp_com.__name__} service"
|
|
))
|
|
|
|
return manifest, pssh_value, additional
|
|
|
|
@staticmethod
|
|
def get_collection_elements(collection_url):
|
|
collection_url = collection_url.rstrip("/")
|
|
if "/watch/" in collection_url:
|
|
return [BaseElement(url=collection_url)]
|
|
if "/livetv?" in collection_url and "q=" in collection_url:
|
|
return [BaseElement(url=collection_url)]
|
|
|
|
if "/details/" not in collection_url:
|
|
return None
|
|
collection_url = collection_url.split("?")[0].split("#")[0].rstrip("/")
|
|
response = requests.get(
|
|
collection_url,
|
|
cookies=midnightpulp_com.LOGIN_COOKIES
|
|
).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)
|
|
|
|
item_details = script["props"]["pageProps"]["itemDetailsData"]
|
|
err_code = item_details.get("code", None)
|
|
err_msg = item_details.get("message", "").lower()
|
|
if err_code == 1001 or "not available" in err_msg:
|
|
raise CustomException(ERR_MSG.format(
|
|
type=USER_ERROR,
|
|
url=collection_url,
|
|
reason="This content isn't available",
|
|
solution='Do not attempt to download it'
|
|
))
|
|
|
|
collection_title = item_details.get("title", None)
|
|
if collection_title is None:
|
|
collection_title = script.get("query", {}).get("item-title", None)
|
|
if collection_title is None:
|
|
collection_title = collection_url.split("/")[-1]
|
|
collection_title = join(
|
|
join(
|
|
str(builtins.CONFIG["DOWNLOAD_COMMANDS"]["OUTPUT_MEDIA_PATH"]),
|
|
midnightpulp_com.__name__
|
|
),
|
|
get_valid_filename(collection_title)
|
|
)
|
|
|
|
seasons = script["props"]["pageProps"].get("seasonEpisodes", [])
|
|
seasons = sorted(seasons, key=lambda s: s["season"])
|
|
collection = []
|
|
|
|
for season in seasons:
|
|
season_index = season["season"]
|
|
check = check_range(True, season_index, None)
|
|
if check is True:
|
|
continue
|
|
elif check is False:
|
|
return collection
|
|
|
|
episodes = season.get("episodes", [])
|
|
episodes = sorted(episodes, key=lambda e: e["episode"])
|
|
|
|
for episode in episodes:
|
|
episode_index = episode["episode"]
|
|
check = check_range(False, season_index, episode_index)
|
|
if check is True:
|
|
continue
|
|
elif check is False:
|
|
return collection
|
|
|
|
episode_title = f'Episode_{episode_index} {episode.get("title", "")}'
|
|
episode_title = get_valid_filename(episode_title)
|
|
episode_url = midnightpulp_com.VIDEO_URL.format(
|
|
video_id=episode["item_id"],
|
|
video_title=slugify(episode.get("title", "title"))
|
|
)
|
|
|
|
collection.append(BaseElement(
|
|
url=episode_url,
|
|
collection=join(collection_title, f'Season_{season_index}'),
|
|
element=episode_title
|
|
))
|
|
return collection
|