<commit_before>package com.twitter.gizzard.scheduler

import scala.collection.mutable
import scala.util.matching.Regex
import com.twitter.json.Json
import net.lag.logging.Logger
import org.codehaus.jackson.map.ObjectMapper
import java.util.{Map => JMap, List => JList}
import scala.collection.jcl

/**
 * Codec for json-encoded jobs.
 *
 * A JsonJob can turn itself into a json-encoded string via 'toJson'/'toMap', so encoding is
 * delegated to the JsonJob itself.
 *
 * For decoding, a set of JsonJobParsers are registered with corresponding regular expressions.
 * If the primary key of the json structure matches the expression, the nested map is passed to
 * that parser to turn it into a JsonJob.
 *
 * Jobs that can't be parsed by the json library are handed to 'unparsableJobHandler'.
 */
class JsonCodec(unparsableJobHandler: Array[Byte] => Unit) extends Codec[JsonJob] {
  private val mapper = new ObjectMapper

  protected val log = Logger.get(getClass.getName)
  protected val processors = {
    val p = mutable.Map.empty[Regex, JsonJobParser]
    p += (("JsonNestedJob".r, new JsonNestedJobParser(this)))
    // for backward compat:
    p += (("JobWithTasks".r, new JsonNestedJobParser(this)))
    p
  }

  def +=(item: (Regex, JsonJobParser)) = processors += item
  def +=(r: Regex, p: JsonJobParser) = processors += ((r, p))

  def flatten(job: JsonJob): Array[Byte] = job.toJson.getBytes

  def inflate(data: Array[Byte]): JsonJob = {
    try {
      val javaMap: JMap[String, Any] = mapper.readValue(new String(data), classOf[JMap[String, Any]])
      val scalaMap = deepConvert(javaMap)
      inflate(scalaMap)
    } catch {
      case e =>
        log.error(e, "Unparsable JsonJob; dropping: " + e.toString)
        unparsableJobHandler(data)
        throw new UnparsableJsonException("Unparsable json", e)
    }
  }

  def inflate(json: Map[String, Any]): JsonJob = {
    val (jobType, attributes) = json.toList.first
    val processor = processors.find { case (processorRegex, _) =>
      processorRegex.findFirstIn(jobType).isDefined
    }.map { case (_, processor) => processor }.getOrElse {
      throw new UnparsableJsonException("Can't find matching processor for '%s' in %s".format(jobType, processors), null)
    }
    try {
      processor.parse(attributes.asInstanceOf[Map[String, Any]])
    } catch {
      case e =>
        throw new UnparsableJsonException("Processor '%s' blew up: %s".format(jobType, e.toString), e)
    }
  }

  private def deepConvert(javaList: JList[Any]): List[Any] = {
    jcl.Buffer(javaList).map { v =>
      v match {
        case jm: JMap[String, Any] => deepConvert(jm)
        case jl: JList[Any] => deepConvert(jl)
        case _ => v
      }
    }.toList
  }

  private def deepConvert(javaMap: JMap[String, Any]): Map[String, Any] = {
    Map(jcl.Map(javaMap).toSeq.map { case (k,v) =>
      v match {
        case jm: JMap[String, Any] => (k, deepConvert(jm))
        case jl: JList[Any] => (k, deepConvert(jl))
        case _ => (k, v)
      }
    }: _*)
  }
}
<commit_msg>add SchedulableWithTasks for backwards compat<commit_after>package com.twitter.gizzard.scheduler

import scala.collection.mutable
import scala.util.matching.Regex
import com.twitter.json.Json
import net.lag.logging.Logger
import org.codehaus.jackson.map.ObjectMapper
import java.util.{Map => JMap, List => JList}
import scala.collection.jcl

/**
 * Codec for json-encoded jobs.
 *
 * A JsonJob can turn itself into a json-encoded string via 'toJson'/'toMap', so encoding is
 * delegated to the JsonJob itself.
 *
 * For decoding, a set of JsonJobParsers are registered with corresponding regular expressions.
 * If the primary key of the json structure matches the expression, the nested map is passed to
 * that parser to turn it into a JsonJob.
 *
 * Jobs that can't be parsed by the json library are handed to 'unparsableJobHandler'.
 */
class JsonCodec(unparsableJobHandler: Array[Byte] => Unit) extends Codec[JsonJob] {
  private val mapper = new ObjectMapper

  protected val log = Logger.get(getClass.getName)
  protected val processors = {
    val p = mutable.Map.empty[Regex, JsonJobParser]
    p += (("JsonNestedJob".r, new JsonNestedJobParser(this)))
    // for backward compat:
    p += (("JobWithTasks".r, new JsonNestedJobParser(this)))
    p += (("SchedulableWithTasks".r, new JsonNestedJobParser(this)))
    p
  }

  def +=(item: (Regex, JsonJobParser)) = processors += item
  def +=(r: Regex, p: JsonJobParser) = processors += ((r, p))

  def flatten(job: JsonJob): Array[Byte] = job.toJson.getBytes

  def inflate(data: Array[Byte]): JsonJob = {
    try {
      val javaMap: JMap[String, Any] = mapper.readValue(new String(data), classOf[JMap[String, Any]])
      val scalaMap = deepConvert(javaMap)
      inflate(scalaMap)
    } catch {
      case e =>
        log.error(e, "Unparsable JsonJob; dropping: " + e.toString)
        unparsableJobHandler(data)
        throw new UnparsableJsonException("Unparsable json", e)
    }
  }

  def inflate(json: Map[String, Any]): JsonJob = {
    val (jobType, attributes) = json.toList.first
    val processor = processors.find { case (processorRegex, _) =>
      processorRegex.findFirstIn(jobType).isDefined
    }.map { case (_, processor) => processor }.getOrElse {
      throw new UnparsableJsonException("Can't find matching processor for '%s' in %s".format(jobType, processors), null)
    }
    try {
      processor.parse(attributes.asInstanceOf[Map[String, Any]])
    } catch {
      case e =>
        throw new UnparsableJsonException("Processor '%s' blew up: %s".format(jobType, e.toString), e)
    }
  }

  private def deepConvert(javaList: JList[Any]): List[Any] = {
    jcl.Buffer(javaList).map { v =>
      v match {
        case jm: JMap[String, Any] => deepConvert(jm)
        case jl: JList[Any] => deepConvert(jl)
        case _ => v
      }
    }.toList
  }

  private def deepConvert(javaMap: JMap[String, Any]): Map[String, Any] = {
    Map(jcl.Map(javaMap).toSeq.map { case (k,v) =>
      v match {
        case jm: JMap[String, Any] => (k, deepConvert(jm))
        case jl: JList[Any] => (k, deepConvert(jl))
        case _ => (k, v)
      }
    }: _*)
  }
}

{'fraction_non_alphanumeric': 0.12684031710079274, 'fraction_numerical': 0.0, 'mean_word_length': 3.3937455579246625, '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': '3107358', 'n_tokens_mistral': 2022, 'n_tokens_neox': 1967, 'n_words': 548}