#!/usr/bin/env python
# Licensed to Cloudera, Inc. under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  Cloudera, Inc. licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
Oozie API classes.

This is mostly just codifying the datastructure of the Oozie REST API.
http://incubator.apache.org/oozie/docs/3.2.0-incubating/docs/WebServicesAPI.html
"""

import re

from cStringIO import StringIO
from time import mktime

from desktop.lib import i18n
from desktop.lib.exceptions_renderable import PopupException
from desktop.log.access import access_warn

import hadoop.confparse
from liboozie.utils import parse_timestamp, format_time

from django.utils.translation import ugettext as _
from django.core.urlresolvers import reverse


class Action(object):

  def __init__(self, json_dict):
    for attr in self._ATTRS:
      setattr(self, attr, json_dict.get(attr))
    self._fixup()

  def _fixup(self): pass

  def is_finished(self):
    return self.status in ('OK', 'SUCCEEDED', 'DONE')

  @classmethod
  def create(self, action_class, action_dict):
    if ControlFlowAction.is_control_flow(action_dict.get('type')):
      return ControlFlowAction(action_dict)
    else:
      return action_class(action_dict)

  def __str__(self):
    return '%s - %s' % (self.type, self.name)


class ControlFlowAction(Action):
  _ATTRS = [
    'errorMessage',
    'status',
    'stats',
    'data',
    'transition',
    'externalStatus',
    'cred',
    'conf',
    'type',
    'endTime',
    'externalId',
    'id',
    'startTime',
    'externalChildIDs',
    'name',
    'errorCode',
    'trackerUri',
    'retries',
    'toString',
    'consoleUrl'
  ]

  @classmethod
  def is_control_flow(self, action_type):
    return action_type is not None and (':' in action_type)

  def _fixup(self):
    """
    Fixup:
      - time fields as struct_time
      - config dict
      - protect externalId
    """
    super(ControlFlowAction, self)._fixup()

    if self.startTime:
      self.startTime = parse_timestamp(self.startTime)
    if self.endTime:
      self.endTime = parse_timestamp(self.endTime)
    if self.retries:
      self.retries = int(self.retries)
    if self.externalId and not re.match('job_.*', self.externalId):
      self.externalId = None

    self.conf_dict = {}


class WorkflowAction(Action):
  _ATTRS = [
    'conf',
    'consoleUrl',
    'data',
    'endTime',
    'errorCode',
    'errorMessage',
    'externalId',
    'externalStatus',
    'id',
    'name',
    'retries',
    'startTime',
    'status',
    'trackerUri',
    'transition',
    'type',
    'externalChildIDs',
  ]

  def _fixup(self):
    """
    Fixup:
      - time fields as struct_time
      - config dict
    """
    super(WorkflowAction, self)._fixup()

    if self.startTime:
      self.startTime = parse_timestamp(self.startTime)
    if self.endTime:
      self.endTime = parse_timestamp(self.endTime)
    if self.retries:
      self.retries = int(self.retries)

    if self.conf:
      xml = StringIO(i18n.smart_str(self.conf))
      self.conf_dict = hadoop.confparse.ConfParse(xml)
    else:
      self.conf_dict = {}

  def get_absolute_url(self):
    related_job_ids = []

    if hasattr(self, 'oozie_coordinator') and self.oozie_coordinator:
      related_job_ids.append('coordinator_job_id=%s' % self.oozie_coordinator.id)
    if hasattr(self, 'oozie_bundle') and self.oozie_bundle:
      related_job_ids.append('bundle_job_id=%s' % self.oozie_bundle.id)

    if related_job_ids:
      extra_params = '?' + '&'.join(related_job_ids)
    else:
      extra_params = ''

    return reverse('oozie:list_oozie_workflow_action', kwargs={'action': self.id}) + extra_params

  def get_absolute_log_url(self):
    url = None
    if self.externalId and re.match('job_.*', self.externalId):
      url = self.externalId and reverse('jobbrowser.views.job_single_logs', kwargs={'job': self.externalId}) or ''
    return url

  def get_external_id_url(self):
    url = None
    if self.externalId and self.externalId.endswith('W'):
      url = reverse('oozie:list_oozie_workflow', kwargs={'job_id': self.externalId}) or ''
    elif self.externalId and re.match('job_.*', self.externalId):
      url = reverse('jobbrowser.views.single_job', kwargs={'job': self.externalId}) or ''
    return url


class CoordinatorAction(Action):
  _ATTRS = [
    'status',
    'runConf',
    'errorMessage',
    'missingDependencies',
    'coordJobId',
    'errorCode',
    'actionNumber',
    'consoleUrl',
    'nominalTime',
    'externalStatus',
    'createdConf',
    'createdTime',
    'externalId',
    'lastModifiedTime',
    'type',
    'id',
    'trackerUri'
  ]

  def _fixup(self):
    """
    Fixup:
      - time fields as struct_time
      - config dict
    """
    super(CoordinatorAction, self)._fixup()

    if self.createdTime:
      self.createdTime = parse_timestamp(self.createdTime)
    if self.nominalTime:
      self.nominalTime = parse_timestamp(self.nominalTime)
    if self.lastModifiedTime:
      self.lastModifiedTime = parse_timestamp(self.lastModifiedTime)

    if self.runConf:
      xml = StringIO(i18n.smart_str(self.runConf))
      self.conf_dict = hadoop.confparse.ConfParse(xml)
    else:
      self.conf_dict = {}

    self.title = ' %s-%s'% (self.actionNumber, format_time(self.nominalTime))


class BundleAction(Action):
  _ATTRS = [
      'startTime',
      'actions',
      'frequency',
      'concurrency',
      'pauseTime',
      'group',
      'toString',
      'consoleUrl',
      'mat_throttling',
      'status',
      'conf',
      'user',
      'timeOut',
      'coordJobPath',
      'timeUnit',
      'coordJobId',
      'coordJobName',
      'nextMaterializedTime',
      'coordExternalId',
      'acl',
      'lastAction',
      'executionPolicy',
      'timeZone',
      'endTime'
  ]

  def _fixup(self):
    """
    Fixup:
      - time fields as struct_time
      - config dict
    """
    super(BundleAction, self)._fixup()

    self.type = 'coord-action'
    self.name = self.coordJobName

    if self.conf:
      xml = StringIO(i18n.smart_str(self.conf))
      self.conf_dict = hadoop.confparse.ConfParse(xml)
    else:
      self.conf_dict = {}

  def get_progress(self):
    """How much more time before the next action."""
    if self.lastAction is None:
      return 0
      
    next = mktime(parse_timestamp(self.lastAction))
    start = mktime(parse_timestamp(self.startTime))
    end = mktime(parse_timestamp(self.endTime))

    if end != start:
      progress = min(int((1 - (end - next) / (end - start)) * 100), 100)
    else:
      progress = 100

    return progress


class Job(object):
  MAX_LOG_SIZE = 3500 * 20 # 20 pages

  """
  Accessing log and definition will trigger Oozie API calls.
  """
  def __init__(self, api, json_dict):
    for attr in self._ATTRS:
      setattr(self, attr, json_dict.get(attr))
    self._fixup()

    self._api = api
    self._log = None
    self._definition = None

  def _fixup(self):
    """
    Fixup fields:
      - expand actions
      - time fields are struct_time
      - run is integer
      - configuration dict
      - log
      - definition
    """
    if self.startTime:
      self.startTime = parse_timestamp(self.startTime)
    if self.endTime:
      self.endTime = parse_timestamp(self.endTime)

    self.actions = [Action.create(self.ACTION, act_dict) for act_dict in self.actions]
    if self.conf is not None:
      xml = StringIO(i18n.smart_str(self.conf))
      self.conf_dict = hadoop.confparse.ConfParse(xml)
    else:
      self.conf_dict = {}

  def _get_log(self):
    """Get the log lazily, trigger Oozie API call at the first access."""
    if self._log is None:
      self._log = self._api.get_job_log(self.id)
    return self._log[-Job.MAX_LOG_SIZE:]
  log = property(_get_log)

  def _get_definition(self):
    """Get the definition lazily, trigger Oozie API call at the first access."""
    if self._definition is None:
      self._definition = self._api.get_job_definition(self.id)
    return self._definition
  definition = property(_get_definition)

  def start(self):
    self._api.job_control(self.id, 'start')

  def suspend(self):
    self._api.job_control(self.id, 'suspend')

  def resume(self):
    self._api.job_control(self.id, 'resume')

  def kill(self):
    self._api.job_control(self.id, 'kill')

  def available_actions(self):
    """
    available_actions() -> Zero or more of [ 'start', 'suspend', 'resume', 'kill' ]
    """
    if self.status in ('SUCCEEDED', 'KILLED', 'FAILED'):
      return []

    res = []
    if self.status == 'PREP':
      res.append('start')
    if self.status == 'RUNNING':
      res.append('suspend')
    if self.status == 'SUSPENDED':
      res.append('resume')
    res.append('kill')
    return res

  def check_request_permission(self, request):
    """Raise PopupException if request user doesn't have permission to modify workflow"""
    if not request.user.is_superuser and request.user.username != self.user:
      access_warn(request, _('Insufficient permission.'))
      raise PopupException(_("Permission denied. User %(username)s cannot modify user %(user)s's job.") %
                           dict(username=request.user.username, user=self.user))

  def get_control_flow_actions(self):
    return [action for action in self.actions if ControlFlowAction.is_control_flow(action.type)]

  def get_working_actions(self):
    return [action for action in self.actions if not ControlFlowAction.is_control_flow(action.type)]

  def is_running(self):
    return self.status in Workflow.RUNNING_STATUSES | Coordinator.RUNNING_STATUSES | Bundle.RUNNING_STATUSES

  def __str__(self):
    return '%s - %s' % (self.id, self.status)

  @property
  def has_sla(self):
    return '<sla:info>' in self.definition


