<commit_before>package passengersplits.s3

import java.io.{InputStream, File => JFile}
import java.nio.file.{Path => JPath}
import java.util.concurrent.TimeUnit
import java.util.zip.ZipInputStream

import akka.actor.{ActorLogging, ActorRef, ActorSystem, Props}
import akka.event.LoggingAdapter
import akka.stream._
import akka.stream.actor.{ActorSubscriber, ActorSubscriberMessage, MaxInFlightRequestStrategy}
import akka.stream.scaladsl.{Flow, Sink, Source, StreamConverters}
import akka.{Done, NotUsed}
import com.amazonaws.ClientConfiguration
import com.amazonaws.auth.profile.ProfileCredentialsProvider
import com.amazonaws.services.s3.S3ClientOptions
import com.mfglabs.commons.aws.s3.{AmazonS3AsyncClient, S3StreamBuilder}
import passengersplits._
import passengersplits.core.ZipUtils.UnzippedFileContent
import passengersplits.core.{Core, CoreActors, CoreLogging, ZipUtils}
import passengersplits.parsing.PassengerInfoParser
import spatutorial.shared.PassengerSplits.VoyagePaxSplits
import spatutorial.shared.SDateLike

import scala.concurrent.duration.{FiniteDuration, _}
import scala.concurrent.{Await, ExecutionContext, Future, Promise}
import scala.language.postfixOps
import scala.util.{Failure, Success, Try}


trait UnzippedFilesProvider {
  def unzippedFilesAsSource: Source[UnzippedFileContent, NotUsed]
}

trait FilenameProvider {

  def fileNameStream: Source[String, NotUsed]

  def zipFileNameFilter(filename: String): Boolean

  def latestFilePaths = fileNameStream.filter(zipFileNameFilter)
}


trait S3Reader extends CoreLogging with UnzippedFilesProvider with FilenameProvider {

  def builder: S3StreamBuilder

  def createBuilder: S3StreamBuilder

  def bucket: String

  def createS3client: AmazonS3AsyncClient

  def numberOfCores = 8

  def unzipTimeout = FiniteDuration(400, TimeUnit.SECONDS)

  override def fileNameStream: Source[String, NotUsed] = builder.listFilesAsStream(bucket).map(_._1)

  def zipFilenameToEventualFileContent(zipFileName: String)(implicit actorMaterializer: Materializer, ec: ExecutionContext): Future[List[UnzippedFileContent]] = Future {
    try {
      log.info(s"Will parse ${zipFileName}")
      val threadSpecificBuilder = createBuilder
      val zippedByteStream = threadSpecificBuilder.getFileAsStream(bucket, zipFileName)
      val inputStream: InputStream = zippedByteStream.runWith(
        StreamConverters.asInputStream(unzipTimeout)
      )(actorMaterializer)
      val unzippedStream = new ZipInputStream(inputStream)
      val unzippedFileContent: List[UnzippedFileContent] = ZipUtils.unzipAllFilesInStream(unzippedStream).toList
      unzippedFileContent.map(_.copy(zipFilename = Some(zipFileName)))
    } catch {
      case e: Throwable =>
        log.error(e, s"Error in S3Poller for ${zipFileName}: ")
        throw e
    }
  }


  def intermediate(paths: Source[String, NotUsed])
                  (implicit actorMaterializer: Materializer, ec: ExecutionContext): Source[List[UnzippedFileContent], NotUsed] = {
    paths
      .mapAsync(numberOfCores) {
        zipFilenameToEventualFileContent
      }
  }

  def unzippedFilesSource(implicit actorMaterializer: Materializer, ec: ExecutionContext): Source[UnzippedFileContent, NotUsed] = {
    intermediate(latestFilePaths).mapConcat {
      t => t
    }
  }

}

