<commit_before>package actors.persistenceManager

import java.io.File
import java.net.URL
import java.nio.file.{Files, Paths}

import akka.actor.{Actor, ActorRef}
import akka.pattern.ask
import akka.util.Timeout
import models.batch.OperationStatus
import models.batch.job._
import models.messages.batchProcessing._
import models.messages.logger.Log
import models.messages.persistenceManaging.DataSetEntry
import models.mining.mlret._
import models.mining.{Algorithm, MinerResult}

import scala.concurrent.duration._
import scala.concurrent.{Await, Future}
import scala.sys.process._


/**
 * Created by chetan on 13/04/15.
 */
class WorkerActor(mediator: ActorRef) extends Actor {

  private implicit val timeout = Timeout(10 seconds)

  /**
   * Checks if a particular file path exists
   * @param incoming path value as string
   * @return unit
   */
  private def checkPathFile(incoming: String) = {
    val path = Paths.get(incoming)
    if (!Files.exists(path)) Files.createFile(path)
  }

  /**
   * Checkes if a particular directory path exists
   * @param incoming path value as string
   * @return unit
   */
  private def checkPathDir(incoming: String) = {
    val path = Paths.get(incoming)
    if (!Files.exists(path)) Files.createDirectories(path)
  }

  /**
   * Handles output operations of the mining results
   * @param operation Details of user and dataset wrapped in GetDsData case object Details of user and dataset wrapped in GetDsData case object Details of user and dataset wrapped in GetDsData case object Details of user and dataset wrapped in GetDsData case object
   * @param dsm Actor reference to datastore manager
   * @return Option[JSON-Output]
   */
  private def handleDsOutput(operation: GetDsData, dsm: ActorRef) = {
    val (uname, name) = (operation.username, operation.Ds)
    val result = Await.result(dsm ? CheckUserDataset(uname, name), 10 seconds).asInstanceOf[Option[DataSetEntry]]
    val filepath = s"./datastore/datasets/$uname/$name"
    result match {
      case Some(x: DataSetEntry) => x.targetAlgorithm match {
        case "als" => new RALS(filepath, name).output
        case "clustering" => new RClustering(filepath, name).output
        case "fpm" => new RFPM(filepath, name).output
        case "svm" => new RSVM(filepath, name).output
        case _ => None
      }
      case None => None
    }
  }

  /**
   * handles all the jobs for the actor reciever
   * @param username username to work on
   * @param job job to be dealt with
   * @param dsm dataset manager actor reference
   * @return OperationStatus
   */
  private def handleDsJob(username: String, job: DataSetOp, dsm: ActorRef) = job match {

    /**
     * Download a file from a source
     */
    case DsAddFromUrl(name, desc, target_algo, url, id) =>
      try {
        checkPathDir("./datastore")
        checkPathDir("./datastore/datasets")
        checkPathDir(s"./datastore/datasets/$username")
        val downloader = new URL(url) #> new File(s"./datasets/$username/$name")
        downloader.run()
        OperationStatus.OpSuccess
      } catch {
        case _: Throwable => OperationStatus.OpFailure
      }

    /**
     * Add a file directly from the supplied resource
     */
    case DsAddDirect(name, desc, target_algo, file, id) =>
      val targetalgo = target_algo match {
        case Algorithm.ALS => "ALS"
        case Algorithm.Clustering => "Clustering"
        case Algorithm.FPgrowth => "FPM"
        case Algorithm.SVM => "SVM"
      }

      try {
        checkPathDir("/.datastore")
        checkPathDir("./datastore/datasets")
        checkPathDir(s"./datastore/datasets/$username")
        val filepath = Paths.get(s"./datastore/datsets/$username/$name")
        if (Files.exists(filepath)) OperationStatus.OpWarning
        else {
          file.moveTo(new File(s"./datastore/datasets/$username/$name"))
          dsm ! AddDatasetRecord(username, DataSetEntry(name, desc, "dataset", targetalgo, "available", ""))
          OperationStatus.OpSuccess
        }
      } catch {
        case _: Throwable => OperationStatus.OpFailure
      }

    /**
     * Removes a dataset on a user's account
     */
    case DsDelete(name, id) =>
      val filepath = Paths.get(s"./datastore/datasets/$username/$name")
      try {
        if (!Files.exists(filepath)) {
          dsm ! RemoveDatasetRecord(username, name)
          Files.delete(filepath)
          OperationStatus.OpSuccess
        } else OperationStatus.OpWarning
      } catch {
        case _: Throwable => OperationStatus.OpFailure
      }

    /**
     * Refreshes the file from the original download location
     */
    case DsRefresh(name, id) =>
      try {
        val url = dsm ? GiveUserData(username) match {
          case x: Future[Any] => Await.result(x, 30 seconds) match {
            case DataSetEntry(_, _, _, _, _, url: String) => url
          }
        }
        val downloader = new URL(url) #> new File(s"./datastore/datasetes/$username/$name")
        downloader.run()
        OperationStatus.OpSuccess
      } catch {
        case _: Throwable => OperationStatus.OpFailure
      }

    case _ => OperationStatus.OpFailure
  }

