//设置package包名称以及导入依赖的类
import com.datastax.spark.connector._
import play.api.libs.json.Json
import java.io.{ByteArrayOutputStream, ByteArrayInputStream}
import java.util.zip.{GZIPOutputStream, GZIPInputStream}
import PreProcessingConfig._

case class VT_results_by_service_name_class(service_name: String, sha256: String)
case class VT_results_by_sha256_class(sha256: String, service_name: String, results: Array[Byte] )
case class VT_join_results_class(sha256: String, service_name: String, results: String)
case class VT_sample_signatures_initial_seq_rdd_class(sha256: String, seq_results: Seq[String])
case class VT_sample_signatures_final_array_rdd_class(sha256:String, array_results:Array[Double])

def unzip(x: Array[Byte]) : String = {		
    val inputStream = new GZIPInputStream(new ByteArrayInputStream(x))
    val output = scala.io.Source.fromInputStream(inputStream).mkString
    return output
}
def deleteNumberInSampleSignatures(x: String): Boolean = {
    val regex = "[0-9]".r
    return regex.findFirstIn(x).isEmpty
}

val VT_results_by_service_name_meta = sc.cassandraTable[VT_results_by_service_name_class](keyspace,service_name_table).where("service_name=?","virustotal")
val VT_results_by_service_name_rdd = VT_results_by_service_name_meta.keyBy(x=> (x.sha256,x.service_name))
val VT_results_by_sha256_meta = sc.cassandraTable[VT_results_by_sha256_class](keyspace,sha256_table)
val VT_results_by_sha256_rdd = VT_results_by_sha256_meta.keyBy(x => (x.sha256,x.service_name))
val VT_join_results = VT_results_by_service_name_rdd.join(VT_results_by_sha256_rdd).map(x => (new VT_join_results_class(x._1._1,x._1._2, unzip(x._2._2.results)))).distinct().cache()
val sample_signatures_rdd = VT_join_results.flatMap(x=>Json.parse(x.results) \ "scans" \\ "result").map(x=>Json.stringify(x)).filter( x=> !(x == "null"))
val sample_signatures_split_rdd = sample_signatures_rdd.flatMap(x=>x.replaceAll("""["]""","").replaceAll("""\![a-zA-Z0-9\s\+]+""","").replaceAll("""@[a-zA-Z0-9\s\+]+""","").replaceAll("""~[a-zA-Z0-9\s\+]+""","").replaceAll("""[\(|\[|{][a-zA-Z0-9\s\+]*[\)|\]|}]""","").replaceAll("""(\.|\!|\:|\_|\-|\\|/|\[|\])"""," ").split(" ")).filter(x=>(x.size>3)).filter(x=>deleteNumberInSampleSignatures(x)).map(x=>x.toLowerCase())
val signatures_prefix_rdd = sc.textFile(VT_signatures_prefix_suffix_file).map(x=>x.toLowerCase())
val family_signatures_subtract_rdd = sample_signatures_split_rdd.subtract(signatures_prefix_rdd)
val family_signatures_sorted_rdd = sc.parallelize(family_signatures_subtract_rdd.countByValue().toSeq).filter(x=>(x._2>50)).sortBy(x=>x._2,false)
val family_signatures_list = family_signatures_sorted_rdd.keys.collect().toList
val VT_sample_signatures_rdd = VT_join_results.map(x=>(x.sha256,(Json.parse(x.results) \ "scans" \\ "result").map(_.toString).filter( s => !(s== "null")).flatMap(x=>x.replaceAll("""["]""","").replaceAll("""\![a-zA-Z0-9\s\+]+""","").replaceAll("""@[a-zA-Z0-9\s\+]+""","").replaceAll("""~[a-zA-Z0-9\s\+]+""","").replaceAll("""[\(|\[|{][a-zA-Z0-9\s\+]*[\)|\]|}]""","").replaceAll("""(\.|\!|\:|\_|\-|\\|/|\[|\])"""," ").split(" ")).filter(x=>(x.size>3)).filter(x=>deleteNumberInSampleSignatures(x)).map(x=>x.toLowerCase())))
val  VT_sample_signatures_initial_seq_rdd = VT_sample_signatures_rdd.map(x=>new VT_sample_signatures_initial_seq_rdd_class(x._1, x._2))

implicit def bool2int(b:Boolean) = if (b) 1 else 0
def findAllInFamilySignatures(sample_signatures_seq : Seq[String]) : Array[Double] ={
    val forlist = for (family <- family_signatures_list) yield {
        (sample_signatures_seq.contains(family):Int).toDouble
    }
    return forlist.toArray
}

