Release v3.2.0

hidive/__init__.py

@@ -1,5 +0,0 @@
-from .client import *
-from .constants import *
-from .utils import *
-
-__version__ = '2.2.7'

hidive/client.py

@@ -1,319 +0,0 @@
-import time
-from urllib.parse import urlencode
-from uuid import UUID
-
-import requests
-from requests import Response
-
-from hidive.constants import View, Filter
-from hidive.utils import jwt_expired, b64encode
-
-
-class Client:
-    USER_AGENT = 'ExoDoris/2.2.7 (Linux;Android 10) AndroidXMedia3/1.0.2'
-
-    def __init__(self, access_token: str = None, refresh_token: str = None):
-        self.api = 'https://dce-frontoffice.imggaming.com/api'
-        self.guide = 'https://guide.imggaming.com'
-        self.algolia = 'https://h99xldr8mj-dsn.algolia.net'
-        self.user = {
-            'authorisationToken': access_token or '',
-            'refreshToken': refresh_token or ''
-        }
-
-    def __request(self, **kwargs) -> Response:
-        headers = {
-            'Accept': '*/*',
-            'User-Agent': 'okhttp/4.9.2',
-            **kwargs.get('headers', {})
-        }
-
-        url: str = kwargs.get('url')
-        params = kwargs.get('params', {})
-
-        if url.startswith(self.api) or url.startswith(self.guide):
-            headers['app'] = 'dice'
-            headers['realm'] = 'dce.hidive'
-            headers['x-api-key'] = '4dc1e8df-5869-41ea-95c2-6f04c67459ed'
-            headers['x-app-var'] = '2.2.7'
-
-            if kwargs.get('auth', True):
-                headers['Authorization'] = self.__token()
-        elif url.startswith(self.algolia):
-            params['x-algolia-agent'] = 'Algolia for JavaScript (3.35.1); React Native'
-            params['x-algolia-application-id'] = 'H99XLDR8MJ'
-            params['x-algolia-api-key'] = 'e55ccb3db0399eabe2bfc37a0314c346'
-
-        r = requests.request(
-            method=kwargs.get('method', 'GET'),
-            url=url,
-            params=params,
-            json=kwargs.get('json'),
-            data=kwargs.get('data'),
-            headers=headers
-        )
-        r.raise_for_status()
-        return r
-
-    def __token(self) -> str:
-        access_token = self.user.get('authorisationToken')
-        if not access_token:
-            self.guest()
-        elif jwt_expired(access_token):
-            self.refresh()
-        return 'Bearer %s' % self.user['authorisationToken']
-
-    # @package: login
-    def guest(self) -> dict:
-        content = self.__request(method='POST', url=f'{self.api}/v2/login/guest/checkin', auth=False).json()
-        self.user.update(content)
-        return content
-
-    def refresh(self) -> dict:
-        refresh_token = self.user['refreshToken']
-        if jwt_expired(refresh_token):
-            raise ValueError('Refresh token expired')
-
-        content = self.__request(
-            method='POST',
-            url=f'{self.api}/v2/token/refresh',
-            json={'refreshToken': refresh_token}
-        ).json()
-        self.user.update(content)
-        return content
-
-    def login(self, email: str, password: str) -> dict:
-        content = self.__request(
-            method='POST',
-            url=f'{self.api}/v2/login',
-            json={'id': email, 'secret': password}
-        ).json()
-        self.user.update(content)
-        return content
-
-    def reset(self, email: str) -> dict:
-        return self.__request(
-            method='POST',
-            url=f'{self.api}/v2/reset-password/create',
-            json={'id': email, 'provider': 'ID'},
-            auth=False
-        ).json()
-
-    def create(self, email: str, password: str) -> dict:
-        content = self.__request(
-            method='POST',
-            url=f'{self.api}/v2/user',
-            json={
-                'email': email,
-                'secret': password,
-                'consentAnswers': [{
-                    'answer': f['required'],
-                    'promptField': f['fieldName']
-                } for f in self.consent()['fields']]
-            }
-        ).json()
-        self.user.update(content)
-        return content
-
-    # @package: realm
-    def providers(self) -> list[dict]:
-        return self.__request(
-            method='GET',
-            url=f'{self.api}/v2/realm/authentication-providers'
-        ).json()['authenticationProviders']
-
-    def settings(self) -> dict:
-        return self.__request(method='GET', url=f'{self.api}/v2/realm-settings/realm/dce.hidive').json()
-
-    def label(self) -> dict:
-        return self.__request(method='GET', url=f'{self.api}/v2/label/dce.hidive').json()
-
-    def licence(self) -> dict:
-        return self.__request(method='GET', url=f'{self.api}/v2/licence').json()
-
-    def consent(self) -> dict:
-        return self.__request(method='GET', url=f'{self.api}/v2/consent-prompt').json()
-
-    # @package: user
-    def preferences(self, auto_play: bool = None) -> dict:
-        method = 'GET' if auto_play is None else 'PUT'
-        data = None if auto_play is None else {'autoAdvance': auto_play}
-        return self.__request(method=method, url=f'{self.api}/v2/user/preferences', json=data).json()
-
-    def profile(self) -> dict:
-        return self.__request(method='GET', url=f'{self.api}/v2/user/profile').json()
-
-    def notification(self, limit: int = 1) -> dict:
-        return self.__request(
-            method='GET',
-            url=f'{self.api}/v2/promo-notification',
-            params={'maxNotifications': limit}
-        ).json()
-
-    def watch_create(self, name: str) -> dict:
-        return self.__request(
-            method='POST',
-            url=f'{self.api}/v3/user/watchlist',
-            json={'name': name}
-        ).json()
-
-    def watch_delete(self, watch_id: int) -> None:
-        href = b64encode('480211|dce.hidive')
-        self.__request(method='DELETE', url=f'{self.api}/v3/user/watchlist/{href}/{watch_id}')
-
-    def watch(self, limit: int = 25) -> dict:
-        return self.__request(method='GET', url=f'{self.api}/v3/user/watchlist', params={'rpp': limit}).json()
-
-    def watch_add(self, watch_id: int, content_type: str, content_id: int) -> dict:
-        return self.__request(
-            method='POST',
-            url=f'{self.api}/v4/user/watchlist/{watch_id}/content',
-            json={'content': [{'contentType': content_type, 'id': content_id}]}
-        ).json()
-
-    def watch_remove(self, watch_id: int, content_type: str, content_id: int) -> None:
-        self.__request(
-            method='DELETE',
-            url=f'{self.api}/v4/user/watchlist/{watch_id}/content/{content_type}/{content_id}'
-        )
-
-    def watch_details(self, watch_id: int, limit: int = 25) -> dict:
-        return self.__request(
-            method='GET',
-            url=f'{self.api}/v4/user/watchlist/{watch_id}',
-            params={'rpp': limit}
-        ).json()
-
-    # @package: content
-    def menu(self) -> list[dict]:
-        return self.__request(method='GET', url=f'{self.api}/v2/menu-items').json()
-
-    def event(self, limit: int = 20) -> dict:
-        return self.__request(method='GET', url=f'{self.api}/v2/event/live', params={'rpp': limit}).json()
-
-    def content(self, title: str = 'home') -> dict:
-        # @Default: home | browse
-        return self.__request(
-            method='GET',
-            url=f'{self.api}/v4/content/{title}',
-            params={
-                'bpp': 10,
-                'rpp': 12,
-                'displaySectionLinkBuckets': 'SHOW',
-                'displayEpgBuckets': 'SHOW',
-                'displayEmptyBucketShortcuts': 'SHOW',
-                'displayGeoblocked': 'SHOW',
-                'bspp': 20
-            }
-        ).json()
-
-    def popular(self) -> dict:
-        return self.__request(method='GET', url=f'{self.api}/v2/popular').json()
-
-    def view(self, view: View, view_id: int) -> dict:
-        return self.__request(
-            method='GET',
-            url=f'{self.api}/v1/view',
-            params={'type': view.value, 'id': view_id}
-        ).json()
-
-    def series(self, series_id: int, limit: int = 1) -> dict:
-        return self.__request(method='GET', url=f'{self.api}/v4/series/{series_id}', params={'rpp': limit}).json()
-
-    def query(
-            self,
-            query: str,
-            facets: list = None,
-            facet_filters: list = None,
-            page: int = 0,
-            filters: Filter = None
-    ) -> dict:
-        params = {
-            'facets': facets or [],
-            'query': query,
-            'facetFilters': facet_filters or []
-        }
-        if filters:
-            params['filters'] = f'type:{filters.value}'
-        else:
-            params['page'] = page
-
-        return self.__request(
-            method='POST',
-            url=f'{self.algolia}/1/indexes/prod-dce.hidive-livestreaming-events/query',
-            json={'params': urlencode(params)}
-        ).json()
-
-    # @package: vod
-    def live(self, video_id: int) -> dict:
-        return self.__request(method='GET', url=f'{self.api}/v4/vods/live/{video_id}').json()
-
-    def stream(self, video_id: int) -> dict:
-        return self.__request(method='GET', url=f'{self.api}/v3/stream/vod/{video_id}').json()
-
-    def details(self, video_id: int, playback: bool = True) -> dict:
-        return self.__request(
-            method='GET',
-            url=f'{self.api}/v2/vod/{video_id}',
-            params={'includePlaybackDetails': 'URL'} if playback else None,
-            headers={
-                'cm-app-bundle': 'com.hidive.android',
-                'cm-app-name': 'HIDIVE',
-                'cm-app-storeid': 'com.hidive.android',
-                'cm-app-version': '2.2.7',
-                'cm-cst-ifa': '',
-                'cm-cst-lat': '0',
-                'cm-cst-tcf': '',
-                'cm-cst-usp': '',
-                'cm-dvc-dnt': '0',
-                'cm-dvc-h': '720',
-                'cm-dvc-lang': 'en_US',
-                'cm-dvc-make': 'Xiaomi',
-                'cm-dvc-model': 'Mi A2',
-                'cm-dvc-os': '2',
-                'cm-dvc-osv': '10',
-                'cm-dvc-type': '4',
-                'cm-dvc-w': '360'
-            }
-        ).json()
-
-    def adjacent(self, video_id: int, limit: int = 20) -> dict:
-        return self.__request(
-            method='GET',
-            url=f'{self.api}/v4/vod/{video_id}/adjacent',
-            params={'size': limit}
-        ).json()
-
-    # @package: player
-    def progress(self, video_id: int, cid: str, progress: int) -> dict:
-        return self.__request(
-            method='PUT',
-            url=f'{self.guide}/prod',
-            params={
-                'action': 2,
-                'cid': cid,
-                'nature': 'last',
-                'progress': progress,
-                'startedAt': round(time.time()),
-                'video': video_id
-            }
-        ).json()
-
-    def playback(self, url: str) -> dict:
-        return self.__request(method='GET', url=url, auth=False).json()
-
-    def drm(self, url: str, token: str, key_ids: list[UUID], challenge: bytes) -> bytes:
-        # @Info: widevine DRM
-        return self.__request(
-            method='POST',
-            url=url,
-            data=challenge,
-            headers={
-                'Authorization': f'Bearer {token}',
-                'User-Agent': 'Dice Shield/2.2.7 (Linux;Android 10) AndroidXMedia3/1.0.2',
-                'X-DRM-INFO': b64encode({
-                    'system': 'com.widevine.alpha',
-                    'key_ids': [str(k) for k in key_ids]
-                })
-            }
-        ).content