class Workflow(Job):
  _ATTRS = [
    'actions',
    'appName',
    'appPath',
    'conf',
    'consoleUrl',
    'createdTime',
    'endTime',
    'externalId',
    'group',
    'id',
    'lastModTime',
    'run',
    'startTime',
    'status',
    'user',
    'acl',
    'parentId'
  ]
  ACTION = WorkflowAction
  RUNNING_STATUSES = set(['PREP', 'RUNNING', 'SUSPENDED'])
  FINISHED_STATUSES = set(['SUCCEEDED', 'KILLED', 'FAILED'])

  def _fixup(self):
    super(Workflow, self)._fixup()

    if self.createdTime:
      self.createdTime = parse_timestamp(self.createdTime)
    if self.lastModTime:
      self.lastModTime = parse_timestamp(self.lastModTime)
    if self.run:
      self.run = int(self.run)

  @property
  def type(self):
    return 'Workflow'

  def get_parent_job_id(self):
    if self.parentId and '@' in self.parentId:
      return self.parentId.split('@')[0]
    return self.parentId

  def get_absolute_url(self, format='html'):
    extra_params = []

    if format == 'json':
      extra_params.append('format=json')
    if hasattr(self, 'oozie_coordinator') and self.oozie_coordinator:
      extra_params.append('coordinator_job_id=%s' % self.oozie_coordinator.id)
    if hasattr(self, 'oozie_bundle') and self.oozie_bundle:
      extra_params.append('bundle_job_id=%s' % self.oozie_bundle.id)

    if extra_params:
      extra_params = '?' + '&'.join(extra_params)
    else:
      extra_params = ''

    return reverse('oozie:list_oozie_workflow', kwargs={'job_id': self.id}) + extra_params

  def get_progress(self, full_node_list=None):
    if self.status in ('SUCCEEDED', 'KILLED', 'FAILED'):
      return 100 # Case of decision nodes
    else:
      if full_node_list is not None:            # Should remove the un-reached branches if decision node
        total_actions = len(full_node_list) - 1 # -1 because of Kill node
      else:
        total_actions = len(self.actions)
      return int(sum([action.is_finished() for action in self.actions]) / float(max(total_actions, 1)) * 100)


