#!/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.

from builtins import str
from builtins import range
from past.builtins import basestring
import json
import logging
import re
import sys

from django.http import Http404

from desktop.lib.django_util import JsonResponse
from desktop.lib.exceptions import StructuredException
from desktop.lib.i18n import force_unicode
from desktop.models import Document

from oozie.forms import WorkflowForm, NodeForm, design_form_by_type
from oozie.models import Workflow, Node, Start, End, Kill,\
                         Link, Decision, Fork, DecisionEnd, Join,\
                         NODE_TYPES, ACTION_TYPES, _STD_PROPERTIES
from oozie.decorators import check_job_access_permission, check_job_edition_permission
from oozie.utils import model_to_dict, format_dict_field_values, format_field_value

if sys.version_info[0] > 2:
  from django.utils.translation import gettext as _
else:
  from django.utils.translation import ugettext as _

LOG = logging.getLogger()


try:
  from jobbrowser.views import job_single_logs
  from jobbrowser.models import LinkJobLogs
except:
  LOG.warning('Oozie is not enabled')


def error_handler(view_fn):
  def decorator(request, *args, **kwargs):
    try:
      return view_fn(request, *args, **kwargs)
    except Http404 as e:
      raise e
    except StructuredException as e:
      error_code = e.error_code
      message = e.message
      details = e.data or {}
    except Exception as e:
      LOG.exception('error in %s' % view_fn)

      error_code = 500
      details = {}
      (type, value, tb) = sys.exc_info()
      if not hasattr(e, 'message') or not e.message:
        message = str(e)
      else:
        message = force_unicode(e.message, strings_only=True, errors='replace')

    response = {
      'status': 1,
      'message': message,
      'details': details
    }

    return JsonResponse(response, status=error_code)
  return decorator


def get_or_create_node(workflow, node_data, save=True):
  node = None
  id = str(node_data['id'])
  separator_index = id.find(':')

  if separator_index == -1:
    return Node.objects.get(id=id, workflow=workflow).get_full_node()

  node_type = id[0:separator_index]
  node_model = NODE_TYPES.get(node_type, None)
  kwargs = {'workflow': workflow, 'node_type': node_data['node_type']}

  if node_model:
    node = node_model(**kwargs)
  else:
    raise StructuredException(code="INVALID_REQUEST_ERROR", message=_('Could not find node of type'), data=node_data, error_code=500)

  if save:
    node.save()

  return node


def _validate_node_links_json(node_type, node_links, errors):
  """
  Validate a single node's links.
  node_type is the node type of the action information passed.
  node_links is list of dictionaries describing the links.
  errors is a dictionary that will be populated with any found errors.
  """
  assert isinstance(errors, dict), "errors must be a dict."
  if not isinstance(node_links, list):
    errors['links'] = _("links must be a list")
    return False

  # Check link counts are accurate.
  if node_type == Start.node_type:
    if len(node_links) != 2:
      errors['links'] = _("Start should have two children: 'related' to end, 'to' to any node but an end.")
      return False
  elif node_type == End.node_type:
    if len(node_links) != 0:
      errors['links'] = _("End should have no children.")
      return False
  elif node_type == Kill.node_type:
    if len(node_links) != 0:
      errors['links'] = _("Kill should have no children.")
      return False
  elif node_type in (Join.node_type, DecisionEnd.node_type):
    if len(node_links) != 1:
      errors['links'] = _("Join and Decision End should have one child: 'to' to any node.")
      return False
  elif node_type in (Fork.node_type, Decision.node_type):
    if len(node_links) < 2:
      errors['links'] = _("Fork and Decision should have at least two children: 'related' to their respective ends, 'start' to any node.")
      return False
  else:
    if len(node_links) != 2:
      errors['links'] = _("Actions should have two children: 'error' to kill, 'ok' to any node.")
      return False

  # Check if link types are okay.
  link_names_by_node_type = {
    'start': {'related': 1, 'to': 1},
    'end': {},
    'kill': {},
    'fork': {'related': 1, 'start': 2},
    'join': {'to': 1},
    'decision': {'related': 1, 'start': 2},
    'decisionend': {'to': 1},
    None: {'ok': 1, 'error': 1},
  }
  link_types = link_names_by_node_type.get(node_type, link_names_by_node_type[None])
  for link in node_links:
    link_name = link.get('name', None)
    if link_name in link_types:
      link_types[link_name] -= 1

  for link_type in link_types:
    if link_types[link_type] > 0:
      errors['links'] = _('%(node_type)s should have %(count)d more %(link_type)s link' % {
        'node_type': node_type,
        'count': link_types[link_type],
        'link_type': link_type
      })
      return False

  return True