hidive/constants.py

@@ -1,20 +0,0 @@
-from enum import Enum
-
-
-class Filter(Enum):
-    SERIES = 'VOD_SERIES'
-    VIDEO = 'VOD_VIDEO'
-    EVENT = 'LIVE_EVENT'
-    PLAYLIST = 'VOD_PLAYLIST'
-
-
-class Format(Enum):
-    VTT = 'vtt'
-    SRT = 'srt'
-    SCC = 'scc'
-
-
-class View(Enum):
-    SEASON = 'season'
-    PLAYLIST = 'playlist'
-    VOD = 'VOD'

hidive/utils.py

@@ -1,31 +0,0 @@
-import base64
-import json
-import re
-import time
-from typing import Union
-
-
-def b64decode(value: str) -> bytes:
-    return base64.b64decode(value + '=' * (4 - len(value) % 4))
-
-
-def b64encode(value: Union[str, bytes, dict, list]) -> str:
-    if isinstance(value, (dict, list)):
-        value = json.dumps(value, separators=(',', ':'))
-    if isinstance(value, str):
-        value = value.encode('utf-8')
-    return base64.b64encode(value).decode('utf-8')
-
-
-def jwt_expired(value: str) -> bool:
-    header, payload, signature = value.split('.', 2)
-    payload = json.loads(b64decode(payload))
-    return payload['exp'] < round(time.time())
-
-
-def parse_pssh(value: str) -> str:
-    lines = value.splitlines()
-    for i, line in enumerate(lines):
-        if 'urn:uuid:edef8ba9-79d6-4ace-a3c8-27dcd51d21ed' in line:
-            return re.search(r'<cenc:pssh>(.+)</cenc:pssh>', lines[i + 1]).group(1)
-    raise ValueError('Could not find pssh')

