package eu.dnetlib.dhp.contextpropagation

import com.fasterxml.jackson.databind.{ObjectMapper, SerializationFeature}
import eu.dnetlib.dhp.contextpropagation.model.{EnrichedEntries, RelationPropagation}
import eu.dnetlib.dhp.provision.scholix.Scholix
import eu.dnetlib.dhp.provision.scholix.summary.ScholixSummary
import org.apache.spark.SparkConf
import org.apache.spark.sql.{Dataset, Encoder, Encoders, SaveMode, SparkSession}
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.Assertions.{assertFalse, assertNotNull}

class ScholixTest extends java.io.Serializable{


  val m: ObjectMapper = new ObjectMapper()
  m.enable(SerializationFeature.INDENT_OUTPUT)

  implicit val scholixEncoder: Encoder[Scholix] = Encoders.kryo[Scholix]
  implicit val summaryEncoder: Encoder[ScholixSummary] = Encoders.kryo[ScholixSummary]
  implicit val propagationEncoder: Encoder[RelationPropagation] = Encoders.kryo[RelationPropagation]
  implicit val enrichedEncoder: Encoder[EnrichedEntries] = Encoders.kryo[EnrichedEntries]
  implicit val tupleForJoinEncoder: Encoder[(String, EnrichedEntries)] = Encoders.tuple(Encoders.STRING, enrichedEncoder)


  @Test
  def selectScholexplorerRelationTest(): Unit ={
    val sourcePath = getClass.getResource("/eu/dnetlib/dhp/contextpropagation/input/scholix-relations-00000.parquet").getPath

    val conf : SparkConf = new SparkConf()

    val spark: SparkSession = SparkSession.builder().appName("SelectScholixRelationTest").master("local").config(conf).getOrCreate()

    val tmp = SparkSelectScholixRelations.getAllowedRelations(sourcePath, spark)

    tmp.write.mode(SaveMode.Overwrite).save("/tmp/temp")
    assert(tmp.count > 0)

  }


  @Test
  def SelectDistinctIDTest(): Unit ={

    val sourcePath = getClass.getResource("/eu/dnetlib/dhp/contextpropagation/producedInfo/selectedRelations.parquet").getPath

    val conf : SparkConf = new SparkConf()

    val spark: SparkSession = SparkSession.builder().appName("SelectDistinctIdsTest").master("local").config(conf).getOrCreate()

    implicit val propagationEncoder: Encoder[RelationPropagation] = Encoders.kryo[RelationPropagation]

    val allowedRelations = spark.read.load(sourcePath).as[RelationPropagation]

    val numberOfNodes = allowedRelations.map(r => r.getSource.getId)(Encoders.STRING)
      .union(allowedRelations.map(r => r.getTarget.getId)(Encoders.STRING)).count()

    val tmp : Dataset[String]= PropagationUtils.getSelectedNodes(sourcePath, spark)


    assert (numberOfNodes > tmp.count())

  }

    @Test
    def mappingScholixOpenAIRETest(): Unit ={
      val sourcePath = getClass.getResource("/eu/dnetlib/dhp/contextpropagation/input/scholix-relations-00000.parquet").getPath

      implicit val scholixEncoder: Encoder[Scholix] = Encoders.kryo[Scholix]

      //val spark: SparkSession = SparkSession.builder().appName("Test").master("local[*]").config("spark.driver.bindAddress", "127.0.0.1").getOrCreate()
      val spark: SparkSession = SparkSession.builder().appName("Test").master("local").config(new SparkConf()).getOrCreate()

      val tmp = SparkEnrichScholixStep2.getMappingScholexplorerOpenAIRE(sourcePath, spark)


      tmp.filter(e => e.getScholixId.contains("dedup"))
        .foreach(e => assertFalse(!(e.getScholixId.substring(17).equals(e.getOaid.substring(17)))))
      tmp.filter(e => !e.getScholixId.contains("dedup"))
        .foreach(e => assertFalse(!(e.getOaid.substring(17).equals(e.getScholixId.substring(3)))))


    }

  @Test
  def enrichScholixTest():Unit = {
    val summaryPath = getClass.getResource("/eu/dnetlib/dhp/contextpropagation/input/part-00000-summaries.parquet").getPath

    val relationPath = getClass.getResource("/eu/dnetlib/dhp/contextpropagation/producedInfo/selectedRelations.parquet").getPath

    val conf : SparkConf = new SparkConf()


    val spark: SparkSession = SparkSession.builder().config(conf) .appName("Test").master("local").getOrCreate()

    val tmp  = SparkEnrichScholixStep1.getEnrichedSubset(relationPath, summaryPath, spark)

    assert(tmp.count() == 5)

    //tmp.write.mode(SaveMode.Overwrite).save("/tmp/scholixEnriched")
  }


  @Test
  def enrichOpenAIRETest():Unit = {

    val conf : SparkConf = new SparkConf()


    val spark: SparkSession = SparkSession.builder().config(conf) .appName("Test").master("local").getOrCreate()

    val scholixPath = getClass.getResource("/eu/dnetlib/dhp/contextpropagation/input/scholix-relations-00000.parquet").getPath
    val relationPath = getClass.getResource("/eu/dnetlib/dhp/contextpropagation/input/relation.json").getPath
    val resultPath = getClass.getResource("/eu/dnetlib/dhp/contextpropagation/input/result/publication").getPath

    val tmp  = SparkEnrichScholixStep2.getEnrichedSubset(scholixPath, relationPath , resultPath , spark)

    print(tmp.count())

    assert(tmp.count() == 1)

    tmp.write.mode(SaveMode.Overwrite).save("/tmp/openaireEnriched")
  }

  @Test
  def mergeEnrichmentsTest():Unit = {

    val conf : SparkConf = new SparkConf()


    val spark: SparkSession = SparkSession.builder().config(conf) .appName("Test").master("local").getOrCreate()

    val scholixPath = getClass.getResource("/eu/dnetlib/dhp/contextpropagation/producedInfo/scholixEnriched.parquet").getPath
    val resultPath = getClass.getResource("/eu/dnetlib/dhp/contextpropagation/producedInfo/result").getPath

    val tmp  = SparkEnrichScholixStep3.getEnriched(scholixPath,  resultPath , spark)


    assert(tmp.count() == 5)

    tmp.write.mode(SaveMode.Overwrite).save("/tmp/mergedEnriched")

    tmp.foreach(r => print(m.writeValueAsString(r)))
  }
}
