mnctl

#!/usr/bin/env python3
# vim: expandtab shiftwidth=2 softtabstop=2

# https://docs.python.org/3/library/argparse.html
# https://docs.python.org/3/library/urllib.request.html
# https://docs.python.org/3/library/xml.ElementTree.elementtree.html

import argparse
import sys
import time
import urllib.request
from xml.etree import ElementTree
from typing import Any, Callable, Dict, Optional


_COMMANDS: Dict[str, Callable[[argparse.Namespace], None]] = {}


def addCommand(func: Any) -> Any:
  _COMMANDS[func.__name__] = func
  return func


def parseArgs() -> argparse.Namespace:
  parser = argparse.ArgumentParser(allow_abbrev=False)
  parser.add_argument(
      'action',
      choices=list(_COMMANDS.keys()))
  parser.add_argument(
      '--url',
      default='http://localhost:28080',
      help='Prefix path to Morrigan HTTP endpoint.')
  parser.add_argument(
      '--authfile',
      required=True,
      help='Path to file containing username:password.')
  parser.add_argument(
      '--library',
      help='Name of the library, e.g. "music".')
  parser.add_argument(
      '--remote',
      help='Name of the remote library, e.g. "music".')
  parser.add_argument(
      '--includeautotags',
      action='store_true',
      help='Will include auto tags in sha1tags output.')

  args = parser.parse_args()

  if not args.url.endswith('/'):
    args.url += '/'

  return args


def fail(msg):
  print(msg)
  sys.exit(1)


def getLibPath(lib: str) -> str:
  if not lib:
    fail('--library not specified.')
  # TODO check lib exists in /mlists ?
  # TODO support other types?
  return 'mlists/LOCALMMDB/%s.local.db3' % lib


@addCommand
def dumpxml(args: argparse.Namespace) -> None:
  path = getLibPath(args.library)
  path += '/items?includeddeletedtags=true'
  resp = callMn(args, path)
  print(resp.read().decode('utf-8'))


@addCommand
def sha1tags(args: argparse.Namespace) -> None:
  path = getLibPath(args.library)
  path += '/sha1tags'
  if args.includeautotags:
    path += '?includeautotags=true'
  resp = callMn(args, path)
  print(resp.read().decode('utf-8'))


@addCommand
def listremotes(args: argparse.Namespace) -> None:
  path = getLibPath(args.library)
  resp = callMn(args, path)
  xml = resp.read().decode('utf-8')
  tree = ElementTree.fromstring(xml)
  for r in tree.findall('remote'):
    print(r.text)


@addCommand
def pullremote(args: argparse.Namespace) -> None:
  remote = args.remote
  if not remote:
    fail('--remote not specified.')
  path = getLibPath(args.library)
  data = 'action=pull&remote=%s' % remote
  resp = callMn(args, path, data=bytes(data, 'utf-8'))
  body = resp.read().decode('utf-8')

  taskid = None
  for line in body.splitlines():
    if line.startswith('id='):
      taskid = line[3:]
      break
  if not taskid:
    fail('Starting pull did not return a task id:\n%s' % body)
  waitForTask(args, taskid)


def waitForTask(args: argparse.Namespace, taskid: str):
  print('Waiting for task: %s' % taskid)
  path = 'status/%s' % taskid
  while True:
    resp = callMn(args, path)
    xml = resp.read().decode('utf-8')
    tree = ElementTree.fromstring(xml)
    state = tree.find('./state')
    if state is None:
      fail('Status missing state:\n%s' % xml)
    elif state.text == 'COMPLETE':
      success = tree.find('./successful')
      if success is None:
        fail('Status missing successful:\n%s' % xml)
      elif success.text == 'true':
        print('Success.')
      else:
        print('Failed.')
        print(xml)
      break
    else:
      print('.', end='', flush=True)
      time.sleep(5)


def mkAuthOpener(args: argparse.Namespace):
  with open(args.authfile) as f:
    auth = f.readline().strip()
  (user, passwd) = auth.split(':')
  passwd_mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
  passwd_mgr.add_password(None, args.url, user, passwd)
  auth_handler = urllib.request.HTTPBasicAuthHandler(passwd_mgr)
  return urllib.request.build_opener(auth_handler)


def callMn(args: argparse.Namespace, path, data=None, headers=None):
  opener = mkAuthOpener(args)
  url = args.url
  url += path
  req = urllib.request.Request(url, data=data, headers=headers or {})
  return opener.open(req)


def main() -> None:
  args = parseArgs()
  func = _COMMANDS[args.action]
  if not func:
    fail('No command ' + args.action)
  func(args)


if __name__ == '__main__':
  main()