Advertisement
Guest User

add_recording.py

a guest
Jan 31st, 2023
67
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Python 10.57 KB | None | 0 0
  1. #!/usr/bin/python3
  2. # -*- coding: utf-8 -*-
  3.  
  4. '''
  5. Create a new recording rule for the title passed on the command line.
  6.  
  7. Use: add_recording_rule.py --help to get started.
  8.  
  9. The --title <title> must match a program name exactly and it will also
  10. become the name of the new recording rule. If the rule already exists,
  11. the program will abort.
  12.  
  13. This is pep8 and pylint compliant. Tested using Python 3.
  14.  
  15. Install as: add_recording_rule.py, chmod a+rx.
  16. '''
  17.  
  18. import argparse
  19. import json
  20. import logging
  21. import sys
  22.  
  23. try:
  24.     from MythTV.services_api import (send as api, utilities as util)
  25. except ImportError:
  26.     sys.exit("Missing python bindings?")
  27.  
  28. TYPES = (
  29.     "Single Record",
  30.     "Record All",
  31.     "Record One",
  32.     "Record Daily",
  33.     "Record Weekly",
  34. )
  35.  
  36.  
  37. # pylint: disable=consider-using-f-string
  38. def process_command_line():
  39.     '''All command line processing is done here.'''
  40.  
  41.     parser = argparse.ArgumentParser(description='Add a recording rule',
  42.                                      epilog='Default values are in ()s')
  43.  
  44.     mandatory = parser.add_argument_group('required arguments')
  45.  
  46.     parser.add_argument('--debug', action='store_true',
  47.                         help='turn on debug messages (%(default)s)')
  48.  
  49.     parser.add_argument('--digest', type=str, metavar='<user:pass>',
  50.                         help='digest username:password (%(default)s)')
  51.  
  52.     mandatory.add_argument('--host', type=str, required=True,
  53.                            metavar='<hostname>', help='backend hostname')
  54.  
  55.     parser.add_argument('--port', type=int, default=6544, metavar='<port>',
  56.                         help='port number of the Services API (%(default)s)')
  57.  
  58.     parser.add_argument('--quiet', action='store_true',
  59.                         help='suppress progress messages (%(default)s)')
  60.  
  61.     parser.add_argument('--template', type=str, required=False,
  62.                         default='Default', metavar='<temp>',
  63.                         help='template name, (%(default)s)')
  64.  
  65.     mandatory.add_argument('--title', type=str, required=True,
  66.                            metavar='<title>',
  67.                            help='full program name, no wild cards/regex')
  68.  
  69.     parser.add_argument('--type', type=str, required=False, choices=(TYPES),
  70.                         default='Record All', metavar='<type>',
  71.                         help='Record <type>, (%(default)s)')
  72.  
  73.     parser.add_argument('--version', action='version', version='%(prog)s 0.14')
  74.  
  75.     parser.add_argument('--wrmi', action='store_true',
  76.                         help='allow data to be changed (%(default)s)')
  77.  
  78.     return vars(parser.parse_args())
  79.  
  80.  
  81. def setup(backend, opts):
  82.     '''
  83.    Make sure the backend is up (GetHostName) and then set the backend's UTC
  84.    offset for other methods to use.
  85.    '''
  86.  
  87.     try:
  88.         backend.send(endpoint='Myth/GetHostName', opts=opts)
  89.         int(util.get_utc_offset(backend=backend, opts=opts))
  90.     except ValueError:
  91.         sys.exit('\nExit, non integer response from get_utc_offset.')
  92.     except RuntimeError as error:
  93.         sys.exit('\nExit on fatal API error: "{}"'.format(error))
  94.  
  95.  
  96. def schedule_already_exists(backend, args, opts):
  97.     '''
  98.    See if there's already a rule for the title.
  99.    '''
  100.  
  101.     endpoint = 'Dvr/GetRecordScheduleList'
  102.  
  103.     try:
  104.         resp_dict = backend.send(endpoint=endpoint, opts=opts)
  105.     except RuntimeError as error:
  106.         sys.exit('\nExit, Get Existing Rule: Fatal error; "{}"'.format(error))
  107.  
  108.     if int(resp_dict['RecRuleList']['Count']) < 1:
  109.         sys.exit('\nExit, no recording rules found.\n')
  110.  
  111.     for rule in resp_dict['RecRuleList']['RecRules']:
  112.         if rule['Title'] == args['title']:
  113.             if args['debug']:
  114.                 print(json.dumps(rule, sort_keys=True, indent=4,
  115.                                  separators=(',', ': ')))
  116.  
  117.             return True
  118.  
  119.     return False
  120.  
  121.  
  122. def get_template(backend, args, opts):
  123.     '''
  124.    Gets the requested (or default) template. This will be modified
  125.    with guide data for the title of interest, then send to the
  126.    backend in a POST. Misspelled template names return the Default
  127.    template.
  128.  
  129.    Only the template name is used, not the trailing: (Template) text.
  130.    '''
  131.  
  132.     endpoint = 'Dvr/GetRecordSchedule'
  133.     rest = 'Template={}'.format(args['template'])
  134.  
  135.     try:
  136.         resp_dict = backend.send(endpoint=endpoint, rest=rest, opts=opts)
  137.     except RuntimeError as error:
  138.         sys.exit('\nExit, Get Template: Fatal error; "{}"'.format(error))
  139.  
  140.     if args['debug']:
  141.         print(json.dumps(resp_dict['RecRule'], sort_keys=True, indent=4,
  142.                          separators=(',', ': ')))
  143.  
  144.     # Templates are always Id -1, just double checking here...
  145.     if resp_dict['RecRule']['Id'] != '-1':
  146.         return None
  147.  
  148.     return resp_dict['RecRule']
  149.  
  150.  
  151. def get_program_data(backend, args, opts):
  152.     '''
  153.    Find matching program(s) from the guide. Note that if --title=Blah,
  154.    then any title with the string Blah in it will be returned by
  155.    GetProgramList.
  156.    '''
  157.  
  158.     endpoint = 'Guide/GetProgramList'
  159.     rest = 'Details=True&TitleFilter={}'.format(args['title'])
  160.  
  161.     try:
  162.         resp_dict = backend.send(endpoint=endpoint, rest=rest, opts=opts)
  163.     except RuntimeError as error:
  164.         sys.exit('\nExit, Get Upcoming: Fatal error; "{}"'.format(error))
  165.  
  166.     count = int(resp_dict['ProgramList']['TotalAvailable'])
  167.  
  168.     if args['debug']:
  169.         print('\nDebug: Programs matching --title {} = {}'
  170.               .format(args['title'], count))
  171.  
  172.     if count < 1:
  173.         sys.exit('\nExit, No programs in the guide matching: {}.\n'
  174.                  .format(args['title']))
  175.  
  176.     for program in resp_dict['ProgramList']['Programs']:
  177.         if args['debug']:
  178.             print('Comparing {} to {}'.format(args['title'], program['Title']))
  179.         if program['Title'] == args['title']:
  180.             if args['debug']:
  181.                 print(json.dumps(program, sort_keys=True, indent=4,
  182.                                  separators=(',', ': ')))
  183.             return program
  184.  
  185.         continue
  186.  
  187.     return None
  188.  
  189.  
  190. def update_template(template, guide_data, args):
  191.     '''
  192.    Put selected guide information into the template to be sent as
  193.    postdata for the new rule.
  194.    '''
  195.  
  196.     try:
  197.         template['StartTime'] = guide_data['StartTime']
  198.         template['EndTime'] = guide_data['EndTime']
  199.         template['Title'] = guide_data['Title']
  200.         template['Type'] = args['type']
  201.         template['Station'] = guide_data['Channel']['CallSign']
  202.         template['ChanId'] = guide_data['Channel']['ChanId']
  203.         template['SearchType'] = 'None'
  204.         template['Category'] = guide_data['Category']
  205.         template['SeriesId'] = guide_data['SeriesId']
  206.         template['ProgramId'] = guide_data['ProgramId']
  207.         template['FindTime'] = util.create_find_time(guide_data['StartTime'])
  208.         template['Description'] = 'Rule created by add_recording_rule.py'
  209.     except KeyError:
  210.         return False
  211.  
  212.     return True
  213.  
  214.  
  215. def add_rule(backend, template, args, opts):
  216.     '''
  217.    Send the changed data to the backend.
  218.    '''
  219.  
  220.     endpoint = 'Dvr/AddRecordSchedule'
  221.  
  222.     params_not_sent = ('AverageDelay', 'CallSign', 'Id', 'LastDeleted',
  223.                        'LastRecorded', 'NextRecording', 'ParentId')
  224.  
  225.     for param in params_not_sent:
  226.         try:
  227.             del template[param]
  228.         except KeyError:
  229.             pass
  230.  
  231.     opts['wrmi'] = args['wrmi']
  232.  
  233.     try:
  234.         resp_dict = backend.send(endpoint=endpoint, postdata=template,
  235.                                  opts=opts)
  236.     except RuntimeWarning as error:
  237.         sys.exit('Exit, Unable to add rule: {}. Warning was: {}.'
  238.                  .format(template['Title'], error))
  239.     except RuntimeError as error:
  240.         sys.exit('\nExit, Fatal API Error response: {}\n'.format(error))
  241.  
  242.     opts['wrmi'] = False
  243.  
  244.     if isinstance(resp_dict, dict) and isinstance(resp_dict['uint'], str):
  245.  
  246.         recording_rule = int(resp_dict['uint'])
  247.  
  248.         if recording_rule < 4294967295:
  249.             vprint('\nAdded: "{}" (RecordId {}).'
  250.                    .format(template['Title'], recording_rule), args)
  251.         else:
  252.             recording_rule = -1
  253.             vprint('Backend failed to add: "{}" (RecordId {}).'
  254.                    .format(template['Title'], recording_rule), args)
  255.     else:
  256.         vprint('Expected a "uint: int" dictionary response, but got {}'
  257.                .format(resp_dict), args)
  258.  
  259.  
  260. def vprint(message, args):
  261.     '''
  262.    Verbose Print: print recording rule information unless --quiet
  263.    was used. Not fully implemented, as there are still lots of
  264.    print()s here.
  265.  
  266.    The intention is that if run out of some other program, this
  267.    will can remain quiet. sys.exit()s will return 1 for failures.
  268.    This may get expanded to put messages in a log...
  269.    '''
  270.  
  271.     if not args['quiet']:
  272.         print(message)
  273.  
  274.  
  275. def main():
  276.     '''
  277.    The primary job of main is to get the arguments from the command line,
  278.    setup logging (and possibly) handle the digest user/password then:
  279.  
  280.        • Create an instance of the Send class
  281.        • See if a rule exists for --title
  282.        • Get the selected template
  283.        • Get data for command line title from the guide
  284.        • Update the template with the guide data
  285.        • Add the rule on the backend.
  286.    '''
  287.  
  288.     args = process_command_line()
  289.  
  290.     opts = {}
  291.  
  292.     logging.basicConfig(level=logging.DEBUG if args['debug'] else logging.INFO)
  293.     logging.getLogger('requests.packages.urllib3').setLevel(logging.WARNING)
  294.     logging.getLogger('urllib3.connectionpool').setLevel(logging.WARNING)
  295.  
  296.     try:
  297.         opts['user'], opts['pass'] = args['digest'].split(':', 1)
  298.     except (AttributeError, ValueError):
  299.         pass
  300.  
  301.     backend = api.Send(host=args['host'], port=args['port'])
  302.  
  303.     setup(backend, opts)
  304.  
  305.     if schedule_already_exists(backend, args, opts):
  306.         sys.exit('\nExit, rule for: {} already exists.'.format(args['title']))
  307.  
  308.     template = get_template(backend, args, opts)
  309.     if not template:
  310.         sys.exit('\nExit, no template found for: {}.'.format(args['template']))
  311.  
  312.     guide_data = get_program_data(backend, args, opts)
  313.     if not guide_data:
  314.         sys.exit('\nExit, no match in guide for: {}'.format(args['title']))
  315.  
  316.     if update_template(template, guide_data, args):
  317.         add_rule(backend, template, args, opts)
  318.     else:
  319.         sys.exit('\nExit, error while copying guide data to template.')
  320.  
  321.  
  322. if __name__ == '__main__':
  323.     main()
  324.  
  325. # vim: set expandtab tabstop=4 shiftwidth=4 smartindent noai colorcolumn=80:
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement