<commit_before>package lila.irwin

import akka.stream.scaladsl._
import play.api.libs.json._

import lila.common.Bus

final class IrwinStream {

  private val classifier = "userSignup"

  private val blueprint =
    Source
      .queue[IrwinRequest](32, akka.stream.OverflowStrategy.dropHead)
      .map(requestJson)
      .map { js =>
        s"${Json.stringify(js)}\n"
      }

  def apply(): Source[String, _] = blueprint mapMaterializedValue { queue =>
    val sub = Bus.subscribeFun(classifier) {
      case req: IrwinRequest =>
        lila.mon.mod.irwin.streamEventType("request").increment()
        queue offer req
    }

    queue.watchCompletion dforeach { _ =>
      Bus.unsubscribe(sub, classifier)
    }
  }

  private def requestJson(req: IrwinRequest) = Json.obj(
    "t"      -> "request",
    "origin" -> req.origin.key,
    "user" -> Json.obj(
      "id"     -> req.suspect.user.id,
      "titled" -> req.suspect.user.hasTitle,
      "engine" -> req.suspect.user.engine,
      "games"  -> req.suspect.user.count.rated
    ),
    "games" -> req.games.map {
      case (game, analysis) =>
        Json.obj(
          "id"    -> game.id,
          "white" -> game.whitePlayer.userId,
          "black" -> game.blackPlayer.userId,
          "pgn"   -> game.pgnMoves.mkString(" "),
          "emts"  -> game.clockHistory.isDefined ?? game.moveTimes.map(_.map(_.centis)),
          "analysis" -> analysis.map {
            _.infos.map { info =>
              info.cp.map { cp =>
                Json.obj("cp" -> cp.value)
              } orElse
                info.mate.map { mate =>
                  Json.obj("mate" -> mate.value)
                } getOrElse
                JsNull
            }
          }
        )
    }
  )
}
<commit_msg>fix irwin stream sub channel<commit_after>package lila.irwin

import akka.stream.scaladsl._
import play.api.libs.json._

import lila.common.Bus

final class IrwinStream {

  private val channel = "irwin"

  private val blueprint =
    Source
      .queue[IrwinRequest](64, akka.stream.OverflowStrategy.dropHead)
      .map(requestJson)
      .map { js =>
        s"${Json.stringify(js)}\n"
      }

  def apply(): Source[String, _] = blueprint mapMaterializedValue { queue =>
    val sub = Bus.subscribeFun(channel) {
      case req: IrwinRequest =>
        lila.mon.mod.irwin.streamEventType("request").increment()
        queue offer req
    }

    queue.watchCompletion dforeach { _ =>
      Bus.unsubscribe(sub, channel)
    }
  }

  private def requestJson(req: IrwinRequest) = Json.obj(
    "t"      -> "request",
    "origin" -> req.origin.key,
    "user" -> Json.obj(
      "id"     -> req.suspect.user.id,
      "titled" -> req.suspect.user.hasTitle,
      "engine" -> req.suspect.user.engine,
      "games"  -> req.suspect.user.count.rated
    ),
    "games" -> req.games.map {
      case (game, analysis) =>
        Json.obj(
          "id"    -> game.id,
          "white" -> game.whitePlayer.userId,
          "black" -> game.blackPlayer.userId,
          "pgn"   -> game.pgnMoves.mkString(" "),
          "emts"  -> game.clockHistory.isDefined ?? game.moveTimes.map(_.map(_.centis)),
          "analysis" -> analysis.map {
            _.infos.map { info =>
              info.cp.map { cp =>
                Json.obj("cp" -> cp.value)
              } orElse
                info.mate.map { mate =>
                  Json.obj("mate" -> mate.value)
                } getOrElse
                JsNull
            }
          }
        )
    }
  )
}

{'fraction_non_alphanumeric': 0.14273528575490474, 'fraction_numerical': 0.0011373329542223485, 'mean_word_length': 1.7898493259318002, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 51, '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': '3092869', 'n_tokens_mistral': 1170, 'n_tokens_neox': 1127, 'n_words': 196}