val VT_sample_signatures_final_array_rdd = VT_sample_signatures_initial_seq_rdd.map(x=>new VT_sample_signatures_final_array_rdd_class(x.sha256,findAllInFamilySignatures(x.seq_results)))
VT_sample_signatures_final_array_rdd.toDF().write.format("parquet").save(VT_sample_signatures_final_array_file) 

//设置package包名称以及导入依赖的类
package at.hazm.quebic

import java.io.{ByteArrayInputStream, ByteArrayOutputStream, InputStream, OutputStream}
import java.util.zip.{GZIPInputStream, GZIPOutputStream}

import scala.annotation.tailrec

sealed abstract class Codec(val id:Byte, val name:String) extends Type {
  def encode(buffer:Array[Byte]):Array[Byte]

  def decode(buffer:Array[Byte]):Array[Byte]
}

object Codec {
  val values:Seq[Codec] = Seq(PLAIN, GZIP)
  private[this] val valuesMap = values.groupBy(_.id).mapValues(_.head)

  def valueOf(id:Byte):Codec = valuesMap(id)

  case object PLAIN extends Codec(0, "plain") {
    def encode(buffer:Array[Byte]):Array[Byte] = buffer

    def decode(buffer:Array[Byte]):Array[Byte] = buffer
  }

  case object GZIP extends Codec(1, "gzip") {
    def encode(buffer:Array[Byte]):Array[Byte] = {
      val baos = new ByteArrayOutputStream()
      val out = new GZIPOutputStream(baos)
      out.write(buffer)
      out.finish()
      out.finish()
      baos.toByteArray
    }

    def decode(buffer:Array[Byte]):Array[Byte] = {
      val in = new GZIPInputStream(new ByteArrayInputStream(buffer))
      val out = new ByteArrayOutputStream()
      _copy(in, out, new Array[Byte](2014))
      out.close()
      out.toByteArray
    }
  }

  @tailrec
  private[Codec] def _copy(in:InputStream, out:OutputStream, buffer:Array[Byte]):Unit = {
    val len = in.read(buffer)
    if(len > 0) {
      out.write(buffer, 0, len)
      _copy(in, out, buffer)
    }
  }

} 

//设置package包名称以及导入依赖的类
package game_of_life.util

import java.io.{ByteArrayInputStream, ByteArrayOutputStream, ObjectInputStream, ObjectOutputStream}

trait Codec {

  def encode(obj: AnyRef): Array[Byte]

  def decode[A >: Null](array: Array[Byte]): A
}

trait SerializationCodec extends Codec {

  def encode(obj: AnyRef): Array[Byte] = {
    val baos = new ByteArrayOutputStream()
    val oos = new ObjectOutputStream(baos)
    try {
      oos writeObject obj
    } finally {
      oos close ()
    }
    baos.toByteArray
  }

  def decode[A >: Null](array: Array[Byte]): A = {
    val ois = new ObjectInputStream(new ByteArrayInputStream(array))
    try {
      (ois readObject ()).asInstanceOf[A]
    } finally {
      ois close ()
    }
  }
}

object SerializationCodec extends SerializationCodec 

//设置package包名称以及导入依赖的类
package org.hammerlab.hadoop.kryo

import java.io.{ ByteArrayInputStream, ByteArrayOutputStream, ObjectInputStream, ObjectOutputStream }

import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.io.{ Input, Output }
import org.hammerlab.test.Suite

class SerializableSerializerTest
  extends Suite {
  test("serde") {
    val kryo = new Kryo()
    kryo.setRegistrationRequired(true)
    val baos = new ByteArrayOutputStream()
    val output = new Output(baos)

    val foo = new Foo
    foo.n = 123
    foo.s = "abc"

    intercept[IllegalArgumentException] {
      kryo.writeClassAndObject(output, foo)
    }
    .getMessage should startWith("Class is not registered: org.hammerlab.hadoop.kryo.Foo")

    kryo.register(classOf[Foo], SerializableSerializer[Foo]())

    kryo.writeClassAndObject(output, foo)

    output.close()

    val bytes = baos.toByteArray
    bytes.length should be(93)

    val bais = new ByteArrayInputStream(bytes)

    val input = new Input(bais)
    val after = kryo.readClassAndObject(input).asInstanceOf[Foo]

    after.n should be(foo.n)
    after.s should be(foo.s)
  }
}

class Foo
  extends Serializable {

  var n = 0
  var s = ""

  private def writeObject(out: ObjectOutputStream): Unit = {
    out.writeInt(n)
    out.writeUTF(s)
  }

  private def readObject(in: ObjectInputStream): Unit = {
    n = in.readInt()
    s = in.readUTF()
  }
} 

//设置package包名称以及导入依赖的类
package edu.uw.at.iroberts.wirefugue.kafka.serdes

import java.io.{ByteArrayInputStream, ByteArrayOutputStream, ObjectInputStream, ObjectOutputStream}
import java.util

import org.apache.kafka.common.serialization._


class PacketKeySerializer extends Serializer[PacketKey] {
  override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = ()
  override def close(): Unit = ()

  override def serialize(topic: String, data: PacketKey): Array[Byte] = {
    val bs = new ByteArrayOutputStream()
    val oos = new ObjectOutputStream(bs)
    oos.writeObject(data)

    bs.toByteArray
  }
}

class PacketKeyDeserializer extends Deserializer[PacketKey] {
  override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = ()
  override def close(): Unit = ()

  override def deserialize(topic: String, data: Array[Byte]): PacketKey = {
    val bs = new ByteArrayInputStream(data)
    val ois = new ObjectInputStream(bs)

    ois.readObject().asInstanceOf[PacketKey]
  }
}

class PacketKeySerde extends Serde[PacketKey] {
  val serializer = new PacketKeySerializer
  val deserializer = new PacketKeyDeserializer

  def configure(configs: util.Map[String, _], isKey: Boolean): Unit = {
    serializer.configure(configs, isKey)
    deserializer.configure(configs, isKey)
  }
  def close(): Unit = {
    serializer.close()
    deserializer.close()
  }
} 

//设置package包名称以及导入依赖的类
package com.wegtam.amws.reports.adt

import java.io.ByteArrayInputStream
import java.nio.charset.StandardCharsets
import java.time.OffsetDateTime

import com.wegtam.amws.reports.ReportType
import eu.cdevreeze.yaidom.parse.DocumentParserUsingSax
import eu.cdevreeze.yaidom.queryapi.HasENameApi.withLocalName
import eu.cdevreeze.yaidom.simple.Elem

import scala.collection.immutable.Seq
import scala.util.Try


  def fromXmlGetReportListResponse(s: String): Seq[ReportInfo] = {
    val eo: Option[Elem] = for {
      p <- Try(DocumentParserUsingSax.newInstance()).toOption
      d <- Try(p.parse(new ByteArrayInputStream(s.getBytes(StandardCharsets.UTF_8)))).toOption
      e <- d.documentElement.findChildElem(withLocalName("GetReportListResult"))
    } yield e
    eo.fold(Seq.empty[ReportInfo]) { e =>
      val it = for {
        c <- e.filterChildElems(withLocalName("ReportInfo"))
        r <- fromXmlElement(c)
      } yield r
      it
    }
  }

} 

//设置package包名称以及导入依赖的类
package kokellab.utils.misc

import java.io.{ByteArrayInputStream, ByteArrayOutputStream}

import com.google.zxing.BarcodeFormat
import org.scalacheck.Gen
import org.scalatest.prop.{GeneratorDrivenPropertyChecks, PropertyChecks}
import org.scalatest.{Matchers, PropSpec}

class BarcoderTest extends PropSpec with GeneratorDrivenPropertyChecks with Matchers {

		def fakeEncodeDecode(text: String, barcodeFormat: BarcodeFormat, dimensions: (Int, Int), imageFormat: String): String =
			if (text.isEmpty) text else encodeDecode(text.toUpperCase, barcodeFormat, dimensions, imageFormat)

		def genBoundedList[T](maxSize: Int, gen: Gen[T]): Gen[List[T]] =
			Gen.choose(0, maxSize) flatMap (size => Gen.listOfN(size, gen))

		def genBoundedString(maxSize: Int, gen: Gen[Char]): Gen[String] =
			Gen.choose(0, maxSize) flatMap (size => Gen.listOfN(size, gen) map (_.mkString))

		def encodeDecode(text: String, codeFormat: BarcodeFormat, dimensions: (Int, Int), imageFormat: String): String = {
			val barcoder = new Barcoder(codeFormat, imageFormat, dimensions._1, dimensions._2)
			val os = new ByteArrayOutputStream()
			barcoder.encode(text, os)
			val is = new ByteArrayInputStream(os.toByteArray)
			barcoder.decode(is)
		}

		val imageFormatGen = Gen.oneOf("png", "jpg", "gif")
		def test(barcodeFormat: BarcodeFormat, dimensionsGen: Gen[(Int, Int)], stringGen: Gen[String]) = {
			property(s"Decoding an encoded string should yield the original string for ${barcodeFormat.name} codes") {
				forAll(imageFormatGen, stringGen, dimensionsGen) { (imageFormat: String, text: String, dimensions: (Int, Int)) =>
					fakeEncodeDecode(text, barcodeFormat, dimensions, imageFormat) should equal (text.toUpperCase)
				}
			}
		}

		val rectangularGen: Gen[(Int, Int)] = for {
			width <- Gen.choose(20, 100)
			height <- Gen.choose(20, 100)
		} yield (width, height)

		val squareGen: Gen[(Int, Int)] = for {
			size <- Gen.choose(20, 100)
		} yield (size, size)

		val code39And93Gen: Gen[String] = genBoundedString(48, Gen.frequency((36, Gen.alphaNumChar), (7, Gen.oneOf('-', '.', '$', '/', '+', '%', ' '))))
		test(BarcodeFormat.CODE_39, rectangularGen, code39And93Gen)
		test(BarcodeFormat.CODE_93, rectangularGen, code39And93Gen)

		// TODO this fails due to https://github.com/zxing/zxing/issues/716
		// there's nothing I can do now
//		test(BarcodeFormat.CODE_128, rectangularGen, genBoundedString(48, Gen.choose[Char](0x20, 127)))

		// TODO QR codes break; also not my fault
//		test(BarcodeFormat.QR_CODE, squareGen, genBoundedString(4296, Gen.frequency((36, Gen.alphaNumChar), (8, Gen.oneOf('-', '.', '$', '/', '+', '%', ' ', ':')))))

} 

//设置package包名称以及导入依赖的类
package kokellab.utils.misc

import java.io.ByteArrayInputStream
import java.nio.file.Paths

import org.scalacheck.Arbitrary
import org.scalacheck.Gen
import org.scalatest.prop.{GeneratorDrivenPropertyChecks, PropertyChecks}
import org.scalatest.{Matchers, PropSpec}

import scala.io.Source

class FileHasherTest extends PropSpec with GeneratorDrivenPropertyChecks with Matchers {

	val hasher = new FileHasher("SHA-256")
	println(hasher.hash("aef46c1e5d854fcfd20ab427b599c86102b2fc6252fcdfe9be84dafdb58d6ffb".getBytes()))

	val anyByteArray: Gen[List[Byte]] = Gen.listOf(Arbitrary.arbByte.arbitrary)
	property("Correct hashes should validate") {
		forAll(anyByteArray) { (array: List[Byte]) =>
			val hex = hasher.hash(array)
			hasher.validate(array, hex)
		}
	}

	property("Incorrect hashes should not validate") {
		// this is a bit weird, but if we generated a byte array and a string hash hex, we couldn't guarantee that the hashes don't match
		forAll(anyByteArray, anyByteArray) { (arrayA: List[Byte], arrayB: List[Byte]) =>
			if (arrayA != arrayB) {
				a [ValidationFailedException] should be thrownBy {
					val hexB = hasher.hash(arrayB)
					hasher.validate(arrayA, hexB)
				}
			}
		}
	}

	property("Can hash from stream") {
		forAll(anyByteArray) { (array: List[Byte]) =>
			val hex = hasher.hash(new ByteArrayInputStream(array.toArray))
			hasher.validate(new ByteArrayInputStream(array.toArray), hex)
		}
	}

	property("Can hash from file") {
		val archive = Paths.get(this.getClass.getResource("test.7z").toURI)
		val file = Paths.get(this.getClass.getResource("expected_file_to_hash.txt").toURI)
		hasher.hash(archive) should equal ("afa411f16a1e7943cb07a57516c593384c097e8521f840b2112d2680877a2b04")
		hasher.hash(file) should equal ("aef46c1e5d854fcfd20ab427b599c86102b2fc6252fcdfe9be84dafdb58d6ffb")
	}

} 

//设置package包名称以及导入依赖的类
import java.io.{ByteArrayOutputStream, ByteArrayInputStream}

import org.scalatest._

class HandlerSpec extends FreeSpec with Matchers {

  "derived class should delegate and write result to output stream" - {
    val input = new ByteArrayInputStream("""{"input": 3}""".getBytes("UTF-8"))
    val output = new ByteArrayOutputStream()
    new MockHandler().handler(input, output)
    output.toString("UTF-8") shouldBe """{"output":6}"""
  }

}

class MockHandler extends Handler(MockHandler.double)

object MockHandler {
  case class TestInput(input: Int)
  case class TestOutput(output: Int)
  def double(input: TestInput): TestOutput = TestOutput(input.input * 2)
} 

//设置package包名称以及导入依赖的类
package uk.pkerrigan.dmarcparser

import java.io.ByteArrayInputStream
import java.nio.charset.CodingErrorAction
import java.util.Properties
import java.util.zip.{GZIPInputStream, ZipInputStream}
import javax.activation.DataSource
import javax.mail.Session
import javax.mail.internet.MimeMessage
import scala.collection.JavaConverters._

import org.apache.commons.mail.util.MimeMessageParser
import uk.pkerrigan.dmarcparser.report.Feedback

import scala.io._

class EmailParser(parser: ParserTrait = new Parser()) extends EmailParserTrait{

  implicit val codec = Codec("UTF-8")
  codec.onMalformedInput(CodingErrorAction.REPLACE)
  codec.onUnmappableCharacter(CodingErrorAction.REPLACE)