main.py

@@ -1,88 +0,0 @@
-import requests
-from pathlib import Path
-
-from hidive.client import Client
-from hidive.constants import Format, View
-from hidive.utils import parse_pssh, b64encode
-
-from pywidevine.cdm import Cdm
-from pywidevine.device import Device
-from pywidevine.pssh import PSSH
-
-DEVICE = Path() / 'DEVICE.wvd'
-
-if __name__ == '__main__':
-    client = Client()
-    client.login(email='EMAIL', password='PASSWORD')
-
-    # @Info: series
-    series = client.series(series_id=1049)
-    print('[+] Series: %s' % series['title'])
-    print('[+] Cover: %s' % series['coverUrl'])
-    print('[+] Poster: %s' % series['posterUrl'])
-
-    # @Info: seasons
-    for view in series['seasons']:
-        print('[+] Season: %s' % view['seasonNumber'])
-        season = client.view(view=View.SEASON, view_id=view['id'])
-
-        # @Info: episodes
-        bucket = next(e for e in season['elements'] if e['$type'] == 'bucket')
-        for episode in bucket['attributes']['items']:
-            video_id = episode['id']
-            print('[+] Title: %s' % episode['title'])
-            print('[+] Description: %s' % episode['description'])
-
-            details = client.details(video_id=video_id, playback=True)
-            print('[+] Thumbnail: %s' % details['thumbnailUrl'])
-
-            # @Info: subtitles
-            playback = client.playback(url=details['playerUrlCallback'])
-            dash = playback['dash'][0]
-            for subtitle in dash['subtitles']:
-                if subtitle['format'] == Format.VTT:
-                    print('[+] Subtitle (%s): %s' % (subtitle['language'], subtitle['url']))
-
-            # @Info: widevine DRM
-            manifest_url = dash['url']
-            print(f'[+] Manifest: {manifest_url}')
-            assert 'WIDEVINE' in dash['drm']['keySystems'], 'Unsupported DRM system'
-
-            manifest = requests.request(method='GET', url=manifest_url, headers={'User-Agent': Client.USER_AGENT}).text
-            pssh = PSSH(parse_pssh(manifest))
-            print(f'[+] PSSH: {pssh}')
-
-            device = Device.load(DEVICE)
-            cdm = Cdm.from_device(device)
-            session_id = cdm.open()
-
-            challenge = cdm.get_license_challenge(session_id, pssh)
-            print('[+] Challenge: %s' % b64encode(challenge))
-
-            licence = client.drm(
-                url=dash['drm']['url'],
-                token=dash['drm']['jwtToken'],
-                key_ids=pssh.key_ids,
-                challenge=challenge
-            )
-            print(f'[+] Licence: {b64encode(licence)}')
-
-            cdm.parse_license(session_id, licence)
-            keys = cdm.get_keys(session_id, type_='CONTENT')
-            if not keys:
-                raise ValueError('Could not find key')
-
-            keys = [f'{k.kid.hex}:{k.key.hex()}' for k in keys]
-            cdm.close(session_id)
-
-            print('[+] Keys: %s' % '|'.join(keys))
-
-            # @Info: download
-            print('[+] Prompt: %s' % ' '.join([
-                'N_m3u8DL-RE',
-                f'"{manifest_url}"',
-                *[f'--key "{k}"' for k in keys],
-                '--header',
-                f'"User-Agent: {Client.USER_AGENT}"'
-            ]))
-            exit(0)

