package eu.dnetlib.dhp.contextpropagation

import eu.dnetlib.dhp.application.ArgumentApplicationParser
import eu.dnetlib.dhp.contextpropagation.model.{DatasetPropagationStructure, EnrichedEntries, MapSxOA, Node, PropagationStructure, PropagationUse, RelationPropagation}
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 the summaries of the scholexplorer nodes involved in propagation
  */
object SparkEnrichScholixStep1 {

  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)

  def getEnrichedSubset(scholixSelectedRelationPath:String, summaryPath: String,  spark:SparkSession): Dataset[EnrichedEntries] = {

    //selects the scholix nodes involved in propagation
    val distinctNodes: Dataset[String] = PropagationUtils.getSelectedNodes(scholixSelectedRelationPath , spark)

    val scholixSummaries = PropagationUtils.enrichScholix(summaryPath, spark).filter(o => o != null)
      .map(e => (e.getScholixId, e))

    //enriches the selected nodes with summary from scholexplorer
    distinctNodes.joinWith(scholixSummaries, distinctNodes("value").equalTo(scholixSummaries("_1"))).map(pair => pair._2._2).filter(o => o != null)


  }


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


    getEnrichedSubset(parser.get("inputPath"), parser.get("scholixSummaryPath"), spark)
      .write.mode(SaveMode.Overwrite).save(parser.get("outputPath"))
  }
}