def _validate_node_json(node_type, node_dict, errors, user, workflow):
  """
  Validates a single node excluding links.
  node_type is the node type of the action information passed.
  node_dict is a dictionary describing the node.
  errors is a dictionary that will be populated with any found errors.
  user is a User object that is associated with the node_type. Only needed for Subworkflow node.
  workflow is the Workflow object associated with the node. Only needed for Subworkflow node.
  Returns Boolean.
  """
  assert isinstance(errors, dict), "errors must be a dict."
  if node_type in ACTION_TYPES:
    form_class = design_form_by_type(node_type, user, workflow)
  else:
    form_class = NodeForm
  form = form_class(data=node_dict)

  if form.is_valid():
    for field in form.fields:
      errors[field] = []

    return True

  else:
    for field in form.fields:
      errors[field] = form[field].errors

    return False


def _validate_nodes_json(json_nodes, errors, user, workflow):
  """
  Validates every node and link in the workflow.
  node_type is the node type of the action information passed.
  node_dict is a dictionary describing the node.
  errors is a dictionary that will be populated with any found errors.
  user is a User object that is associated with the node_type. Only needed for Subworkflow node.
  workflow is the Workflow object associated with the node. Only needed for Subworkflow node.
  Returns Boolean.
  """
  assert isinstance(errors, dict), "errors must be a dict."
  result = True

  for node in json_nodes:
    _errors = {}
    node_dict = format_dict_field_values(node)
    if node['node_type'] in ACTION_TYPES:
      node_result = _validate_node_json(node['node_type'], node_dict, _errors, user, workflow)
    else:
      node_result = True
    link_result = _validate_node_links_json(node['node_type'], node_dict['child_links'], _errors)
    result = result and node_result and link_result
    if 'name' not in node and ( 'node_type' not in node or 'id' not in node ):
      raise StructuredException(code="INVALID_REQUEST_ERROR", message=_('Error saving workflow'), data={'errors': 'Node is missing a name.'}, error_code=400)
    errors[node.get('name', '%s-%s' % ( node.get('node_type'), node.get('id')))] = _errors

  return result


def _update_workflow_nodes_json(workflow, json_nodes, id_map, user):
  """Ideally would get objects from form validation instead."""
  nodes = []

  for json_node in json_nodes:
    node = get_or_create_node(workflow, json_node, save=False)

    if node.node_type == 'subworkflow':
      try:
        node.sub_workflow = Workflow.objects.get(id=int(json_node['sub_workflow']))
      except TypeError:
        # sub_workflow is None
        node.sub_workflow = None
      except Workflow.DoesNotExist:
        raise StructuredException(code="INVALID_REQUEST_ERROR", message=_('Error saving workflow'), data={'errors': 'Chosen subworkflow does not exist.'}, error_code=400)
    elif node.node_type == 'fork' and json_node['node_type'] == 'decision':
      node.save() # Need to save in case database throws error when performing delete.
      node = node.convert_to_decision()
    node.save()

    id_map[str(json_node['id'])] = node.id

    for key in json_node:
      if key == 'data':
        if isinstance(json_node[key], basestring):
          node.data = json_node[key]
        else:
          node.data = json.dumps(json_node[key])
      elif key not in ('node_ptr', 'child_nodes', 'workflow', 'id', 'sub_workflow'):
        setattr(node, key, format_field_value(key, json_node[key]))

    node.workflow = workflow
    node.save()

    # Keep track of nodes in order of received list
    # so that we may iterate over them again in the same order
    # when we handle links
    nodes.append(node)

  # Delete unused nodes from workflow
  old_nodes = Node.objects.filter(workflow=workflow).exclude(id__in=[x.id for x in nodes])
  for node in old_nodes:
    node.get_full_node().delete()

  return nodes


