<commit_before>package manifests.passengers

import akka.actor.{ActorSystem, Cancellable}
import akka.stream.QueueOfferResult.Enqueued
import akka.stream.scaladsl.SourceQueueWithComplete
import drt.server.feeds.api.ApiProviderLike
import drt.shared.CrunchApi.MillisSinceEpoch
import drt.shared.{DqEventCodes, SDateLike}
import org.slf4j.{Logger, LoggerFactory}
import passengersplits.parsing.VoyageManifestParser
import passengersplits.parsing.VoyageManifestParser.VoyageManifest
import server.feeds.{ManifestsFeedFailure, ManifestsFeedResponse, ManifestsFeedSuccess}
import services.SDate
import services.graphstages.DqManifests

import scala.concurrent.Await
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
import scala.language.postfixOps
import scala.util.matching.Regex
import scala.util.{Failure, Success, Try}


class S3ManifestPoller(sourceQueue: SourceQueueWithComplete[ManifestsFeedResponse], portCode: String, initialLastSeenFileName: String, provider: ApiProviderLike)
                      (implicit actorSystem: ActorSystem) {

  val log: Logger = LoggerFactory.getLogger(getClass)

  val dqRegex: Regex = "(drt_dq_[0-9]{6}_[0-9]{6})(_[0-9]{4}\\.zip)".r

  var liveManifestsBuffer: Option[ManifestsFeedSuccess] = None
  var lastSeenFileName: String = initialLastSeenFileName
  var lastFetchedMillis: MillisSinceEpoch = 0

  def fetchNewManifests(startingFileName: String): ManifestsFeedResponse = {
    log.info(s"Fetching manifests from files newer than $startingFileName")
    val eventualFileNameAndManifests = provider
      .manifestsFuture(startingFileName)
      .map(fetchedFilesAndManifests => {
        val (latestFileName, fetchedManifests) = if (fetchedFilesAndManifests.nonEmpty) {
          val lastSeen = fetchedFilesAndManifests.map { case (fileName, _) => fileName }.max
          val manifests = fetchedFilesAndManifests.map { case (_, manifest) => jsonStringToManifest(manifest) }.toSet
          log.info(s"Got ${manifests.size} manifests")
          (lastSeen, manifests)
        }
        else (startingFileName, Set[Option[VoyageManifest]]())

        (latestFileName, fetchedManifests)
      })

    Try {
      Await.result(eventualFileNameAndManifests, 30 minute)
    } match {
      case Success((latestFileName, maybeManifests)) =>
        log.info(s"Fetched ${maybeManifests.count(_.isDefined)} manifests up to file $latestFileName")
        ManifestsFeedSuccess(DqManifests(latestFileName, maybeManifests.flatten), SDate.now())
      case Failure(t) =>
        log.warn(s"Failed to fetch new manifests", t)
        ManifestsFeedFailure(t.toString, SDate.now())
    }
  }

  def jsonStringToManifest(content: String): Option[VoyageManifest] = {
    VoyageManifestParser.parseVoyagePassengerInfo(content) match {
      case Success(m) =>
        if (m.EventCode == DqEventCodes.DepartureConfirmed && m.ArrivalPortCode == portCode) {
          log.info(s"Using ${m.EventCode} manifest for ${m.ArrivalPortCode} arrival ${m.flightCode}")
          Option(m)
        }
        else None
      case Failure(t) =>
        log.error(s"Failed to parse voyage manifest json", t)
        None
    }
  }

  def startPollingForManifests(): Cancellable = {
    actorSystem.scheduler.schedule(0 seconds, 1 minute, new Runnable {
      def run(): Unit = {
        fetchNewManifests(lastSeenFileName) match {
          case ManifestsFeedSuccess(DqManifests(latestFileName, manifests), createdAt) =>
            log.info(s"Received live manifests")
            liveManifestsBuffer = updateManifestsBuffer(latestFileName, manifests, createdAt)
            lastSeenFileName = latestFileName
            tryOfferManifests()
          case ManifestsFeedFailure(msg, _) =>
            log.warn(s"Failed to fetch new live manifests: $msg")
        }
      }
    })
  }

  def tryOfferManifests(): Unit = {
    liveManifestsBuffer.foreach { manifestsToSend =>
      sourceQueue.offer(manifestsToSend).map {
        case Enqueued =>
          liveManifestsBuffer = None
          log.info(s"Enqueued live manifests: $manifestsToSend")
        case _ => log.info(s"Couldn't enqueue live manifests. Leaving them in the buffer")
      }.recover {
        case t => log.error(s"Couldn't enqueue live manifests. Leaving them in the buffer", t)
      }
    }
  }

  def updateManifestsBuffer(lastFilename: String, manifests: Set[VoyageManifest], createdAt: SDateLike): Option[ManifestsFeedSuccess] = {
    liveManifestsBuffer match {
      case None =>
        log.info(s"live manifest buffer was empty")
        Option(ManifestsFeedSuccess(DqManifests(lastFilename, manifests), createdAt))
      case Some(ManifestsFeedSuccess(DqManifests(_, existingManifests), _)) =>
        log.info(s"live manifest buffer was not empty. Adding new manifests to it")
        Option(ManifestsFeedSuccess(DqManifests(lastFilename, existingManifests ++ manifests), createdAt))
    }
  }
}
<commit_msg>Remove verbose logging dump of manifests..<commit_after>package manifests.passengers

