text: # 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.import logging
  
  from builtins import object
  import logging
  import os
  import sys
  
  from collections import deque
  
  from django.urls import reverse
  from mako.lookup import TemplateLookup
  
  from desktop.models import Document2
  from notebook.connectors.base import get_api
  from notebook.models import Notebook, make_notebook
  from useradmin.models import User
  
  from indexer.conf import CONFIG_INDEXING_TEMPLATES_PATH, CONFIG_INDEXER_LIBS_PATH
  from indexer.fields import get_field_type
  from indexer.file_format import get_file_format_instance, get_file_format_class
  from indexer.indexers.morphline_operations import get_checked_args
  from indexer.solr_client import SolrClient
  
  if sys.version_info[0] > 2:
    from django.utils.translation import gettext as _
  else:
    from django.utils.translation import ugettext as _
  
  
  LOG = logging.getLogger(__name__)
  
  
  class MorphlineIndexer(object):
  
    def __init__(self, user, fs=None, jt=None, solr_client=None):
      self.fs = fs
      self.jt = jt
      self.user = user
      self.username = user.username
      self.solr_client = solr_client if solr_client is not None else SolrClient(self.user)
  
    def _upload_workspace(self, morphline):
      from oozie.models2 import Job
  
      hdfs_workspace_path = Job.get_workspace(self.username)
      hdfs_morphline_path = os.path.join(hdfs_workspace_path, "morphline.conf")
      hdfs_log4j_properties_path = os.path.join(hdfs_workspace_path, "log4j.properties")
  
      log4j_template_path = os.path.join(CONFIG_INDEXING_TEMPLATES_PATH.get(), "log4j.properties")
  
      # Create workspace on hdfs
      self.fs.do_as_user(self.username, self.fs.mkdir, hdfs_workspace_path)
  
      self.fs.do_as_user(self.username, self.fs.create, hdfs_morphline_path, data=morphline)
      self.fs.do_as_user(self.username, self.fs.create, hdfs_log4j_properties_path, data=open(log4j_template_path).read())
  
      return hdfs_workspace_path
  
    def run_morphline(self, request, collection_name, morphline, input_path, query=None, start_time=None, lib_path=None):
      workspace_path = self._upload_workspace(morphline)
  
      task = make_notebook(
        name=_('Indexing into %s') % collection_name,
        editor_type='notebook',
        on_success_url=reverse('search:browse', kwargs={'name': collection_name}),
        pub_sub_url='assist.collections.refresh',
        is_task=True,
        is_notebook=True,
        last_executed=start_time
      )
  
      if query:
        q = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=query))
        notebook_data = q.get_data()
        snippet = notebook_data['snippets'][0]
  
        api = get_api(request, snippet)
  
        destination = '__hue_%s' % notebook_data['uuid'][:4]
        location = '/user/%s/__hue-%s' % (request.user, notebook_data['uuid'][:4])
        sql, _success_url = api.export_data_as_table(notebook_data, snippet, destination, is_temporary=True, location=location)
        input_path = '${nameNode}%s' % location
  
        task.add_hive_snippet(snippet['database'], sql)
  
      client = SolrClient(self.user)
  
      extra_args = ['-Dmapreduce.job.user.classpath.first=true'] if client.is_solr_six_or_more() else []
  
      task.add_java_snippet(
        clazz='org.apache.solr.hadoop.MapReduceIndexerTool',
        app_jar=lib_path if lib_path is not None else CONFIG_INDEXER_LIBS_PATH.get(),
        arguments=extra_args + [
            u'--morphline-file',
            u'morphline.conf',
            u'--output-dir',
            u'${nameNode}/user/%s/indexer' % self.username,
            u'--log4j',
            u'log4j.properties',
            u'--go-live',
            u'--zk-host',
            client.get_zookeeper_host(),
            u'--collection',
            collection_name,
            input_path,
        ],
        files=[
            {u'path': u'%s/log4j.properties' % workspace_path, u'type': u'file'},
            {u'path': u'%s/morphline.conf' % workspace_path, u'type': u'file'}
        ]
      )
  
      return task.execute(request, batch=True)
  
    def guess_format(self, data):
      file_format = get_file_format_instance(data['file'])
      return file_format.get_format()
  
    def guess_field_types(self, data):
      file_format = get_file_format_instance(data['file'], data['format'])
      return file_format.get_fields() if file_format else {'columns': []}
  
    # Breadth first ordering of fields
    @classmethod
    def get_field_list(self, field_data, is_converting_types=False):
      fields = []
  
      queue = deque(field_data)
  
      while len(queue):
        curr_field = queue.popleft()
        curr_field['type'] = curr_field['type']
        if is_converting_types:
          SolrClient._port_field_types(curr_field)
        fields.append(curr_field)
  
        for operation in curr_field["operations"]:
          for field in operation["fields"]:
            queue.append(field)
  
      return fields
  
    @classmethod
    def get_kept_field_list(cls, field_data):
      return [field for field in cls.get_field_list(field_data) if field['keep']]
  
    def get_unique_field(self, format_):
      unique_fields = [column['name'] for column in format_['columns'] if column['unique']]
  
      if unique_fields:
        return unique_fields[0]
      else:
        base_name = "_uuid"
        field_names = set([column['name'] for column in format_['columns']])
  
        while base_name in field_names:
          base_name = '_' + base_name
  
        return base_name
  
    def is_unique_generated(self, format_):
      unique_fields = [column['name'] for column in format_['columns'] if column['unique']]
  
      return len(unique_fields) == 0
  
    @staticmethod
    def _get_regex_for_type(type_name):
      field_type = get_field_type(type_name)
  
      return field_type.regex.replace('\\', '\\\\')
  
    def generate_morphline_config(self, collection_name, data, uuid_name=None, lib_path=None):
      if lib_path is None:
        lib_path = CONFIG_INDEXER_LIBS_PATH.get()
  
      geolite_loc = os.path.join(lib_path, "GeoLite2-City.mmdb")
      grok_dicts_loc = os.path.join(lib_path, "grok_dictionaries")
  
      properties = {
        "collection_name": collection_name,
        "fields": self.get_field_list(data['columns'], is_converting_types=True),
        "num_base_fields": len(data['columns']),
        "uuid_name": uuid_name,
        "get_regex": MorphlineIndexer._get_regex_for_type,
        "format_settings": data['format'],
        "format_class": get_file_format_class(data['format']['type']),
        "get_kept_args": get_checked_args,
        "grok_dictionaries_location": grok_dicts_loc if self.fs and self.fs.exists(grok_dicts_loc) else None,
        "geolite_db_location": geolite_loc if self.fs and self.fs.exists(geolite_loc) else None,
        "zk_host": self.solr_client.get_zookeeper_host() ## offline test?
      }
  
      oozie_workspace = CONFIG_INDEXING_TEMPLATES_PATH.get()
  
      lookup = TemplateLookup(directories=[oozie_workspace])
      morphline = lookup.get_template("morphline_template.conf").render(**properties)
  
      return morphline
  
  
  def _create_solr_collection(user, fs, client, destination, index_name, kwargs):
    unique_key_field = destination['indexerPrimaryKey'] and destination['indexerPrimaryKey'][0] or None
    df = destination['indexerDefaultField'] and destination['indexerDefaultField'][0] or None
  
    indexer = MorphlineIndexer(user, fs)
    fields = indexer.get_field_list(destination['columns'])
    skip_fields = [field['name'] for field in fields if not field['keep']]
  
    kwargs['fieldnames'] = ','.join([field['name'] for field in fields])
    for field in fields:
      for operation in field['operations']:
        if operation['type'] == 'split':
          field['multiValued'] = True # Solr requires multiValued to be set when splitting
          kwargs['f.%(name)s.split' % field] = 'true'
          kwargs['f.%(name)s.separator' % field] = operation['settings']['splitChar'] or ','
  
    if skip_fields:
      kwargs['skip'] = ','.join(skip_fields)
      fields = [field for field in fields if field['name'] not in skip_fields]
  
    if not unique_key_field:
      unique_key_field = 'hue_id'
      fields += [{"name": unique_key_field, "type": "string"}]
      kwargs['rowid'] = unique_key_field
  
    if not destination['hasHeader']:
      kwargs['header'] = 'false'
    else:
      kwargs['skipLines'] = 1
  
    if not client.exists(index_name):
      client.create_index(
          name=index_name,
          config_name=destination.get('indexerConfigSet'),
          fields=fields,
          unique_key_field=unique_key_field,
          df=df,
          shards=destination['indexerNumShards'],
          replication=destination['indexerReplicationFactor']
      )
  