class Coordinator(Job):
  _ATTRS = [
    'acl',
    'actions',
    'conf',
    'concurrency',
    'consoleUrl',
    'coordExternalId',
    'coordJobId',
    'coordJobName',
    'coordJobPath',
    'endTime',
    'executionPolicy',
    'frequency',
    'group',
    'lastAction',
    'mat_throttling',
    'nextMaterializedTime',
    'pauseTime',
    'startTime',
    'status',
    'timeOut',
    'timeUnit',
    'timeZone',
    'user',
    'bundleId',
    'total'
  ]
  ACTION = CoordinatorAction
  RUNNING_STATUSES = set(['PREP', 'RUNNING', 'RUNNINGWITHERROR', 'PREPSUSPENDED', 'SUSPENDED', 'SUSPENDEDWITHERROR', 'PREPPAUSED', 'PAUSED', 'PAUSEDWITHERROR'])
  FINISHED_STATUSES = set(['SUCCEEDED', 'DONEWITHERROR', 'KILLED', 'FAILED'])

  def _fixup(self):
    super(Coordinator, self)._fixup()

    if self.nextMaterializedTime is not None:
      self.nextMaterializedTime = parse_timestamp(self.nextMaterializedTime)
    else:
      self.nextMaterializedTime = self.startTime

    if self.pauseTime:
      self.pauseTime = parse_timestamp(self.pauseTime)


    # For when listing/mixing all the jobs together
    self.id = self.coordJobId
    self.appName = self.coordJobName

  @property
  def type(self):
    return 'Coordinator'

  def get_absolute_url(self, oozie_bundle=None, format='html'):
    extra_params = []

    if format == 'json':
      extra_params.append('format=json')
    if oozie_bundle:
      extra_params.append('bundle_job_id=%s' % oozie_bundle.id)
    if hasattr(self, 'bundleId') and self.bundleId:
      extra_params.append('bundle_job_id=%s' % self.bundleId)

    if extra_params:
      extra_params = '?' + '&'.join(extra_params)
    else:
      extra_params = ''

    return reverse('oozie:list_oozie_coordinator', kwargs={'job_id': self.id}) + extra_params

  def get_progress(self):
    """How much more time before the final materialization."""
    next = mktime(self.nextMaterializedTime)
    start = mktime(self.startTime)
    end = mktime(self.endTime)

    if end != start:
      progress = min(int((1 - (end - next) / (end - start)) * 100), 100)
    else:
      progress = 100

    # Manage case of a rerun
    action_count = float(len(self.actions))
    if action_count != 0 and progress == 100:
      progress = int(sum([action.is_finished() for action in self.actions]) / action_count * 100)

    return progress

  @classmethod
  def aggreate(cls, actions):
    if not actions:
      return []

    result = []
    first = prev = actions[0]

    for a in actions[1:]:
      if int(a) != int(prev) + 1:
        result.append('-'.join((first, prev)))
        first = a
      prev = a

    result.append('-'.join((first, prev)))

    return result

  @property
  def human_frequency(self):
    from oozie.models import Coordinator
    return Coordinator.CRON_MAPPING.get(self.frequency, self.frequency)