def _update_workflow_json(json_workflow):
  workflow = Workflow.objects.get(id=json_workflow['id'])
  for key in json_workflow:
    if key == 'data':
      if isinstance(json_workflow[key], basestring):
        workflow.data = json_workflow[key]
      else:
        workflow.data = json.dumps(json_workflow[key])
    elif key not in ('nodes', 'start', 'end', 'job_ptr', 'owner'):
      setattr(workflow, key, json_workflow[key])

  workflow.save()
  workflow.doc.update(name=workflow.name, description=workflow.description)
  return workflow


def _workflow(request, workflow):
  response = {'status': -1, 'data': 'None'}

  workflow_dict = model_to_dict(workflow)

  node_list = [node.get_full_node() for node in workflow.node_list]
  nodes = [model_to_dict(node) for node in node_list]

  for index in range(0, len(node_list)):
    nodes[index]['child_links'] = [model_to_dict(link) for link in node_list[index].get_all_children_links()]

  workflow_dict['nodes'] = nodes

  response['status'] = 0
  response['data'] = workflow_dict
  return JsonResponse(response)


@error_handler
@check_job_access_permission(exception_class=(lambda x: StructuredException(code="UNAUTHORIZED_REQUEST_ERROR", message=x, data=None, error_code=401)))
@check_job_edition_permission(exception_class=(lambda x: StructuredException(code="UNAUTHORIZED_REQUEST_ERROR", message=x, data=None, error_code=401)))
def workflow_validate_node(request, workflow, node_type):
  response = {'status': -1, 'data': {}}

  node_dict = format_dict_field_values(json.loads(request.POST.get('node')))

  if _validate_node_json(node_type, node_dict, response['data'], request.user, workflow):
    response['status'] = 0
  else:
    response['status'] = -1

  return JsonResponse(response)


# Workflow and child links are SPECIAL.
@error_handler
@check_job_access_permission(exception_class=(lambda x: StructuredException(code="UNAUTHORIZED_REQUEST_ERROR", message=x, data=None, error_code=401)))
@check_job_edition_permission(exception_class=(lambda x: StructuredException(code="UNAUTHORIZED_REQUEST_ERROR", message=x, data=None, error_code=401)))
def workflow_save(request, workflow):
  if request.method != 'POST':
    raise StructuredException(code="METHOD_NOT_ALLOWED_ERROR", message=_('Must be POST request.'), error_code=405)

  json_workflow = format_dict_field_values(json.loads(request.POST.get('workflow')))
  json_workflow.setdefault('schema_version', workflow.schema_version)

  form = WorkflowForm(data=json_workflow)

  if not form.is_valid():
    raise StructuredException(code="INVALID_REQUEST_ERROR", message=_('Error saving workflow'), data={'errors': form.errors}, error_code=400)

  json_nodes = json_workflow['nodes']
  id_map = {}
  errors = {}

  if not _validate_nodes_json(json_nodes, errors, request.user, workflow):
    raise StructuredException(code="INVALID_REQUEST_ERROR", message=_('Error saving workflow'), data={'errors': errors}, error_code=400)

  workflow = _update_workflow_json(json_workflow)
  nodes = _update_workflow_nodes_json(workflow, json_nodes, id_map, request.user)

  # Update links
  index = 0
  for json_node in json_nodes:
    child_links = json_node['child_links']
    Link.objects.filter(parent=nodes[index]).delete()

    for child_link in child_links:
      link = Link()
      link.id = getattr(child_link, 'id', None)
      link.name = child_link['name']

      id = str(child_link['parent'])
      link.parent = Node.objects.get(id=id_map[id])

      id = str(child_link['child'])
      link.child = Node.objects.get(id=id_map[id])

      link.comment = child_link.get('comment', '')

      link.save()

    index += 1

  # Make sure workflow HDFS permissions are correct
  Workflow.objects.check_workspace(workflow, request.fs)

  return _workflow(request, workflow=workflow)


