package eu.dnetlib.dhp.contextpropagation

/**
  * Selects all the Scholexplorer relations not involving nodes with prefix 70 (unknown) and for which the source node
  * is different from the target node
  */

import java.util

import eu.dnetlib.dhp.application.ArgumentApplicationParser
import eu.dnetlib.dhp.contextpropagation.model.{Node, RelationPropagation}
import eu.dnetlib.dhp.provision.SparkConvertDatasetToJson
import eu.dnetlib.dhp.provision.scholix.{Scholix, ScholixEntityId}
import org.apache.commons.io.IOUtils
import org.apache.spark.SparkConf
import org.apache.spark.sql.{Dataset, Encoder, Encoders, SaveMode, SparkSession}
import scala.collection.JavaConverters._

object SparkSelectScholixRelations {

  implicit val scholixEncoder: Encoder[Scholix] = Encoders.kryo[Scholix]

  implicit val propagationEncoder: Encoder[RelationPropagation] = Encoders.kryo[RelationPropagation]


  def getPublisherList(item: List[ScholixEntityId]) : util.List[String] =
  {
    item.map(p=>p.getName).asJava
  }


  def getAllowedRelations(scholixPath: String, spark:SparkSession): Dataset[RelationPropagation] = {
    spark.read.load(scholixPath).as[Scholix]
      .filter(s => !s.getSource().getDnetIdentifier().substring(0,2).equals("70") )
      .filter(s => !s.getTarget().getDnetIdentifier().substring(0,2).equals("70"))
      .filter(s => !s.getSource.getDnetIdentifier.equals(s.getTarget.getDnetIdentifier))
      .map(s => {
        val rp = new RelationPropagation
        if(s.getSource.getPublisher != null)
          rp.setSource(Node.newInstance(s.getSource.getDnetIdentifier, getPublisherList(s.getSource.getPublisher.asScala.toList)))
        else
          rp.setSource(Node.newInstance(s.getSource.getDnetIdentifier, new util.ArrayList()))
        if(s.getTarget.getPublisher != null)
          rp.setTarget(Node.newInstance(s.getTarget.getDnetIdentifier, getPublisherList(s.getTarget.getPublisher.asScala.toList)))
        else
          rp.setTarget(Node.newInstance(s.getTarget.getDnetIdentifier, new util.ArrayList()))
        rp.setSemantics(s.getRelationship.getName.toLowerCase())
        rp
      })
  }

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


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

}