  def parseEmail(email: String): Option[Feedback] = {
    val s = Session.getDefaultInstance(new Properties())
    val is = new ByteArrayInputStream(email.getBytes)
    val message = new MimeMessage(s, is)
    val messageParser = new MimeMessageParser(message).parse()

    messageParser.getAttachmentList.asScala.headOption.flatMap(extract)
  }

  private def extract(a: DataSource): Option[Feedback] = a match {
    case `a` if a.getContentType.equals("application/gzip") => extractGzip(a)
    case `a` if a.getContentType.equals("application/x-gzip") => extractGzip(a)
    case `a` if a.getContentType.equals("application/zip") => extractZip(a)
    case `a` if a.getContentType.equals("application/x-zip-compressed") => extractZip(a)
    case _ => None
  }

  private def extractZip(a: DataSource): Option[Feedback] = {
    val zip = new ZipInputStream(a.getInputStream)
    zip.getNextEntry
    val rawXml = Source.fromInputStream(zip).mkString
    if (rawXml == "") None else Some(parser.parse(rawXml))
  }
  private def extractGzip(a: DataSource): Option[Feedback] = {
    val zip = new GZIPInputStream(a.getInputStream)
    val rawXml = Source.fromInputStream(zip).mkString
    if (rawXml == "") None else Some(parser.parse(rawXml))
  }
} 

//设置package包名称以及导入依赖的类
package serialisation

import java.io.{ByteArrayInputStream, ObjectInputStream}
import java.util

import org.apache.kafka.common.serialization.Deserializer

import scala.collection.mutable

class SeqDeserialiser[ELEMENT] extends Deserializer[Seq[ELEMENT]] {

  override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = {
  }

  override def deserialize(topic: String, data: Array[Byte]): Seq[ELEMENT] = {
    val byteStream = new ByteArrayInputStream(data)
    val objectStream = new ObjectInputStream(byteStream)
    val result = mutable.Seq[ELEMENT]()
    while (objectStream.available() > 0) {
      result :+ objectStream.readObject().asInstanceOf[ELEMENT]
    }
    objectStream.close()
    result
  }

  override def close(): Unit = {
  }
} 

//设置package包名称以及导入依赖的类
package korolev.server

import java.io.{ByteArrayInputStream, InputStream}
import java.nio.charset.StandardCharsets


sealed trait Response

object Response {

  case class Http(status: Status,
                  body: Option[InputStream] = None,
                  headers: Seq[(String, String)] = Seq.empty)
      extends Response

  object Http {
    def apply(status: Status, message: String): Http = {
      val bytes = message.getBytes(StandardCharsets.UTF_8)
      val body = new ByteArrayInputStream(bytes)
      Http(status, Some(body))
    }
  }

  case class WebSocket(publish: String => Unit,
                       subscribe: (String => Unit) => Unit,
                       destroyHandler: () => Unit)
      extends Response

  sealed trait Status {
    def code: Int
    def phrase: String
  }

  object Status {
    case object Ok extends Status {
      val code = 200
      val phrase = "OK"
    }
    case object BadRequest extends Status {
      val code = 400
      val phrase = "Bad Request"
    }
    case object Gone extends Status {
      val code = 410
      val phrase = "Gone"
    }
  }
} 

//设置package包名称以及导入依赖的类
package edu.goldlok.minio_scala.s3v4

import java.io.ByteArrayInputStream
import java.nio.charset.StandardCharsets

import akka.stream.scaladsl.Source
import akka.util.ByteString
import com.squareup.okhttp.mockwebserver.{MockResponse, MockWebServer}
import edu.goldlok.minio_scala.mio.MioClient
import org.scalatest.mockito.MockitoSugar
import org.scalatest.{FlatSpec, Matchers}

import scala.concurrent.Await


class TestPutObject extends FlatSpec with Matchers with MockitoSugar {
  import edu.goldlok.minio_scala.s3v4.TestElem._

  private def obtPutObjectServer(): MockWebServer = {
    val server = new MockWebServer()
    val response = new MockResponse()
    response.addHeader("Date", MON_29_JUN_2015_22_01_10_GMT)
    response.addHeader(LAST_MODIFIED, MON_04_MAY_2015_07_58_51_GMT)
    response.addHeader("ETag", MD5_HASH_STRING)
    response.setResponseCode(200)
    server.enqueue(response)
    server.start()
    server
  }

  private def testPutObject() =  {
    val server = obtPutObjectServer()
    val source = Source.single(ByteString(HELLO_WORLD))
    val response = MioClient(server.getHostName, server.getPort, keys).putObject(BUCKET, OBJECT, source, HELLO_WORLD.length)
    val uploadResult = Await.result(response, timeout)
    uploadResult.isSuccess should be (true)
    server.shutdown()
  }

