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.0.0 Authorization: Cookies (No login require) + Credential (No need but can be provide) Security: FHD@L3 \b Notes: - Dont care about account or vip. can bypass all content """ TITLE_RE = r"^(?:https?://(?:www\.)?flixerapp\.com/video/)?(?P\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("-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,cdn): super().__init__(ctx) self.title=self.parse_series_id(title) self.original_lang = original_lang self.season = season self.is_movie=movie 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 credential: 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) 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) else: self.log.warning("Credential not found") 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()