package eu.dnetlib.dhp.contextpropagation

import java.util

import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
import eu.dnetlib.dhp.contextpropagation.model.{DatasetPropagationStructure, EnrichedEntries, MapSxOA, Node, PropagationStructure, PropagationUse, RelationPropagation}
import eu.dnetlib.dhp.provision.scholix.{Scholix, ScholixEntityId}
import eu.dnetlib.dhp.provision.scholix.summary.{SchemeValue, ScholixSummary}
import org.apache.spark.sql.{Dataset, Encoder, Encoders, SparkSession}

import scala.collection.mutable.ListBuffer
import scala.collection.JavaConverters._

object PropagationUtils {

  implicit val enrichedEntitiesEncoder: Encoder[EnrichedEntries] = Encoders.kryo[EnrichedEntries]
  implicit val scholixEncoder: Encoder[Scholix] = Encoders.kryo[Scholix]
  implicit val propagationEncoder: Encoder[RelationPropagation] = Encoders.kryo[RelationPropagation]


  def getSelectedNodes(path: String, spark:SparkSession): Dataset[String] ={

    implicit val stringEncoder: Encoder[String] = Encoders.STRING

    val allowedRelations = spark.read.load(path).as[RelationPropagation]


    val pubs_rel : Dataset[RelationPropagation] = allowedRelations
      .filter(r => r.getSource.getId.startsWith("50")
        && r.getTarget.getId.startsWith("60")
        && 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)

    pubs_rel.map(r => r.getSource.getId).union(pubs_rel.map(r => r.getTarget.getId))
      .union(dats_rel.map(r => r.getSource.getId)).union(dats_rel.map(r => r.getTarget.getId)).distinct()
  }


  def getSubjectList(value: util.List[String], scheme: util.List[String]): util.List[SchemeValue] = {
    var subjects = new ListBuffer[SchemeValue]()

    var i = 0
    for (elem <- value.asScala) {
      val sv :SchemeValue = new SchemeValue()
      sv.setScheme(scheme.get(i))
      sv.setValue(elem)
      subjects += sv
      i += 1
    }

    subjects.toList.asJava

  }