  private def testPutObjectByStream() = {
    val server = obtPutObjectServer()
    val stream = new ByteArrayInputStream(HELLO_WORLD.getBytes(StandardCharsets.UTF_8))
    val response = MioClient(server.getHostName, server.getPort, keys).putObject(BUCKET, OBJECT, stream, HELLO_WORLD.length)
    val uploadResult = Await.result(response, timeout)
    uploadResult.isSuccess should be (true)
    server.shutdown()
  }

  "put object " should "return" in {
    testPutObject()
    testPutObjectByStream()
  }
} 

//设置package包名称以及导入依赖的类
import com.datastax.spark.connector._
import play.api.libs.json.Json
import play.api.libs.json._
import java.io.{ByteArrayOutputStream, ByteArrayInputStream}
import java.util.zip.{GZIPOutputStream, GZIPInputStream}
import PreProcessingConfig._

case class objdump_results_by_service_name_class(service_name: String, sha256: String)
case class objdump_results_by_sha256_class(sha256: String, service_name: String, results: Array[Byte])
case class objdump_join_results_class(sha256: String, service_name: String, results: String)
case class objdump_binaray_final_array_rdd_class(sha256: String, array_results: Array[Double])
 
val objdump_main_list = sc.textFile(objdump_x86Opcodes_file).collect.toList
def unzip(x: Array[Byte]) : String = {		
    val inputStream = new GZIPInputStream(new ByteArrayInputStream(x))
    val output = scala.io.Source.fromInputStream(inputStream).mkString
    return output
}
def combineAllObjdumpInOne( malwarelist :Seq[play.api.libs.json.JsValue]) : List[String] ={
    if (malwarelist(0).toString() == "null") return List("null")
    var begin = malwarelist(0).as[List[String]]
    for (i <- 1 to (malwarelist.size-1)){
        if (malwarelist(i).toString() == "null") begin = begin
        else begin = begin ::: malwarelist(i).as[List[String]]
    }
    return  begin
}
def convertToList( malwarelist :Seq[play.api.libs.json.JsValue]) : List[String] = {
    if (malwarelist(0).toString() == "null") return List("null")
    else {
        return malwarelist(0).as[List[String]]
    } 
    
}
def findAllBininobjdump_main_list(malware :List[String]) : Array[Double] ={
    if (malware == List("null")) return (List.fill(10000)(0.0)).toArray
    else {
        val forlist = for ( one  <- malware ) yield {
            objdump_main_list.indexOf(one) + 1.0
        }
        if (forlist.size < 10000){
            return  (List.concat(forlist,List.fill(10000-forlist.size)(0.0))).toArray
        }
        else return forlist.toArray
    }
}

val objdump_results_by_service_name_meta = sc.cassandraTable[objdump_results_by_service_name_class](keyspace,service_name_table).where("service_name=?","objdump")
val objdump_results_by_service_name_rdd = objdump_results_by_service_name_meta.keyBy(x=> (x.sha256,x.service_name))
val objdump_results_by_sha256_meta = sc.cassandraTable[objdump_results_by_sha256_class](keyspace,sha256_table)
val objdump_results_by_sha256_rdd = objdump_results_by_sha256_meta.keyBy(x => (x.sha256,x.service_name))
val objdump_join_results = objdump_results_by_service_name_rdd.join(objdump_results_by_sha256_rdd).map(x=> (new objdump_join_results_class(x._1._1,x._1._2, unzip(x._2._2.results)))).distinct()
val objdump_binaray_final_array_rdd = objdump_join_results.map(x=>(x.sha256,(Json.parse(x.results) \\ "opcodes"))).filter(x=> (x._2.size > 0)).map(x=>(x._1,if ( x._2.size == 1 ) convertToList(x._2) else combineAllObjdumpInOne(x._2))).map(x=>(x._1,findAllBininobjdump_main_list(x._2)))
objdump_binaray_final_array_rdd.toDF().write.format("parquet").save(objdump_binaray_final_array_file) 

//设置package包名称以及导入依赖的类
package knot.data.serialization.j

import java.io.{ByteArrayInputStream, InputStream, ObjectInputStream}

import knot.data.serialization.{Decoder, Deserializer, DeserializerFactory}

import scala.reflect.runtime.universe

class JavaDecoder(in: InputStream) extends ObjectInputStream(in) with Decoder{
}

class JavaDeserializerFactory extends DeserializerFactory[JavaDecoder] {

  override def get[T: universe.TypeTag](): Deserializer[JavaDecoder, T] = {
    new Deserializer[JavaDecoder, T] {
      override def deserialize(bytes: Array[Byte]): T = {
        val jd = new JavaDecoder(new ByteArrayInputStream(bytes))
        try {
          deserialize(jd)
        } finally {
          jd.close()
        }
      }

      override def deserialize(decoder: JavaDecoder): T = {
        decoder.readObject().asInstanceOf[T]
      }
    }
  }

