mirror of
https://github.com/movixcorp/MovixOpenSource.git
synced 2026-07-29 02:42:45 +00:00
247 lines
9.4 KiB
Python
247 lines
9.4 KiB
Python
import builtins
|
|
import json
|
|
import re
|
|
from os.path import join
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
import requests
|
|
|
|
from utils.constants.macros import ERR_MSG, USER_ERROR, 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, clean_url
|
|
|
|
|
|
class stirr_com(BaseService):
|
|
DEMO_URLS = [
|
|
"https://stirr.com/watch/5726/nautical-channel",
|
|
"https://stirr.com/watch/5691/the-surprising-world-of-rabbits",
|
|
"https://stirr.com/watch/6423/the-bagman-official-trailer",
|
|
"https://stirr.com/watch/6134/of-souls-plus-water-the-elder",
|
|
"https://stirr.com/watch/4246?type=series&series_id=174&season_id=238",
|
|
"https://stirr.com/live?channel_id=5735",
|
|
"https://stirr.com/tv-shows/174/beyond-the-beaten-path",
|
|
"https://stirr.com/tv-shows/170/epic-trails",
|
|
]
|
|
|
|
PLAYABLE_URL = 'https://stirr.com/api/v2/videos/{video_id}/playable'
|
|
SERIES_URL = 'https://stirr.com/api/series/list/{series_id}'
|
|
SEASONS_URL = 'https://stirr.com/api/season/list/{series_id}'
|
|
EPISODES_URL = 'https://stirr.com/api/season/data?series_id={series_id}&season_id={season_id}&page={page}&paginate=1'
|
|
EPISODE_URL = "https://stirr.com/watch/{video_id}"
|
|
BASE_URL = 'https://stirr.com'
|
|
|
|
@staticmethod
|
|
def test_service():
|
|
main_service.run_service(stirr_com)
|
|
|
|
@staticmethod
|
|
def is_content_livestream(content, additional):
|
|
return "/live?channel_id=" in content or additional.get("IS_LIVE", False) is True
|
|
|
|
@staticmethod
|
|
def credentials_needed():
|
|
return {"NONE_NEEDED": True}
|
|
|
|
@staticmethod
|
|
def initialize_service():
|
|
return stirr_com
|
|
|
|
@staticmethod
|
|
def get_keys(challenge, additional):
|
|
raise Exception(
|
|
f'{BaseService.get_keys.__name__} must not be called '
|
|
f'for the {stirr_com.__name__} service'
|
|
)
|
|
|
|
@staticmethod
|
|
def get_video_data(source_element):
|
|
if "/live?channel_id" in source_element.url:
|
|
is_live = True
|
|
params_dict = parse_qs(urlparse(source_element.url).query)
|
|
video_id = params_dict["channel_id"][0]
|
|
else:
|
|
is_live = None
|
|
video_id = re.search(r"/watch/([^/?]*)", source_element.url).group(1)
|
|
response = requests.post(stirr_com.PLAYABLE_URL.format(video_id=video_id))
|
|
|
|
status_code = response.status_code
|
|
response = response.content.decode()
|
|
response = json.loads(response)
|
|
error = response.get("error_code", "").lower()
|
|
message = response.get("message", "").lower()
|
|
|
|
if status_code == 403 and "region" in message:
|
|
raise CustomException(ERR_MSG.format(
|
|
type=f'{USER_ERROR}',
|
|
url=source_element.url,
|
|
reason="Need American IP to access content",
|
|
solution="Use a VPN"
|
|
))
|
|
|
|
if 400 <= status_code < 500 or "video_not_found" in error:
|
|
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"
|
|
))
|
|
|
|
response = response["data"]
|
|
if type(response) is not list:
|
|
response = [response]
|
|
response = [r for r in response if str(r["content"]["videoid"]) == video_id][0]
|
|
if is_live is None:
|
|
is_live = response["content"].get("content_type_label", "").lower() in ["live"]
|
|
|
|
if source_element.element is None:
|
|
title = response["content"].get("title", None)
|
|
if title in ["", None]:
|
|
title = source_element.url.split("/")[-1]
|
|
|
|
if is_live:
|
|
title = f'Livestream_{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"]),
|
|
stirr_com.__name__
|
|
)
|
|
|
|
if response["content"].get("drm_protected", "").lower() not in ["", "no"]:
|
|
raise CustomException(ERR_MSG.format(
|
|
type=APP_ERROR,
|
|
url=source_element.url,
|
|
reason=f"DRM not supported. Widevine found",
|
|
solution=f"Extend the {stirr_com.__name__} service"
|
|
))
|
|
|
|
manifest = None
|
|
if manifest is None:
|
|
for k, f in [("media", None), ("_media", "src"), ("media_v2", "url")]:
|
|
if response.get(k, []) not in [None, []]:
|
|
if type(response[k]) is not list:
|
|
response[k] = [response[k]]
|
|
|
|
manifest = response[k][0]
|
|
if f is not None:
|
|
manifest = manifest.get(f, None)
|
|
if manifest is not None:
|
|
break
|
|
|
|
if manifest is None and is_live:
|
|
manifest = response["content"]["live"]
|
|
assert manifest not in [None, ""]
|
|
return manifest, None, {"IS_LIVE": is_live}
|
|
|
|
@staticmethod
|
|
def get_collection_elements(collection_url):
|
|
if "/watch/" in collection_url or "/live?channel_id=" in collection_url:
|
|
return [BaseElement(url=collection_url)]
|
|
if "/tv-shows/" not in collection_url:
|
|
return None
|
|
|
|
collection_url = clean_url(collection_url)
|
|
series_id = re.search(r"/tv-shows/([^/?]*)", collection_url).group(1)
|
|
response = requests.get(stirr_com.SERIES_URL.format(series_id=series_id))
|
|
|
|
status_code = response.status_code
|
|
response = response.content.decode()
|
|
response = json.loads(response)
|
|
message = response.get("message", "").lower()
|
|
|
|
if 400 <= status_code < 500 and "found" in message:
|
|
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 = None
|
|
for series in response.get("series", {}).get("series_by_id", []):
|
|
if str(series["series_id"]) != series_id:
|
|
continue
|
|
|
|
collection_name = series.get("series_name", None)
|
|
if collection_name not in ["", None]:
|
|
break
|
|
|
|
if collection_name in ["", None]:
|
|
collection_name = collection_url.split("/")[-1]
|
|
collection_name = join(
|
|
join(
|
|
str(builtins.CONFIG["DOWNLOAD_COMMANDS"]["OUTPUT_MEDIA_PATH"]),
|
|
stirr_com.__name__
|
|
),
|
|
get_valid_filename(collection_name)
|
|
)
|
|
|
|
response = requests.get(stirr_com.SEASONS_URL.format(series_id=series_id))
|
|
response = response.content.decode()
|
|
response = json.loads(response)
|
|
seasons = response.get("data", {}).get("seasons", [])
|
|
seasons = sorted(seasons, key=lambda s: s["sequence"])
|
|
|
|
collection = []
|
|
episode_count = 0
|
|
for season in seasons:
|
|
season_index = season["sequence"]
|
|
check = check_range(True, season_index, None)
|
|
if check is True:
|
|
continue
|
|
elif check is False:
|
|
return collection
|
|
|
|
season_id = season["season_id"]
|
|
page = 0
|
|
episode_index = 0
|
|
while True:
|
|
page += 1
|
|
response = requests.get(stirr_com.EPISODES_URL.format(
|
|
series_id=series_id, season_id=season_id, page=page
|
|
))
|
|
response = response.content.decode()
|
|
response = json.loads(response)
|
|
|
|
episodes = response.get("data", {}).get("data", [])
|
|
if len(episodes) == 0:
|
|
break
|
|
|
|
for episode in episodes:
|
|
# episode_index += 1
|
|
episode_index = episode["sequence"]
|
|
episode_index = episode_index - episode_count
|
|
check = check_range(False, season_index, episode_index)
|
|
if check in [True, False]:
|
|
continue
|
|
|
|
episode_id = episode["videoid"]
|
|
episode_title = episode.get("title", None)
|
|
episode_url = None
|
|
|
|
for f in ["watch_url", "url", "shareable"]:
|
|
if episode.get(f, "") in ["", None]:
|
|
continue
|
|
episode_url = episode[f]
|
|
break
|
|
|
|
if episode_url is None:
|
|
episode_url = stirr_com.EPISODE_URL.format(video_id=episode_id)
|
|
elif not episode_url.startswith("http"):
|
|
episode_url = stirr_com.BASE_URL + episode_url
|
|
|
|
if episode_title in ["", None]:
|
|
episode_title = clean_url(episode_url).split("/")[-1]
|
|
episode_title = f'Episode_{episode_index}_{episode_title}'
|
|
|
|
collection.append(BaseElement(
|
|
url=episode_url,
|
|
collection=join(collection_name, f'Season_{season_index}'),
|
|
element=get_valid_filename(episode_title)
|
|
))
|
|
episode_count += season["episode_count"]
|
|
|
|
return collection
|