import re import uuid from typing import Optional from http.cookiejar import CookieJar from langcodes import Language import click from unshackle.core.search_result import SearchResult from unshackle.core.credential import Credential 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 Chapter, Tracks class VIDO(Service): """ Vidio.com service, Series and Movies, login required. Version: 1.3.0 Supports URLs like: • https://www.vidio.com/premier/2978/giligilis (Series) • https://www.vidio.com/watch/7454613-marantau-short-movie (Movie) Note: Login is mandatory. Even free content requires valid session tokens for stream access (as per API behavior). """ # Updated regex to support both series and movies TITLE_RE = r"^https?://(?:www\.)?vidio\.com/(?:premier|series|watch)/(?P\d+)" NO_SUBTITLES = True @staticmethod @click.command(name="VIDO", short_help="https://vidio.com (login required)") @click.argument("title", type=str) @click.pass_context def cli(ctx, **kwargs): return VIDO(ctx, **kwargs) def __init__(self, ctx, title: str): super().__init__(ctx) match = re.match(self.TITLE_RE, title) if not match: raise ValueError(f"Unsupported or invalid Vidio URL: {title}") self.content_id = match.group("id") # Determine if it's a movie or series based on URL pattern self.is_movie = "watch" in title # Static app identifiers from Android traffic self.API_AUTH = "laZOmogezono5ogekaso5oz4Mezimew1" self.USER_AGENT = "vidioandroid/7.14.6-e4d1de87f2 (3191683)" self.API_APP_INFO = "android/15/7.14.6-e4d1de87f2-3191683" self.VISITOR_ID = str(uuid.uuid4()) # Auth state self._email = None self._user_token = None self._access_token = None def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None: if not credential or not credential.username or not credential.password: raise ValueError("Vidio requires email and password login.") self._email = credential.username password = credential.password headers = { "referer": "android-app://com.vidio.android", "x-api-platform": "app-android", "x-api-auth": self.API_AUTH, "user-agent": self.USER_AGENT, "x-api-app-info": self.API_APP_INFO, "accept-language": "en", "content-type": "application/x-www-form-urlencoded", "x-visitor-id": self.VISITOR_ID, } data = f"login={self._email}&password={password}" r = self.session.post("https://api.vidio.com/api/login", headers=headers, data=data) r.raise_for_status() auth_data = r.json() self._user_token = auth_data["auth"]["authentication_token"] self._access_token = auth_data["auth_tokens"]["access_token"] self.log.info(f"Authenticated as {self._email}") def _headers(self): if not self._user_token or not self._access_token: raise RuntimeError("Not authenticated. Call authenticate() first.") return { "referer": "android-app://com.vidio.android", "x-api-platform": "app-android", "x-api-auth": self.API_AUTH, "user-agent": self.USER_AGENT, "x-api-app-info": self.API_APP_INFO, "x-visitor-id": self.VISITOR_ID, "x-user-email": self._email, "x-user-token": self._user_token, "x-authorization": self._access_token, "accept-language": "en", "accept": "application/json", "accept-charset": "UTF-8", "content-type": "application/vnd.api+json", } def get_titles(self) -> Titles_T: headers = self._headers() if self.is_movie: # For movies, we need to get video details directly r = self.session.get(f"https://api.vidio.com/api/videos/{self.content_id}/detail", headers=headers) r.raise_for_status() video_data = r.json()["video"] # Extract year from publish_date if available year = None if video_data.get("publish_date"): try: year = int(video_data["publish_date"][:4]) except (ValueError, TypeError): pass return Movies([ Movie( id_=video_data["id"], service=self.__class__, name=video_data["title"], description=video_data.get("description", ""), year=year, language=Language.get("id"), data=video_data, ) ]) else: # For series, use the existing logic r = self.session.get(f"https://api.vidio.com/content_profiles/{self.content_id}", headers=headers) r.raise_for_status() root = r.json()["data"] series_title = root["attributes"]["title"] playlists = root["relationships"]["playlists"]["data"] if not playlists: raise ValueError("No season/playlist found for this series.") playlist_id = playlists[0]["id"] # Fetch all episodes episodes = [] page = 1 while True: r_eps = self.session.get( f"https://api.vidio.com/content_profiles/{self.content_id}/playlists/{playlist_id}/videos", params={"page[number]": page, "page[size]": 20, "sort": "order", "included": "upcoming_videos"}, headers=headers, ) r_eps.raise_for_status() page_data = r_eps.json() for raw_ep in page_data["data"]: attrs = raw_ep["attributes"] episodes.append( Episode( id_=int(raw_ep["id"]), service=self.__class__, title=series_title, season=1, number=len(episodes) + 1, name=attrs["title"], description=attrs.get("description", ""), language=Language.get("id"), data=raw_ep, ) ) if not page_data["links"].get("next"): break page += 1 return Series(episodes) def get_tracks(self, title: Title_T) -> Tracks: headers = self._headers() headers.update({ "x-device-brand": "samsung", "x-device-model": "SM-A525F", "x-device-form-factor": "phone", "x-device-soc": "Qualcomm SM7125", "x-device-os": "Android 15 (API 35)", "x-device-android-mpc": "0", "x-device-cpu-arch": "arm64-v8a", }) # Use the correct ID attribute based on title type video_id = str(title.id_) if hasattr(title, 'id_') else str(title.id) r = self.session.get( f"https://api.vidio.com/api/stream/v1/video_data/{video_id}?initialize=true", headers=headers, ) r.raise_for_status() stream = r.json() hls_url = stream.get("stream_hls_url") if not hls_url: raise ValueError("Stream URL not available. Possibly geo-blocked or subscription required.") return HLS.from_url(hls_url, session=self.session).to_tracks(language=title.language) def get_chapters(self, title: Title_T) -> list[Chapter]: return [] def search(self): raise NotImplementedError("Search not implemented for Vidio.")