@error_handler
@check_job_access_permission(exception_class=(lambda x: StructuredException(code="UNAUTHORIZED_REQUEST_ERROR", message=x, data=None, error_code=401)))
def workflow(request, workflow):
  if request.method != 'GET':
    raise StructuredException(code="METHOD_NOT_ALLOWED_ERROR", message=_('Must be GET request.'), error_code=405)

  return _workflow(request, workflow)


@error_handler
@check_job_access_permission(exception_class=(lambda x: StructuredException(code="UNAUTHORIZED_REQUEST_ERROR", message=x, data=None, error_code=401)))
def workflow_actions(request, workflow):
  if request.method != 'GET':
    raise StructuredException(code="METHOD_NOT_ALLOWED_ERROR", message=_('Must be GET request.'), error_code=405)

  action_list = [action.get_full_node() for action in workflow.actions]
  response = {
    'status': 0,
    'data': {
      'actions': [model_to_dict(action) for action in action_list]
    }
  }
  return JsonResponse(response)


@error_handler
def workflows(request):
  if request.method not in ['GET']:
    raise StructuredException(code="METHOD_NOT_ALLOWED_ERROR", message=_('Must be GET request.'), error_code=405)

  if request.GET.get('managed', 'false').lower() == 'false':
    extra='jobsub'
  else:
    extra=''

  workflow_docs = Document.objects.get_docs(request.user, Workflow, extra=extra)

  response = {
    'status': 0,
    'data': {
      'workflows': [model_to_dict(workflow.content_object) for workflow in workflow_docs]
    }
  }

  return JsonResponse(response)


def autocomplete_properties(request):
  return JsonResponse({ 'properties': _STD_PROPERTIES })


@error_handler
def get_log(request, oozie_workflow, make_links=True, log_start_pattern=None, log_end_pattern=None):
  logs = {}
  is_really_done = False

  for action in oozie_workflow.get_working_actions():
    try:
      if action.externalId:
        data = job_single_logs(request, **{'job': action.externalId})

        if data and 'logs' in data:
          action_logs = data['logs'][1]

          if log_start_pattern:
            re_log_start = re.compile(log_start_pattern, re.M | re.DOTALL)
            if re_log_start.search(action_logs):
              action_logs = re.search(re_log_start, action_logs).group(1).strip()
            else:
              LOG.debug('Failed to find given start log pattern in logs: %s' % log_start_pattern)

          if log_end_pattern:
            re_log_end = re.compile(log_end_pattern)
            is_really_done = re_log_end.search(action_logs) is not None or oozie_workflow.status == 'KILLED'
            if is_really_done and not action_logs:
              LOG.warning('Unable to scrape full logs, try increasing the jobbrowser log_offset configuration value.')

          if make_links:
            action_logs = LinkJobLogs._make_links(action_logs)

          logs[action.name] = action_logs

    except Exception:
      LOG.exception('An error occurred while watching the job running')
      is_really_done = True

  workflow_actions = _get_workflow_actions(oozie_workflow, logs, is_really_done)

  return logs, workflow_actions, is_really_done


def _get_workflow_actions(oozie_workflow, logs, is_really_done=False):
  workflow_actions = []

  # Return metadata for workflow actions (required for Pig)
  for action in oozie_workflow.get_working_actions():
    progress = _get_progress(oozie_workflow, logs.get(action.name, ''))
    appendable = {
      'name': action.name,
      'status': action.status,
      'logs': logs.get(action.name, ''),
      'isReallyDone': is_really_done,
      'progress': progress,
      'progressPercent': '%d%%' % progress,
      'absoluteUrl': oozie_workflow.get_absolute_url(),
    }
    workflow_actions.append(appendable)

  return workflow_actions


def _get_progress(job, log):
  if job.status in ('SUCCEEDED', 'KILLED', 'FAILED'):
    return 100
  else:
    try:
      return int(re.findall("MapReduceLauncher  - (1?\d?\d)% complete", log)[-1])
    except:
      return 0