import akka.actor.{ActorSystem, Cancellable}
import akka.stream.QueueOfferResult.Enqueued
import akka.stream.scaladsl.SourceQueueWithComplete
import drt.server.feeds.api.ApiProviderLike
import drt.shared.CrunchApi.MillisSinceEpoch
import drt.shared.{DqEventCodes, SDateLike}
import org.slf4j.{Logger, LoggerFactory}
import passengersplits.parsing.VoyageManifestParser
import passengersplits.parsing.VoyageManifestParser.VoyageManifest
import server.feeds.{ManifestsFeedFailure, ManifestsFeedResponse, ManifestsFeedSuccess}
import services.SDate
import services.graphstages.DqManifests

import scala.concurrent.Await
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
import scala.language.postfixOps
import scala.util.matching.Regex
import scala.util.{Failure, Success, Try}


class S3ManifestPoller(sourceQueue: SourceQueueWithComplete[ManifestsFeedResponse], portCode: String, initialLastSeenFileName: String, provider: ApiProviderLike)
                      (implicit actorSystem: ActorSystem) {

  val log: Logger = LoggerFactory.getLogger(getClass)

  val dqRegex: Regex = "(drt_dq_[0-9]{6}_[0-9]{6})(_[0-9]{4}\\.zip)".r

  var liveManifestsBuffer: Option[ManifestsFeedSuccess] = None
  var lastSeenFileName: String = initialLastSeenFileName
  var lastFetchedMillis: MillisSinceEpoch = 0

  def fetchNewManifests(startingFileName: String): ManifestsFeedResponse = {
    log.info(s"Fetching manifests from files newer than $startingFileName")
    val eventualFileNameAndManifests = provider
      .manifestsFuture(startingFileName)
      .map(fetchedFilesAndManifests => {
        val (latestFileName, fetchedManifests) = if (fetchedFilesAndManifests.nonEmpty) {
          val lastSeen = fetchedFilesAndManifests.map { case (fileName, _) => fileName }.max
          val manifests = fetchedFilesAndManifests.map { case (_, manifest) => jsonStringToManifest(manifest) }.toSet
          log.info(s"Got ${manifests.size} manifests")
          (lastSeen, manifests)
        }
        else (startingFileName, Set[Option[VoyageManifest]]())

        (latestFileName, fetchedManifests)
      })

    Try {
      Await.result(eventualFileNameAndManifests, 30 minute)
    } match {
      case Success((latestFileName, maybeManifests)) =>
        log.info(s"Fetched ${maybeManifests.count(_.isDefined)} manifests up to file $latestFileName")
        ManifestsFeedSuccess(DqManifests(latestFileName, maybeManifests.flatten), SDate.now())
      case Failure(t) =>
        log.warn(s"Failed to fetch new manifests", t)
        ManifestsFeedFailure(t.toString, SDate.now())
    }
  }

  def jsonStringToManifest(content: String): Option[VoyageManifest] = {
    VoyageManifestParser.parseVoyagePassengerInfo(content) match {
      case Success(m) =>
        if (m.EventCode == DqEventCodes.DepartureConfirmed && m.ArrivalPortCode == portCode) {
          log.info(s"Using ${m.EventCode} manifest for ${m.ArrivalPortCode} arrival ${m.flightCode}")
          Option(m)
        }
        else None
      case Failure(t) =>
        log.error(s"Failed to parse voyage manifest json", t)
        None
    }
  }

  def startPollingForManifests(): Cancellable = {
    actorSystem.scheduler.schedule(0 seconds, 1 minute, new Runnable {
      def run(): Unit = {
        fetchNewManifests(lastSeenFileName) match {
          case ManifestsFeedSuccess(DqManifests(latestFileName, manifests), createdAt) =>
            log.info(s"Received live manifests")
            liveManifestsBuffer = updateManifestsBuffer(latestFileName, manifests, createdAt)
            lastSeenFileName = latestFileName
            tryOfferManifests()
          case ManifestsFeedFailure(msg, _) =>
            log.warn(s"Failed to fetch new live manifests: $msg")
        }
      }
    })
  }

  def tryOfferManifests(): Unit = {
    liveManifestsBuffer.foreach { manifestsToSend =>
      sourceQueue.offer(manifestsToSend).map {
        case Enqueued =>
          liveManifestsBuffer = None
          log.info(s"Enqueued live manifests")
        case _ => log.info(s"Couldn't enqueue live manifests. Leaving them in the buffer")
      }.recover {
        case t => log.error(s"Couldn't enqueue live manifests. Leaving them in the buffer", t)
      }
    }
  }

  def updateManifestsBuffer(lastFilename: String, manifests: Set[VoyageManifest], createdAt: SDateLike): Option[ManifestsFeedSuccess] = {
    liveManifestsBuffer match {
      case None =>
        log.info(s"live manifest buffer was empty")
        Option(ManifestsFeedSuccess(DqManifests(lastFilename, manifests), createdAt))
      case Some(ManifestsFeedSuccess(DqManifests(_, existingManifests), _)) =>
        log.info(s"live manifest buffer was not empty. Adding new manifests to it")
        Option(ManifestsFeedSuccess(DqManifests(lastFilename, existingManifests ++ manifests), createdAt))
    }
  }
}

{'fraction_non_alphanumeric': 0.0970383099161023, 'fraction_numerical': 0.003234610330536743, 'mean_word_length': 3.84287812041116, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 33, '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': '6035834', 'n_tokens_mistral': 3054, 'n_tokens_neox': 2968, 'n_words': 594}