<commit_before>package de.uni_potsdam.hpi.coheel.programs

import java.io.File
import java.lang.Iterable
import java.net.URI
import java.util.Date

import de.uni_potsdam.hpi.coheel.Params
import de.uni_potsdam.hpi.coheel.datastructures.TrieHit
import de.uni_potsdam.hpi.coheel.debugging.FreeMemory
import de.uni_potsdam.hpi.coheel.io.OutputFiles._
import de.uni_potsdam.hpi.coheel.io.Sample
import de.uni_potsdam.hpi.coheel.ml.CoheelClassifier
import de.uni_potsdam.hpi.coheel.ml.CoheelClassifier.POS_TAG_GROUPS
import de.uni_potsdam.hpi.coheel.programs.DataClasses._
import de.uni_potsdam.hpi.coheel.util.Util
import org.apache.flink.api.common.functions.{Partitioner, RichGroupReduceFunction}
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.{Path, FileSystem}
import org.apache.flink.util.Collector
import weka.classifiers.Classifier
import weka.core.SerializationHelper

import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.io.Source


class DocumentPartitioner extends Partitioner[Int] {
	override def partition(index: Int, numPartitions: Int): Int = {
		index
	}
}

/**
  * Finds entities in texts.
  *
  * IMPORTANT NOTE: For Flink, this requires taskmanager.numberOfTaskSlots
  * to be set to 1. See http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Distribute-DataSet-to-subset-of-nodes-td2814.html,
  * if you need more details.
  */
class ClassificationProgram extends NoParamCoheelProgram with Serializable {

	import CoheelLogger._

	override def getDescription: String = "CohEEL Classification"

	// Select, which neighbours file to use
	val NEIGHBOURS_FILE = fullNeighboursPath
//	val NEIGHBOURS_FILE = reciprocalNeighboursPath

	val neighboursCreationMethod = Map(fullNeighboursPath -> buildFullNeighbours _, reciprocalNeighboursPath -> buildReciprocalNeighbours _)

	override def buildProgram(env: ExecutionEnvironment): Unit = {
		val documentStrings = List(4).map { x =>
			Source.fromFile(s"src/main/resources/classification-documents/$x", "UTF-8").mkString
		}
		val documents = if (runsOffline())
				env.fromElements(Sample.ANGELA_MERKEL_SAMPLE_TEXT_3).name("Documents")
			else {
//				env.fromCollection(documentStrings).name("Documents")
				env.readTextFile(newYorkTimesDataPath).flatMap { l =>
					val split = l.split('\t')
					val text = split(4).replace("\n", " ")

					val words = text.split(' ')
					if (words.length > 60)
						Option(words.slice(0, 60).mkString(" "))
					else
						None
				}
			}

		val inputDocuments = documents
			.first(100)
			.flatMap(new InputDocumentDistributorFlatMap(params, runsOffline())).name("Input-Documents")

		val partitionedDocuments = inputDocuments.partitionCustom(new DocumentPartitioner, "index").name("Partitioned-Documents")

		val classifiables = partitionedDocuments
			.flatMap(new RunTrieOverDocumentsFlatMap(params))
			.name("Possible links")

		// fill the classifiables with all feature information
		val featuresPerGroup = FeatureHelper.buildFeaturesPerGroup(env, classifiables)
		val basicClassifierResults = featuresPerGroup.reduceGroup(new ClassificationReduceGroup(params)).name("Basic Classifier Results")

		val fileExists = if (runsOffline())
			new File(NEIGHBOURS_FILE).exists()
		else {
			val f = FileSystem.get(new URI("hdfs://tenemhead2"))
			f.exists(new Path(NEIGHBOURS_FILE.replace("hdfs://tenemhead2", "")))
		}

		val preprocessedNeighbours = if (fileExists) {
			log.info("Neighbourhood-File exists")
			loadNeighboursFromDisk(env, NEIGHBOURS_FILE)
		} else {
			log.info("Neighbourhood-File does not exist")
			neighboursCreationMethod(NEIGHBOURS_FILE)(env)
		}

		val withNeighbours = basicClassifierResults.join(preprocessedNeighbours)
			.where("candidateEntity")
			.equalTo("entity")
			.name("Join With Neighbours")
			.map { joinResult => joinResult match {
					case (classifierResult, neighbours) =>
						ClassifierResultWithNeighbours(
							classifierResult.documentId,
							classifierResult.classifierType,
							classifierResult.candidateEntity,
							classifierResult.trieHit,
							neighbours.in,
							neighbours.out)
				}
			}

		/*
		 * OUTPUT
		 */
		inputDocuments.filter(_.replication == 0).writeAsTsv(inputDocumentsPath)

		if (fileExists) {
			preprocessedNeighbours.map(serializeNeighboursToString _).writeAsText(NEIGHBOURS_FILE.replace(".wiki", ".wiki2"), FileSystem.WriteMode.OVERWRITE)
		} else {
			preprocessedNeighbours.map(serializeNeighboursToString _).writeAsText(NEIGHBOURS_FILE                           , FileSystem.WriteMode.OVERWRITE)
		}

		// Write trie hits for debugging
		val trieHitOutput = classifiables.map { trieHit =>
			val posTags = trieHit.info.posTags
			(trieHit.id, trieHit.surfaceRepr, trieHit.info.trieHit, s"PosTags(${posTags.mkString(", ")})", s">>>${trieHit.context.mkString(" ")}<<<")
		}
		trieHitOutput.writeAsTsv(trieHitPath)

		// Write raw features for debugging
		featuresPerGroup.reduceGroup { (classifiablesIt, out: Collector[(TrieHit, String, Double, Double)]) =>
			classifiablesIt.foreach { classifiable =>
				import classifiable._
				out.collect((info.trieHit, candidateEntity, surfaceProb, contextProb))
			}
		}.writeAsTsv(rawFeaturesPath)

		// Write candidate classifier results for debugging
		basicClassifierResults.map { res =>
			(res.documentId, res.classifierType, res.candidateEntity, res.trieHit)
		}.writeAsTsv(classificationPath)

		withNeighbours.groupBy("documentId").reduceGroup(new RandomWalkReduceGroup).name("Random Walk").writeAsTsv(randomWalkResultsPath)

	}

	def buildReciprocalNeighbours(env: ExecutionEnvironment): DataSet[Neighbours] = {
		val fullNeighbours = buildFullNeighbours(env)
		fullNeighbours.map { neighbours =><commit_msg>Better names in classification program<commit_after>package de.uni_potsdam.hpi.coheel.programs

import java.io.File
import java.lang.Iterable
import java.net.URI
import java.util.Date

import de.uni_potsdam.hpi.coheel.Params
import de.uni_potsdam.hpi.coheel.datastructures.TrieHit
import de.uni_potsdam.hpi.coheel.debugging.FreeMemory
import de.uni_potsdam.hpi.coheel.io.OutputFiles._
import de.uni_potsdam.hpi.coheel.io.Sample
import de.uni_potsdam.hpi.coheel.ml.CoheelClassifier
import de.uni_potsdam.hpi.coheel.ml.CoheelClassifier.POS_TAG_GROUPS
import de.uni_potsdam.hpi.coheel.programs.DataClasses._
import de.uni_potsdam.hpi.coheel.util.Util
import org.apache.flink.api.common.functions.{Partitioner, RichGroupReduceFunction}
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.{Path, FileSystem}
import org.apache.flink.util.Collector
import weka.classifiers.Classifier
import weka.core.SerializationHelper

import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.io.Source


class DocumentPartitioner extends Partitioner[Int] {
	override def partition(index: Int, numPartitions: Int): Int = {
		index
	}
}

/**
  * Finds entities in texts.
  *
  * IMPORTANT NOTE: For Flink, this requires taskmanager.numberOfTaskSlots
  * to be set to 1. See http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Distribute-DataSet-to-subset-of-nodes-td2814.html,
  * if you need more details.
  */
class ClassificationProgram extends NoParamCoheelProgram with Serializable {

	import CoheelLogger._

	override def getDescription: String = "CohEEL Classification"

	// Select, which neighbours file to use
	val NEIGHBOURS_FILE = fullNeighboursPath
//	val NEIGHBOURS_FILE = reciprocalNeighboursPath

	val neighboursCreationMethod = Map(fullNeighboursPath -> buildFullNeighbours _, reciprocalNeighboursPath -> buildReciprocalNeighbours _)

	override def buildProgram(env: ExecutionEnvironment): Unit = {
		val documentStrings = List(4).map { x =>
			Source.fromFile(s"src/main/resources/classification-documents/$x", "UTF-8").mkString
		}
		val documents = if (runsOffline())
				env.fromElements(Sample.ANGELA_MERKEL_SAMPLE_TEXT_3).name("Documents")
			else {
//				env.fromCollection(documentStrings).name("Documents")
				env.readTextFile(newYorkTimesDataPath).flatMap { l =>
					val split = l.split('\t')
					val text = split(4).replace("\n", " ")

					val words = text.split(' ')
					if (words.length > 60)
						Option(words.slice(0, 60).mkString(" "))
					else
						None
				}
			}

		val inputDocuments = documents
			.first(100)
			.flatMap(new InputDocumentDistributorFlatMap(params, runsOffline())).name("Input-Documents")

		val partitionedDocuments = inputDocuments.partitionCustom(new DocumentPartitioner, "index").name("Partitioned-Documents")

		val classifiables = partitionedDocuments
			.flatMap(new RunTrieOverDocumentsFlatMap(params))
			.name("Possible links")

		// fill the classifiables with all feature information
		val featuresPerGroup = FeatureHelper.buildFeaturesPerGroup(env, classifiables)
		val basicClassifierResults = featuresPerGroup.reduceGroup(new ClassificationReduceGroup(params)).name("Basic Classifier Results")

		val fileExists = if (runsOffline())
			new File(NEIGHBOURS_FILE).exists()
		else {
			val f = FileSystem.get(new URI("hdfs://tenemhead2"))
			f.exists(new Path(NEIGHBOURS_FILE.replace("hdfs://tenemhead2", "")))
		}

		val preprocessedNeighbours = if (fileExists) {
			log.info("Neighbourhood-File exists")
			loadNeighboursFromDisk(env, NEIGHBOURS_FILE)
		} else {
			log.info("Neighbourhood-File does not exist")
			neighboursCreationMethod(NEIGHBOURS_FILE)(env)
		}

		val withNeighbours = basicClassifierResults.join(preprocessedNeighbours)
			.where("candidateEntity")
			.equalTo("entity")
			.name("Join With Neighbours")
			.map { joinResult => joinResult match {
					case (classifierResult, neighbours) =>
						ClassifierResultWithNeighbours(
							classifierResult.documentId,
							classifierResult.classifierType,
							classifierResult.candidateEntity,
							classifierResult.trieHit,
							neighbours.in,
							neighbours.out)
				}
			}

		/*
		 * OUTPUT
		 */
		inputDocuments.filter(_.replication == 0).name("Input Documents").writeAsTsv(inputDocumentsPath)

		if (!fileExists) {
			preprocessedNeighbours.map(serializeNeighboursToString _).name("Serialized Neighbours").writeAsText(NEIGHBOURS_FILE, FileSystem.WriteMode.OVERWRITE)
		}

		// Write trie hits for debugging
		val trieHitOutput = classifiables.map { trieHit =>
			val posTags = trieHit.info.posTags
			(trieHit.id, trieHit.surfaceRepr, trieHit.info.trieHit, s"PosTags(${posTags.mkString(", ")})", s">>>${trieHit.context.mkString(" ")}<<<")
		}.name("Trie-Hits")
		trieHitOutput.writeAsTsv(trieHitPath)

		// Write raw features for debugging
		featuresPerGroup.reduceGroup { (classifiablesIt, out: Collector[(TrieHit, String, Double, Double)]) =>
			classifiablesIt.foreach { classifiable =>
				import classifiable._
				out.collect((info.trieHit, candidateEntity, surfaceProb, contextProb))
			}
		}.name("Raw-Features").writeAsTsv(rawFeaturesPath)

		// Write candidate classifier results for debugging
		basicClassifierResults.map { res =>
			(res.documentId, res.classifierType, res.candidateEntity, res.trieHit)
		}.name("Classifier-Results").writeAsTsv(classificationPath)

		withNeighbours.groupBy("documentId").reduceGroup(new RandomWalkReduceGroup).name("Random Walk").writeAsTsv(randomWalkResultsPath)

	}

	def buildReciprocalNeighbours(env: ExecutionEnvironment): DataSet[Neighbours] = {
		val fullNeighbours = buildFullNeighbours(env)
		fullNeighbours.map { neighbours =>
{'fraction_non_alphanumeric': 0.11115935735996527, 'fraction_numerical': 0.004950065132435953, 'mean_word_length': 4.860559796437659, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '261640', 'n_tokens_mistral': 4284, 'n_tokens_neox': 3711, 'n_words': 632}