profiles/.idea/.gitignore

@@ -1,8 +0,0 @@
-# Default ignored files
-/shelf/
-/workspace.xml
-# Editor-based HTTP Client requests
-/httpRequests/
-# Datasource local storage ignored files
-/dataSources/
-/dataSources.local.xml

profiles/.idea/inspectionProfiles/profiles_settings.xml

@@ -1,6 +0,0 @@
-<component name="InspectionProjectProfileManager">
-  <settings>
-    <option name="USE_PROJECT_PROFILE" value="false" />
-    <version value="1.0" />
-  </settings>
-</component>

profiles/.idea/misc.xml

@@ -1,7 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project version="4">
-  <component name="Black">
-    <option name="sdkName" value="Python 3.10" />
-  </component>
-  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.10" project-jdk-type="Python SDK" />
-</project>

profiles/.idea/modules.xml

@@ -1,8 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project version="4">
-  <component name="ProjectModuleManager">
-    <modules>
-      <module fileurl="file://$PROJECT_DIR$/.idea/profiles.iml" filepath="$PROJECT_DIR$/.idea/profiles.iml" />
-    </modules>
-  </component>
-</project>

profiles/.idea/profiles.iml

@@ -1,8 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<module type="PYTHON_MODULE" version="4">
-  <component name="NewModuleRootManager">
-    <content url="file://$MODULE_DIR$" />
-    <orderEntry type="inheritedJdk" />
-    <orderEntry type="sourceFolder" forTests="false" />
-  </component>
-</module>