# -*- coding: utf-8 -*- """ Basic access utilities. """ from datetime import datetime, timedelta import re, sys try: import requests from requests.auth import HTTPDigestAuth except ImportError: sys.exit('Debian: apt-get install python-requests or python3-requests') try: from urllib import quote except ImportError: from urllib.parse import quote __version__ = '0.28' ServerVersion = 'Set to back/frontend version after calls to Send()' global session global RecStatusCache global RecTypeCache global DupMethodCache RecStatusCache = {} RecTypeCache = {} DupMethodCache = {} session = None def Send(host='', port=6544, endpoint='', postdata={}, rest='', opts={}): """ Form a URL and send it to the back/frontend. Error handling is done here too. Examples: ========= import Utilities as api api.Send(host='someHostName', endpoint='Myth/GetHostName') Returns: {'String': 'someHostName'} api.Send(host='someFEName', port=6547, endpoint='Frontend/GetStatus') Returns: {'FrontendStatus': {'AudioTracks':... Input: ====== host: Must be set and is the hostname or IP of the back/frontend. port: Only needed if the backend is using a different port (unlikely) or set to 6547 for frontend endpoints. Defaults to 6544. endpoint: Must be set. Example: Myth/GetHostName postdata: May be set if the endpoint allows it. Used when information is to be added/changed/deleted. postdata is passed as a Python dict e.g. {'ChanId':1071, ...}. Don't use if rest is used. The HTTP method will be a POST (as opposed to a GET.) If using postdata, TAKE EXTREME CAUTION!!! Use opts['wrmi']=False 1st, set opts['debug']=True and watch what's sent. When happy with the data, make wrmi True. N.B. The MythTV Services API is still evolving and the wise user will backup their DB before including postdata. rest: May be set if the endpoint allows it. For example, endpoint= Myth/GetRecordedList, rest='Count=10&StorageGroup=Sports' Don't use if postdata is used. The HTTP method will be a GET. opts SHORT DESCRIPTION: It's OK to call this function without any options set and: • If there's postdata, nothing will be sent to the server • No "Debug:..." messages will print from this function • The server response will be gzipped • It will fail if the backend requires authorization ( user/pass would be required) DETAILS: opts is a dictionary of options that may be set in the calling program. Default values will be used if callers don't pass all or some of their own. The defaults are all False except for the user and pass as above. opts['debug']: Set to True and some informational messages will be printed. opts['noetag']: Don't request the back/frontend to check for a matching ETag. Mostly for testing. opts['nogzip']: Don't request the back/frontend to gzip it's response. Useful if watching protocol with a tool that doesn't uncompress it. opts['user']: Digest authentication. opts['pass']: opts['usexml']: For testing only! If True, causes the backend to send its response in XML rather than JSON. This will force an error when parsing the response. Defaults to False. opts['wrmi']: If True and there is postdata, the URL is actually sent to the server. If opts['wrmi'] is False and there is postdata, *NOTHING* is sent to the server. This is a failsafe that allows testing. Users can examine what's about to be sent before doing it (wrmi = We Really Mean It.) opts['wsdl']: If True return WSDL from the back/frontend. Accepts no rest or postdata. Just set endpoint, e.g. Content/wsdl Output: ====== Either the response from the server in a Python dict format or an error message in a dict (currently with an 'Abort' or 'Warning' key.) Callers can handle the response like this: response = api.Send(...) if list(response.keys())[0] in ['Abort', 'Warning']: sys.exit('{}'.format(list(response.values())[0])) normal processing... However, some errors returned by the server are in XML, e.g. if an endpoint is invalid. That will cause the JSON decoder to fail. Use the debug opt to view the failing response. Whenever Send() is used, the global 'ServerVersion' is set to the value returned by the back/frontend in the HTTP Server: header. It is saved as just the version, e.g. 0.28. Callers can check it and *may* choose to adjust their code work with other versions. """ global session global ServerVersion #!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!# # The version should never be changed without testing. If # # you're just getting data, no harm will be done. But if # # you Add/Delete/Update anything, then all bets are off! # # Anything requiring an HTTP POST is potentially dangerous. # #!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!# version = '29' version = '30' ############################################################## # Set missing options to False and if debug is True, tell # # the user if which options were changed. # ############################################################## optionList = [ 'debug', 'noetag', 'nogzip', 'usexml', 'wrmi', 'wsdl' ] missingList = '' for option in optionList: try: opts[option] except: missingList = missingList + option + ', ' opts[option] = False if opts['debug'] and missingList: print('Debug: Missing opts set to False: {}'.format(missingList[:-2])) ############################################################## # Make sure there's an endpoint and optionally *either* rest # # or postdata. If so, then form the URL. # ############################################################## if endpoint == '': return { 'Abort':'No endpoint (e.g. Myth/GetHostName.)' } if postdata and rest: return { 'Abort':'Use either postdata or rest' } if rest == '': qmark = '' else: qmark = '?' url='http://{}:{}/{}{}{}'.format(host, port, endpoint, qmark, rest) ############################################################## # Create a session. If postdata was supplied and wrmi wasn't # # set,then return immediately. Make sure postdata was passed # # as a dict. Add headers here. # ############################################################## if not session: session = requests.Session() session.headers.update({'User-Agent':'Python Services API Client v{}' .format(version)}) if opts['noetag']: session.headers.update({'Cache-Control':'no-store'}) session.headers.update({'If-None-Match':''}) if opts['nogzip']: session.headers.update({'Accept-Encoding':''}) else: session.headers.update({'Accept-Encoding':'gzip,deflate'}) if opts['usexml']: session.headers.update({'Accept':None}) else: session.headers.update({'Accept':'application/json'}) if opts['debug']: print('Debug: New session: {}'.format(session)) # TODO: Problem with the BE not accepting postdata in the initial # authorized query, Using a GET first as a workaround. try: opts['user'] and opts['pass'] session.auth = HTTPDigestAuth(opts['user'], opts['pass']) Send(host=host, port=port, endpoint='Myth/version', opts=opts) except: # Proceed without authentication. pass if opts['debug']: print('Debug: URL = {}'.format(url)) if postdata: print('Debug: The following postdata was included:') for key in postdata: print(' {:30} {}'.format(key, postdata[key])) if postdata and not opts['wrmi']: return { 'Warning':'wrmi=False' } if postdata: if not isinstance(postdata, dict): return { 'Abort':'usage: postdata must be passed as a dict' } if opts['wsdl'] and (rest or postdata): return { 'Abort':'usage: rest/postdata aren\'t allowed with WSDL' } ############################################################## # Actually try to get the data and handle errors. # ############################################################## try: if postdata: response = session.post(url, data=postdata) else: response = session.get(url) except requests.exceptions.HTTPError: return { 'Abort':'HTTP Error. URL was: {}'.format(url) } except requests.exceptions.URLRequired: return { 'Abort':'URL Required. URL was: {}'.format(url) } except requests.exceptions.Timeout: return { 'Abort':'Connect/Read Timeout: URL was {}'.format(url) } except requests.exceptions.ConnectionError: return { 'Abort':'Connection Error: URL was {}'.format(url) } except requests.exceptions.InvalidURL: return { 'Abort':'Invalid URL: URL was {}'.format(url) } except KeyboardInterrupt: return { 'Abort':'Keyboard Interrupt' } except: return { 'Abort':'Unexpected error: URL was: {}'.format(url) } if response.status_code == 401: return { 'Abort':'Unauthorized (401) error. Need valid user/password.' } # TODO: Should handle redirects here: if response.status_code > 299: return { 'Abort':'Unexpected status returned: {}: URL was: {}'.format( response.status_code, url) } ################################################################## # Process the contents of the HTTP Server: header. Try to see # # what version the server is running on. As of this writing the # # expected contents for 29, 0.28 and 0.27 are similar to: # # # #MythTV/29-pre-5-g6865940-dirty Linux/3.13.0-85-generic UPnP/1.0.# #MythTV/0.28-pre-3094-g349d3a4 Linux/3.13.0-66-generic UPnP/1.0 # #MythTV/28.0-10-g57c1afb Linux/4.4.0-21-generic UPnP/1.0. # #Linux 3.13.0-65-generic, UPnP/1.0, MythTV 0.27.20150622-1 # ################################################################## server = response.headers['Server'] if server == None: return { 'Abort':'No HTTP "Server:" header returned.' } else: if re.search('30', server): ServerVersion = '30' elif re.search('29', server): ServerVersion = '29' elif re.search('0.28', server) or re.search('28.', server): ServerVersion = '0.28' elif re.search('0.27', server): ServerVersion = '0.27' else: return { 'Abort':'Tested on 0.27/0.28/29/30, not: {}.'.format(server) } ############################################################## # Finally, return the response after converting the JSON to # # a dict. Or, if the wsdl option is set return that # ############################################################## if opts['wsdl']: return { 'WSDL':response.text} if opts['debug']: print('Debug: 1st 60 bytes of response: {}'.format(response.text[:60])) try: return response.json() except Exception as err: return { 'Abort':err } def URLEncode(value=''): """ This is really unnecessary. It's more of a reminder about how to use urllib.[parse]quote(). At least as of this writing, 0.28-pre doesn't decode the escaped values and the endpoints just get the percent encoded text. E.g. don't use it. How show titles with & or = in them work isn't clear. Input: A string. E.g a program's title or anything that has special characters like ?, & and UTF characters beyond the ASCII set. Output: The URL encoded string. E.g. ó becomes: %C3%B3 or ? becomes %3F. """ if value == '': print('Warning: Utilities.URLEncode() called without any value') return value return quote(value) def CreateFindTime(time=''): """ Normally used to take a starttime and convert it for use in adding new recordings. GetUTCOffset() should be called before this is, but that only needs to be done once. TODO: shouldn't this be removed and just use UTCToLocal with omityear=True? Input: Full UTC timestamp, e.g. 2014-08-12T22:00:00 (with or without the trailing 'Z'.) Output: Time portion of the above in local time. """ global UTCOffset if time == '': print('Warning: CreateFindTime called without any time') return None try: int(UTCOffset) utc_offset = UTCOffset except (NameError, ValueError): print('Warning: Run GetUTCOffset() first. using UTC offset of 0.') utc_offset = 0 time = time.replace('Z', '') dt = datetime.strptime(time, '%Y-%m-%dT%H:%M:%S') return (dt + timedelta(seconds = utc_offset)).strftime('%H:%M:%S') def UTCToLocal(utctime='', omityear=False): """ Does exactly that conversion. GetUTCOffset() should be run once before calling this function. A UTC offset of 0 will be used if UTCOffset isn't available, so the function won't abort. Inputs: utctime = Full UTC timestamp, e.g. 2014-08-12T22:00:00[Z]. omityear = If True, then drop the 4 digit year and following -. Output: Local time, also a string. Possibly without the year- and always without the T between the data/time and no trailing Z. """ global UTCOffset try: int(UTCOffset) utc_offset = UTCOffset except (NameError, ValueError): print('Warning: Run GetUTCOffset() first, using UTC offset of 0.') utc_offset = 0 if utctime == '': return 'Error: UTCToLocal(): utctime is empty!' utctime = utctime.replace('Z', '').replace('T', ' ') dt = datetime.strptime(utctime, '%Y-%m-%d %H:%M:%S') if omityear: fromstring = '%m-%d %H:%M:%S' else: fromstring = '%Y-%m-%d %H:%M:%S' return (dt + timedelta(seconds = utc_offset)).strftime(fromstring) def GetUTCOffset(host='', port=6544, opts={}): """ Get the backend's offset from UTC. Once retrieved, it's saved value is available in UTCOffset and is returned too. Additional calls to this function aren't necessary, but if made, won't query the backend again. Input: host, optional port. Output: The offset (in seconds) or -1 and a message prints """ global UTCOffset if host == '': print('GetUTCOffset(): Error: host is empty.') return -1 try: int(UTCOffset) return UTCOffset except (NameError, ValueError): resp_dict = Send(host=host, port=port, endpoint='Myth/GetTimeZone', opts=opts) if list(resp_dict.keys())[0] in ['Abort', 'Warning']: print('GetUTCOffset(): {}'.format(resp_dict)) return -1 else: UTCOffset = int(resp_dict['TimeZoneInfo']['UTCOffset']) return UTCOffset def RecStatusToString(host='', port=6544, recStatus=0, opts={}): """ Convert a signed integer to a Recording Status String and cache the result. recStatus defaults to 0, which currently (29.0) means 'Unknown' """ global RecStatusCache if host == '': print('RecStatusToString(): Error: host is empty.') return -1 try: str(RecStatusCache[recStatus]) return RecStatusCache[recStatus] except (KeyError, NameError, ValueError): endpoint='Dvr/RecStatusToString' rest = 'RecStatus={}'.format(recStatus) resp_dict = Send(host=host, port=port, endpoint=endpoint, rest=rest, \ opts=opts) if list(resp_dict.keys())[0] in ['Abort', 'Warning']: print('RecStatusToString(): {}'.format(resp_dict)) return resp_dict.keys()[0] else: RecStatusCache[recStatus] = resp_dict['String'] return RecStatusCache[recStatus] def RecTypeToString(host='', port=6544, recType=0, opts={}): """ Convert a signed integer to a Recording Type String and cache the result. recType defaults to 0, which currently (29.0) means 'Not Recording' """ global RecTypeCache if host == '': print('RecTypeToString(): Error: host is empty.') return -1 try: str(RecTypeCache[recType]) return RecTypeCache[recType] except (KeyError, NameError, ValueError): endpoint='Dvr/RecTypeToString' rest = 'RecType={}'.format(recType) resp_dict = Send(host=host, port=port, endpoint=endpoint, rest=rest, \ opts=opts) if list(resp_dict.keys())[0] in ['Abort', 'Warning']: print('RecTypeToString(): {}'.format(resp_dict)) return resp_dict.keys()[0] else: RecTypeCache[recType] = resp_dict['String'] return RecTypeCache[recType] def DupMethodToString(host='', port=6544, dupMethod=0, opts={}): """ Convert a signed integer to a Duplicate Method String and cache the result. dupMethod defaults to 0, which currently (29.0) means 'No Search' """ global DupMethodCache if host == '': print('DupMethodToString(): Error: host is empty.') return -1 try: str(DupMethodCache[dupMethod]) return DupMethodCache[dupMethod] except (KeyError, NameError, ValueError): endpoint='Dvr/DupMethodToString' rest = 'DupMethod={}'.format(dupMethod) resp_dict = Send(host=host, port=port, endpoint=endpoint, rest=rest, \ opts=opts) if list(resp_dict.keys())[0] in ['Abort', 'Warning']: print('DupMethodToString(): {}'.format(resp_dict)) return resp_dict.keys()[0] else: DupMethodCache[dupMethod] = resp_dict['String'] return DupMethodCache[dupMethod]