  override def register[T: universe.TypeTag](): Unit = {
  }

  override def register[T: universe.TypeTag](deser: Deserializer[JavaDecoder, T]): Unit = {
  }
} 

//设置package包名称以及导入依赖的类
package pulse.kafka.avro

import java.io.{ByteArrayInputStream, ByteArrayOutputStream, DataInputStream, File}

import com.twitter.util.Future
import org.apache.avro.Schema
import org.apache.avro.file.DataFileWriter
import org.apache.avro.generic.{GenericDatumReader, GenericDatumWriter, GenericRecord}
import org.apache.avro.io.DecoderFactory
import pulse.kafka.extensions.managedByteArrayInputStream
import pulse.kafka.extensions.managedByteArrayOutputStream
import pulse.kafka.extensions.catsStdInstancesForFuture
import scala.concurrent.ExecutionContext.Implicits._

object AvroUtils {

  import pulse.common.syntax._


  def jsonToAvroBytes(json: String, schemaFile: File): Future[Array[Byte]] =
    use(new ByteArrayOutputStream()) { output =>
      for {
        s <- loadSchema(schemaFile)
        _ <- convertImpl(json, output, s)
      } yield output.toByteArray
    }

  private def convertImpl(json: String, output: ByteArrayOutputStream, schemaSpec: Schema): Future[GenericDatumReader[GenericRecord]] =

    use(new ByteArrayInputStream(json.getBytes)) { input =>
      for {
        w <- getWriter(output, schemaSpec)
        r <- getReader(input, schemaSpec, w)
      } yield r
    }

  def getReader(input: ByteArrayInputStream, schemaSpec: Schema, w: DataFileWriter[GenericRecord]) = Future.value {
    val reader = new GenericDatumReader[GenericRecord](schemaSpec)
    val datum = reader.read(null, getJsonDecoder(input, schemaSpec))
    w.append(datum)
    w.flush()
    reader
  }

  private def getJsonDecoder(input: ByteArrayInputStream, schema: Schema) =
    DecoderFactory.get.jsonDecoder(schema, new DataInputStream(input))

  private def getWriter(output: ByteArrayOutputStream, schemaSpec: Schema) = {
    Future.value {
      val writer = new DataFileWriter[GenericRecord](new GenericDatumWriter[GenericRecord]())
      writer.create(schemaSpec, output)
    }
  }

  private def loadSchema(schemaFile: File): Future[Schema] =
    Future {
      new Schema.Parser().parse(schemaFile)
    }
} 

//设置package包名称以及导入依赖的类
package pulse.kafka.extensions

import java.io.{ByteArrayInputStream, ByteArrayOutputStream}

import pulse.common.Managed
import pulse.kafka.client.producer.EnvelopeProducer

trait Resources {
  implicit def managedByteArrayOutputStream = new Managed[ByteArrayOutputStream] {
    override def close(instance: ByteArrayOutputStream): Unit = instance.close()
  }

  implicit def managedEnvelopeProducer = new Managed[EnvelopeProducer] {
    override def close(instance: EnvelopeProducer): Unit = instance.close()
  }

  implicit def managedByteArrayInputStream = new Managed[ByteArrayInputStream] {
    override def close(instance: ByteArrayInputStream): Unit = instance.close()
  }
} 

//设置package包名称以及导入依赖的类
package com.dataoptimo.imgprocessing
import org.opencv.core.Mat
import javax.imageio.ImageIO
import java.io.ByteArrayInputStream
import java.awt.image.DataBufferByte
import org.opencv.core.CvType
import org.opencv.core.Core
import org.opencv.features2d.FeatureDetector
import org.opencv.core.MatOfKeyPoint
import org.opencv.features2d.Features2d
import org.opencv.highgui.Highgui
import java.awt.image.BufferedImage
import java.io.File



object OpenCVOps {
  
  System.loadLibrary(Core.NATIVE_LIBRARY_NAME);
  
  def imageToMat(byteArray: Array[Byte]): MatSer ={
      val bufferedImage = ImageIO.read(new ByteArrayInputStream(byteArray))
      val mat = new Mat(bufferedImage.getHeight(), bufferedImage.getWidth(), CvType.CV_8UC3);
      val data = bufferedImage.getRaster().getDataBuffer.asInstanceOf[DataBufferByte].getData();
      mat.put(0, 0, data);
      new MatSer(mat);
  }
  
  def matToImage(mat: Mat,file: String): Boolean ={
    
      val imageType = BufferedImage.TYPE_3BYTE_BGR;
      val image = new BufferedImage(mat.cols(),mat.rows(), imageType);
      val x = image.getRaster.getDataBuffer.asInstanceOf[DataBufferByte].getData
      mat.get(0,0,x)
      val fileName = new File("file")
      ImageIO.write(image, "jpg", fileName)
    
  }
  
