monkee.ch/SickGear
1
0
Fork 0
mirror of https://github.com/SickGear/SickGear.git synced 2024-11-25 06:15:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
SickGear/sickgear/notifiers/kodi.py
SickGear e56303798c Change bump to major version 3.xx to signal that this branch supports Python3+ only. 
Initial SickGear for Python 3.
2023-02-09 13:41:15 +00:00

458 lines
19 KiB
Python
Raw Blame History

# coding=utf-8
#
# This file is part of SickGear.
#
# SickGear is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# SickGear is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with SickGear. If not, see <http://www.gnu.org/licenses/>.
import time
from .generic import Notifier
from .. import logger
import sickgear
import sickgear.helpers
from exceptions_helper import ex
from json_helper import json_dumps
from _23 import decode_str, etree, quote, unquote, unquote_plus
class KodiNotifier(Notifier):
def __init__(self):
super(KodiNotifier, self).__init__()
self.username, self.password = (None, None)
self.response = None
self.prefix = ''
def _get_kodi_version(self, host):
""" Return Kodi JSON-RPC API version (odd # = dev, even # = stable)
Communicate with Kodi hosts using JSON-RPC to determine whether to use the legacy API or the JSON-RPC API.
Fallback to testing legacy HTTP API before assuming it is a badly configured host.
Returns:
Returns API number or False
API | Kodi Version
-----+---------------
2 | v10 (Dharma)
3 | (pre Eden)
4 | v11 (Eden)
5 | (pre Frodo)
6 | v12 (Frodo) / v13 (Gotham)
"""
timeout = 10
response = self._send_json(host, dict(method='JSONRPC.Version'), timeout)
if self.response and 401 == self.response.get('status_code'):
return False
if response.get('version'):
version = response.get('version')
return isinstance(version, dict) and version.get('major') or version
# fallback to legacy HTTPAPI method
test_command = {'command': 'Help'}
if self._send(host, test_command, timeout):
# return fake version number to use the legacy method
return 1
if self.response and 404 == self.response.get('status_code'):
self.prefix = 'xbmc'
if self._send(host, test_command, timeout):
# return fake version number to use the legacy method
return 1
return False
def update_library(self, show_name=None, **kwargs):
""" Wrapper for the update library functions
Call either the JSON-RPC over HTTP or the legacy HTTP API methods depending on the Kodi API version.
Uses a list of comma delimited hosts where only one is updated, the first to respond with success. This is a
workaround for SQL backend users because updating multiple clients causes duplicate entries.
Future plan is to revisit how host/ip/username/pw/options are stored so that this may become more flexible.
Args:
show_name: Name of a TV show to target for a library update
Returns: True if processing succeeded with no issues else False if any issues found
"""
if not sickgear.KODI_HOST:
self._log_warning(u'No Kodi hosts specified, check your settings')
return False
# either update each host, or only attempt to update until one successful result
result = 0
only_first = dict(show='', first='', first_note='')
show_name and only_first.update(show=' for show;"%s"' % show_name)
sickgear.KODI_UPDATE_ONLYFIRST and only_first.update(dict(
first=' first', first_note=' in line with the "Only update first host"%s' % ' setting'))
for cur_host in [x.strip() for x in sickgear.KODI_HOST.split(',')]:
response = self._send_json(cur_host, dict(method='Profiles.GetCurrentProfile'))
if self.response and 401 == self.response.get('status_code'):
self._log_debug(u'Failed to authenticate with %s' % cur_host)
continue
if not response:
self._maybe_log_failed_detection(cur_host)
continue
if self._send_library_update(cur_host, show_name):
only_first.update(dict(profile=response.get('label') or 'Master', host=cur_host))
self._log('Success: profile;' +
u'"%(profile)s" at%(first)s host;%(host)s updated%(show)s%(first_note)s' % only_first)
else:
self._maybe_log_failed_detection(cur_host)
result += 1
if sickgear.KODI_UPDATE_ONLYFIRST:
return True
# needed for the 'update kodi' submenu command as it only cares of the final result vs the individual ones
return 0 == result
def _send_library_update(self, host, show_name=None):
""" Internal wrapper for the update library function
Call either the JSON-RPC over HTTP or the legacy HTTP API methods depending on the Kodi API version.
Args:
show_name: Name of a TV show to specifically target the library update for
Return:
True if the update was successful else False
"""
api_version = self._get_kodi_version(host)
if api_version:
# try to update just the show, if it fails, do full update if enabled
__method_update = (self._update, self._update_json)[4 < api_version]
if __method_update(host, show_name):
return True
failed_msg = 'Single show update failed,'
if sickgear.KODI_UPDATE_FULL:
self._log_debug(u'%s falling back to full update' % failed_msg)
return __method_update(host)
self._log_debug(u'%s consider enabling "Perform full library update" in config/notifications' % failed_msg)
return False
##############################################################################
# Legacy HTTP API (pre Kodi 12) methods
##############################################################################
def _send(self, host, command, timeout=30):
""" Handle communication to Kodi servers via HTTP API
Args:
command: Dictionary encoded via urllib and passed to the Kodi API via HTTP
Return:
response.result for successful commands or False if there was an error
"""
if not host:
self._log_warning(u'No host specified, aborting update')
return False
args = {}
if not sickgear.KODI_ALWAYS_ON and not self._testing:
args['mute_connect_err'] = True
if self.password or sickgear.KODI_PASSWORD:
args['auth'] = (self.username or sickgear.KODI_USERNAME, self.password or sickgear.KODI_PASSWORD)
url = 'http://%s/%sCmds/%sHttp' % (host, self.prefix or 'kodi', self.prefix or 'kodi')
response = sickgear.helpers.get_url(url=url, params=command,
timeout=timeout, hooks=dict(response=self.cb_response), **args)
return response or False
def _update(self, host=None, show_name=None):
""" Handle updating Kodi host via HTTP API
Update the video library for a specific tv show if passed, otherwise update the whole library if option enabled.
Args:
show_name: Name of a TV show to target for a library update
Return:
True or False
"""
if not host:
self._log_warning(u'No host specified, aborting update')
return False
self._log_debug(u'Updating library via HTTP method for host: %s' % host)
# if we're doing per-show
if show_name:
self._log_debug(u'Updating library via HTTP method for show %s' % show_name)
# noinspection SqlResolve
path_sql = 'SELECT path.strPath' \
' FROM path, tvshow, tvshowlinkpath' \
' WHERE tvshow.c00 = "%s"' % show_name \
+ ' AND tvshowlinkpath.idShow = tvshow.idShow' \
' AND tvshowlinkpath.idPath = path.idPath'
# set xml response format, if this fails then don't bother with the rest
if not self._send(
host, {'command': 'SetResponseFormat(webheader;false;webfooter;false;header;<xml>;footer;</xml>;'
'opentag;<tag>;closetag;</tag>;closefinaltag;false)'}):
return False
# sql used to grab path(s)
response = self._send(host, {'command': 'QueryVideoDatabase(%s)' % path_sql})
if not response:
self._log_debug(u'Invalid response for %s on %s' % (show_name, host))
return False
try:
et = etree.fromstring(quote(response, ':\\/<>'))
except SyntaxError as e:
self._log_error(u'Unable to parse XML in response: %s' % ex(e))
return False
paths = et.findall('.//field')
if not paths:
self._log_debug(u'No valid path found for %s on %s' % (show_name, host))
return False
for path in paths:
# we do not need it double-encoded, gawd this is dumb
un_enc_path = decode_str(unquote(path.text), sickgear.SYS_ENCODING)
self._log_debug(u'Updating %s on %s at %s' % (show_name, host, un_enc_path))
if not self._send(
host, dict(command='ExecBuiltIn', parameter='Kodi.updatelibrary(video, %s)' % un_enc_path)):
self._log_error(u'Update of show directory failed for %s on %s at %s'
% (show_name, host, un_enc_path))
return False
# sleep for a few seconds just to be sure kodi has a chance to finish each directory
if 1 < len(paths):
time.sleep(5)
# do a full update if requested
else:
self._log_debug(u'Full library update on host: %s' % host)
if not self._send(host, dict(command='ExecBuiltIn', parameter='Kodi.updatelibrary(video)')):
self._log_error(u'Failed full library update on: %s' % host)
return False
return True
##############################################################################
# JSON-RPC API (Kodi 12+) methods
##############################################################################
def _send_json(self, host, command, timeout=30):
""" Handle communication to Kodi installations via JSONRPC
Args:
command: Kodi JSON-RPC command to send via HTTP
Return:
response.result dict for successful commands or empty dict if there was an error
"""
result = {}
if not host:
self._log_warning(u'No host specified, aborting update')
return result
if isinstance(command, dict):
command.setdefault('jsonrpc', '2.0')
command.setdefault('id', 'SickGear')
args = dict(post_json=command)
else:
args = dict(data=command)
if not sickgear.KODI_ALWAYS_ON and not self._testing:
args['mute_connect_err'] = True
if self.password or sickgear.KODI_PASSWORD:
args['auth'] = (self.username or sickgear.KODI_USERNAME, self.password or sickgear.KODI_PASSWORD)
response = sickgear.helpers.get_url(url='http://%s/jsonrpc' % host, timeout=timeout,
headers={'Content-type': 'application/json'}, parse_json=True,
hooks=dict(response=self.cb_response), **args)
if response:
if not response.get('error'):
return 'OK' == response.get('result') and {'OK': True} or response.get('result')
self._log_error(u'API error; %s from %s in response to command: %s'
% (json_dumps(response['error']), host, json_dumps(command)))
return result
def _update_json(self, host=None, show_name=None):
""" Handle updating Kodi host via HTTP JSON-RPC
Update the video library for a specific tv show if passed, otherwise update the whole library if option enabled.
Args:
show_name: Name of a TV show to target for a library update
Return:
True or False
"""
if not host:
self._log_warning(u'No host specified, aborting update')
return False
# if we're doing per-show
if show_name:
self._log_debug(u'JSON library update. Host: %s Show: %s' % (host, show_name))
# try fetching tvshowid using show_name with a fallback to getting show list
show_name = unquote_plus(show_name)
commands = [dict(method='VideoLibrary.GetTVShows',
params={'filter': {'field': 'title', 'operator': 'is', 'value': '%s' % show_name},
'properties': ['title']}),
dict(method='VideoLibrary.GetTVShows')]
shows = None
for command in commands:
response = self._send_json(host, command)
shows = response.get('tvshows')
if shows:
break
if not shows:
self._log_debug(u'No items in GetTVShows response')
return False
tvshowid = -1
path = ''
# noinspection PyTypeChecker
for show in shows:
if show_name == show.get('title') or show_name == show.get('label'):
tvshowid = show.get('tvshowid', -1)
path = show.get('file', '')
break
del shows
# we didn't find the show (exact match), thus revert to just doing a full update if enabled
if -1 == tvshowid:
self._log_debug(u'Doesn\'t have "%s" in it\'s known shows, full library update required' % show_name)
return False
# lookup tv-show path if we don't already know it
if not len(path):
command = dict(method='VideoLibrary.GetTVShowDetails',
params={'tvshowid': tvshowid, 'properties': ['file']})
response = self._send_json(host, command)
path = 'tvshowdetails' in response and response['tvshowdetails'].get('file', '') or ''
if not len(path):
self._log_warning(u'No valid path found for %s with ID: %s on %s' % (show_name, tvshowid, host))
return False
self._log_debug(u'Updating %s on %s at %s' % (show_name, host, path))
command = dict(method='VideoLibrary.Scan',
params={'directory': '%s' % json_dumps(path)[1:-1].replace('\\\\', '\\')})
response_scan = self._send_json(host, command)
if not response_scan.get('OK'):
self._log_error(u'Update of show directory failed for %s on %s at %s response: %s' %
(show_name, host, path, response_scan))
return False
# do a full update if requested
else:
self._log_debug(u'Full library update on host: %s' % host)
response_scan = self._send_json(host, dict(method='VideoLibrary.Scan'))
if not response_scan.get('OK'):
self._log_error(u'Failed full library update on: %s response: %s' % (host, response_scan))
return False
return True
# noinspection PyUnusedLocal
def cb_response(self, r, *args, **kwargs):
self.response = dict(status_code=r.status_code)
return r
def _maybe_log(self, msg, log_level=logger.WARNING):
if msg and (sickgear.KODI_ALWAYS_ON or self._testing):
self._log(msg + (not sickgear.KODI_ALWAYS_ON and self._testing and
' (Test mode ignores "Always On")' or ''), log_level)
def _maybe_log_failed_detection(self, host, msg='connect to'):
self._maybe_log(u'Failed to %s %s, check device(s) and config' % (msg, host), logger.ERROR)
def _notify(self, title, body, hosts=None, username=None, password=None, **kwargs):
""" Internal wrapper for the notify_snatch and notify_download functions
Call either the JSON-RPC over HTTP or the legacy HTTP API methods depending on the Kodi API version.
Args:
title: Title of the notice to send
body: Message body of the notice to send
Return:
A list of results in the format of host:ip:result, where result will either be 'OK' or False.
"""
self.username, self.password = username, password
title = title or 'SickGear'
hosts = self._choose(hosts, sickgear.KODI_HOST)
success = True
message = []
for host in [x.strip() for x in hosts.split(',')]:
cur_host = unquote_plus(host)
api_version = self._get_kodi_version(cur_host)
if self.response and 401 == self.response.get('status_code'):
success = False
message += ['Fail: Cannot authenticate with %s' % cur_host]
self._log_debug(u'Failed to authenticate with %s' % cur_host)
elif not api_version:
success = False
message += ['Fail: No supported Kodi found at %s' % cur_host]
self._maybe_log_failed_detection(cur_host, 'connect and detect version for')
else:
if 4 >= api_version:
self._log_debug(u'Detected %sversion <= 11, using HTTP API'
% self.prefix and ' ' + self.prefix.capitalize())
__method_send = self._send
command = dict(command='ExecBuiltIn',
parameter='Notification(%s,%s)' % (title, body))
else:
self._log_debug(u'Detected version >= 12, using JSON API')
__method_send = self._send_json
command = dict(method='GUI.ShowNotification', params=dict(
[('title', title), ('message', body), ('image', self._sg_logo_url)]
+ ([], [('displaytime', 8000)])[self._testing]))
response_notify = __method_send(cur_host, command, 10)
if response_notify:
message += ['%s: %s' % ((response_notify, 'OK')['OK' in response_notify], cur_host)]
return self._choose(('Success, all hosts tested', '<br />\n'.join(message))[not success], success)
notifier = KodiNotifier
Reference in a new issue View git blame Copy permalink