260 lines
11 KiB
Python
260 lines
11 KiB
Python
import re
|
|
import click
|
|
|
|
from urllib.parse import urlparse ,urljoin
|
|
import hashlib
|
|
|
|
|
|
from unshackle.core.manifests import HLS
|
|
from unshackle.core.service import Service
|
|
from unshackle.core.titles import Episode, Movie, Movies, Series, Title_T, Titles_T
|
|
from unshackle.core.tracks import Subtitle, Tracks,Chapters,Track
|
|
from unshackle.core.downloaders import requests
|
|
|
|
import m3u8
|
|
|
|
from langcodes import Language
|
|
from hashlib import md5
|
|
|
|
|
|
class FLX(Service):
|
|
"""
|
|
Service code for Flixer streaming service (https://www.flixerapp.com/).
|
|
|
|
\b
|
|
Original-Author: [SeFree]
|
|
Version: 1.1.0
|
|
Authorization: Credentials + Cookies
|
|
Security: FHD@L3
|
|
\b
|
|
"""
|
|
|
|
TITLE_RE = r"^(?:https?://(?:www\.)?flixerapp\.com/video/)?(?P<id>\d+)"
|
|
LANGUAGE_MAP = {
|
|
"en": "en",
|
|
"th": "th",
|
|
"jp": "ja",
|
|
"ko": "ko",
|
|
"zh": "zh",
|
|
}
|
|
THAI_SOUND_FILTER={
|
|
"(Thai Sound)": "th"
|
|
}
|
|
@staticmethod
|
|
@click.command(name="Flixer", short_help="https://www.flixerapp.com/")
|
|
@click.argument("title", type=str, required=False)
|
|
@click.option("-ol", "--original_lang", default='ja', required=False, type=str,
|
|
help="If the title is foreigner audio language, specify the original language.")
|
|
@click.option("-SE", "--season", default=1, required=False, type=int,
|
|
help="BiliBili not provide Season in info, so specify it manually.")
|
|
@click.option("-m", "--movie", is_flag=True, default=False,
|
|
help="Download series as movie if listed as a series.)")
|
|
@click.option("-d", "--device_type", default='android', required=False, type=str,
|
|
help="Device type to use for API requests. Default is 'android'.")
|
|
|
|
@click.option("-c", "--cdn", default=None, required=False, type=str,
|
|
help="Overwrite CDN settings. Default is 'vps1'.")
|
|
@click.pass_context
|
|
def cli(ctx, **kwargs):
|
|
return FLX(ctx, **kwargs)
|
|
|
|
def parse_series_id(self, title_input: str) -> str:
|
|
"""Parse series ID from URL or direct ID input."""
|
|
match = re.match(self.TITLE_RE, title_input, re.IGNORECASE)
|
|
if not match:
|
|
raise ValueError(f"Could not parse series ID from: {title_input}")
|
|
series_id = match.group("id")
|
|
return series_id
|
|
|
|
def __init__(self, ctx, title, original_lang, season,movie,device_type,cdn):
|
|
super().__init__(ctx)
|
|
self.title=self.parse_series_id(title)
|
|
|
|
self.original_lang = original_lang
|
|
self.season = season
|
|
|
|
self.is_movie=movie
|
|
self.device_type = device_type
|
|
self.cdn=cdn
|
|
|
|
# def get_session(self):
|
|
|
|
# # Create a session using curl_cffi as it can impersonate browsers and avoid bot detection by Crunchyroll
|
|
# return session("chrome124")
|
|
|
|
def authenticate(self, cookies=None, credential=None) -> None:
|
|
"""Authenticate using username and password credentials, with refresh token support."""
|
|
super().authenticate(cookies, credential)
|
|
|
|
if self.device_type == "android":
|
|
|
|
data = f'screen-density=xhdpi&password={credential.password}&email={credential.username}'
|
|
|
|
# self.session.headers.update(self.config['headers'])
|
|
response = self.session.post(self.config['endpoints']['login'], headers=self.config['headers_android'], data=data)
|
|
|
|
elif self.device_type == "web":
|
|
|
|
|
|
files = {
|
|
'email': (None, credential.username),
|
|
'password': (None, credential.password),
|
|
'devicie_id': (None, 'web'),
|
|
'screen_density': (None, 'xxhdpi'),
|
|
'mac_address': (None, ''),
|
|
}
|
|
|
|
response=self.session.post(self.config['endpoints']['login'], files=files)
|
|
|
|
if response.status_code == 200 and response.json()['error_code'] == 110:
|
|
self.log.info("Login successful")
|
|
else:
|
|
self.log.error("Login failed. Please check credential")
|
|
exit(1)
|
|
|
|
def get_titles(self) -> Titles_T:
|
|
response=self.session.get(self.config['endpoints']['info'].format(id=self.title)).json()
|
|
if not response.get('error', True):
|
|
self.log.info("Success")
|
|
else:
|
|
self.log.error(f"Error fetching title info: {response.get('message', 'Unknown error')}")
|
|
exit(1)
|
|
media_title:str=response.get('result', {}).get('title', '').get('name', '')
|
|
media_title=media_title.replace("👑", "").strip()
|
|
episodes=response.get('result', {}).get('episodes', '')
|
|
titles=[]
|
|
if not self.is_movie:
|
|
|
|
for ep in episodes:
|
|
titles.append(
|
|
Episode(
|
|
id_=hashlib.md5(ep["player_api"].encode()).hexdigest()[0:6],
|
|
service=self.__class__,
|
|
title=media_title,
|
|
season=self.season,
|
|
number=int(ep['episode']),
|
|
data=ep,
|
|
language=self.original_lang
|
|
)
|
|
)
|
|
return Series(titles)
|
|
else:
|
|
return Movies(
|
|
[Movie(
|
|
id_=hashlib.md5(media_title.encode()).hexdigest()[0:6],
|
|
service=self.__class__,
|
|
name=media_title,
|
|
data=episodes[0],
|
|
language=self.original_lang
|
|
)]
|
|
)
|
|
|
|
|
|
def get_tracks(self, title: Title_T) -> Tracks:
|
|
tracks = Tracks()
|
|
media_info=self.session.get(self.config['endpoints']['episode'].format(id=self.title,ep=title.data['episode'])).json()
|
|
if media_info.get('error_code', None) == "":
|
|
self.log.info("Success")
|
|
else:
|
|
self.log.error(f"Error fetching title info: {media_info.get('error_message', 'Unknown error')}")
|
|
exit(1)
|
|
# print(media_info['result'])
|
|
# with open("test.json", 'w') as f:
|
|
# json.dump(media_info['result'], f,indent=4, ensure_ascii=False)
|
|
if self.cdn:
|
|
# Regex pattern to match vpsX or iX (where X is a number)
|
|
pattern = r"(https://)(vps[0-9]+|i[0-9]+)(\.flixerapp\.com)"
|
|
media_info['result']['playlist'] = re.sub(pattern, r"\1" + self.cdn + r"\3", media_info['result']['playlist'])
|
|
r = self.session.get(url=media_info['result']['playlist'])
|
|
res = r.text
|
|
|
|
parsed_url = urlparse(media_info['result']['playlist_android'])
|
|
for _, resolution in media_info['result']['source']['video'].items():
|
|
if _ == "auto":
|
|
continue
|
|
for lang, url in resolution.items():
|
|
if self.cdn:
|
|
# Regex pattern to match vpsX or iX (where X is a number)
|
|
pattern = r"(https://)(vps[0-9]+|i[0-9]+)(\.flixerapp\.com)"
|
|
url = re.sub(pattern, r"\1" + self.cdn + r"\3", url)
|
|
r = self.session.get(url=url)
|
|
res = r.text
|
|
if r.status_code != 200:
|
|
continue
|
|
tracks.add(HLS.from_text(res, url).to_tracks(Language.get(self.LANGUAGE_MAP[lang])))
|
|
|
|
for lang, url in media_info['result']['source'].get('audio', {}).items():
|
|
if self.cdn:
|
|
# Regex pattern to match vpsX or iX (where X is a number)
|
|
pattern = r"(https://)(vps[0-9]+|i[0-9]+)(\.flixerapp\.com)"
|
|
url = re.sub(pattern, r"\1" + self.cdn + r"\3", url)
|
|
r = self.session.get(url=url)
|
|
|
|
res = r.text
|
|
if r.status_code != 200 :
|
|
continue
|
|
tracks.add(HLS.from_text(res, url).to_tracks(Language.get(self.LANGUAGE_MAP[lang])))
|
|
tracks.subtitles.clear()
|
|
for lang, url in media_info['result']['source'].get('subtitle', {}).items():
|
|
url= urljoin(media_info['result']['playlist_android'], parsed_url.path.removesuffix("playlist.m3u8"))+urlparse(url).path.split('/')[-1].replace('_','__')+"_subtitle.m3u8?"+parsed_url.query
|
|
if self.cdn:
|
|
# Regex pattern to match vpsX or iX (where X is a number)
|
|
pattern = r"(https://)(vps[0-9]+|i[0-9]+)(\.flixerapp\.com)"
|
|
url = re.sub(pattern, r"\1" + self.cdn + r"\3", url)
|
|
|
|
r = self.session.get(url=url)
|
|
res = r.text
|
|
|
|
if 'placeholder' in res or r.status_code != 200:
|
|
continue
|
|
sub_load=m3u8.loads(res, url)
|
|
base_url= urljoin(media_info['result']['playlist_android'], parsed_url.path.removesuffix("playlist.m3u8"))
|
|
subtitle_urls=[]
|
|
|
|
for x in (sub_load.segments):
|
|
sub_url=urljoin(base_url,x.uri)
|
|
rr=self.session.get(url=sub_url)
|
|
if rr.status_code != 200:
|
|
continue
|
|
subtitle_urls.append(sub_url)
|
|
# print(subtitle_urls)
|
|
tracks.add(Subtitle(
|
|
id_=md5("_".join(subtitle_urls).encode()).hexdigest()[0:6],
|
|
url=subtitle_urls,
|
|
codec=Subtitle.Codec.from_mime(media_info['result'].get('subtitle_type',"srt")),
|
|
language=Language.get(self.LANGUAGE_MAP.get(lang, lang)),
|
|
))
|
|
# for track in tracks:
|
|
|
|
# if track.__class__.__name__ == "Subtitle":
|
|
# continue
|
|
# base_url= urljoin(media_info['result']['playlist'], parsed_url.path.removesuffix("playlist.m3u8"))
|
|
# res = self.session.get(track.url).text
|
|
# url_load=m3u8.loads(res, track.url)
|
|
# track_url=[]
|
|
# for url in (url_load.segments):
|
|
# track_url.append(url.absolute_uri)
|
|
# track.url=track_url
|
|
# track.descriptor=Track.Descriptor.URL
|
|
# # track.downloader=requests
|
|
|
|
return tracks
|
|
|
|
def get_chapters(self, title: Title_T) -> Chapters:
|
|
"""
|
|
Get Chapters for the Title.
|
|
|
|
Parameters:
|
|
title: The current Title from `get_titles` that is being processed.
|
|
|
|
You must return a Chapters object containing 0 or more Chapter objects.
|
|
|
|
You do not need to set a Chapter number or sort/order the chapters in any way as
|
|
the Chapters class automatically handles all of that for you. If there's no
|
|
descriptive name for a Chapter then do not set a name at all.
|
|
|
|
You must not set Chapter names to "Chapter {n}" or such. If you (or the user)
|
|
wants "Chapter {n}" style Chapter names (or similar) then they can use the config
|
|
option `chapter_fallback_name`. For example, `"Chapter {i:02}"` for "Chapter 01".
|
|
"""
|
|
return Chapters() |