package eu.dnetlib.dhp.contextpropagation

import java.util

import com.fasterxml.jackson.databind.ObjectMapper
import eu.dnetlib.dhp.application.ArgumentApplicationParser
import eu.dnetlib.dhp.provision.SparkConvertDatasetToJson
import eu.dnetlib.dhp.provision.scholix.{Scholix, ScholixEntityId}
import eu.dnetlib.dhp.provision.scholix.summary.ScholixSummary
import org.apache.commons.io.IOUtils
import org.apache.hadoop.io.compress.GzipCodec
import org.apache.spark.SparkConf
import org.apache.spark.sql.{Dataset, Encoder, Encoders, SparkSession}
import scala.collection.JavaConverters._

object SparkContextPropagation {


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

    implicit val summaryEncoder: Encoder[ScholixSummary] = Encoders.kryo[ScholixSummary]
    implicit val scholixEncoder: Encoder[Scholix] = Encoders.kryo[Scholix]
    implicit val propagationEncoder: Encoder[RelationPropagation] = Encoders.kryo[RelationPropagation]
    implicit val mapEncoderPub: Encoder[PropagationStructure] = Encoders.kryo[PropagationStructure]
    implicit val mapEncoderDats: Encoder[DatasetPropagationStructure] = Encoders.kryo[DatasetPropagationStructure]
    implicit val tupleForPropagation: Encoder[(String, PropagationStructure)] = Encoders.tuple(Encoders.STRING, mapEncoderPub)
    implicit val tupleForPropagationDars: Encoder[(String, DatasetPropagationStructure)] = Encoders.tuple(Encoders.STRING, mapEncoderDats)
    implicit val stringEncoder: Encoder[String] = Encoders.STRING

    val workingPath = parser.get("workingPath")

    def getPublisherList(item: List[ScholixEntityId]) : util.List[String] =
    {

      item.map(entry => entry.getName).asJava
    }

    def propagateDataset (item: ((String, PropagationStructure), (String, DatasetPropagationStructure))) : (String, PropagationStructure) = {
      val propagation = item._1._2.getPropagation.asScala
      val dsprob : DatasetPropagationStructure= item._2._2
      val source = dsprob.getPropagation.keySet().iterator().next()
      val dic = new scala.collection.mutable.HashMap[String, util.List[PropagationUse]]

      propagation.keysIterator.foreach(key  => {
        val entries = propagation.get(key).get.asScala

        entries.foreach(entry => {
          if((entry.getUse == dsprob.getPropagation.get(source).getUse || dsprob.getPropagation.get(source).getUse == "proxy")
            && !entry.getPath.contains(source)) {
            var new_p = Integer.valueOf(entry.getWeight) * Integer.valueOf(dsprob.getPropagation.get(source).getWeight)
            if (new_p > 0.3){
              var newentry : PropagationUse =  PropagationUse.copyInstance(entry)
              newentry.setWeight(String.valueOf(new_p))
              newentry.getPath.add(source)

              dic(key).add(newentry)
            }
          }
        })
      })
      var ps: PropagationStructure = new PropagationStructure
      ps.setPropagation(dic.asJava)
      (source, ps)

    }

    spark.read.load(s"$workingPath/summary").as[ScholixSummary]
      .map(s => new ObjectMapper().writeValueAsString(s))(Encoders.STRING)
      .rdd.repartition(500).saveAsTextFile(s"$workingPath/summary_json", classOf[GzipCodec])

    val allowedRelations : Dataset[RelationPropagation] = spark.read.load(s"$workingPath/scholix").as[Scholix]
      .filter(s => !s.getSource().getDnetIdentifier().substring(0,2).equals("70") )
      .filter(s => !s.getTarget().getDnetIdentifier().substring(0,2).equals("70"))
      .map(s => {
        val rp = new RelationPropagation
        rp.setSource(Node.newInstance(s.getSource.getDnetIdentifier, getPublisherList(s.getSource.getPublisher.asScala.toList)))
        rp.setTarget(Node.newInstance(s.getTarget.getDnetIdentifier, getPublisherList(s.getTarget.getPublisher.asScala.toList)))
        rp.setSemantics(s.getRelationship.getName)
        rp
      })

    val pubs_rel : Dataset[RelationPropagation] = allowedRelations
      .filter(r => r.getSource.getId.substring(0,2) == "50"
        && r.getTarget.getId.substring(0,2) == "60"
        && Costants.containedInPubSem(r.getSemantics))
    val dats_rel : Dataset[RelationPropagation] = allowedRelations
      .filter(r => r.getSource.getId.substring(0,2) == "60"
        && r.getTarget.getId.substring(0,2) == "60"
        && Costants.containedInDatsSem(r.getSemantics)
        && r.getSource.getId != r.getTarget.getId)

    val publication_dataset : Dataset[(String, PropagationStructure)] = pubs_rel.map(r => {
      val ps = new PropagationStructure

      val pv : List[PropagationUse] = List(PropagationUse.copyInstance(Costants.getPublicationValue(r.getSemantics)))
      ps.add(r.getSource.getId, pv.asJava)
      (r.getTarget.getId, ps)

    })

    val dataset_dataset : Dataset[(String, DatasetPropagationStructure)] = dats_rel.map(r => {
      val ps = new DatasetPropagationStructure

      ps.add(r.getTarget.getId, PropagationUse.copyInstance(Costants.getDatasetValue(r.getSemantics)))
      (r.getSource.getId, ps)

    })


    val pl1 : Dataset[(String, PropagationStructure)] = publication_dataset.groupByKey(_._1)
      .agg(PropagationAggregator.getDatasetAggregator().toColumn)


    val pl2_step1 : Dataset [(String, PropagationStructure)] = pl1.joinWith(dataset_dataset, pl1("_1").equalTo(dataset_dataset("_1")))
      .map(propagateDataset)

    val pl2 : Dataset [(String, PropagationStructure)] = pl1.union(pl2_step1).groupByKey(_._1)
      .agg(PropagationAggregator.getDatasetAggregator().toColumn)
  }



}
