[mythtv] Services API: Does Dvr/AddRecordSchedule work?

[Bill Meek]

On 10/15/2018 02:55 PM, John P Poet wrote:
> Yes, I am making use of those tools that you have written.  They are a huge
> help -- although I had not noticed your `api` application, and will need to
> check it out.

api only exists on my box, but attached here. I'll look at the
tool Peter just mentioned too.

-- 
Bill
-------------- next part --------------
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
A Utilities API test tool.

Examples:

api.py --host yourBackendHostNameOrIP --endpoint Myth/GetHostName

Same thing only easier to type:
api.py --ho BE --en Myth/GetHostName

api.py --ho BE --en Dvr/GetRecordedList --rest Count=1

api.py --ho frontendHostName --port 6547 --end Frontend/SendNotification \
--wrmi --post 'Message=Test Error=true'

Bill Meek 2015
"""

from __future__ import print_function
from __future__ import absolute_import
import argparse
import json
import logging
import sys
from mythtv_services_api import send as api


# pylint: disable=anomalous-backslash-in-string
def get_program_arguments():
    """
    Process the command line.
    """

    parser = argparse.ArgumentParser(
        description='Services API Tester',
        epilog='* E.g. Id=1 TitleRegEx=\*News\*. Default values are in ()s.')

    mandatory = parser.add_argument_group('required arguments')
    m_exclusive = parser.add_mutually_exclusive_group()

    parser.add_argument(
        '--debug', action='store_true',
        help='output additional information (%(default)s)')

    parser.add_argument(
        '--digest', type=str, metavar='<user:pass>', default='admin:mythtv',
        help='digest username:password')

    mandatory.add_argument(
        '--host', type=str, nargs='+', metavar='<host>',
        help='server hostname(s) or IP address(s)')

    mandatory.add_argument(
        '--endpoint', type=str, required=True,
        metavar='<ep>', help='endpoint to test')

    parser.add_argument(
        '--noetag', action='store_true',
        help='tell the backend not to return a ETag (%(default)s)')

    parser.add_argument(
        '--nogzip', action='store_true',
        help='tell the backend not to return gzipped data (%(default)s)')

    parser.add_argument(
        '--oneline', action='store_true',
        help='print the response on one line, no formatting (%(default)s)')

    parser.add_argument(
        '--port', type=int, metavar='<port>', default=6544,
        help='back/frontend port (%(default)i)')

    m_exclusive.add_argument(
        '--postdata', type=str, nargs='+', metavar='<pd>',
        help='Tag=Value *')

    m_exclusive.add_argument(
        '--rest', type=str, nargs='+', metavar='<rest>',
        help='Tag=Value *')

    parser.add_argument(
        '--quiet', action='store_true',
        help='don\'t print any responses (%(default)s)')

    parser.add_argument(
        '--timeout', type=int, metavar='<seconds>', default=10,
        help='seconds to wait for a response (%(default)s)')

    parser.add_argument(
        '--usexml', action='store_true',
        help='ask for an XML response (%(default)s)')

    parser.add_argument(
        '--version', action='version', version='%(prog)s 0.9')

    parser.add_argument(
        '--wrmi', action='store_true',
        help='allows postdata to be sent to the server (%(default)s)')

    return vars(parser.parse_args())
# pylint: enable=anomalous-backslash-in-string


def postdata_to_dict(postdata_items):
    """Convert Key=Value sets to Python dicts."""

    postdata = dict()

    for kv_pair in postdata_items:

        try:
            key, value = kv_pair.split('=', 1)
        except ValueError:
            sys.exit('usage: postdata format is: key=value [key=value...]')

        postdata[key] = value

    return postdata


def rest_to_string(rest_items):
    """
    Convert Key=Value sets to an '&' separated string. Makes
    sure there is a single = between the key/value.
    """

    rest = ''

    for kv_pair in rest_items:

        try:
            key, value = kv_pair.split('=', 1)
            rest += '{}={}'.format(key, value)
        except ValueError:
            sys.exit('usage: rest format is: key=value [key=value...]')

        rest += '&'

    return rest[:-1]

# pylint: disable=too-many-arguments
def process_host(host='', port=6544, endpoint='', opts=None, postdata=None,
                 rest='', oneline=False, quiet=False):
    """Send the query for a single host."""

    backend = api.Send(host=host, port=port)

    try:
        if postdata:
            resp_dict = backend.send(endpoint=endpoint, postdata=postdata,
                                     opts=opts)
        elif rest:
            resp_dict = backend.send(endpoint=endpoint, rest=rest, opts=opts)
        else:
            resp_dict = backend.send(endpoint=endpoint, opts=opts)

        if oneline:
            print('{}'.format(resp_dict))
        elif quiet:
            pass
        else:
            print(json.dumps(resp_dict, sort_keys=True, indent=4,
                             separators=(',', ': ')))
    except RuntimeWarning as warning:
        print('\nWarning: "{}"'.format(warning))
    except RuntimeError as error:
        sys.exit('\nFatal error: "{}"'.format(error))

    backend.close_session()
# pylint: enable=too-many-arguments


def main():
    """Basic stuff, fix this comment"""
    args = get_program_arguments()

    opts = {'noetag': args['noetag'], 'nogzip': args['nogzip'],
            'usexml': args['usexml'], 'wrmi': args['wrmi'], 'wsdl': False}

    if args['timeout']:
        opts['timeout'] = args['timeout']

    logging.basicConfig(level=logging.DEBUG if args['debug'] else logging.INFO)
    logging.getLogger('requests.packages.urllib3').setLevel(logging.WARNING)
    logging.getLogger('urllib3.connectionpool').setLevel(logging.WARNING)

    try:
        opts['user'], opts['pass'] = args['digest'].split(':')
    except (AttributeError, ValueError):
        pass

    postdata = dict()
    if args['postdata']:
        postdata = postdata_to_dict(args['postdata'])

    rest = ''
    if args['rest']:
        rest = rest_to_string(args['rest'])

    for hostname in args['host']:
        process_host(host=hostname, port=args['port'],
                     endpoint=args['endpoint'], opts=opts,
                     postdata=postdata, rest=rest,
                     oneline=args['oneline'], quiet=args['quiet'])

if __name__ == '__main__':
    main()

# :!% --host mc0 ofc0 --endpoint=Dvr/version --oneline
# vim: set expandtab tabstop=4 shiftwidth=4 smartindent colorcolumn=80: