package eu.dnetlib.dhp.contextpropagation

import eu.dnetlib.dhp.application.ArgumentApplicationParser

import eu.dnetlib.dhp.contextpropagation.model.{EnrichedEntries, MapSxOA}
import eu.dnetlib.dhp.provision.SparkConvertDatasetToJson
import org.apache.commons.io.IOUtils
import org.apache.spark.SparkConf
import org.apache.spark.sql.{Dataset, Encoder, Encoders, SaveMode, SparkSession}
/**
  * It puts together the outcome of the two previous step to get all the enrichemnts in one single entry
  */
object SparkEnrichScholixStep3 {

  implicit val enrichedEncoder: Encoder[EnrichedEntries] = Encoders.kryo[EnrichedEntries]
  implicit val tupleForJoinEncoder: Encoder[(String, EnrichedEntries)] = Encoders.tuple(Encoders.STRING, enrichedEncoder)


  def getEnriched(scholixPath:String, openairePath: String,  spark:SparkSession): Dataset[EnrichedEntries] = {

    spark.read.load(scholixPath).as[EnrichedEntries]
      .union(spark.read.load(s"$openairePath/publication").as[EnrichedEntries])
      .union(spark.read.load(s"$openairePath/dataset").as[EnrichedEntries])
      .union(spark.read.load(s"$openairePath/software").as[EnrichedEntries])
      .union(spark.read.load(s"$openairePath/otherresearchproduct").as[EnrichedEntries])
      .map(ee => (ee.getScholixId, ee))
      .groupByKey(_._1)(Encoders.STRING)
      .agg(PropagationAggregator.mergeEnrichedEntries().toColumn).map(c => c._2)
  }

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


    getEnriched(parser.get("scholixEnrichedPath"), parser.get("openaireEnrichedPath"), spark)
      .write.mode(SaveMode.Overwrite).save(parser.get("outputPath"))
  }

}