trait SimpleS3Reader extends S3Reader with Core {
  this: CoreActors =>
  val bucket: String = "drt-deveu-west-1"<commit_msg>DRT-4524 Close zip input stream when finished reading<commit_after>package passengersplits.s3

import java.io.{InputStream, File => JFile}
import java.nio.file.{Path => JPath}
import java.util.concurrent.TimeUnit
import java.util.zip.ZipInputStream

import akka.actor.{ActorLogging, ActorRef, ActorSystem, Props}
import akka.event.LoggingAdapter
import akka.stream._
import akka.stream.actor.{ActorSubscriber, ActorSubscriberMessage, MaxInFlightRequestStrategy}
import akka.stream.scaladsl.{Flow, Sink, Source, StreamConverters}
import akka.{Done, NotUsed}
import com.amazonaws.ClientConfiguration
import com.amazonaws.auth.profile.ProfileCredentialsProvider
import com.amazonaws.services.s3.S3ClientOptions
import com.mfglabs.commons.aws.s3.{AmazonS3AsyncClient, S3StreamBuilder}
import passengersplits._
import passengersplits.core.ZipUtils.UnzippedFileContent
import passengersplits.core.{Core, CoreActors, CoreLogging, ZipUtils}
import passengersplits.parsing.PassengerInfoParser
import spatutorial.shared.PassengerSplits.VoyagePaxSplits
import spatutorial.shared.SDateLike

import scala.concurrent.duration.{FiniteDuration, _}
import scala.concurrent.{Await, ExecutionContext, Future, Promise}
import scala.language.postfixOps
import scala.util.{Failure, Success, Try}


trait UnzippedFilesProvider {
  def unzippedFilesAsSource: Source[UnzippedFileContent, NotUsed]
}

trait FilenameProvider {

  def fileNameStream: Source[String, NotUsed]

  def zipFileNameFilter(filename: String): Boolean

  def latestFilePaths = fileNameStream.filter(zipFileNameFilter)
}


trait S3Reader extends CoreLogging with UnzippedFilesProvider with FilenameProvider {

  def builder: S3StreamBuilder

  def createBuilder: S3StreamBuilder

  def bucket: String

  def createS3client: AmazonS3AsyncClient

  def numberOfCores = 8

  def unzipTimeout = FiniteDuration(400, TimeUnit.SECONDS)

  override def fileNameStream: Source[String, NotUsed] = builder.listFilesAsStream(bucket).map(_._1)

  def zipFilenameToEventualFileContent(zipFileName: String)(implicit actorMaterializer: Materializer, ec: ExecutionContext): Future[List[UnzippedFileContent]] = Future {
    try {
      log.info(s"Will parse ${zipFileName}")
      val threadSpecificBuilder = createBuilder
      val zippedByteStream = threadSpecificBuilder.getFileAsStream(bucket, zipFileName)
      val inputStream: InputStream = zippedByteStream.runWith(
        StreamConverters.asInputStream(unzipTimeout)
      )(actorMaterializer)

      ZipUtils.usingZip(new ZipInputStream(inputStream)) {
        unzippedStream =>
          val unzippedFileContent: List[UnzippedFileContent] = ZipUtils.unzipAllFilesInStream(unzippedStream).toList
          unzippedFileContent.map(_.copy(zipFilename = Some(zipFileName)))
      }

    } catch {
      case e: Throwable =>
        log.error(e, s"Error in S3Poller for ${zipFileName}: ")
        throw e
    }
  }


  def intermediate(paths: Source[String, NotUsed])
                  (implicit actorMaterializer: Materializer, ec: ExecutionContext): Source[List[UnzippedFileContent], NotUsed] = {
    paths
      .mapAsync(numberOfCores) {
        zipFilenameToEventualFileContent
      }
  }

  def unzippedFilesSource(implicit actorMaterializer: Materializer, ec: ExecutionContext): Source[UnzippedFileContent, NotUsed] = {
    intermediate(latestFilePaths).mapConcat {
      t => t
    }
  }

}

trait SimpleS3Reader extends S3Reader with Core {
  this: CoreActors =>
  val bucket: String = "drt-deveu-west-1"
{'fraction_non_alphanumeric': 0.08403239803297657, 'fraction_numerical': 0.006363899334683252, 'mean_word_length': 5.224122412241224, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 14, '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': '7317227', 'n_tokens_mistral': 2103, 'n_tokens_neox': 2048, 'n_words': 421}