  def propagateDataset (item: ((String, PropagationStructure), (String, DatasetPropagationStructure))) : List[(String, PropagationStructure)] = {
    val lst = new ListBuffer[(String,PropagationStructure)]()
    lst += item._1
    if(item._2 != null){

      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 = entry.getWeight.toDouble * dsprob.getPropagation.get(source).getWeight.toDouble
            if (new_p > 0.3) {
              var newentry: PropagationUse = PropagationUse.copyInstance(entry)
              newentry.setWeight(String.valueOf(new_p))
              if(!newentry.getPath.contains(item._1._1))
                newentry.getPath.add(item._1._1)
              if (!dic.keySet.contains(key)) {
                dic.put(key, new util.ArrayList[PropagationUse]())
              }

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

  def enrichScholix(summary_path: String, spark: SparkSession): Dataset[EnrichedEntries] = {

    implicit val summaryEncoder: Encoder[ScholixSummary] = Encoders.kryo[ScholixSummary]

    spark.read.load(summary_path).as[ScholixSummary]
      .map(ss => {
        val ee: EnrichedEntries = new EnrichedEntries()
        ee.setScholixId(ss.getId)
        ee.setTitle(ss.getTitle)
        if (ss.getDescription != null) {
          ee.setDescription(List(ss.getDescription).asJava)
        } else {
          ee.setDescription(new util.ArrayList[String]())
        }
        if (ss.getSubject != null) {
          ee.setSubject(ss.getSubject)
        } else {
          ee.setSubject(new util.ArrayList[SchemeValue]())
        }
        if (ee.getDescription.size() > 0 && ee.getSubject.size() > 0){
          ee
        }
        else{
          null
        }
      })
  }







  def mergeEnrichments(item : ((String, EnrichedEntries), (String, EnrichedEntries)) ): EnrichedEntries = {
    if (item._1 == null)
      item._2._2

    if (item._2 == null)
      item._1._2

    item._2._2.mergeWith(item._1._2)
  }


      /*#reads the scholixexplorer scholix dump. It filters out records with prefix 70 (unknown)
input = sc.textFile('/user/dnet.scholexplorer/scholix/provision/scholix_dump/scholix_json').map(json.loads).filter(lambda x: x['source']['dnetIdentifier'][0:2]!='70')
sources = input.map(lambda x: x['source']['dnetIdentifier']).distinct()

#le relazioni sono bidirezionali quindi l'insieme di nodi e' identificato univocamente dai sources
nodes = input.map(lambda x: x['source']['dnetIdentifier']).distinct().map(lambda x: prefix + x[3:] if not 'dedup' in x else prefix + x[17:]).distinct().map(lambda x : {'scholix' : x})

#createa a mapping between the original scholexplorer ids and the ids the records will have in OpenAIRE
scholexplorerMapOpenaire = sources.map(lambda x: {'scholexplorer': x, 'openaire': prefix + x[3:] if 'dedup' not in x else prefix + x[17:]})
scholexplorerMapOpenaire.map(json.dumps).saveAsTextFile(path = '/tmp/miriam/context_propagation/scholexplorerIdsMapOA', compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec")

#reads the summaries (containing title, subject and description ) for the scholexplorer records
summaries = sc.textFile('/user/dnet.scholexplorer/scholix/provision/summary_json').map(json.loads).map(lambda x: {"title": x['title'], "description":[x['description']] if x['description'] is not None else [], "subject":x['subject'], "sid":x['id']})
sources = sources.map(lambda x: {'id':x})

#enriches data with summaries information from scholexplorer
sdf = sources.toDF()
smdf = summaries.toDF()
enriched_sx = sdf.join(smdf, sdf.id == smdf.sid).rdd.map(lambda x: {"id" : x['id'], "abstract": x['description'], "title":x['title'], "subject":x['subject']})
enriched_sx.map(json.dumps).saveAsTextFile(path = '/tmp/miriam/context_propagation/scholexplorerEnrichedSX', compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec")


nodesdf = nodes.toDF()
nodesdf.createOrReplaceTempView("ids")

#associates the dedup openaire identifier to ingested scholix ids
dfr = spark.read.json('/tmp/beta_provision/graph/13_graph_blacklisted/relation')
relation = dfr.createOrReplaceTempView("relation")
mergedIds = spark.sql("SELECT source, target from relation join ids on relation.target = ids.scholix  where datainfo.deletedbyinference = false and relclass = 'merges' ")
mergedIds.rdd.map(lambda x: {'dedup':x['source'], 'scholixId':x['target']}).map(json.dumps).saveAsTextFile(path='/tmp/miriam/context_propagation/scholixIdsMergedInOA', compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec")

#new execution on dfr = spark.read.json('/tmp/miriam/context_propagation/14_graph_blacklisted/relation')
#this will be saved as the one above with the postfix 'production'

#replace the ingested scholix id with the deduped one
mergedIds.createOrReplaceTempView("merged")
changeInMerged = spark.sql("Select * from ids left join merged on ids.scholix = merged.target")
enrich = changeInMerged.rdd.map(lambda x: x['source'] if x['source'] is not None else x['scholix']).distinct().map(lambda x:{"enrich":x})

edf = enrich.toDF()
edf.createOrReplaceTempView("enrich_ids")

ddf = spark.read.json('/tmp/beta_provision/graph/13_graph_blacklisted/dataset')
ddf.createOrReplaceTempView("dataset")

#new execution on ddf = spark.read.json('/tmp/miriam/context_propagation/14_graph_blacklisted/dataset')

#enriches the scholix ingested records with information for title abstract and subject found in OpenAIRE
enriched_dataset = spark.sql("select a.* from (select id, title.value title, description.value description, collect_set(named_struct('scheme', MyS.qualifier.classid, 'value', MyS.value)) as subjects from dataset lateral view explode (subject)s as MyS where datainfo.deletedbyinference = false group by title.value, description.value, id) as a join enrich_ids on a.id = enrich_ids.enrich")

pdf = spark.read.json('/tmp/beta_provision/graph/13_graph_blacklisted/publication')
#new execution on pdf = spark.read.json('/tmp/miriam/context_propagation/14_graph_blacklisted/publication')
pdf.createOrReplaceTempView("publication")
enriched_publication = spark.sql("select a.* from (select id, title.value title, description.value description, collect_set(named_struct('scheme', MyS.qualifier.classid, 'value', MyS.value)) as subjects from publication lateral view explode (subject)s as MyS where datainfo.deletedbyinference = false group by title.value, description.value, id) as a join enrich_ids on a.id = enrich_ids.enrich")

enriched = enriched_dataset.rdd.map(lambda x: {"id":x['id'], 'title':x['title'], 'abstract':x['description'], 'subject':[{'scheme': subject['scheme'], "value": subject['value']} for subject in x['subjects']]}).union(enriched_publication.rdd.map(lambda x: {"id":x['id'], 'title':x['title'], 'abstract':x['description'], 'subject':[{'scheme': subject['scheme'], "value": subject['value']} for subject in x['subjects']]}))
enriched.map(json.dumps).saveAsTextFile(path='/tmp/miriam/context_propagation/scholixIdsEnrichedInOA', compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec")
enriched.toDF().createOrReplaceTempView("enriched")
changeInMerged.createOrReplaceTempView("merged_ids")

#associo gli openaireid eventualmente deduplicati come l'id scholix originale in openaire
scholixIds = spark.sql("select scholix, id, title, abstract, subject from enriched left join merged_ids on id = source").rdd.map(lambda x: {"id":x['scholix'] if x['scholix'] is not None else x['id'], "title":x['title'], "abstract":x['abstract'],'subject':x['subject']})
scholixIds.map(json.dumps).saveAsTextFile(path='/tmp/miriam/context_propagation/scholixIdsEnrichedOA', compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec")
sids=scholixIds.toDF()

sids.createOrReplaceTempView("scholix")
mdf = scholexplorerMapOpenaire.toDF()
mdf.createOrReplaceTempView("map")

#original scholexplorer ids with the enrichment from openaire
scholexplorerEnrichedOA = spark.sql("Select scholexplorer, title, abstract, subject from scholix join map on id = map.openaire")
scholexplorerEnrichedOA = scholexplorerEnrichedOA.rdd.map(lambda x: {'id': x['scholexplorer'], 'title':x['title'], 'abstract':x['abstract'],'subject':x['subject']} )
scholexplorerEnrichedOA.map(json.dumps).saveAsTextFile(path='/tmp/miriam/context_propagation/scholexplorerEnrichedOA', compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec")


sxEnriched = enriched_sx.union(scholexplorerEnrichedOA).map(lambda x: (x['id'],x)).groupByKey().map(groupFunction)

sxEnriched.map(json.dumps).saveAsTextFile(path='/tmp/miriam/context_propagation/scholexplorerEnriched', compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec")

#select the set of relevant relationships
relation = sc.textFile('/user/dnet.scholexplorer/scholix/provision/scholix_dump/scholix_json').map(json.loads).filter(lambda x: x['source']['dnetIdentifier'][0:2]!='70' and x['target']['dnetIdentifier'][0:2] != '70')
pub_rel = relation.filter(lambda x: x['source']['objectType'] == 'publication' and x['target']['objectType'] == 'dataset').map(lambda x: {'source':x['source']['dnetIdentifier'], 'semantics':x['relationship']['name'].lower(), 'target':x['target']['dnetIdentifier']})
#167,210,655
#contare le semantiche nel sottografo identificato dalle publication con relzioni verso dataset
pub_rel.map(lambda x: (x['semantics'], 1)).reduceByKey(lambda a,b: a+b).collect()
#[(u'iscitedby', 11542), (u'reviews', 2051), (u'iscompiledby', 499), (u'unknown', 111706), (u'isnewversionof', 27977), (u'requires', 1), (u'isdocumentedby', 747), (u'describes', 211), (u'issourceof', 30877), (u'ismetadataof', 11), (u'isversionof', 269006), (u'ispartof', 454244), (u'issupplementedby', 1517666), (u'obsoletes', 5), (u'isreferencedby', 89986753), (u'isvariantformof', 3688), (u'hasassociationwith', 30), (u'isidenticalto', 293876), (u'haspart', 621177), (u'ismetadatafor', 121), (u'isrelatedto', 70310923), (u'issupplementto', 85460), (u'isoriginalformof', 476), (u'iscontinuedby', 356407), (u'cites', 200336), (u'ispreviousversionof', 24119), (u'hasversion', 273427), (u'isdescribedby', 5), (u'continues', 356582), (u'isreviewedby', 53), (u'documents', 265636), (u'compiles', 177), (u'references', 2004247), (u'isobsoletedby', 2), (u'isderivedfrom', 617)]

pub_dats_sem = {'issupplementedby':{'use':'latent', 'weight':1.0, 'path':set()}, 'cites':{'use':'reuse', 'weight':1.0, 'path':set()}, 'describes':{'use':'latent', 'weight':1.0, 'path':set()},'references':{'use':'reuse', 'weight':1.0, 'path':set()}, 'documents':{'use':'latent','weight':1, 'path':set()}}
pub_rel_subset = pub_rel.filter(lambda x: x['semantics'] in pub_dats_sem)
pub_rel_subset.count()
#3,988,096

pubdf = pub_rel_subset.toDF()
sxdf = sxEnriched.toDF()

pubs_enriched = pubdf.join(sxdf, pubdf.source == sxdf.id)


pubs_with_abst = pubs_enriched.rdd.filter(lambda x: x['abstract'] != [] or x['subject'] != []).map(lambda x: {'source':x['source'], 'semantics':x['semantics'], 'target': x['target']})

pubs_with_abst.count()
*/





}