  def detectFeatures(mat: Mat)
  {
    val featureDetector = FeatureDetector.create(FeatureDetector.SIFT)
    val matKeyPoint = new MatOfKeyPoint()
    featureDetector.detect(mat,matKeyPoint)
    println(mat.get(0, 0))
    println(matKeyPoint.toList())
    //writeToImage(mat,matKeyPoint)
    
  }
  
  def writeToImage(mat: Mat, matKeyPoint: MatOfKeyPoint){
    val outImage = new Mat()
    Features2d.drawKeypoints(mat, matKeyPoint, outImage)
    Highgui.imwrite("myfile.jpg",outImage)
    
  }
  
} 

//设置package包名称以及导入依赖的类
package util

import java.io.ByteArrayInputStream
import java.nio.charset.{Charset, CodingErrorAction}

import text.StringOption

import scala.collection.mutable.ListBuffer
import scala.io.{Codec, Source}
import scala.sys.process.ProcessBuilder


object ProcessBuilderUtils {
  implicit def processToProcessUtils(repr: ProcessBuilder): ProcessBuilderUtils = {
    new ProcessBuilderUtils(repr)
  }
}

class ProcessBuilderUtils(repr: ProcessBuilder) {
  def lineStream(encoding: Charset,
                 onMalformedInput: CodingErrorAction,
                 onUnmappableCharacter: CodingErrorAction,
                 replacementOpt: StringOption): Iterator[String] = {
    val lines: Iterator[String] = repr.lineStream_!.iterator
    val byteBuffer = ListBuffer.empty[Byte]
    while (lines.hasNext) {
      val line: String = lines.next.trim concat "\n"
      byteBuffer ++= line.getBytes
    }
    implicit val codec = Codec(encoding).
      onMalformedInput(onMalformedInput).
      onUnmappableCharacter(onUnmappableCharacter)
    if (replacementOpt.nonEmpty) {
      codec.decodingReplaceWith(replacementOpt.get)
    }
    Source.fromInputStream(new ByteArrayInputStream(byteBuffer.toArray)).getLines
  }
} 

//设置package包名称以及导入依赖的类
package aws.s3

import java.io.ByteArrayInputStream
import com.amazonaws.services.s3.AmazonS3Client
import com.amazonaws.services.s3.model.{AmazonS3Exception, ListObjectsV2Request, ObjectMetadata}
import org.apache.commons.compress.utils.IOUtils

import scala.collection.JavaConverters._
import play.api.Logger

case class S3FileDetails(contents: Array[Byte], key: String, bucket: String)

class AmazonS3ClientWrapper(client: AmazonS3Client) {

  def uploadFile(fileDetails: S3FileDetails): Either[String, String] = {
    val stream = new ByteArrayInputStream(fileDetails.contents)
    try {
      val meta = new ObjectMetadata()
      meta.setContentLength(fileDetails.contents.length)
      client.putObject(fileDetails.bucket, fileDetails.key, stream, meta)
      Logger.info(s"Uploaded file to S3: ${fileDetails.bucket} - ${fileDetails.key}")
      Right(client.getResourceUrl(fileDetails.bucket, fileDetails.key))
    } catch {
      case e: AmazonS3Exception =>
        Left(s"Failed to upload to aws.s3 with error: ${e.getMessage} for file: ${fileDetails.key} ")
    } finally {
      IOUtils.closeQuietly(stream)
    }
  }

  def downloadFile(bucket: String, key: String): Either[String, Array[Byte]] = {
    try {
      val obj    = client.getObject(bucket, key)
      val stream = obj.getObjectContent
      try {
        Right(IOUtils.toByteArray(stream))
      } finally {
        stream.close()
      }
    } catch {
      case e: AmazonS3Exception =>
        // either the object does not exist or something went really wrong
        Logger.warn(s"Failed to download aws.s3://$bucket/$key", e)
        Left(s"Failed to download s3://$bucket/$key, with status code ${e.getStatusCode}")

    }
  }

  // Returns keys of all the files in specified s3 bucket with the given prefix
  def listFiles(bucket: String, prefix: String): Either[String, Seq[String]] = {
    try {
      val request = new ListObjectsV2Request().withBucketName(bucket).withPrefix(prefix)
      val result  = client.listObjectsV2(request).getObjectSummaries.asScala.map(_.getKey)
      Right(result)
    } catch {
      case e: AmazonS3Exception =>
        Logger.warn(s"Failed to list objects under s3://$bucket/$prefix", e)
        Left(s"Failed to retrieve template files from s3://$bucket/$prefix")
    }
  }
}