#!/usr/bin/env python
# encoding: utf-8

"""Azkaban CLI: a lightweight command line interface for Azkaban.

Usage:
  azkaban build [-cp PROJECT] [-a ALIAS | -u URL | [-r] ZIP] [-o OPTION ...]
  azkaban info [-p PROJECT] [-f | -o OPTION ... | [-i] JOB ...]
  azkaban log [-a ALIAS | -u URL] EXECUTION [JOB]
  azkaban run [-bkp PROJECT] [-a ALIAS | -u URL] [-e EMAIL ...]
              [-o OPTION ...] FLOW [JOB ...]
  azkaban schedule [-bkp PROJECT] [-a ALIAS | -u URL] [-e EMAIL ...]
                   [-o OPTION ...] [-s SPAN] (-d DATE) (-t TIME)
                   FLOW [JOB ...]
  azkaban upload [-cp PROJECT] [-a ALIAS | -u URL] ZIP
  azkaban -h | --help | -l | --log | -v | --version

Commmands:
  build*                        Build project and upload to Azkaban or save
                                locally the resulting archive.
  info*                         View information about jobs or files.
  log                           View workflow or job execution logs.
  run                           Run jobs or workflows. If no job is specified,
                                the entire workflow will be executed.
  schedule                      Schedule a workflow to be run at a specified
                                date and time.
  upload                        Upload archive to Azkaban server.

Arguments:
  EXECUTION                     Execution ID.
  JOB                           Job name.
  FLOW                          Workflow name. Recall that in the Azkaban world
                                this is simply a job without children.
  ZIP                           For `upload` command, the path to an existing
                                project zip archive. For `build`, the path
                                where the output archive will be built. If it
                                points to a directory, the archive will be
                                named after the project name (and version, if
                                present) and created in said directory.

Options:
  -a ALIAS --alias=ALIAS        Alias to saved URL and username. Will also try
                                to reuse session IDs for later connections.
  -b --bounce                   Skip execution if workflow is already running.
  -c --create                   Create the project if it does not exist.
  -d DATE --date=DATE           Date used for first run of a schedule. It must
                                be in the format `MM/DD/YYYY`.
  -e EMAIL --email=EMAIL        Email address to be notified when the workflow
                                finishes (can be specified multiple times).
  -f --files                    List project files instead of jobs. The first
                                column is the local path of the file, the
                                second the path of the file in the archive.
  -h --help                     Show this message and exit.
  -i --include-properties       Include project properties with job options.
  -k --kill                     Kill worfklow on first job failure.
  -l --log                      Show path to current log file and exit.
  -o OPTION --option=OPTION     Azkaban properties. Can either be the path to
                                a properties file or a single parameter
                                formatted as `key=value`, e.g. `-o
                                user.to.proxy=foo`. For the `build` and `run`
                                commands, these will be added to the project's
                                or run's properties respectively (potentially
                                overriding existing ones). For the `info`
                                command, this will cause only jobs with these
                                exact parameters to be displayed.
  -p PROJECT --project=PROJECT  Azkaban project. Can either be a project name
                                or a path to a python module/package defining
                                an `azkaban.Project` instance. Commands which
                                are followed by an asterisk will only work in
                                the latter case. If multiple projects are
                                registered, you can disambiguate as follows:
                                `--project=module:project_name`.
  -r --replace                  Overwrite any existing file.
  -s SPAN --span=SPAN           Period to repeat the scheduled flow. Must be
                                in format `1d`, a combination of magnitude and
                                unit of repetition. If not specified, the flow
                                will be run only once.
  -t TIME --time=TIME           Time when a schedule should be run. Must be of
                                the format `hh,mm,(AM|PM),(PDT|UTC|..)`.
  -u URL --url=URL              Azkaban endpoint (with protocol, and optionally
                                a username): '[user@]protocol:endpoint'. E.g.
                                'http://azkaban.server'. The username defaults
                                to the current user, as determined by `whoami`.
                                If you often use the same url, consider using
                                the `--alias` option instead.
  -v --version                  Show version and exit.

Azkaban CLI returns with exit code 1 if an error occurred and 0 otherwise.

"""

from azkaban import __version__, CLI_ARGS
from azkaban.project import Project
from azkaban.remote import Execution, Session
from azkaban.util import (AzkabanError, Config, catch, flatten, human_readable,
temppath, read_properties, suppress_urllib_warnings, write_properties)
from docopt import docopt
from traceback import format_exc
from requests.exceptions import HTTPError
import logging as lg
import os
import os.path as osp
import sys


_logger = lg.getLogger(__name__)


def _forward(args, names):
  """Forward subset of arguments from initial dictionary.

  :param args: Dictionary of parsed arguments (output of `docopt.docopt`).
  :param names: List of names that will be included.

  """
  names = set(names)
  return dict(
    ('_%s' % (k.lower().lstrip('-').replace('-', '_'), ), v)
    for (k, v) in args.items() if k in names
  )

def _parse_option(_option):
  """Parse `--option` argument.

  :param _option: `--option` argument.

  Returns a dictionary.

  """
  paths = (opt for opt in _option if not '=' in opt)
  opts = read_properties(*paths)
  try:
    opts.update(dict(s.split('=', 1) for s in _option if '=' in s))
  except ValueError:
    raise AzkabanError('Invalid `--option` flag.')
  return opts

def _parse_project(_project, require_project=False):
  """Parse `--project` argument into `(name, project)`.

  :param _project: `--project` argument.
  :param require_project: Fail if we fail to load the project.

  Note that `name` is guaranteed to be non-`None` (this function will throw an
  exception otherwise) but `project` can be.

  The rules are as follows:

  + If at least one `':'` is found in `_project` then the rightmost one is
    interpreted as delimitor between the path to the module and the project
    name.

  + Else:

    + We first try to interpret `_project` as a module path and find a unique
      project inside.

    + If the above attempt raises an `ImportError`, we interpret it as a name.

  """
  default_project = Config().get_option('azkaban', 'default.project', 'jobs')
  exceptions = {}
  projects = {}

  def try_load(path):
    try:
      projects.update(Project.load(path))
      return True
    except Exception: