<commit_before>package core.actor

import java.util.UUID.randomUUID

import akka.actor.{ActorLogging, Actor, Props}
import core.actor.utils._
import models.Tag
import play.modules.reactivemongo.ReactiveMongoPlugin._
import play.modules.reactivemongo.json.collection.JSONCollection
import play.api.Play.current
import akka.pattern._
import models.JsonFormats.tagFormat

import scala.concurrent.Future

object TagPersister {

  case object TagsPersisted

  def actorName = s"tag_persister_$randomUUID"

  def props(tags: Set[Tag]) = Props(classOf[TagPersister], tags)

}

class TagPersister(tags: Set[Tag]) extends Actor with RandomIdGenerator with RestartLogging {

  import TagPersister._
  import context.dispatcher

  val collection: JSONCollection = db.collection[JSONCollection]("tag")

  val saveStatuses = tags.map(collection.save(_).withFilter(!_.inError))
  Future.sequence(saveStatuses).map(_ => TagsPersisted).pipeTo(self)

  override def receive: Receive = {
    case TagsPersisted =>
      context.parent ! TagsPersisted
      context.stop(self)
  }

  override def unhandled(message: Any) = {
    (throwOnNonTerminated orElse PartialFunction(super.unhandled _))(message)
  }

}<commit_msg>Improved persisting tags<commit_after>package core.actor

import java.util.UUID.randomUUID

import akka.actor.{Actor, Props}
import core.actor.utils._
import models.Tag

import play.modules.reactivemongo.ReactiveMongoPlugin._
import play.api.Play.current
import akka.pattern._

import play.api.libs.json.Json
import models.JsonFormats.tagFormat
import play.modules.reactivemongo.json.BSONFormats._
import reactivemongo.bson.{BSONArray, BSONDocument}
import reactivemongo.core.commands.RawCommand


object TagPersister {

  case object TagsPersisted

  def actorName = s"tag_persister_$randomUUID"

  def props(tags: Set[Tag]) = Props(classOf[TagPersister], tags)

}

class TagPersister(tags: Set[Tag]) extends Actor with RandomIdGenerator with RestartLogging {

  import TagPersister._
  import context.dispatcher

  val updateStatements = BSONArray(
    for {
      tag <- tags
      tagBson = Json.toJson(tag).as[BSONDocument]
    } yield BSONDocument(
      "q" -> BSONDocument("_id" -> tag._id),
      "u" -> tagBson,
      "upsert" -> true
    )
  )

  val updateCommand = BSONDocument(
    "update" -> "tag",
    "updates" -> updateStatements,
    "ordered" -> true
  )

  db.command(RawCommand(updateCommand)).map(_ => TagsPersisted).pipeTo(self)

  override def receive: Receive = {
    case TagsPersisted =>
      context.parent ! TagsPersisted
      context.stop(self)
  }

  override def unhandled(message: Any) = {
    (throwOnNonTerminated orElse PartialFunction(super.unhandled _))(message)
  }

}
{'fraction_non_alphanumeric': 0.1032067821599705, 'fraction_numerical': 0.0, 'mean_word_length': 4.449799196787149, 'pattern_counts': {'":': 0, '<': 4, '<?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': '4532742', 'n_tokens_mistral': 888, 'n_tokens_neox': 860, 'n_words': 177}