package eu.dnetlib.dhp.contextpropagation

import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
import eu.dnetlib.dhp.application.ArgumentApplicationParser
import eu.dnetlib.dhp.contextpropagation.model.{EnrichedEntries, MapSxOA}
import eu.dnetlib.dhp.provision.SparkConvertDatasetToJson
import eu.dnetlib.dhp.provision.scholix.Scholix
import org.apache.commons.io.IOUtils
import org.apache.spark.SparkConf
import org.apache.spark.sql.{Dataset, Encoder, Encoders, SaveMode, SparkSession}

/**
  * it takes enrichment from OpenAIRE. It considers only deduped entities since those not deduped have their
  * enrichment directly from scholexplorer
  *
  * One step for each result type
  */
object SparkEnrichScholixStep2 {

  implicit val enrichedEncoder: Encoder[EnrichedEntries] = Encoders.kryo[EnrichedEntries]
  implicit val scholixEncoder: Encoder[Scholix] = Encoders.kryo[Scholix]
  implicit val tupleForJoinEncoder: Encoder[(String, EnrichedEntries)] = Encoders.tuple(Encoders.STRING, enrichedEncoder)
  implicit val mapEncoder: Encoder[MapSxOA] = Encoders.kryo[MapSxOA]
  implicit val tupleForJoinMap: Encoder[(String, MapSxOA)] = Encoders.tuple(Encoders.STRING, mapEncoder)

  def getMappingScholexplorerOpenAIRE(scolixPath: String, spark:SparkSession): Dataset[MapSxOA] ={
    implicit val scholixEncoder: Encoder[Scholix] = Encoders.kryo[Scholix]
    implicit val mapEncoder: Encoder[MapSxOA] = Encoders.kryo[MapSxOA]

    var prefix = "50|scholix_____::"
    spark.read.load(scolixPath).as[Scholix]
      .map(s => s.getSource.getDnetIdentifier)(Encoders.STRING)
      .filter(id => !id.startsWith("70|"))
      .distinct()
      .map(id => {
        val  map : MapSxOA = new MapSxOA()
        if(id.contains("dedup")){
          map.setOaid(prefix + id.substring(17))
        }else{
          map.setOaid(prefix + id.substring(3))
        }
        map.setScholixId(id)
        map
      })
  }

  def  enrichOpenAIRE(resourcePath: String, relationPath : String,  spark:SparkSession): Dataset[EnrichedEntries] = {

    val mapper = new ObjectMapper()
    mapper.getDeserializationConfig.withFeatures(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)

    val result = spark.read.json(resourcePath)
    val relation = spark.read.json(relationPath)


    result.createOrReplaceTempView("result")
    relation.createOrReplaceTempView("relation")

    spark.sql("SELECT id, target, title.value title, description.value description, subject.value svalues, subject.qualifier.classid sscheme  " +
      "  FROM relation" +
      " JOIN result " +
      "   ON relation.source = result.id " +
      "    WHERE relation.datainfo.deletedbyinference = false " +
      "   AND relclass = 'merges'" +
      "    AND relation.target like '50|scholix%' "

    )
      .map(line => {
        val ee : EnrichedEntries = new EnrichedEntries()
        ee.setOpenAireId(line.getString(1))
        ee.setTitle(line.getList(2))
        ee.setDescription(line.getList(3))
        ee.setSubject(PropagationUtils.getSubjectList(line.getList(4), line.getList(5)))
        ee
      })

  }

  def getEnrichedSubset(scholixPath:String, relationPath:String, resultPath: String,  spark:SparkSession): Dataset[EnrichedEntries] = {

        val openAireInfo = enrichOpenAIRE(resultPath, relationPath,  spark)
          .map(r => (r.getOpenAireId, r))

        val mapping = getMappingScholexplorerOpenAIRE(scholixPath , spark).map(m => (m.getOaid, m))

       mapping.joinWith(openAireInfo, mapping("_1").equalTo(openAireInfo("_1"))).map(t => {
          val ret :EnrichedEntries = t._2._2
          ret.setScholixId(t._1._2.getScholixId)
          ret
        })

  }


  def main(args: Array[String]): Unit = {
    val parser = new ArgumentApplicationParser(IOUtils.toString(SparkEnrichScholixStep2.getClass.getResourceAsStream("/eu/dnetlib/dhp/contextpropagation/enrich-openaire.json")))
    parser.parseArgument(args)
    val conf = new SparkConf
    val spark = SparkSession.builder.config(conf).appName(SparkEnrichScholixStep2.getClass.getSimpleName).master(parser.get("master")).getOrCreate


    getEnrichedSubset(parser.get("scholixPath"), parser.get("relationPath"), parser.get("resultPath"), spark)
      .filter(o => o != null)
      .write.mode(SaveMode.Overwrite).save(parser.get("outputPath"))
  }

}