  def receive = {

    /**
     * Manages all the incoming Dataset related jobs
     *
     * Modified by Anish,
     * The reply should be a JobStatus message
     */
    case (dsm: ActorRef, SubmitDsOpJob(username: String, job: DataSetOp)) => handleDsJob(username, job, dsm) match {
      case s @ OperationStatus.OpSuccess =>
        sender ! JobStatus(username,s)
        mediator ! Log(s, username, "Dataset stored on disk successfully.", job)

      case s @ OperationStatus.OpWarning =>
        sender ! JobStatus(username,s)
        mediator ! Log(s, username, "Warning: No fatal error, but operation could not be completed.", job)

      case s @ OperationStatus.OpFailure =>
        sender ! JobStatus(username,s)
        mediator ! Log(s, username, "Fatal error: Operation could not be completed", job)
    }

    /**
     * Save the incoming miner result to the disk
     */
    case MinerResult(al: Algorithm.Algorithm, user: String, name: String, save: (String => Unit), job) => {<commit_msg>patch for WorkerActor<commit_after>package actors.persistenceManager

import java.io.File
import java.net.URL
import java.nio.file.{Files, Paths}

import akka.actor.{Actor, ActorRef}
import akka.pattern.ask
import akka.util.Timeout
import models.batch.OperationStatus
import models.batch.job._
import models.messages.batchProcessing._
import models.messages.logger.Log
import models.messages.persistenceManaging.DataSetEntry
import models.mining.mlret._
import models.mining.{Algorithm, MinerResult}

import scala.concurrent.duration._
import scala.concurrent.{Await, Future}
import scala.sys.process._


/**
 * Created by chetan on 13/04/15.
 */
class WorkerActor(mediator: ActorRef) extends Actor {

  private implicit val timeout = Timeout(10 seconds)

  /**
   * Checks if a particular file path exists
   * @param incoming path value as string
   * @return unit
   */
  private def checkPathFile(incoming: String) = {
    val path = Paths.get(incoming)
    if (!Files.exists(path)) Files.createFile(path)
  }

  /**
   * Checkes if a particular directory path exists
   * @param incoming path value as string
   * @return unit
   */
  private def checkPathDir(incoming: String) = {
    val path = Paths.get(incoming)
    if (!Files.exists(path)) Files.createDirectories(path)
  }

  /**
   * Handles output operations of the mining results
   * @param operation Details of user and dataset wrapped in GetDsData case object Details of user and dataset wrapped in GetDsData case object Details of user and dataset wrapped in GetDsData case object Details of user and dataset wrapped in GetDsData case object
   * @param dsm Actor reference to datastore manager
   * @return Option[JSON-Output]
   */
  private def handleDsOutput(operation: GetDsData, dsm: ActorRef) = {
    val (uname, name) = (operation.username, operation.Ds)
    val result = Await.result(dsm ? CheckUserDataset(uname, name), 10 seconds).asInstanceOf[Option[DataSetEntry]]
    val filepath = s"./datastore/datasets/$uname/$name"
    result match {
      case Some(x: DataSetEntry) => x.targetAlgorithm match {
        case "als" => new RALS(filepath, name).output
        case "clustering" => new RClustering(filepath, name).output
        case "fpm" => new RFPM(filepath, name).output
        case "svm" => new RSVM(filepath, name).output
        case _ => None
      }
      case None => None
    }
  }

  /**
   * handles all the jobs for the actor reciever
   * @param username username to work on
   * @param job job to be dealt with
   * @param dsm dataset manager actor reference
   * @return OperationStatus
   */
  private def handleDsJob(username: String, job: DataSetOp, dsm: ActorRef) = job match {

    /**
     * Download a file from a source
     */
    case DsAddFromUrl(name, desc, target_algo, url, id) =>
      try {
        checkPathDir("./datastore")
        checkPathDir("./datastore/datasets")
        checkPathDir(s"./datastore/datasets/$username")
        val downloader = new URL(url) #> new File(s"./datasets/$username/$name")
        downloader.run()
        OperationStatus.OpSuccess
      } catch {
        case _: Throwable => OperationStatus.OpFailure
      }

    /**
     * Add a file directly from the supplied resource
     */
    case DsAddDirect(name, desc, target_algo, file, id) =>
      val targetalgo = target_algo match {
        case Algorithm.ALS => "ALS"
        case Algorithm.Clustering => "Clustering"
        case Algorithm.FPgrowth => "FPM"
        case Algorithm.SVM => "SVM"
      }

      try {
        checkPathDir("/.datastore")
        checkPathDir("./datastore/datasets")
        checkPathDir(s"./datastore/datasets/$username")
        val filepath = Paths.get(s"./datastore/datsets/$username/$name")
        if (Files.exists(filepath)) OperationStatus.OpWarning
        else {
          file.moveTo(new File(s"./datastore/datasets/$username/$name"))
          dsm ! AddDatasetRecord(username, DataSetEntry(name, desc, "dataset", targetalgo, "available", ""))
          OperationStatus.OpSuccess
        }
      } catch {
        case _: Throwable => OperationStatus.OpFailure
      }

    /**
     * Removes a dataset on a user's account
     */
    case DsDelete(name, id) =>
      val filepath = Paths.get(s"./datastore/datasets/$username/$name")
      try {
        if (!Files.exists(filepath)) {
          dsm ! RemoveDatasetRecord(username, name)
          Files.delete(filepath)
          OperationStatus.OpSuccess
        } else OperationStatus.OpWarning
      } catch {
        case _: Throwable => OperationStatus.OpFailure
      }

    /**
     * Refreshes the file from the original download location
     */
    case DsRefresh(name, id) =>
      try {
        val url = dsm ? GiveUserData(username) match {
          case x: Future[Any] => Await.result(x, 30 seconds) match {
            case DataSetEntry(_, _, _, _, _, url: String) => url
          }
        }
        val downloader = new URL(url) #> new File(s"./datastore/datasetes/$username/$name")
        downloader.run()
        OperationStatus.OpSuccess
      } catch {
        case _: Throwable => OperationStatus.OpFailure
      }

    case _ => OperationStatus.OpFailure
  }

  def receive = {

    /**
     * Manages all the incoming Dataset related jobs
     *
     * Modified by Anish,
     * The reply should be a JobStatus message
     */
    case (dsm: ActorRef, SubmitDsOpJob(username: String, job: DataSetOp)) =>
      println("inside presWorker, got submitDsOp")
      handleDsJob(username, job, dsm) match {
      case s @ OperationStatus.OpSuccess =>
        mediator ! JobStatus(username,s)
        mediator ! Log(s, username, "Dataset stored on disk successfully.", job)

      case s @ OperationStatus.OpWarning =>
        mediator ! JobStatus(username,s)
        mediator ! Log(s, username, "Warning: No fatal error, but operation could not be completed.", job)

      case s @ OperationStatus.OpFailure =>
        mediator ! JobStatus(username,s)
        mediator ! Log(s, username, "Fatal error: Operation could not be completed", job)
    }

    /**
     * Save the incoming miner result to the disk
     */
    case MinerResult(al: Algorithm.Algorithm, user: String, name: String, save: (String => Unit), job) => {
{'fraction_non_alphanumeric': 0.10742203239584172, 'fraction_numerical': 0.0019340801031509389, 'mean_word_length': 2.787000305157156, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6337442', 'n_tokens_mistral': 3708, 'n_tokens_neox': 3600, 'n_words': 1016}