From 6a93794737981247a1acb72704c172ef858153ad Mon Sep 17 00:00:00 2001 From: Sam Chudnick Date: Sun, 17 Jul 2022 20:20:23 -0400 Subject: Initial commit --- jellyfin_apiclient_python/__init__.py | 133 +++++ jellyfin_apiclient_python/api.py | 653 ++++++++++++++++++++++++ jellyfin_apiclient_python/client.py | 87 ++++ jellyfin_apiclient_python/configuration.py | 53 ++ jellyfin_apiclient_python/connection_manager.py | 379 ++++++++++++++ jellyfin_apiclient_python/credentials.py | 128 +++++ jellyfin_apiclient_python/exceptions.py | 11 + jellyfin_apiclient_python/http.py | 267 ++++++++++ jellyfin_apiclient_python/keepalive.py | 20 + jellyfin_apiclient_python/timesync_manager.py | 140 +++++ jellyfin_apiclient_python/ws_client.py | 140 +++++ 11 files changed, 2011 insertions(+) create mode 100644 jellyfin_apiclient_python/__init__.py create mode 100644 jellyfin_apiclient_python/api.py create mode 100644 jellyfin_apiclient_python/client.py create mode 100644 jellyfin_apiclient_python/configuration.py create mode 100644 jellyfin_apiclient_python/connection_manager.py create mode 100644 jellyfin_apiclient_python/credentials.py create mode 100644 jellyfin_apiclient_python/exceptions.py create mode 100644 jellyfin_apiclient_python/http.py create mode 100644 jellyfin_apiclient_python/keepalive.py create mode 100644 jellyfin_apiclient_python/timesync_manager.py create mode 100644 jellyfin_apiclient_python/ws_client.py (limited to 'jellyfin_apiclient_python') diff --git a/jellyfin_apiclient_python/__init__.py b/jellyfin_apiclient_python/__init__.py new file mode 100644 index 0000000..dcc660f --- /dev/null +++ b/jellyfin_apiclient_python/__init__.py @@ -0,0 +1,133 @@ +# -*- coding: utf-8 -*- +from __future__ import division, absolute_import, print_function, unicode_literals + +################################################################################################# + +import logging + +from .client import JellyfinClient + +################################################################################################# + + +class NullHandler(logging.Handler): + def emit(self, record): + print(self.format(record)) + + +loghandler = NullHandler +LOG = logging.getLogger('Jellyfin') + +################################################################################################# + + +def config(level=logging.INFO): + + logger = logging.getLogger('Jellyfin') + logger.addHandler(Jellyfin.loghandler()) + logger.setLevel(level) + +def has_attribute(obj, name): + try: + object.__getattribute__(obj, name) + return True + except AttributeError: + return False + +def ensure_client(): + + def decorator(func): + def wrapper(self, *args, **kwargs): + + if self.client.get(self.server_id) is None: + self.construct() + + return func(self, *args, **kwargs) + + return wrapper + return decorator + + +class Jellyfin(object): + + ''' This is your Jellyfinclient, you can create more than one. The server_id is only a temporary thing + to communicate with the JellyfinClient(). + + from jellyfin import Jellyfin + + Jellyfin('123456').config.data['app'] + + # Permanent client reference + client = Jellyfin('123456').get_client() + client.config.data['app'] + ''' + + # Borg - multiple instances, shared state + _shared_state = {} + client = {} + server_id = "default" + loghandler = loghandler + + def __init__(self, server_id=None): + self.__dict__ = self._shared_state + self.server_id = server_id or "default" + + def get_client(self): + return self.client[self.server_id] + + @classmethod + def set_loghandler(cls, func=loghandler, level=logging.INFO): + + for handler in logging.getLogger('Jellyfin').handlers: + if isinstance(handler, cls.loghandler): + logging.getLogger('Jellyfin').removeHandler(handler) + + cls.loghandler = func + config(level) + + def close(self): + + if self.server_id not in self.client: + return + + self.client[self.server_id].stop() + self.client.pop(self.server_id, None) + + LOG.info("---[ STOPPED JELLYFINCLIENT: %s ]---", self.server_id) + + @classmethod + def close_all(cls): + + for client in cls.client: + cls.client[client].stop() + + cls.client = {} + LOG.info("---[ STOPPED ALL JELLYFINCLIENTS ]---") + + @classmethod + def get_active_clients(cls): + return cls.client + + @ensure_client() + def __setattr__(self, name, value): + + if has_attribute(self, name): + return super(Jellyfin, self).__setattr__(name, value) + + setattr(self.client[self.server_id], name, value) + + @ensure_client() + def __getattr__(self, name): + return getattr(self.client[self.server_id], name) + + def construct(self): + + self.client[self.server_id] = JellyfinClient() + + if self.server_id == 'default': + LOG.info("---[ START JELLYFINCLIENT ]---") + else: + LOG.info("---[ START JELLYFINCLIENT: %s ]---", self.server_id) + + +config() diff --git a/jellyfin_apiclient_python/api.py b/jellyfin_apiclient_python/api.py new file mode 100644 index 0000000..a6df708 --- /dev/null +++ b/jellyfin_apiclient_python/api.py @@ -0,0 +1,653 @@ +# -*- coding: utf-8 -*- +from __future__ import division, absolute_import, print_function, unicode_literals +from datetime import datetime +import requests +import json +import logging + +LOG = logging.getLogger('JELLYFIN.' + __name__) + + +def jellyfin_url(client, handler): + return "%s/%s" % (client.config.data['auth.server'], handler) + + +def basic_info(): + return "Etag" + + +def info(): + return ( + "Path,Genres,SortName,Studios,Writer,Taglines,LocalTrailerCount," + "OfficialRating,CumulativeRunTimeTicks,ItemCounts," + "Metascore,AirTime,DateCreated,People,Overview," + "CriticRating,CriticRatingSummary,Etag,ShortOverview,ProductionLocations," + "Tags,ProviderIds,ParentId,RemoteTrailers,SpecialEpisodeNumbers," + "MediaSources,VoteCount,RecursiveItemCount,PrimaryImageAspectRatio" + ) + + +def music_info(): + return ( + "Etag,Genres,SortName,Studios,Writer," + "OfficialRating,CumulativeRunTimeTicks,Metascore," + "AirTime,DateCreated,MediaStreams,People,ProviderIds,Overview,ItemCounts" + ) + + +class API(object): + + ''' All the api calls to the server. + ''' + def __init__(self, client, *args, **kwargs): + self.client = client + self.config = client.config + self.default_timeout = 5 + + def _http(self, action, url, request={}): + request.update({'type': action, 'handler': url}) + + return self.client.request(request) + + def _http_url(self, action, url, request={}): + request.update({"type": action, "handler": url}) + + return self.client.request_url(request) + + def _http_stream(self, action, url, dest_file, request={}): + request.update({'type': action, 'handler': url}) + + self.client.request(request, dest_file=dest_file) + + def _get(self, handler, params=None): + return self._http("GET", handler, {'params': params}) + + def _get_url(self, handler, params=None): + return self._http_url("GET", handler, {"params": params}) + + def _post(self, handler, json=None, params=None): + return self._http("POST", handler, {'params': params, 'json': json}) + + def _delete(self, handler, params=None): + return self._http("DELETE", handler, {'params': params}) + + def _get_stream(self, handler, dest_file, params=None): + self._http_stream("GET", handler, dest_file, {'params': params}) + + ################################################################################################# + + # Bigger section of the Jellyfin api + + ################################################################################################# + + def try_server(self): + return self._get("System/Info/Public") + + def sessions(self, handler="", action="GET", params=None, json=None): + if action == "POST": + return self._post("Sessions%s" % handler, json, params) + elif action == "DELETE": + return self._delete("Sessions%s" % handler, params) + else: + return self._get("Sessions%s" % handler, params) + + def users(self, handler="", action="GET", params=None, json=None): + if action == "POST": + return self._post("Users/{UserId}%s" % handler, json, params) + elif action == "DELETE": + return self._delete("Users/{UserId}%s" % handler, params) + else: + return self._get("Users/{UserId}%s" % handler, params) + + def items(self, handler="", action="GET", params=None, json=None): + if action == "POST": + return self._post("Items%s" % handler, json, params) + elif action == "DELETE": + return self._delete("Items%s" % handler, params) + else: + return self._get("Items%s" % handler, params) + + def user_items(self, handler="", params=None): + return self.users("/Items%s" % handler, params=params) + + def shows(self, handler, params): + return self._get("Shows%s" % handler, params) + + def videos(self, handler): + return self._get("Videos%s" % handler) + + def artwork(self, item_id, art, max_width, ext="jpg", index=None): + params = {"MaxWidth": max_width, "format": ext} + handler = ("Items/%s/Images/%s" % (item_id, art) if index is None + else "Items/%s/Images/%s/%s" % (item_id, art, index) + ) + + return self._get_url(handler, params) + + def audio_url(self, item_id, container=None, audio_codec=None, max_streaming_bitrate=140000000): + params = { + "UserId": "{UserId}", + "DeviceId": "{DeviceId}", + "MaxStreamingBitrate": max_streaming_bitrate, + } + + if container: + params["Container"] = container + + if audio_codec: + params["AudioCodec"] = audio_codec + + return self._get_url("Audio/%s/universal" % item_id, params) + + def video_url(self, item_id, media_source_id=None): + params = { + "static": "true", + "DeviceId": "{DeviceId}" + } + if media_source_id is not None: + params["MediaSourceId"] = media_source_id + + return self._get_url("Videos/%s/stream" % item_id, params) + + def download_url(self, item_id): + params = {} + return self._get_url("Items/%s/Download" % item_id, params) + + ################################################################################################# + + # More granular api + + ################################################################################################# + + def get_users(self): + return self._get("Users") + + def get_public_users(self): + return self._get("Users/Public") + + def get_user(self, user_id=None): + return self.users() if user_id is None else self._get("Users/%s" % user_id) + + def get_user_settings(self, client="emby"): + return self._get("DisplayPreferences/usersettings", params={ + "userId": "{UserId}", + "client": client + }) + + def get_views(self): + return self.users("/Views") + + def get_media_folders(self): + return self.users("/Items") + + def get_item(self, item_id): + return self.users("/Items/%s" % item_id) + + def get_items(self, item_ids): + return self.users("/Items", params={ + 'Ids': ','.join(str(x) for x in item_ids), + 'Fields': info() + }) + + def get_sessions(self): + return self.sessions(params={'ControllableByUserId': "{UserId}"}) + + def get_device(self, device_id): + return self.sessions(params={'DeviceId': device_id}) + + def post_session(self, session_id, url, params=None, data=None): + return self.sessions("/%s/%s" % (session_id, url), "POST", params, data) + + def get_images(self, item_id): + return self.items("/%s/Images" % item_id) + + def get_suggestion(self, media="Movie,Episode", limit=1): + return self.users("/Suggestions", params={ + 'Type': media, + 'Limit': limit + }) + + def get_recently_added(self, media=None, parent_id=None, limit=20): + return self.user_items("/Latest", { + 'Limit': limit, + 'UserId': "{UserId}", + 'IncludeItemTypes': media, + 'ParentId': parent_id, + 'Fields': info() + }) + + def get_next(self, index=None, limit=1): + return self.shows("/NextUp", { + 'Limit': limit, + 'UserId': "{UserId}", + 'StartIndex': None if index is None else int(index) + }) + + def get_adjacent_episodes(self, show_id, item_id): + return self.shows("/%s/Episodes" % show_id, { + 'UserId': "{UserId}", + 'AdjacentTo': item_id, + 'Fields': "Overview" + }) + + def get_season(self, show_id, season_id): + return self.shows("/%s/Episodes" % show_id, { + 'UserId': "{UserId}", + 'SeasonId': season_id + }) + + def get_genres(self, parent_id=None): + return self._get("Genres", { + 'ParentId': parent_id, + 'UserId': "{UserId}", + 'Fields': info() + }) + + def get_recommendation(self, parent_id=None, limit=20): + return self._get("Movies/Recommendations", { + 'ParentId': parent_id, + 'UserId': "{UserId}", + 'Fields': info(), + 'Limit': limit + }) + + # Modified + def get_items_by_letter(self, parent_id=None, media=None, letter=None, recurse=True): + return self.user_items(params={ + 'ParentId': parent_id, + 'NameStartsWith': letter, + 'Fields': info(), + 'Recursive': recurse, + 'IncludeItemTypes': media + }) + + def search_media_items(self, term=None, media=None, limit=20): + return self.user_items(params={ + 'searchTerm': term, + 'Recursive': True, + 'IncludeItemTypes': media, + 'Limit': limit + }) + + def get_channels(self): + return self._get("LiveTv/Channels", { + 'UserId': "{UserId}", + 'EnableImages': True, + 'EnableUserData': True + }) + + def get_intros(self, item_id): + return self.user_items("/%s/Intros" % item_id) + + def get_additional_parts(self, item_id): + return self.videos("/%s/AdditionalParts" % item_id) + + def delete_item(self, item_id): + return self.items("/%s" % item_id, "DELETE") + + def get_local_trailers(self, item_id): + return self.user_items("/%s/LocalTrailers" % item_id) + + def get_transcode_settings(self): + return self._get('System/Configuration/encoding') + + def get_ancestors(self, item_id): + return self.items("/%s/Ancestors" % item_id, params={ + 'UserId': "{UserId}" + }) + + def get_items_theme_video(self, parent_id): + return self.users("/Items", params={ + 'HasThemeVideo': True, + 'ParentId': parent_id + }) + + def get_themes(self, item_id): + return self.items("/%s/ThemeMedia" % item_id, params={ + 'UserId': "{UserId}", + 'InheritFromParent': True + }) + + def get_items_theme_song(self, parent_id): + return self.users("/Items", params={ + 'HasThemeSong': True, + 'ParentId': parent_id + }) + + def get_plugins(self): + return self._get("Plugins") + + def check_companion_installed(self): + try: + self._get("/Jellyfin.Plugin.KodiSyncQueue/GetServerDateTime") + return True + except Exception: + return False + + def get_seasons(self, show_id): + return self.shows("/%s/Seasons" % show_id, params={ + 'UserId': "{UserId}", + 'EnableImages': True, + 'Fields': info() + }) + + def get_date_modified(self, date, parent_id, media=None): + return self.users("/Items", params={ + 'ParentId': parent_id, + 'Recursive': False, + 'IsMissing': False, + 'IsVirtualUnaired': False, + 'IncludeItemTypes': media or None, + 'MinDateLastSaved': date, + 'Fields': info() + }) + + def get_userdata_date_modified(self, date, parent_id, media=None): + return self.users("/Items", params={ + 'ParentId': parent_id, + 'Recursive': True, + 'IsMissing': False, + 'IsVirtualUnaired': False, + 'IncludeItemTypes': media or None, + 'MinDateLastSavedForUser': date, + 'Fields': info() + }) + + def refresh_item(self, item_id): + return self.items("/%s/Refresh" % item_id, "POST", json={ + 'Recursive': True, + 'ImageRefreshMode': "FullRefresh", + 'MetadataRefreshMode': "FullRefresh", + 'ReplaceAllImages': False, + 'ReplaceAllMetadata': True + }) + + def favorite(self, item_id, option=True): + return self.users("/FavoriteItems/%s" % item_id, "POST" if option else "DELETE") + + def get_system_info(self): + return self._get("System/Configuration") + + def post_capabilities(self, data): + return self.sessions("/Capabilities/Full", "POST", json=data) + + def session_add_user(self, session_id, user_id, option=True): + return self.sessions("/%s/Users/%s" % (session_id, user_id), "POST" if option else "DELETE") + + def session_playing(self, data): + return self.sessions("/Playing", "POST", json=data) + + def session_progress(self, data): + return self.sessions("/Playing/Progress", "POST", json=data) + + def session_stop(self, data): + return self.sessions("/Playing/Stopped", "POST", json=data) + + def item_played(self, item_id, watched): + return self.users("/PlayedItems/%s" % item_id, "POST" if watched else "DELETE") + + def get_sync_queue(self, date, filters=None): + return self._get("Jellyfin.Plugin.KodiSyncQueue/{UserId}/GetItems", params={ + 'LastUpdateDT': date, + 'filter': filters or None + }) + + def get_server_time(self): + return self._get("Jellyfin.Plugin.KodiSyncQueue/GetServerDateTime") + + def get_play_info(self, item_id, profile, aid=None, sid=None, start_time_ticks=None, is_playback=True): + args = { + 'UserId': "{UserId}", + 'DeviceProfile': profile, + 'AutoOpenLiveStream': is_playback, + 'IsPlayback': is_playback + } + if sid: + args['SubtitleStreamIndex'] = sid + if aid: + args['AudioStreamIndex'] = aid + if start_time_ticks: + args['StartTimeTicks'] = start_time_ticks + return self.items("/%s/PlaybackInfo" % item_id, "POST", json=args) + + def get_live_stream(self, item_id, play_id, token, profile): + return self._post("LiveStreams/Open", json={ + 'UserId': "{UserId}", + 'DeviceProfile': profile, + 'OpenToken': token, + 'PlaySessionId': play_id, + 'ItemId': item_id + }) + + def close_live_stream(self, live_id): + return self._post("LiveStreams/Close", json={ + 'LiveStreamId': live_id + }) + + def close_transcode(self, device_id): + return self._delete("Videos/ActiveEncodings", params={ + 'DeviceId': device_id + }) + + def get_audio_stream(self, dest_file, item_id, play_id, container, max_streaming_bitrate=140000000, audio_codec=None): + self._get_stream("Audio/%s/universal" % item_id, dest_file, params={ + 'UserId': "{UserId}", + 'DeviceId': "{DeviceId}", + 'PlaySessionId': play_id, + 'Container': container, + 'AudioCodec': audio_codec, + "MaxStreamingBitrate": max_streaming_bitrate, + }) + + def get_default_headers(self): + auth = "MediaBrowser " + auth += "Client=%s, " % self.config.data['app.name'] + auth += "Device=%s, " % self.config.data['app.device_name'] + auth += "DeviceId=%s, " % self.config.data['app.device_id'] + auth += "Version=%s" % self.config.data['app.version'] + + return { + "Accept": "application/json", + "Content-type": "application/x-www-form-urlencoded; charset=UTF-8", + "X-Application": "%s/%s" % (self.config.data['app.name'], self.config.data['app.version']), + "Accept-Charset": "UTF-8,*", + "Accept-encoding": "gzip", + "User-Agent": self.config.data['http.user_agent'] or "%s/%s" % (self.config.data['app.name'], self.config.data['app.version']), + "x-emby-authorization": auth + } + + def send_request(self, url, path, method="get", timeout=None, headers=None, data=None, session=None): + request_method = getattr(session or requests, method.lower()) + url = "%s/%s" % (url, path) + request_settings = { + "timeout": timeout or self.default_timeout, + "headers": headers or self.get_default_headers(), + "data": data + } + + # Changed to use non-Kodi specific setting. + if self.config.data.get('auth.ssl') == False: + request_settings["verify"] = False + + LOG.info("Sending %s request to %s" % (method, path)) + LOG.debug(request_settings['timeout']) + LOG.debug(request_settings['headers']) + + return request_method(url, **request_settings) + + def login(self, server_url, username, password=""): + path = "Users/AuthenticateByName" + authData = { + "username": username, + "Pw": password + } + + headers = self.get_default_headers() + headers.update({'Content-type': "application/json"}) + + try: + LOG.info("Trying to login to %s/%s as %s" % (server_url, path, username)) + response = self.send_request(server_url, path, method="post", headers=headers, + data=json.dumps(authData), timeout=(5, 30)) + + if response.status_code == 200: + return response.json() + else: + LOG.error("Failed to login to server with status code: " + str(response.status_code)) + LOG.error("Server Response:\n" + str(response.content)) + LOG.debug(headers) + + return {} + except Exception as e: # Find exceptions for likely cases i.e, server timeout, etc + LOG.error(e) + + return {} + + def validate_authentication_token(self, server): + authTokenHeader = { + 'X-MediaBrowser-Token': server['AccessToken'] + } + headers = self.get_default_headers() + headers.update(authTokenHeader) + + response = self.send_request(server['address'], "system/info", headers=headers) + return response.json() if response.status_code == 200 else {} + + def get_public_info(self, server_address): + response = self.send_request(server_address, "system/info/public") + return response.json() if response.status_code == 200 else {} + + def check_redirect(self, server_address): + ''' Checks if the server is redirecting traffic to a new URL and + returns the URL the server prefers to use + ''' + response = self.send_request(server_address, "system/info/public") + url = response.url.replace('/system/info/public', '') + return url + + + + ################################################################################################# + + # Syncplay + + ################################################################################################# + + def _parse_precise_time(self, time): + # We have to remove the Z and the least significant digit. + return datetime.strptime(time[:-2], "%Y-%m-%dT%H:%M:%S.%f") + + def utc_time(self): + # Measure time as close to the call as is possible. + server_address = self.config.data.get("auth.server") + session = self.client.session + + response = self.send_request(server_address, "GetUTCTime", session=session) + response_received = datetime.utcnow() + request_sent = response_received - response.elapsed + + response_obj = response.json() + request_received = self._parse_precise_time(response_obj["RequestReceptionTime"]) + response_sent = self._parse_precise_time(response_obj["ResponseTransmissionTime"]) + + return { + "request_sent": request_sent, + "request_received": request_received, + "response_sent": response_sent, + "response_received": response_received + } + + def get_sync_play(self, item_id=None): + params = {} + if item_id is not None: + params["FilterItemId"] = item_id + return self._get("SyncPlay/List", params) + + def join_sync_play(self, group_id): + return self._post("SyncPlay/Join", { + "GroupId": group_id + }) + + def leave_sync_play(self): + return self._post("SyncPlay/Leave") + + def play_sync_play(self): + """deprecated (<= 10.7.0)""" + return self._post("SyncPlay/Play") + + def pause_sync_play(self): + return self._post("SyncPlay/Pause") + + def unpause_sync_play(self): + """10.7.0+ only""" + return self._post("SyncPlay/Unpause") + + def seek_sync_play(self, position_ticks): + return self._post("SyncPlay/Seek", { + "PositionTicks": position_ticks + }) + + def buffering_sync_play(self, when, position_ticks, is_playing, item_id): + return self._post("SyncPlay/Buffering", { + "When": when.isoformat() + "Z", + "PositionTicks": position_ticks, + "IsPlaying": is_playing, + "PlaylistItemId": item_id + }) + + def ready_sync_play(self, when, position_ticks, is_playing, item_id): + """10.7.0+ only""" + return self._post("SyncPlay/Ready", { + "When": when.isoformat() + "Z", + "PositionTicks": position_ticks, + "IsPlaying": is_playing, + "PlaylistItemId": item_id + }) + + def reset_queue_sync_play(self, queue_item_ids, position=0, position_ticks=0): + """10.7.0+ only""" + return self._post("SyncPlay/SetNewQueue", { + "PlayingQueue": queue_item_ids, + "PlayingItemPosition": position, + "StartPositionTicks": position_ticks + }) + + def ignore_sync_play(self, should_ignore): + """10.7.0+ only""" + return self._post("SyncPlay/SetIgnoreWait", { + "IgnoreWait": should_ignore + }) + + def next_sync_play(self, item_id): + """10.7.0+ only""" + return self._post("SyncPlay/NextItem", { + "PlaylistItemId": item_id + }) + + def prev_sync_play(self, item_id): + """10.7.0+ only""" + return self._post("SyncPlay/PreviousItem", { + "PlaylistItemId": item_id + }) + + def set_item_sync_play(self, item_id): + """10.7.0+ only""" + return self._post("SyncPlay/SetPlaylistItem", { + "PlaylistItemId": item_id + }) + + def ping_sync_play(self, ping): + return self._post("SyncPlay/Ping", { + "Ping": ping + }) + + def new_sync_play(self): + """deprecated (< 10.7.0)""" + return self._post("SyncPlay/New") + + def new_sync_play_v2(self, group_name): + """10.7.0+ only""" + return self._post("SyncPlay/New", { + "GroupName": group_name + }) + diff --git a/jellyfin_apiclient_python/client.py b/jellyfin_apiclient_python/client.py new file mode 100644 index 0000000..474c47c --- /dev/null +++ b/jellyfin_apiclient_python/client.py @@ -0,0 +1,87 @@ +# -*- coding: utf-8 -*- +from __future__ import division, absolute_import, print_function, unicode_literals + +################################################################################################# + +import logging + +from . import api +from .configuration import Config +from .http import HTTP +from .ws_client import WSClient +from .connection_manager import ConnectionManager, CONNECTION_STATE +from .timesync_manager import TimeSyncManager + +################################################################################################# + +LOG = logging.getLogger('JELLYFIN.' + __name__) + +################################################################################################# + + +def callback(message, data): + + ''' Callback function should received message, data + message: string + data: json dictionary + ''' + pass + + +class JellyfinClient(object): + + logged_in = False + + def __init__(self, allow_multiple_clients=False): + LOG.debug("JellyfinClient initializing...") + + self.config = Config() + self.http = HTTP(self) + self.wsc = WSClient(self, allow_multiple_clients) + self.auth = ConnectionManager(self) + self.jellyfin = api.API(self.http) + self.callback_ws = callback + self.callback = callback + self.timesync = TimeSyncManager(self) + + def set_credentials(self, credentials=None): + self.auth.credentials.set_credentials(credentials or {}) + + def get_credentials(self): + return self.auth.credentials.get_credentials() + + def authenticate(self, credentials=None, options=None, discover=True): + + self.set_credentials(credentials or {}) + state = self.auth.connect(options or {}, discover) + + if state['State'] == CONNECTION_STATE['SignedIn']: + + LOG.info("User is authenticated.") + self.logged_in = True + self.callback("ServerOnline", {'Id': self.auth.server_id}) + + state['Credentials'] = self.get_credentials() + + return state + + def start(self, websocket=False, keep_alive=True): + + if not self.logged_in: + raise ValueError("User is not authenticated.") + + self.http.start_session() + + if keep_alive: + self.http.keep_alive = True + + if websocket: + self.start_wsc() + + def start_wsc(self): + self.wsc.start() + + def stop(self): + self.wsc.stop_client() + self.http.stop_session() + self.timesync.stop_ping() diff --git a/jellyfin_apiclient_python/configuration.py b/jellyfin_apiclient_python/configuration.py new file mode 100644 index 0000000..1d93ef9 --- /dev/null +++ b/jellyfin_apiclient_python/configuration.py @@ -0,0 +1,53 @@ +# -*- coding: utf-8 -*- +from __future__ import division, absolute_import, print_function, unicode_literals + +''' This will hold all configs from the client. + Configuration set here will be used for the HTTP client. +''' + +################################################################################################# + +import logging + +################################################################################################# + +DEFAULT_HTTP_MAX_RETRIES = 3 +DEFAULT_HTTP_TIMEOUT = 30 +LOG = logging.getLogger('JELLYFIN.' + __name__) + +################################################################################################# + + +class Config(object): + + def __init__(self): + + LOG.debug("Configuration initializing...") + self.data = {} + self.http() + + def app(self, name, version, device_name, device_id, capabilities=None, device_pixel_ratio=None): + + LOG.debug("Begin app constructor.") + self.data['app.name'] = name + self.data['app.version'] = version + self.data['app.device_name'] = device_name + self.data['app.device_id'] = device_id + self.data['app.capabilities'] = capabilities + self.data['app.device_pixel_ratio'] = device_pixel_ratio + self.data['app.default'] = False + + def auth(self, server, user_id, token=None, ssl=None): + + LOG.debug("Begin auth constructor.") + self.data['auth.server'] = server + self.data['auth.user_id'] = user_id + self.data['auth.token'] = token + self.data['auth.ssl'] = ssl + + def http(self, user_agent=None, max_retries=DEFAULT_HTTP_MAX_RETRIES, timeout=DEFAULT_HTTP_TIMEOUT): + + LOG.debug("Begin http constructor.") + self.data['http.max_retries'] = max_retries + self.data['http.timeout'] = timeout + self.data['http.user_agent'] = user_agent diff --git a/jellyfin_apiclient_python/connection_manager.py b/jellyfin_apiclient_python/connection_manager.py new file mode 100644 index 0000000..9c6a6af --- /dev/null +++ b/jellyfin_apiclient_python/connection_manager.py @@ -0,0 +1,379 @@ +# -*- coding: utf-8 -*- +from __future__ import division, absolute_import, print_function, unicode_literals + +################################################################################################# + +import json +import logging +import socket +from datetime import datetime +from operator import itemgetter + +import urllib3 + +from .credentials import Credentials +from .api import API +import traceback + +################################################################################################# + +LOG = logging.getLogger('JELLYFIN.' + __name__) +CONNECTION_STATE = { + 'Unavailable': 0, + 'ServerSelection': 1, + 'ServerSignIn': 2, + 'SignedIn': 3 +} + +################################################################################################# + +class ConnectionManager(object): + + user = {} + server_id = None + + def __init__(self, client): + + LOG.debug("ConnectionManager initializing...") + + self.client = client + self.config = client.config + self.credentials = Credentials() + + self.API = API(client) + + def clear_data(self): + + LOG.info("connection manager clearing data") + + self.user = None + credentials = self.credentials.get_credentials() + credentials['Servers'] = list() + self.credentials.get_credentials(credentials) + + self.config.auth(None, None) + + def revoke_token(self): + + LOG.info("revoking token") + + self['server']['AccessToken'] = None + self.credentials.set_credentials(self.credentials.get()) + + self.config.data['auth.token'] = None + + def get_available_servers(self, discover=True): + + LOG.info("Begin getAvailableServers") + + # Clone the credentials + credentials = self.credentials.get() + found_servers = [] + + if discover: + found_servers = self.process_found_servers(self._server_discovery()) + + if not found_servers and not credentials['Servers']: # back out right away, no point in continuing + LOG.info("Found no servers") + return list() + + servers = list(credentials['Servers']) + + # Merges servers we already knew with newly found ones + for found_server in found_servers: + try: + self.credentials.add_update_server(servers, found_server) + except KeyError: + continue + + servers.sort(key=itemgetter('DateLastAccessed'), reverse=True) + credentials['Servers'] = servers + self.credentials.set(credentials) + + return servers + + def login(self, server_url, username, password=None, clear=None, options=None): + + if not username: + raise AttributeError("username cannot be empty") + + if not server_url: + raise AttributeError("server url cannot be empty") + + if clear is not None: + LOG.warn("The clear option on login() has no effect.") + + if options is not None: + LOG.warn("The options option on login() has no effect.") + + data = self.API.login(server_url, username, password) # returns empty dict on failure + + if not data: + LOG.info("Failed to login as `"+username+"`") + return {} + + LOG.info("Succesfully logged in as %s" % (username)) + # TODO Change when moving to database storage of server details + credentials = self.credentials.get() + + self.config.data['auth.user_id'] = data['User']['Id'] + self.config.data['auth.token'] = data['AccessToken'] + + for server in credentials['Servers']: + if server['Id'] == data['ServerId']: + found_server = server + break + else: + return {} # No server found + + found_server['DateLastAccessed'] = datetime.now().strftime('%Y-%m-%dT%H:%M:%SZ') + found_server['UserId'] = data['User']['Id'] + found_server['AccessToken'] = data['AccessToken'] + + self.credentials.add_update_server(credentials['Servers'], found_server) + + info = { + 'Id': data['User']['Id'], + 'IsSignedInOffline': True + } + self.credentials.add_update_user(server, info) + + self.credentials.set_credentials(credentials) + + return data + + + def connect_to_address(self, address, options={}): + + if not address: + return False + + address = self._normalize_address(address) + + try: + response_url = self.API.check_redirect(address) + if address != response_url: + address = response_url + LOG.info("connect_to_address %s succeeded", address) + server = { + 'address': address, + } + server = self.connect_to_server(server, options) + if server is False: + LOG.error("connect_to_address %s failed", address) + return { 'State': CONNECTION_STATE['Unavailable'] } + + return server + except Exception: + LOG.error("connect_to_address %s failed", address) + return { 'State': CONNECTION_STATE['Unavailable'] } + + + def connect_to_server(self, server, options={}): + + LOG.info("begin connect_to_server") + + try: + result = self.API.get_public_info(server.get('address')) + + if not result: + LOG.error("Failed to connect to server: %s" % server.get('address')) + return { 'State': CONNECTION_STATE['Unavailable'] } + + LOG.info("calling onSuccessfulConnection with server %s", server.get('Name')) + + self._update_server_info(server, result) + credentials = self.credentials.get() + return self._after_connect_validated(server, credentials, result, True, options) + + except Exception as e: + LOG.error(traceback.format_exc()) + LOG.error("Failing server connection. ERROR msg: {}".format(e)) + return { 'State': CONNECTION_STATE['Unavailable'] } + + def connect(self, options={}, discover=True): + + LOG.info("Begin connect") + + servers = self.get_available_servers(discover) + LOG.info("connect has %s servers", len(servers)) + + if not (len(servers)): # No servers provided + return { + 'State': ['ServerSelection'] + } + + result = self.connect_to_server(servers[0], options) + LOG.debug("resolving connect with result: %s", result) + + return result + + def jellyfin_user_id(self): + return self.get_server_info(self.server_id)['UserId'] + + def jellyfin_token(self): + return self.get_server_info(self.server_id)['AccessToken'] + + def get_server_info(self, server_id): + + if server_id is None: + LOG.info("server_id is empty") + return {} + + servers = self.credentials.get()['Servers'] + + for server in servers: + if server['Id'] == server_id: + return server + + def get_public_users(self): + return self.client.jellyfin.get_public_users() + + def get_jellyfin_url(self, base, handler): + return "%s/%s" % (base, handler) + + def _server_discovery(self): + MULTI_GROUP = ("", 7359) + MESSAGE = b"who is JellyfinServer?" + + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.settimeout(1.0) # This controls the socket.timeout exception + + sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 20) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + sock.setsockopt(socket.SOL_IP, socket.IP_MULTICAST_LOOP, 1) + sock.setsockopt(socket.IPPROTO_IP, socket.SO_REUSEADDR, 1) + + LOG.debug("MultiGroup : %s", str(MULTI_GROUP)) + LOG.debug("Sending UDP Data: %s", MESSAGE) + + servers = [] + + try: + sock.sendto(MESSAGE, MULTI_GROUP) + except Exception as error: + LOG.exception(traceback.format_exc()) + LOG.exception(error) + return servers + + while True: + try: + data, addr = sock.recvfrom(1024) # buffer size + servers.append(json.loads(data)) + + except socket.timeout: + LOG.info("Found Servers: %s", servers) + return servers + + except Exception as e: + LOG.error(traceback.format_exc()) + LOG.exception("Error trying to find servers: %s", e) + return servers + + def process_found_servers(self, found_servers): + + servers = [] + + for found_server in found_servers: + + server = self._convert_endpoint_address_to_manual_address(found_server) + + info = { + 'Id': found_server['Id'], + 'address': server or found_server['Address'], + 'Name': found_server['Name'] + } + + servers.append(info) + else: + return servers + + # TODO: Make IPv6 compatable + def _convert_endpoint_address_to_manual_address(self, info): + + if info.get('Address') and info.get('EndpointAddress'): + address = info['EndpointAddress'].split(':')[0] + + # Determine the port, if any + parts = info['Address'].split(':') + if len(parts) > 1: + port_string = parts[len(parts) - 1] + + try: + address += ":%s" % int(port_string) + return self._normalize_address(address) + except ValueError: + pass + + return None + + def _normalize_address(self, address): + # TODO: Try HTTPS first, then HTTP if that fails. + if '://' not in address: + address = 'http://' + address + + # Attempt to correct bad input + url = urllib3.util.parse_url(address.strip()) + + if url.scheme is None: + url = url._replace(scheme='http') + + if url.scheme == 'http' and url.port == 80: + url = url._replace(port=None) + + if url.scheme == 'https' and url.port == 443: + url = url._replace(port=None) + + return url.url + + def _after_connect_validated(self, server, credentials, system_info, verify_authentication, options): + if options.get('enableAutoLogin') is False: + + self.config.data['auth.user_id'] = server.pop('UserId', None) + self.config.data['auth.token'] = server.pop('AccessToken', None) + + elif verify_authentication and server.get('AccessToken'): + system_info = self.API.validate_authentication_token(server) + if system_info: + + self._update_server_info(server, system_info) + self.config.data['auth.user_id'] = server['UserId'] + self.config.data['auth.token'] = server['AccessToken'] + + return self._after_connect_validated(server, credentials, system_info, False, options) + + server['UserId'] = None + server['AccessToken'] = None + return { 'State': CONNECTION_STATE['Unavailable'] } + + self._update_server_info(server, system_info) + + server['DateLastAccessed'] = datetime.now().strftime('%Y-%m-%dT%H:%M:%SZ') + self.credentials.add_update_server(credentials['Servers'], server) + self.credentials.set(credentials) + self.server_id = server['Id'] + + # Update configs + self.config.data['auth.server'] = server['address'] + self.config.data['auth.server-name'] = server['Name'] + self.config.data['auth.server=id'] = server['Id'] + self.config.data['auth.ssl'] = options.get('ssl', self.config.data['auth.ssl']) + + result = { + 'Servers': [server] + } + + result['State'] = CONNECTION_STATE['SignedIn'] if server.get('AccessToken') else CONNECTION_STATE['ServerSignIn'] + # Connected + return result + + def _update_server_info(self, server, system_info): + + if server is None or system_info is None: + return + + server['Name'] = system_info['ServerName'] + server['Id'] = system_info['Id'] + + if system_info.get('address'): + server['address'] = system_info['address'] diff --git a/jellyfin_apiclient_python/credentials.py b/jellyfin_apiclient_python/credentials.py new file mode 100644 index 0000000..715abb5 --- /dev/null +++ b/jellyfin_apiclient_python/credentials.py @@ -0,0 +1,128 @@ +# -*- coding: utf-8 -*- +from __future__ import division, absolute_import, print_function, unicode_literals + +################################################################################################# + +import logging +import time +from datetime import datetime + +################################################################################################# + +LOG = logging.getLogger('JELLYFIN.' + __name__) + +################################################################################################# + + +class Credentials(object): + + credentials = None + + def __init__(self): + LOG.debug("Credentials initializing...") + self.credentials = {} + + def set_credentials(self, credentials): + self.credentials = credentials + + def get_credentials(self): + return self.get() + + def _ensure(self): + + if not self.credentials: + try: + LOG.info(self.credentials) + if not isinstance(self.credentials, dict): + raise ValueError("invalid credentials format") + + except Exception as e: # File is either empty or missing + LOG.warning(e) + self.credentials = {} + + LOG.debug("credentials initialized with: %s", self.credentials) + self.credentials['Servers'] = self.credentials.setdefault('Servers', []) + + def get(self): + self._ensure() + + return self.credentials + + def set(self, data): + + if data: + self.credentials.update(data) + else: + self._clear() + + LOG.debug("credentialsupdated") + + def _clear(self): + self.credentials.clear() + + def add_update_user(self, server, user): + + for existing in server.setdefault('Users', []): + if existing['Id'] == user['Id']: + # Merge the data + existing['IsSignedInOffline'] = True + break + else: + server['Users'].append(user) + + def add_update_server(self, servers, server): + + if server.get('Id') is None: + raise KeyError("Server['Id'] cannot be null or empty") + + # Add default DateLastAccessed if doesn't exist. + server.setdefault('DateLastAccessed', "2001-01-01T00:00:00Z") + + for existing in servers: + if existing['Id'] == server['Id']: + + # Merge the data + if server.get('DateLastAccessed'): + if self._date_object(server['DateLastAccessed']) > self._date_object(existing['DateLastAccessed']): + existing['DateLastAccessed'] = server['DateLastAccessed'] + + if server.get('UserLinkType'): + existing['UserLinkType'] = server['UserLinkType'] + + if server.get('AccessToken'): + existing['AccessToken'] = server['AccessToken'] + existing['UserId'] = server['UserId'] + + if server.get('ExchangeToken'): + existing['ExchangeToken'] = server['ExchangeToken'] + + if server.get('ManualAddress'): + existing['ManualAddress'] = server['ManualAddress'] + + if server.get('LocalAddress'): + existing['LocalAddress'] = server['LocalAddress'] + + if server.get('Name'): + existing['Name'] = server['Name'] + + if server.get('LastConnectionMode') is not None: + existing['LastConnectionMode'] = server['LastConnectionMode'] + + if server.get('ConnectServerId'): + existing['ConnectServerId'] = server['ConnectServerId'] + + return existing + else: + servers.append(server) + return server + + def _date_object(self, date): + # Convert string to date + try: + date_obj = time.strptime(date, "%Y-%m-%dT%H:%M:%SZ") + except (ImportError, TypeError): + # TypeError: attribute of type 'NoneType' is not callable + # Known Kodi/python error + date_obj = datetime(*(time.strptime(date, "%Y-%m-%dT%H:%M:%SZ")[0:6])) + + return date_obj diff --git a/jellyfin_apiclient_python/exceptions.py b/jellyfin_apiclient_python/exceptions.py new file mode 100644 index 0000000..6cd5051 --- /dev/null +++ b/jellyfin_apiclient_python/exceptions.py @@ -0,0 +1,11 @@ +# -*- coding: utf-8 -*- +from __future__ import division, absolute_import, print_function, unicode_literals + +################################################################################################# + + +class HTTPException(Exception): + # Jellyfin HTTP exception + def __init__(self, status, message): + self.status = status + self.message = message diff --git a/jellyfin_apiclient_python/http.py b/jellyfin_apiclient_python/http.py new file mode 100644 index 0000000..4cb7ae1 --- /dev/null +++ b/jellyfin_apiclient_python/http.py @@ -0,0 +1,267 @@ +# -*- coding: utf-8 -*- +from __future__ import division, absolute_import, print_function, unicode_literals + +################################################################################################# + +import json +import logging +import time +import urllib + +import requests +from six import string_types + +from .exceptions import HTTPException + +################################################################################################# + +LOG = logging.getLogger('Jellyfin.' + __name__) + +################################################################################################# + + +class HTTP(object): + + session = None + keep_alive = False + + def __init__(self, client): + + self.client = client + self.config = client.config + + def start_session(self): + + self.session = requests.Session() + + max_retries = self.config.data['http.max_retries'] + self.session.mount("http://", requests.adapters.HTTPAdapter(max_retries=max_retries)) + self.session.mount("https://", requests.adapters.HTTPAdapter(max_retries=max_retries)) + + def stop_session(self): + + if self.session is None: + return + + try: + LOG.info("--<[ session/%s ]", id(self.session)) + self.session.close() + except Exception as error: + LOG.warning("The requests session could not be terminated: %s", error) + + def _replace_user_info(self, string): + + if '{server}' in string: + if self.config.data.get('auth.server', None): + string = string.replace("{server}", self.config.data['auth.server']) + else: + LOG.debug("Server address not set") + + if '{UserId}'in string: + if self.config.data.get('auth.user_id', None): + string = string.replace("{UserId}", self.config.data['auth.user_id']) + else: + LOG.debug("UserId is not set.") + + if '{DeviceId}'in string: + if self.config.data.get('app.device_id', None): + string = string.replace("{DeviceId}", self.config.data['app.device_id']) + else: + LOG.debug("DeviceId is not set.") + + return string + + def request_url(self, data): + if not data: + raise AttributeError("Request cannot be empty") + + data = self._request(data) + + params = data["params"] + if "api_key" not in params: + params["api_key"] = self.config.data.get('auth.token') + + encoded_params = urllib.parse.urlencode(data["params"]) + return "%s?%s" % (data["url"], encoded_params) + + def request(self, data, session=None, dest_file=None): + + ''' Give a chance to retry the connection. Jellyfin sometimes can be slow to answer back + data dictionary can contain: + type: GET, POST, etc. + url: (optional) + handler: not considered when url is provided (optional) + params: request parameters (optional) + json: request body (optional) + headers: (optional), + verify: ssl certificate, True (verify using device built-in library) or False + ''' + if not data: + raise AttributeError("Request cannot be empty") + + data = self._request(data) + LOG.debug("--->[ http ] %s", json.dumps(data, indent=4)) + retry = data.pop('retry', 5) + stream = dest_file is not None + + while True: + + try: + r = self._requests(session or self.session or requests, data.pop('type', "GET"), **data, stream=stream) + if stream: + for chunk in r.iter_content(chunk_size=8192): + if chunk: # filter out keep-alive new chunks + dest_file.write(chunk) + else: + r.content # release the connection + + if not self.keep_alive and self.session is not None: + self.stop_session() + + r.raise_for_status() + + except requests.exceptions.ConnectionError as error: + if retry: + + retry -= 1 + time.sleep(1) + + continue + + LOG.error(error) + self.client.callback("ServerUnreachable", {'ServerId': self.config.data['auth.server-id']}) + + raise HTTPException("ServerUnreachable", error) + + except requests.exceptions.ReadTimeout as error: + if retry: + + retry -= 1 + time.sleep(1) + + continue + + LOG.error(error) + + raise HTTPException("ReadTimeout", error) + + except requests.exceptions.HTTPError as error: + LOG.error(error) + + if r.status_code == 401: + + if 'X-Application-Error-Code' in r.headers: + self.client.callback("AccessRestricted", {'ServerId': self.config.data['auth.server-id']}) + + raise HTTPException("AccessRestricted", error) + else: + self.client.callback("Unauthorized", {'ServerId': self.config.data['auth.server-id']}) + self.client.auth.revoke_token() + + raise HTTPException("Unauthorized", error) + + elif r.status_code == 500: # log and ignore. + LOG.error("--[ 500 response ] %s", error) + + return + + elif r.status_code == 502: + if retry: + + retry -= 1 + time.sleep(1) + + continue + + raise HTTPException(r.status_code, error) + + except requests.exceptions.MissingSchema as error: + LOG.error("Request missing Schema. " + str(error)) + raise HTTPException("MissingSchema", {'Id': self.config.data.get('auth.server', "None")}) + + except Exception as error: + raise + + else: + try: + if stream: + return + self.config.data['server-time'] = r.headers['Date'] + elapsed = int(r.elapsed.total_seconds() * 1000) + response = r.json() + LOG.debug("---<[ http ][%s ms]", elapsed) + LOG.debug(json.dumps(response, indent=4)) + + return response + except ValueError: + return + + def _request(self, data): + + if 'url' not in data: + data['url'] = "%s/%s" % (self.config.data.get("auth.server", ""), data.pop('handler', "")) + + self._get_header(data) + data['timeout'] = data.get('timeout') or self.config.data['http.timeout'] + data['verify'] = data.get('verify') or self.config.data.get('auth.ssl', False) + data['url'] = self._replace_user_info(data['url']) + self._process_params(data.get('params') or {}) + self._process_params(data.get('json') or {}) + + return data + + def _process_params(self, params): + + for key in params: + value = params[key] + + if isinstance(value, dict): + self._process_params(value) + + if isinstance(value, string_types): + params[key] = self._replace_user_info(value) + + def _get_header(self, data): + + data['headers'] = data.setdefault('headers', {}) + + if not data['headers']: + data['headers'].update({ + 'Content-type': "application/json", + 'Accept-Charset': "UTF-8,*", + 'Accept-encoding': "gzip", + 'User-Agent': self.config.data['http.user_agent'] or "%s/%s" % (self.config.data.get('app.name', 'Jellyfin for Kodi'), self.config.data.get('app.version', "0.0.0")) + }) + + if 'x-emby-authorization' not in data['headers']: + self._authorization(data) + + return data + + def _authorization(self, data): + + auth = "MediaBrowser " + auth += "Client=%s, " % self.config.data.get('app.name', "Jellyfin for Kodi") + auth += "Device=%s, " % self.config.data.get('app.device_name', 'Unknown Device') + auth += "DeviceId=%s, " % self.config.data.get('app.device_id', 'Unknown Device id') + auth += "Version=%s" % self.config.data.get('app.version', '0.0.0') + + data['headers'].update({'x-emby-authorization': auth}) + + if self.config.data.get('auth.token') and self.config.data.get('auth.user_id'): + + auth += ', UserId=%s' % self.config.data.get('auth.user_id') + data['headers'].update({'x-emby-authorization': auth, 'X-MediaBrowser-Token': self.config.data.get('auth.token')}) + + return data + + def _requests(self, session, action, **kwargs): + + if action == "GET": + return session.get(**kwargs) + elif action == "POST": + return session.post(**kwargs) + elif action == "HEAD": + return session.head(**kwargs) + elif action == "DELETE": + return session.delete(**kwargs) diff --git a/jellyfin_apiclient_python/keepalive.py b/jellyfin_apiclient_python/keepalive.py new file mode 100644 index 0000000..7d9e40e --- /dev/null +++ b/jellyfin_apiclient_python/keepalive.py @@ -0,0 +1,20 @@ +import threading + +class KeepAlive(threading.Thread): + def __init__(self, timeout, ws): + self.halt = threading.Event() + self.timeout = timeout + self.ws = ws + + threading.Thread.__init__(self) + + def stop(self): + self.halt.set() + self.join() + + def run(self): + while not self.halt.is_set(): + if self.halt.wait(self.timeout/2): + break + else: + self.ws.send("KeepAlive") diff --git a/jellyfin_apiclient_python/timesync_manager.py b/jellyfin_apiclient_python/timesync_manager.py new file mode 100644 index 0000000..5c4e98e --- /dev/null +++ b/jellyfin_apiclient_python/timesync_manager.py @@ -0,0 +1,140 @@ +# This is based on https://github.com/jellyfin/jellyfin-web/blob/master/src/components/syncPlay/timeSyncManager.js +import threading +import logging +import datetime + +LOG = logging.getLogger('Jellyfin.' + __name__) + +number_of_tracked_measurements = 8 +polling_interval_greedy = 1 +polling_interval_low_profile = 60 +greedy_ping_count = 3 + + +class Measurement: + def __init__(self, request_sent, request_received, response_sent, response_received): + self.request_sent = request_sent + self.request_received = request_received + self.response_sent = response_sent + self.response_received = response_received + + def get_offset(self): + """Time offset from server.""" + return ((self.request_received - self.request_sent) + (self.response_sent - self.response_received)) / 2.0 + + def get_delay(self): + """Get round-trip delay.""" + return (self.response_received - self.request_sent) - (self.response_sent - self.request_received) + + def get_ping(self): + """Get ping time.""" + return self.get_delay() / 2.0 + + +class _TimeSyncThread(threading.Thread): + def __init__(self, manager): + self.manager = manager + self.halt = threading.Event() + threading.Thread.__init__(self) + + def run(self): + while not self.halt.wait(self.manager.polling_interval): + try: + measurement = self.manager.client.jellyfin.utc_time() + measurement = Measurement(measurement["request_sent"], measurement["request_received"], + measurement["response_sent"], measurement["response_received"]) + + self.manager.update_time_offset(measurement) + + if self.manager.pings > greedy_ping_count: + self.manager.polling_interval = polling_interval_low_profile + else: + self.manager.pings += 1 + + self.manager._notify_subscribers() + except Exception: + LOG.error("Timesync call failed.", exc_info=True) + + def stop(self): + self.halt.set() + self.join() + + +class TimeSyncManager: + def __init__(self, client): + self.ping_stop = True + self.polling_interval = polling_interval_greedy + self.poller = None + self.pings = 0 # number of pings + self.measurement = None # current time sync + self.measurements = [] + self.client = client + self.timesync_thread = None + self.subscribers = set() + + def is_ready(self): + """Gets status of time sync.""" + return self.measurement is not None + + def get_time_offset(self): + """Gets time offset with server.""" + return self.measurement.get_offset() if self.measurement is not None else datetime.timedelta(0) + + def get_ping(self): + """Gets ping time to server.""" + return self.measurement.get_ping() if self.measurement is not None else datetime.timedelta(0) + + def update_time_offset(self, measurement): + """Updates time offset between server and client.""" + self.measurements.append(measurement) + if len(self.measurements) > number_of_tracked_measurements: + self.measurements.pop(0) + + self.measurement = min(self.measurements, key=lambda x: x.get_delay()) + + def reset_measurements(self): + """Drops accumulated measurements.""" + self.measurement = None + self.measurements = [] + + def start_ping(self): + """Starts the time poller.""" + if not self.timesync_thread: + self.timesync_thread = _TimeSyncThread(self) + self.timesync_thread.start() + + def stop_ping(self): + """Stops the time poller.""" + if self.timesync_thread: + self.timesync_thread.stop() + self.timesync_thread = None + + def force_update(self): + """Resets poller into greedy mode.""" + self.stop_ping() + self.polling_interval = polling_interval_greedy + self.pings = 0 + self.start_ping() + + def server_date_to_local(self, server): + """Converts server time to local time.""" + return server - self.get_time_offset() + + def local_date_to_server(self, local): + """Converts local time to server time.""" + return local + self.get_time_offset() + + def subscribe_time_offset(self, subscriber_callable): + """Pass a callback function to get notified about time offset changes.""" + self.subscribers.add(subscriber_callable) + + def remove_subscriber(self, subscriber_callable): + """Remove a callback function from notifications.""" + self.subscribers.remove(subscriber_callable) + + def _notify_subscribers(self): + for subscriber in self.subscribers: + try: + subscriber(self.get_time_offset(), self.get_ping()) + except Exception: + LOG.error("Exception in subscriber callback.") diff --git a/jellyfin_apiclient_python/ws_client.py b/jellyfin_apiclient_python/ws_client.py new file mode 100644 index 0000000..d36310b --- /dev/null +++ b/jellyfin_apiclient_python/ws_client.py @@ -0,0 +1,140 @@ +# -*- coding: utf-8 -*- +from __future__ import division, absolute_import, print_function, unicode_literals + +################################################################################################# + +import json +import logging +import threading +import ssl +import certifi + +import websocket + +from .keepalive import KeepAlive + +################################################################################################## + +LOG = logging.getLogger('JELLYFIN.' + __name__) + +################################################################################################## + + +class WSClient(threading.Thread): + multi_client = False + global_wsc = None + global_stop = False + + def __init__(self, client, allow_multiple_clients=False): + + LOG.debug("WSClient initializing...") + + self.client = client + self.keepalive = None + self.wsc = None + self.stop = False + self.message_ids = set() + + if self.multi_client or allow_multiple_clients: + self.multi_client = True + + threading.Thread.__init__(self) + + def send(self, message, data=""): + if self.wsc is None: + raise ValueError("The websocket client is not started.") + + self.wsc.send(json.dumps({'MessageType': message, "Data": data})) + + def run(self): + + token = self.client.config.data['auth.token'] + device_id = self.client.config.data['app.device_id'] + server = self.client.config.data['auth.server'] + server = server.replace('https', "wss") if server.startswith('https') else server.replace('http', "ws") + wsc_url = "%s/socket?api_key=%s&device_id=%s" % (server, token, device_id) + verify = self.client.config.data.get('auth.ssl', False) + + LOG.info("Websocket url: %s", wsc_url) + + self.wsc = websocket.WebSocketApp(wsc_url, + on_message=lambda ws, message: self.on_message(ws, message), + on_error=lambda ws, error: self.on_error(ws, error)) + self.wsc.on_open = lambda ws: self.on_open(ws) + + if not self.multi_client: + if self.global_wsc is not None: + self.global_wsc.close() + self.global_wsc = self.wsc + + while not self.stop and not self.global_stop: + if not verify: + # https://stackoverflow.com/questions/48740053/ + self.wsc.run_forever( + ping_interval=10, sslopt={"cert_reqs": ssl.CERT_NONE} + ) + else: + self.wsc.run_forever(ping_interval=10, sslopt={"ca_certs": certifi.where()}) + + if not self.stop: + break + + LOG.info("---<[ websocket ]") + self.client.callback('WebSocketDisconnect', None) + + def on_error(self, ws, error): + LOG.error(error) + self.client.callback('WebSocketError', error) + + def on_open(self, ws): + LOG.info("--->[ websocket ]") + self.client.callback('WebSocketConnect', None) + + def on_message(self, ws, message): + + message = json.loads(message) + + # If a message is received multiple times, ignore repeats. + message_id = message.get("MessageId") + if message_id is not None: + if message_id in self.message_ids: + return + self.message_ids.add(message_id) + + data = message.get('Data', {}) + + if message['MessageType'] == "ForceKeepAlive": + self.send("KeepAlive") + if self.keepalive is not None: + self.keepalive.stop() + self.keepalive = KeepAlive(data, self) + self.keepalive.start() + LOG.debug("ForceKeepAlive received from server.") + return + elif message['MessageType'] == "KeepAlive": + LOG.debug("KeepAlive received from server.") + return + + if data is None: + data = {} + elif type(data) is not dict: + data = {"value": data} + + if not self.client.config.data['app.default']: + data['ServerId'] = self.client.auth.server_id + + self.client.callback(message['MessageType'], data) + + def stop_client(self): + + self.stop = True + + if self.keepalive is not None: + self.keepalive.stop() + + if self.wsc is not None: + self.wsc.close() + + if not self.multi_client: + self.global_stop = True + self.global_wsc = None -- cgit v1.2.3