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类LOGWARNING的实例源码
def log(msg, level=xbmc.LOGNOTICE):
"""
Outputs message to log file
:param msg: message to output
:param level: debug levelxbmc. Values:
xbmc.LOGDEBUG = 0
xbmc.LOGERROR = 4
xbmc.LOGFATAL = 6
xbmc.LOGINFO = 1
xbmc.LOGNONE = 7
xbmc.LOGNOTICE = 2
xbmc.LOGSEVERE = 5
xbmc.LOGWARNING = 3
"""
log_message = u'{0}: {1}'.format(addonID, msg)
xbmc.log(log_message.encode("utf-8"), level)
def get_track_url(self, track_id, quality=None, cut_id=None, fallback=True):
oldSessionId = self.session_id
self.session_id = self.stream_session_id
soundQuality = quality if quality else self._config.quality
#if soundQuality == Quality.lossless and self._config.codec == 'MQA' and not cut_id:
# soundQuality = Quality.hi_res
media = Session.get_track_url(self, track_id, quality=soundQuality, cut_id=cut_id)
if fallback and soundQuality == Quality.lossless and (media == None or media.isEncrypted):
log(media.url, level=xbmc.LOGWARNING)
if media:
log('Got encryptionKey "%s" for track %s, trying HIGH Quality ...' % (media.encryptionKey, track_id), level=xbmc.LOGWARNING)
else:
log('No Lossless stream for track %s, trying HIGH Quality ...' % track_id, level=xbmc.LOGWARNING)
media = self.get_track_url(track_id, quality=Quality.high, cut_id=cut_id, fallback=False)
if media:
if quality == Quality.lossless and media.codec not in ['FLAC', 'ALAC', 'MQA']:
xbmcgui.Dialog().notification(plugin.name, _T(30504) , icon=xbmcgui.NOTIFICATION_WARNING)
log('Got stream with soundQuality:%s, codec:%s' % (media.soundQuality, media.codec))
self.session_id = oldSessionId
return media
def emit(self, record):
if record.levelno < logging.WARNING and self._modules and not record.name in self._modules:
# Log INFO and DEBUG only with enabled modules
return
levels = {
logging.CRITICAL: xbmc.LOGFATAL,
logging.ERROR: xbmc.LOGERROR,
logging.WARNING: xbmc.LOGWARNING,
logging.INFO: xbmc.LOGNOTICE,
logging.DEBUG: xbmc.LOGSEVERE,
logging.NOTSET: xbmc.LOGNONE,
}
try:
xbmc.log(self.format(record), levels[record.levelno])
except:
try:
xbmc.log(self.format(record).encode('utf-8', 'ignore'), levels[record.levelno])
except:
xbmc.log(b"[%s] Unicode Error in message text" % self.pluginName, levels[record.levelno])
def search(self, artist, album, track):
'''get musicbrainz id by query of artist, album and/or track'''
albumid = ""
artistid = ""
try:
# lookup with artist and album (preferred method)
if artist and album:
artistid, albumid = self.search_release_group_match(artist, album)
# lookup with artist and track (if no album provided)
if not (artistid and albumid) and artist and track:
artistid, albumid = self.search_recording_match(artist, track)
# last resort: lookup with trackname as album
if not (artistid and albumid) and artist and track:
artistid, albumid = self.search_release_group_match(artist, track)
except Exception as exc:
log_msg("Error in musicbrainz.search: %s" % str(exc), xbmc.LOGWARNING)
return (artistid, albumid)
def log(txt, log_level=xbmc.LOGDEBUG):
"""
Log something to the kodi.log file
:param txt: Text to write to the log
:param log_level: Severity of the log text
:return: None
"""
if (_addon.getSetting("debug") == "true") or (log_level != xbmc.LOGDEBUG):
if isinstance(txt, str):
try:
txt = txt.decode("utf-8")
except UnicodeDecodeError:
xbmc.log('Could not decode to Unicode: {0}'.format(txt), level=xbmc.LOGWARNING)
return
message = u'%s: %s' % (_addonid, txt)
xbmc.log(msg=message.encode("utf-8"), level=log_level)
def log(message,loglevel=xbmc.LOGNOTICE):
""""save message to kodi.log.
Args:
message: has to be unicode, http://wiki.xbmc.org/index.php?title=Add-on_unicode_paths#Logging
loglevel: xbmc.LOGDEBUG, xbmc.LOGINFO, xbmc.LOGNOTICE, xbmc.LOGWARNING, xbmc.LOGERROR, xbmc.LOGFATAL
"""
xbmc.log(encode(__addon_id__ + u": " + message), level=loglevel)
def log_exception(modulename, exceptiondetails):
'''helper to properly log an exception'''
log_msg(format_exc(sys.exc_info()), xbmc.LOGDEBUG)
log_msg("Exception in %s ! --> %s" % (modulename, exceptiondetails), xbmc.LOGWARNING)
def log_warning(self, message):
"""
Add WARNING message to the Kodi log
:param message: message to write to the Kodi log
:type message: str
"""
self.log(message, xbmc.LOGWARNING)
def i18n(string_id):
try:
return addon.getLocalizedString(strings.STRINGS[string_id]).encode('utf-8', 'ignore')
except Exception as e:
xbmc.log('%s: Failed String Lookup: %s (%s)' % (get_name(), string_id, e), xbmc.LOGWARNING)
return string_id
def warning(self, message):
message = prep_log_message(message)
if xbmc:
self._log(message, xbmc.LOGWARNING)
else:
self._log.debug(message)
def _get_database(self):
'''get reference to our sqllite _database - performs basic integrity check'''
addon = xbmcaddon.Addon(ADDON_ID)
dbpath = addon.getAddonInfo('profile')
dbfile = xbmc.translatePath("%s/simplecache.db" % dbpath).decode('utf-8')
if not xbmcvfs.exists(dbpath):
xbmcvfs.mkdirs(dbpath)
del addon
try:
connection = sqlite3.connect(dbfile, timeout=30, isolation_level=None)
connection.execute('SELECT * FROM simplecache LIMIT 1')
return connection
except Exception as error:
# our _database is corrupt or doesn't exist yet, we simply try to recreate it
if xbmcvfs.exists(dbfile):
xbmcvfs.delete(dbfile)
try:
connection = sqlite3.connect(dbfile, timeout=30, isolation_level=None)
connection.execute(
"""CREATE TABLE IF NOT EXISTS simplecache(
id TEXT UNIQUE, expires INTEGER, data TEXT, checksum INTEGER)""")
return connection
except Exception as error:
self._log_msg("Exception while initializing _database: %s" % str(error), xbmc.LOGWARNING)
self.close()
return None
def _execute_sql(self, query, data=None):
'''little wrapper around execute and executemany to just retry a db command if db is locked'''
retries = 0
result = None
error = None
# always use new db object because we need to be sure that data is available for other simplecache instances
with self._get_database() as _database:
while not retries == 10:
if self._exit:
return None
try:
if isinstance(data, list):
result = _database.executemany(query, data)
elif data:
result = _database.execute(query, data)
else:
result = _database.execute(query)
return result
except sqlite3.OperationalError as error:
if "_database is locked" in error:
self._log_msg("retrying DB commit...")
retries += 1
self._monitor.waitForAbort(0.5)
else:
break
except Exception as error:
break
self._log_msg("_database ERROR ! -- %s" % str(error), xbmc.LOGWARNING)
return None
def _get_token(self, refresh=False):
'''get jwt token for api'''
# get token from memory cache first
if self._token and not refresh:
return self._token
token = self._win.getProperty("script.module.thetvdb.token").decode('utf-8')
if token and not refresh:
return token
# refresh previous token
prev_token = self._addon.getSetting("token")
if prev_token:
url = 'https://api.thetvdb.com/refresh_token'
headers = {'Content-Type': 'application/json', 'Accept': 'application/json',
'User-agent': 'Mozilla/5.0', 'Authorization': 'Bearer %s' % prev_token}
response = requests.get(url, headers=headers)
if response and response.content and response.status_code == 200:
data = json.loads(response.content.decode('utf-8', 'replace'))
token = data["token"]
if token:
self._win.setProperty("script.module.thetvdb.token", token)
self._token = token
return token
# do first login to get initial token
url = 'https://api.thetvdb.com/login'
payload = {'apikey': self.api_key}
headers = {'Content-Type': 'application/json', 'Accept': 'application/json', 'User-agent': 'Mozilla/5.0'}
response = requests.post(url, data=json.dumps(payload), headers=headers)
if response and response.content and response.status_code == 200:
data = json.loads(response.content.decode('utf-8', 'replace'))
token = data["token"]
self._addon.setSetting("token", token)
self._win.setProperty("script.module.thetvdb.token", token)
self._token = token
return token
else:
self._log_msg("Error getting JWT token!", xbmc.LOGWARNING)
return None
def _get_local_time(self, timestr):
'''returns the correct localized representation of the time provided by the api'''
result = ""
try:
if timestr and ":" in timestr:
timestr = timestr.replace(".", ":")
if "H" in xbmc.getRegion('time'):
time_format = "HH:mm"
else:
time_format = "h:mm A"
if " AM" in timestr or " PM" in timestr:
result = arrow.get(timestr, 'h:mm A').format(time_format, locale=KODI_LANGUAGE)
elif " am" in timestr or " pm" in timestr:
result = arrow.get(timestr, 'h:mm a').format(time_format, locale=KODI_LANGUAGE)
elif "AM" in timestr or "PM" in timestr:
result = arrow.get(timestr, 'h:mmA').format(time_format, locale=KODI_LANGUAGE)
elif "am" in timestr or "pm" in timestr:
result = arrow.get(timestr, 'h:mma').format(time_format, locale=KODI_LANGUAGE)
elif len(timestr.split(":")[0]) == 1:
result = arrow.get(timestr, 'h:mm').format(time_format, locale=KODI_LANGUAGE)
else:
result = arrow.get(timestr, 'HH:mm').format(time_format, locale=KODI_LANGUAGE)
except Exception as exc:
self._log_msg(str(exc), xbmc.LOGWARNING)
return timestr
return result
def emit(self, record):
levels = {
logging.CRITICAL: xbmc.LOGFATAL,
logging.ERROR: xbmc.LOGERROR,
logging.WARNING: xbmc.LOGWARNING,
logging.INFO: xbmc.LOGINFO,
logging.DEBUG: xbmc.LOGDEBUG,
logging.NOTSET: xbmc.LOGNONE,
}
if get_setting_as_bool('debug'):
try:
xbmc.log(self.format(record), levels[record.levelno])
except UnicodeEncodeError:
xbmc.log(self.format(record).encode(
'utf-8', 'ignore'), levels[record.levelno])
def _get_database(self):
'''get reference to our sqllite _database - performs basic integrity check'''
addon = xbmcaddon.Addon(ADDON_ID)
dbpath = addon.getAddonInfo('profile')
dbfile = xbmc.translatePath("%s/simplecache.db" % dbpath).decode('utf-8')
if not xbmcvfs.exists(dbpath):
xbmcvfs.mkdirs(dbpath)
del addon
try:
connection = sqlite3.connect(dbfile, timeout=30, isolation_level=None)
connection.execute('SELECT * FROM simplecache LIMIT 1')
return connection
except Exception as error:
# our _database is corrupt or doesn't exist yet, we simply try to recreate it
if xbmcvfs.exists(dbfile):
xbmcvfs.delete(dbfile)
try:
connection = sqlite3.connect(dbfile, timeout=30, isolation_level=None)
connection.execute(
"""CREATE TABLE IF NOT EXISTS simplecache(
id TEXT UNIQUE, expires INTEGER, data TEXT, checksum INTEGER)""")
return connection
except Exception as error:
self._log_msg("Exception while initializing _database: %s" % str(error), xbmc.LOGWARNING)
self.close()
return None
def _execute_sql(self, query, data=None):
'''little wrapper around execute and executemany to just retry a db command if db is locked'''
retries = 0
result = None
error = None
# always use new db object because we need to be sure that data is available for other simplecache instances
with self._get_database() as _database:
while not retries == 10:
if self._exit:
return None
try:
if isinstance(data, list):
result = _database.executemany(query, data)
elif data:
result = _database.execute(query, data)
else:
result = _database.execute(query)
return result
except sqlite3.OperationalError as error:
if "_database is locked" in error:
self._log_msg("retrying DB commit...")
retries += 1
self._monitor.waitForAbort(0.5)
else:
break
except Exception as error:
break
self._log_msg("_database ERROR ! -- %s" % str(error), xbmc.LOGWARNING)
return None
def __init__(self, pluginName, detailLevel=0, enableTidalApiLog=False):
''' Initialize Error Logging with a given Log Level
detailLevel = 0 : xbmc.LOGERROR and xbmc.LOGNOTICE
detailLevel = 1 : as level 0 plus xbmc.LOGWARNING
detailLevel = 2 : as level 1 plus xbmc.LOGDEBUG
detailLevel = 3 : as level 2 plus xbmc.LOGSEVERE
'''
self.pluginName = pluginName
self.detailLevel = detailLevel
self.debugServer = 'localhost'
# Set Log Handler for tidalapi
self.addTidalapiLogger(pluginName, enableDebug=enableTidalApiLog)
def log_warning(self, message):
"""
Add WARNING message to the Kodi log
:param message: message to write to the Kodi log
:type message: str
"""
self.log(message, xbmc.LOGWARNING)
def emit(self, record):
levels = {
logging.CRITICAL: xbmc.LOGFATAL,
logging.ERROR: xbmc.LOGERROR,
logging.WARNING: xbmc.LOGWARNING,
logging.INFO: xbmc.LOGINFO,
logging.DEBUG: xbmc.LOGDEBUG,
logging.NOTSET: xbmc.LOGNONE,
}
if get_setting_as_bool('debug'):
try:
xbmc.log(self.format(record), levels[record.levelno])
except UnicodeEncodeError:
xbmc.log(self.format(record).encode(
'utf-8', 'ignore'), levels[record.levelno])
def log_exception(modulename, exceptiondetails):
'''helper to properly log an exception'''
log_msg(format_exc(sys.exc_info()), xbmc.LOGWARNING)
log_msg("ERROR in %s ! --> %s" % (modulename, exceptiondetails), xbmc.LOGERROR)
def get_albuminfo(self, mbalbumid):
'''get album info from musicbrainz'''
result = {}
try:
data = self.mbrainz.get_release_group_by_id(mbalbumid, includes=["tags", "ratings"])
if data and data.get("release-group"):
data = data["release-group"]
result["year"] = data.get("first-release-date", "").split("-")[0]
result["title"] = data.get("title", "")
if data.get("rating") and data["rating"].get("rating"):
result["rating"] = data["rating"].get("rating")
result["votes"] = data["rating"].get("votes", 0)
if data.get("tag-list"):
result["tags"] = []
result["genre"] = []
for tag in data["tag-list"]:
if tag["count"] and int(tag["count"]) > 1:
taglst = []
if " / " in tag["name"]:
taglst = tag["name"].split(" / ")
elif "/" in tag["name"]:
taglst = tag["name"].split("/")
elif " - " in tag["name"]:
taglst = tag["name"].split(" - ")
elif "-" in tag["name"]:
taglst = tag["name"].split("-")
else:
taglst = [tag["name"]]
for item in taglst:
if not item in result["tags"]:
result["tags"].append(item)
if not item in result["genre"] and int(tag["count"]) > 4:
result["genre"].append(item)
except Exception as exc:
log_msg("Error in musicbrainz - get album details: %s" % str(exc), xbmc.LOGWARNING)
return result
def i18n(string_id):
try:
return addon.getLocalizedString(strings.STRINGS[string_id]).encode('utf-8', 'ignore')
except Exception as e:
xbmc.log('%s: Failed String Lookup: %s (%s)' % (get_name(), string_id, e), xbmc.LOGWARNING)
return string_id
def warn(self, message, *args):
self._log(xbmc.LOGWARNING, message, *args)
def get_image_url_for_series(series_id):
series_items = get_items(0, series=series_id, limit=1)
if series_items:
series_item = series_items[0]
if 'partOfSeries' in series_item:
if 'available' in series_item['partOfSeries']['image']:
if series_item['partOfSeries']['image']['available']:
image_url = '{0}/{1}/{2}.{3}'.format(
_image_cdn_url, _image_transformation, series_item['partOfSeries']['image']['id'], 'png')
return image_url
else:
return 'NO_ITEMS'
log('Could not find image URL for series {0}'.format(series_id), xbmc.LOGWARNING)
return None
def get_app_id():
app_id = _addon.getSetting("appID")
if app_id == '':
try:
import credentials
app_id = credentials._appId
except ImportError:
credentials = None
log('Could not find the app_id. Either set it from the setting menu or create credentials.py file.',
xbmc.LOGWARNING)
return app_id
def get_app_key():
app_key = _addon.getSetting("appKey")
if app_key == '':
try:
import credentials
app_key = credentials._appKey
except ImportError:
credentials = None
log('Could not find the app_key. Either set it from the setting menu or create credentials.py file.',
xbmc.LOGWARNING)
return app_key
def get_color(setting):
color = _addon.getSetting(setting)
colors = ["aliceblue", "antiquewhite", "aqua", "aquamarine", "azure", "beige", "bisque", "black", "blanchedalmond",
"blue", "blueviolet", "brown", "burlywood", "cadetblue", "chartreuse", "chocolate", "coral",
"cornflowerblue", "cornsilk", "crimson", "cyan", "darkblue", "darkcyan", "darkgoldenrod", "darkgray",
"darkgreen", "darkgrey", "darkkhaki", "darkmagenta", "darkolivegreen", "darkorange", "darkorchid",
"darkred", "darksalmon", "darkseagreen", "darkslateblue", "darkslategray", "darkslategrey",
"darkturquoise", "darkviolet", "deeppink", "deepskyblue", "dimgray", "dimgrey", "dodgerblue", "firebrick",
"floralwhite", "forestgreen", "fuchsia", "gainsboro", "ghostwhite", "gold", "goldenrod", "gray", "green",
"greenyellow", "grey", "honeydew", "hotpink", "indianred", "indigo", "ivory", "khaki", "lavender",
"lavenderblush", "lawngreen", "lemonchiffon", "lightblue", "lightcoral", "lightcyan",
"lightgoldenrodyellow", "lightgray", "lightgreen", "lightgrey", "lightpink", "lightsalmon",
"lightseagreen", "lightskyblue", "lightslategray", "lightslategrey", "lightsteelblue", "lightyellow",
"lime", "limegreen", "linen", "magenta", "maroon", "mediumaquamarine", "mediumblue", "mediumorchid",
"mediumpurple", "mediumseagreen", "mediumslateblue", "mediumspringgreen", "mediumturquoise",
"mediumvioletred", "midnightblue", "mintcream", "mistyrose", "moccasin", "navajowhite", "navy", "none",
"oldlace", "olive", "olivedrab", "orange", "orangered", "orchid", "palegoldenrod", "palegreen",
"paleturquoise", "palevioletred", "papayawhip", "peachpuff", "peru", "pink", "plum", "powderblue",
"purple", "red", "rosybrown", "royalblue", "saddlebrown", "salmon", "sandybrown", "seagreen", "seashell",
"sienna", "silver", "skyblue", "slateblue", "slategray", "slategrey", "snow", "springgreen", "steelblue",
"tan", "teal", "thistle", "tomato", "transparent", "turquoise", "violet", "wheat", "white", "whitesmoke",
"yellow", "yellowgreen"]
if _addon.getSetting('randomColors') == 'true':
import random
return random.choice(colors)
if color not in colors:
log('Unknown color "{0}."'.format(color), xbmc.LOGWARNING)
log('Available colors: {0}'.format(colors))
return 'black'
return color
def log_warning(self, message):
"""
Add WARNING message to the Kodi log
:param message: message to write to the Kodi log
:type message: str
"""
self.log(message, xbmc.LOGWARNING)
