<commit_before>package io.hydrosphere.mist.worker.runners.python

import java.io.File

import io.hydrosphere.mist.core.CommonData.RunJobRequest
import io.hydrosphere.mist.utils.Logger
import io.hydrosphere.mist.worker.NamedContext
import io.hydrosphere.mist.worker.runners.JobRunner
import io.hydrosphere.mist.worker.runners.python.wrappers._
import py4j.GatewayServer

import scala.concurrent.{ExecutionContext, Future}
import scala.sys.process._

class PythonEntryPoint(req: RunJobRequest, context: NamedContext) {

  val errorWrapper: ErrorWrapper = new ErrorWrapper
  val dataWrapper: DataWrapper = new DataWrapper
  val sparkContextWrapper: NamedContext = context
  val configurationWrapper: ConfigurationWrapper = new ConfigurationWrapper(req.params)
  val sparkStreamingWrapper: SparkStreamingWrapper = new SparkStreamingWrapper(context.setupConfiguration(req.id))

}

class PythonRunner(jobFile: File) extends JobRunner with Logger {

  override def run(
    req: RunJobRequest,
    context: NamedContext
  ): Either[Throwable, Map[String, Any]] = {

    try {
      val selfJarPath = new File(getClass.getProtectionDomain.getCodeSource.getLocation.toURI.getPath)
      var cmd = "python " + selfJarPath
      val entryPoint = new PythonEntryPoint(req, context)

      val gatewayServer: GatewayServer = new GatewayServer(entryPoint, 0)
      try {
        gatewayServer.start()
        val boundPort = gatewayServer.getListeningPort

        if (boundPort == -1) {
          logger.error("GatewayServer to Python exception")
          throw new Exception("GatewayServer to Python exception")
        } else {
          logger.info(s" Started PythonGatewayServer on port $boundPort")
          cmd += s" $boundPort"
        }
        logger.info(s"Running python task: $cmd")

        val exitCode = cmd.!
        if (exitCode != 0 || entryPoint.errorWrapper.get().nonEmpty) {
          val errmsg = entryPoint.errorWrapper.get()
          logger.error(errmsg)
          throw new Exception("Error in python code: " + errmsg)
        }
      } finally {
        // We must shutdown gatewayServer
        gatewayServer.shutdown()
      }

      Right(entryPoint.dataWrapper.get)
    } catch {
      case e: Throwable => Left(e)
    }
  }
}
<commit_msg>fix issue when python job uploaded remotely and local path in worker differs from endpoint path<commit_after>package io.hydrosphere.mist.worker.runners.python

import java.io.File

import io.hydrosphere.mist.core.CommonData.RunJobRequest
import io.hydrosphere.mist.utils.Logger
import io.hydrosphere.mist.worker.NamedContext
import io.hydrosphere.mist.worker.runners.JobRunner
import io.hydrosphere.mist.worker.runners.python.wrappers._
import py4j.GatewayServer

import scala.concurrent.{ExecutionContext, Future}
import scala.sys.process._

class PythonEntryPoint(req: RunJobRequest, context: NamedContext) {

  val errorWrapper: ErrorWrapper = new ErrorWrapper
  val dataWrapper: DataWrapper = new DataWrapper
  val sparkContextWrapper: NamedContext = context
  val configurationWrapper: ConfigurationWrapper = new ConfigurationWrapper(req.params)
  val sparkStreamingWrapper: SparkStreamingWrapper = new SparkStreamingWrapper(context.setupConfiguration(req.id))

}

class PythonRunner(jobFile: File) extends JobRunner with Logger {

  override def run(
    req: RunJobRequest,
    context: NamedContext
  ): Either[Throwable, Map[String, Any]] = {

    try {
      val selfJarPath = new File(getClass.getProtectionDomain.getCodeSource.getLocation.toURI.getPath)
      var cmd = "python " + selfJarPath
      val entryPoint = new PythonEntryPoint(req.copy(params = req.params.copy(filePath = jobFile.toString)), context)

      val gatewayServer: GatewayServer = new GatewayServer(entryPoint, 0)
      try {
        gatewayServer.start()
        val boundPort = gatewayServer.getListeningPort

        if (boundPort == -1) {
          logger.error("GatewayServer to Python exception")
          throw new Exception("GatewayServer to Python exception")
        } else {
          logger.info(s" Started PythonGatewayServer on port $boundPort")
          cmd += s" $boundPort"
        }
        logger.info(s"Running python task: $cmd")

        val exitCode = cmd.!
        if (exitCode != 0 || entryPoint.errorWrapper.get().nonEmpty) {
          val errmsg = entryPoint.errorWrapper.get()
          logger.error(errmsg)
          throw new Exception("Error in python code: " + errmsg)
        }
      } finally {
        // We must shutdown gatewayServer
        gatewayServer.shutdown()
      }

      Right(entryPoint.dataWrapper.get)
    } catch {
      case e: Throwable => Left(e)
    }
  }
}

{'fraction_non_alphanumeric': 0.08969957081545064, 'fraction_numerical': 0.0017167381974248926, 'mean_word_length': 3.397169811320755, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1570120', 'n_tokens_mistral': 1355, 'n_tokens_neox': 1295, 'n_words': 315}