class Bundle(Job):
  _ATTRS = [
    'status',
    'toString',
    'group',
    'conf',
    'bundleJobName',
    'startTime',
    'bundleCoordJobs',
    'kickoffTime',
    'acl',
    'bundleJobPath',
    'createdTime',
    'timeOut',
    'consoleUrl',
    'bundleExternalId',
    'timeUnit',
    'pauseTime',
    'bundleJobId',
    'endTime',
    'user',
  ]

  ACTION = BundleAction
  RUNNING_STATUSES = set(['PREP', 'RUNNING', 'RUNNINGWITHERROR', 'SUSPENDED', 'PREPSUSPENDED', 'SUSPENDEDWITHERROR', 'PAUSED', 'PAUSEDWITHERROR', 'PREPPAUSED'])
  FINISHED_STATUSES = set(['SUCCEEDED', 'DONEWITHERROR', 'KILLED', 'FAILED'])

  def _fixup(self):
    self.actions = self.bundleCoordJobs

    super(Bundle, self)._fixup()

    # For when listing/mixing all the jobs together
    self.id = self.bundleJobId
    self.appName = self.bundleJobName

  @property
  def type(self):
    return 'Bundle'

  def get_absolute_url(self, format='html'):
    extra_params = ''
    if format == 'json':
      extra_params = '?format=json'

    return reverse('oozie:list_oozie_bundle', kwargs={'job_id': self.id}) + extra_params

  def get_progress(self):
    progresses = [action.get_progress() for action in self.actions]
    count = len(progresses)

    if count != 0:
      return sum(progresses) / float(count)
    else:
      return 0


class JobList(object):
  """
  Represents a list of Oozie jobs (Workflows or Coordinators or Bundles).
  """
  _ATTRS = [
    'offset',
    'len',
    'total',
    'jobs',
  ]

  def __init__(self, klass, jobs_key, api, json_dict, filters=None):
    """
    json_dict is the oozie json.
    filters is (optionally) the list of filters used to select this list
    """
    self._api = api
    self.offset = int(json_dict['offset'])
    self.total = int(json_dict['total'])
    self.jobs = [klass(self._api, wf_dict) for wf_dict in json_dict[jobs_key]]
    self.filters = filters


class WorkflowList(JobList):
  def __init__(self, api, json_dict, filters=None):
    super(WorkflowList, self).__init__(Workflow, 'workflows', api, json_dict, filters)


class CoordinatorList(JobList):
  def __init__(self, api, json_dict, filters=None):
    super(CoordinatorList, self).__init__(Coordinator, 'coordinatorjobs', api, json_dict, filters)


class BundleList(JobList):
  def __init__(self, api, json_dict, filters=None):
    super(BundleList, self).__init__(Bundle, 'bundlejobs', api, json_dict, filters)

