package eu.dnetlib.dhp.contextpropagation

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

import scala.collection.JavaConverters._


class TestProva extends java.io.Serializable{



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




//  @Test
//  def testFunderRelationshipsMapping(): Unit = {
//
//
//    def findInDats(dats: Dataset[(String, DatasetPropagationStructure)], elem:String) : Dataset[(String, DatasetPropagationStructure)] = {
//      dats.filter(dats("_1") === elem)
//    }
//
//
//    val sourcePath = getClass.getResource("/eu/dnetlib/dhp/contextpropagation/part-00000.parquet").getPath
//
//
//    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 spark: SparkSession = SparkSession.builder().appName("Test").master("local[*]").config("spark.driver.bindAddress", "127.0.0.1").getOrCreate()
//
//
//    val ds: Dataset[Scholix] =  spark.read.load(sourcePath).as[Scholix]
//
//    val allowedRelations : Dataset[RelationPropagation] = ds
//      .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
//      })
//
//
//    //println(allowedRelations.count())
//
//    val pubs_rel : Dataset[RelationPropagation] = allowedRelations.filter(r => r.getSource.getId.startsWith("50"))
//        .filter(r => r.getTarget.getId.startsWith("60")).filter(r => Costants.containedInPubSem(r.getSemantics.toLowerCase()))
//
//    val dats_rel : Dataset[RelationPropagation] = allowedRelations
//      .filter(r => r.getSource.getId.startsWith("60")
//        && r.getTarget.getId.startsWith("60")
//        && Costants.containedInDatsSem(r.getSemantics.toLowerCase())
//        && 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.toLowerCase())))
//      ps.add(r.getSource.getId, pv.asJava)
//      (r.getTarget.getId, ps)
//
//    })
//
//
//    val pl1 : Dataset[(String, PropagationStructure)] = publication_dataset.groupByKey(_._1)(Encoders.STRING)
//      .agg(PropagationAggregator.getDatasetAggregator().toColumn)
//
//
//
//
//
//   // print(pl1.count)
//
//    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.toLowerCase())))
//      (r.getSource.getId, ps)
//
//    })
////
////    //pl1.foreach(r => print(m.writeValueAsString(r._1)))
////
////
////
//    val dataset_dataset_modified : Dataset[(String, DatasetPropagationStructure)] =
//      dataset_dataset.map(ds => {
//        if(ds._1 == "60|4b5e9fa8e91b206001589993179f69d1"){
//          ("60|82368200e90cf75c714b58288a371bbe", ds._2)
//        }
//        else{
//          ds
//        }
//      })
////
////   // findInDats(dataset_dataset_modified, "60|82368200e90cf75c714b58288a371bbe").show(false)
////
////
//    val pl2_step1  = pl1.joinWith(dataset_dataset_modified, pl1("value")
//      .equalTo(dataset_dataset_modified("_1")), "left")
//      .flatMap(PropagationUtils.propagateDataset)
//
//
//
//
//    val pl2= pl2_step1.groupByKey(_._1)(Encoders.STRING).agg(PropagationAggregator.getDatasetAggregator().toColumn)
//    print(pl2.count())
//
//
////    pl1.foreach(i=> {
////      if (i._1 =="60|b91b1296e3e37523887c2eaaf3f2e673")
////        print(m.writeValueAsString(i))
////    })
////
////    print(pl1.count)
//
////
//
//   // print(m.writeValueAsString(dsprob.getPropagation.get(source).getUse))
//
////    print(dataset_dataset.map(d => {
////      var found : Boolean = false
////      for (elem <- d._2.getPropagation.keySet().asScala){
////        if (d._2.getPropagation.get(elem).getUse == "proxy"){
////          found = true
////        }
////      }
////      if (found){
////        d
////      }else{
////        null
////      }
////    }).filter(o => o != null).first()._1)
//
//
////    dataset_dataset.foreach(d => {
////
////      for (elem <- d._2.getPropagation.keySet().asScala){
////        if (d._2.getPropagation.get(elem).getUse == "reuse"){
////          print("reuse")
////        }
////      }
////      println()
////    })
//
//  }

}

