def kodi_json_request(params):
data = json.dumps(params)
request = xbmc.executeJSONRPC(data)
try:
response = json.loads(request)
except UnicodeDecodeError:
response = json.loads(request.decode('utf-8', 'ignore'))
try:
if 'result' in response:
return response['result']
return None
except KeyError:
log("[%s] %s" %
(params['method'], response['error']['message']),level=xbmc.LOGWARNING)
return None
python类executeJSONRPC()的实例源码
def kodi_json(jsonmethod, params=None):
'''get info from the kodi json api'''
import json
kodi_json = {}
kodi_json["jsonrpc"] = "2.0"
kodi_json["method"] = jsonmethod
if not params:
params = {}
kodi_json["params"] = params
kodi_json["id"] = 1
json_response = xbmc.executeJSONRPC(json.dumps(kodi_json).encode("utf-8"))
json_object = json.loads(json_response.decode('utf-8', 'replace'))
result = None
if 'result' in json_object:
# look for correct returntype
if isinstance(json_object['result'], dict):
for key, value in json_object['result'].iteritems():
if not key == "limits":
result = value
break
else:
return json_object['result']
return result
def JSONRPC(method, params={}):
"""
Execute JSON-RPC method
:param method: name of the method. Example "Input.Back"
:type method: str
:param params: dictionary with parameter values
:type params: dict
:return: response as a string or None (in case of exception)
:rtype: str or None
"""
data = {
"jsonrpc": "2.0",
"id": 1,
"method": method,
"params": params
}
command = json.dumps(data)
try:
response = xbmc.executeJSONRPC(command)
except Exception, e:
log_exception("Error executing JSON RPC method " + method)
log_exception("Params: " + str(params))
log_exception(str(e))
return None
return response
def get_kodi_version(cls):
"""
Retrieves the Kodi version (Defaults to 17)
:returns: string - Kodi version
"""
version = 17
payload = {
'jsonrpc': '2.0',
'method': 'Application.GetProperties',
'params': {
'properties': ['version', 'name']
},
'id': 1
}
response = xbmc.executeJSONRPC(json.dumps(payload))
responses_uni = unicode(response, 'utf-8', errors='ignore')
response_serialized = json.loads(responses_uni)
if 'error' not in response_serialized.keys():
result = response_serialized.get('result', {})
version_raw = result.get('version', {})
version = version_raw.get('major', 17)
return version
def get_inputstream_version(cls):
"""
Retrieves the Inputsteam version (Defaults to 1.0.0)
:returns: string - Inputsteam version
"""
payload = {
'jsonrpc': '2.0',
'id': 1,
'method': 'Addons.GetAddonDetails',
'params': {
'addonid': 'inputstream.adaptive',
'properties': ['enabled', 'version']
}
}
# execute the request
response = xbmc.executeJSONRPC(json.dumps(payload))
responses_uni = unicode(response, 'utf-8', errors='ignore')
response_serialized = json.loads(responses_uni)
if 'error' not in response_serialized.keys():
result = response_serialized.get('result', {})
addon = result.get('addon', {})
if addon.get('enabled', False) is True:
return addon.get('version', '1.0.0')
return '1.0.0'
def get_inputstream_addon():
"""Checks if the inputstream addon is installed & enabled.
Returns the type of the inputstream addon used and if it's enabled,
or None if not found.
Returns
-------
:obj:`tuple` of obj:`str` and bool, or None
Inputstream addon and if it's enabled, or None
"""
type = 'inputstream.adaptive'
payload = {
'jsonrpc': '2.0',
'id': 1,
'method': 'Addons.GetAddonDetails',
'params': {
'addonid': type,
'properties': ['enabled']
}
}
response = xbmc.executeJSONRPC(json.dumps(payload))
data = json.loads(response)
if 'error' not in data.keys():
return type, data['result']['addon']['enabled']
return None, None
def kodi_json_request(params):
data = json.dumps(params)
request = xbmc.executeJSONRPC(data)
try:
response = json.loads(request)
except UnicodeDecodeError:
response = json.loads(request.decode('utf-8', 'ignore'))
try:
if 'result' in response:
return response['result']
return None
except KeyError:
logger.warn("[%s] %s" %
(params['method'], response['error']['message']))
return None
def get_details_from_player():
"""Docstring"""
logging.debug("addictedutils.get_details_from_player() begin")
# Create request
request = {
"jsonrpc": "2.0",
"method": "Player.GetItem",
"params": {
"properties": [
"showtitle",
"season",
"episode",
"duration",
"file"],
"playerid": 1
},
"id": "VideoGetItem"
}
# Make request
response = xbmc.executeJSONRPC(json.dumps(request))
logging.debug("response: " + pretty_json(response))
parsed_response = json.loads(response)['result']['item']
logging.debug("parsed_response: " + json.dumps(parsed_response, indent=2, sort_keys=True))
return parsed_response
def kodi_json_request(params):
data = json.dumps(params)
request = xbmc.executeJSONRPC(data)
try:
response = json.loads(request)
except UnicodeDecodeError:
response = json.loads(request.decode('utf-8', 'ignore'))
try:
if 'result' in response:
return response['result']
return None
except KeyError:
logger.warn("[%s] %s" %
(params['method'], response['error']['message']))
return None
def get_details_from_player():
# Create request
request = {
"jsonrpc": "2.0",
"method": "Player.GetItem",
"params": {
"properties": ["showtitle", "season", "episode", "duration", "file"],
"playerid": 1
},
"id": "VideoGetItem"
}
# Make request
response = xbmc.executeJSONRPC(json.dumps(request))
print(response)
parsed_response = json.loads(response)['result']['item']
print("---------")
print(parsed_response)
print("---------")
return parsed_response
def channel_to_favs(channel_name, channel_id):
params = {
'action': 'play_channel',
'channel_id': channel_id,
}
cmd = {
'jsonrpc': '2.0',
'method': 'Favourites.AddFavourite',
'params': {
'title': channel_name,
'type': 'media',
'path': _url + '?' + urllib.urlencode(params)
},
'id': '1'
}
debug_dict = json.loads(xbmc.executeJSONRPC(json.dumps(cmd)))
addon_log('channel_to_favs response: %s' % debug_dict)
def addon_enabled(self, addon_id):
rpc_request = json.dumps({"jsonrpc": "2.0",
"method": "Addons.GetAddonDetails",
"id": 1,
"params": {"addonid": "%s" % addon_id,
"properties": ["enabled"]}
})
response = json.loads(xbmc.executeJSONRPC(rpc_request))
try:
return response['result']['addon']['enabled'] is True
except KeyError:
message = response['error']['message']
code = response['error']['code']
error = 'Requested |%s| received error |%s| and code: |%s|' % (rpc_request, message, code)
xbmc.log(error, xbmc.LOGDEBUG)
return False
def set_addon_enabled(self, addon_id, enabled=True):
rpc_request = json.dumps({"jsonrpc": "2.0",
"method": "Addons.SetAddonEnabled",
"id": 1,
"params": {"addonid": "%s" % addon_id,
"enabled": enabled}
})
response = json.loads(xbmc.executeJSONRPC(rpc_request))
try:
return response['result'] == 'OK'
except KeyError:
message = response['error']['message']
code = response['error']['code']
error = 'Requested |%s| received error |%s| and code: |%s|' % (rpc_request, message, code)
xbmc.log(error, xbmc.LOGDEBUG)
return False
def __getattr__(self, method):
def handler(**kwargs):
command = {
'jsonrpc': '2.0',
'id': 1,
'method': '{0}.{1}'.format(self.family, method)
}
if kwargs:
command['params'] = kwargs
# xbmc.log(json.dumps(command))
ret = json.loads(xbmc.executeJSONRPC(json.dumps(command)))
if ret:
if 'error' in ret:
raise self.Exception(ret['error'])
else:
return ret['result']
else:
return None
return handler
def addon_enabled(self, addon_id):
rpc_request = json.dumps({"jsonrpc": "2.0",
"method": "Addons.GetAddonDetails",
"id": 1,
"params": {"addonid": "%s" % addon_id,
"properties": ["enabled"]}
})
response = json.loads(xbmc.executeJSONRPC(rpc_request))
try:
return response['result']['addon']['enabled'] is True
except KeyError:
message = response['error']['message']
code = response['error']['code']
error = 'Requested |%s| received error |%s| and code: |%s|' % (rpc_request, message, code)
xbmc.log(error, xbmc.LOGDEBUG)
return False
def set_addon_enabled(self, addon_id, enabled=True):
rpc_request = json.dumps({"jsonrpc": "2.0",
"method": "Addons.SetAddonEnabled",
"id": 1,
"params": {"addonid": "%s" % addon_id,
"enabled": enabled}
})
response = json.loads(xbmc.executeJSONRPC(rpc_request))
try:
return response['result'] == 'OK'
except KeyError:
message = response['error']['message']
code = response['error']['code']
error = 'Requested |%s| received error |%s| and code: |%s|' % (rpc_request, message, code)
xbmc.log(error, xbmc.LOGDEBUG)
return False
def get_episode_details(episode_id):
postdata = json.dumps({"jsonrpc": "2.0",
"id": 1,
'method': 'VideoLibrary.GetEpisodeDetails',
"params": {
'episodeid': episode_id,
"properties": ["season", "episode", "tvshowid", "playcount", "file", "resume"]
}})
json_query = xbmc.executeJSONRPC(postdata)
json_query = unicode(json_query, 'utf-8', errors='ignore')
json_query = json.loads(json_query)
if 'error' in json_query:
xbmc.log('%s: ERROR: %s' % (ADDONID, json_query['error']['stack']['message']))
return None
json_query = json_query['result']['episodedetails']
return json_query
def executeJSON(request):
"""Execute JSON-RPC Command
Args:
request: Dictionary with JSON-RPC Commands
Found code in xbmc-addon-service-watchedlist
"""
rpccmd = json.dumps(request) # create string from dict
json_query = xbmc.executeJSONRPC(rpccmd)
json_query = unicode(json_query, 'utf-8', errors='ignore')
json_response = json.loads(json_query)
return json_response
# This class is the interface between the internal default settings and the user.
# The user adjust the settings to his/her likings in Kodi. This class will make
# sure that the addon knows that the user changed a setting.
def execute_jsonrpc(command):
if not isinstance(command, basestring):
command = json.dumps(command)
response = xbmc.executeJSONRPC(command)
return json.loads(response)
def _get_kodi_json(method, params):
'''helper to get data from the kodi json database'''
json_response = xbmc.executeJSONRPC('{ "jsonrpc": "2.0", "method" : "%s", "params": %s, "id":1 }'
% (method, params.encode("utf-8")))
jsonobject = json.loads(json_response.decode('utf-8', 'replace'))
if 'result' in jsonobject:
jsonobject = jsonobject['result']
return jsonobject
def Kodi17():
xbmc_version = re.search('^(\d+)', xbmc.getInfoLabel( "System.BuildVersion" ))
if xbmc_version:
xbmc_version = int(xbmc_version.group(1))
else:
xbmc_version = 1
if xbmc_version >= 16.9:
dependencies = []
import glob
folder = xbmc.translatePath('special://home/addons/')
for DEPEND in glob.glob(folder+'*.*'):
try:dependencies.append(DEPEND.rsplit('\\', 1)[1])
except:dependencies.append(DEPEND.rsplit('/', 1)[1])
for THEPLUGIN in dependencies:
query = '{"jsonrpc":"2.0", "method":"Addons.SetAddonEnabled","params":{"addonid":"%s","enabled":true}, "id":1}' % (THEPLUGIN)
xbmc.executeJSONRPC(query)
xbmc.executebuiltin('UpdateLocalAddons')
xbmc.executebuiltin("UpdateAddonRepos")
def getBandwidthLimit():
json_result = xbmc.executeJSONRPC(
'{"jsonrpc":"2.0","method":"Settings.GetSettingValue","params":{"setting":"network.bandwidth"},"id":1}')
data_object = json.loads(json_result)
return data_object['result']['value']
def get_inputstream_addon(self):
"""Checks if the inputstream addon is installed & enabled.
Returns the type of the inputstream addon used and if it's enabled,
or None if not found.
Returns
-------
:obj:`tuple` of obj:`str` and bool, or None
Inputstream addon and if it's enabled, or None
"""
is_type = 'inputstream.adaptive'
is_enabled = False
payload = {
'jsonrpc': '2.0',
'id': 1,
'method': 'Addons.GetAddonDetails',
'params': {
'addonid': is_type,
'properties': ['enabled']
}
}
response = xbmc.executeJSONRPC(json.dumps(payload))
data = json.loads(response)
if 'error' not in data.keys():
if isinstance(data.get('result'), dict):
if isinstance(data.get('result').get('addon'), dict):
is_enabled = data.get('result').get('addon').get('enabled')
return (is_type, is_enabled)
return (None, is_enabled)
def movietitle_to_id(self, title):
query = {
"jsonrpc": "2.0",
"method": "VideoLibrary.GetMovies",
"params": {
"properties": ["title"]
},
"id": "libMovies"
}
try:
rpc_result = xbmc.executeJSONRPC(
jsonrpccommand=json.dumps(query, encoding='utf-8'))
json_result = json.loads(rpc_result)
if 'result' in json_result and 'movies' in json_result['result']:
json_result = json_result['result']['movies']
for movie in json_result:
# Switch to ascii/lowercase and remove special chars and spaces
# to make sure best possible compare is possible
titledb = movie['title'].encode('ascii', 'ignore')
titledb = re.sub(r'[?|$|!|:|#|\.|\,|\'| ]', r'', titledb).lower().replace('-', '')
if '(' in titledb:
titledb = titledb.split('(')[0]
titlegiven = title.encode('ascii','ignore')
titlegiven = re.sub(r'[?|$|!|:|#|\.|\,|\'| ]', r'', titlegiven).lower().replace('-', '')
if '(' in titlegiven:
titlegiven = titlegiven.split('(')[0]
if titledb == titlegiven:
return movie['movieid']
return '-1'
except Exception:
return '-1'
def showtitle_to_id(self, title):
query = {
"jsonrpc": "2.0",
"method": "VideoLibrary.GetTVShows",
"params": {
"properties": ["title", "genre"]
},
"id": "libTvShows"
}
try:
rpc_result = xbmc.executeJSONRPC(
jsonrpccommand=json.dumps(query, encoding='utf-8'))
json_result = json.loads(rpc_result)
if 'result' in json_result and 'tvshows' in json_result['result']:
json_result = json_result['result']['tvshows']
for tvshow in json_result:
# Switch to ascii/lowercase and
# remove special chars and spaces
# to make sure best possible compare is possible
titledb = tvshow['label'].encode('ascii', 'ignore')
titledb = re.sub(
pattern=r'[?|$|!|:|#|\.|\,|\'| ]',
repl=r'',
string=titledb).lower().replace('-', '')
if '(' in titledb:
titledb = titledb.split('(')[0]
titlegiven = title.encode('ascii', 'ignore')
titlegiven = re.sub(
pattern=r'[?|$|!|:|#|\.|\,|\'| ]',
repl=r'',
string=titlegiven).lower().replace('-', '')
if '(' in titlegiven:
titlegiven = titlegiven.split('(')[0]
if titledb == titlegiven:
return tvshow['tvshowid'], tvshow['genre']
return '-1', ''
except Exception:
return '-1', ''
def get_show_content_by_id(self, showid, showseason, showepisode):
showseason = int(showseason)
showepisode = int(showepisode)
props = ["season", "episode", "plot", "fanart", "art"]
query = {
"jsonrpc": "2.0",
"method": "VideoLibrary.GetEpisodes",
"params": {
"properties": props,
"tvshowid": int(showid[0])
},
"id": "1"
}
try:
rpc_result = xbmc.executeJSONRPC(
jsonrpccommand=json.dumps(query, encoding='utf-8'))
json_result = json.loads(rpc_result)
result = json_result.get('result', None)
if result is not None and 'episodes' in result:
result = result['episodes']
for episode in result:
in_season = episode['season'] == showseason
in_episode = episode['episode'] == showepisode
if in_season and in_episode:
infos = {}
if 'plot' in episode and len(episode['plot']) > 0:
infos.update({
'plot': episode['plot'],
'genre': showid[1]})
art = {}
if 'fanart' in episode and len(episode['fanart']) > 0:
art.update({'fanart': episode['fanart']})
if 'art' in episode and len(episode['art']['season.poster']) > 0:
art.update({
'thumb': episode['art']['season.poster']})
return infos, art
return False
except Exception:
return False
def get_movie_content_by_id(self, movieid):
query = {
"jsonrpc": "2.0",
"method": "VideoLibrary.GetMovieDetails",
"params": {
"movieid": movieid,
"properties": [
"genre",
"plot",
"fanart",
"thumbnail",
"art"]
},
"id": "libMovies"
}
try:
rpc_result = xbmc.executeJSONRPC(
jsonrpccommand=json.dumps(query, encoding='utf-8'))
json_result = json.loads(rpc_result)
result = json_result.get('result', None)
if result is not None and 'moviedetails' in result:
result = result.get('moviedetails', {})
infos = {}
if 'genre' in result and len(result['genre']) > 0:
infos.update({'genre': json_result['genre']})
if 'plot' in result and len(result['plot']) > 0:
infos.update({'plot': result['plot']})
art = {}
if 'fanart' in result and len(result['fanart']) > 0:
art.update({'fanart': result['fanart']})
if 'thumbnail' in result and len(result['thumbnail']) > 0:
art.update({'thumb': result['thumbnail']})
if 'art' in json_result and len(result['art']['poster']) > 0:
art.update({'poster': result['art']['poster']})
return infos, art
return False
except Exception:
return False
def executeJSON(request):
# =================================================================
# Execute JSON-RPC Command
# Args:
# request: Dictionary with JSON-RPC Commands
# Found code in xbmc-addon-service-watchedlist
# =================================================================
rpccmd = json.dumps(request) # create string from dict
json_query = xbmc.executeJSONRPC(rpccmd)
json_query = unicode(json_query, 'utf-8', errors='ignore')
json_response = json.loads(json_query)
return json_response
def json_query(query):
if not "jsonrpc" in query:
query["jsonrpc"] = "2.0"
if not "id" in query:
query["id"] = 1
xbmc_request = json.dumps(query)
raw = xbmc.executeJSONRPC(xbmc_request)
clean = unicode(raw, 'utf-8', errors='ignore')
response = json.loads(clean)
if "error" in response:
raise RPCError(response["error"])
return response.get('result', response)
def set_json(jsonmethod, params):
'''method to set info in the kodi json api'''
kodi_json = {}
kodi_json["jsonrpc"] = "2.0"
kodi_json["method"] = jsonmethod
kodi_json["params"] = params
kodi_json["id"] = 1
json_response = xbmc.executeJSONRPC(try_encode(json.dumps(kodi_json)))
return json.loads(json_response.decode('utf-8', 'replace'))
