<commit_before>package de.avgl.dmp.converter.flow;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.io.StringWriter;
import java.io.UnsupportedEncodingException;
import java.util.Iterator;
import java.util.Set;

import org.culturegraph.mf.framework.DefaultObjectPipe;
import org.culturegraph.mf.framework.ObjectPipe;
import org.culturegraph.mf.framework.ObjectReceiver;
import org.culturegraph.mf.framework.StreamReceiver;
import org.culturegraph.mf.morph.Metamorph;
import org.culturegraph.mf.stream.converter.JsonEncoder;
import org.culturegraph.mf.stream.sink.ObjectJavaIoWriter;
import org.culturegraph.mf.stream.source.ResourceOpener;
import org.culturegraph.mf.stream.source.StringReader;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Optional;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Sets;
import com.google.inject.Provider;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.ModelFactory;

import de.avgl.dmp.converter.DMPConverterException;
import de.avgl.dmp.converter.mf.stream.RDFModelReceiver;
import de.avgl.dmp.converter.mf.stream.TripleEncoder;
import de.avgl.dmp.converter.mf.stream.reader.JsonNodeReader;
import de.avgl.dmp.converter.morph.MorphScriptBuilder;
import de.avgl.dmp.converter.pipe.StreamJsonCollapser;
import de.avgl.dmp.converter.pipe.StreamUnflattener;
import de.avgl.dmp.converter.reader.QucosaReader;
import de.avgl.dmp.init.util.DMPStatics;
import de.avgl.dmp.persistence.DMPPersistenceException;
import de.avgl.dmp.persistence.model.internal.rdf.RDFModel;
import de.avgl.dmp.persistence.model.job.Task;
import de.avgl.dmp.persistence.model.resource.DataModel;
import de.avgl.dmp.persistence.model.schema.Clasz;
import de.avgl.dmp.persistence.model.schema.Schema;
import de.avgl.dmp.persistence.model.types.Tuple;
import de.avgl.dmp.persistence.service.InternalModelService;
import de.avgl.dmp.persistence.service.InternalModelServiceFactory;
import de.avgl.dmp.persistence.util.DMPPersistenceUtil;

/**
 * Flow that executes a given set of transformations on data of a given data model.
 *
 * @author phorn
 * @author tgaengler
 */
public class TransformationFlow {

	private static final org.apache.log4j.Logger		LOG						= org.apache.log4j.Logger.getLogger(TransformationFlow.class);

	public static final String							DEFAULT_RESOURCE_PATH	= "qucosa_record.xml";

	private final Metamorph								transformer;

	private final String								script;

	private final Optional<DataModel>					outputDataModel;

	private final Provider<InternalModelServiceFactory>	internalModelServiceFactoryProvider;

	public TransformationFlow(final Metamorph transformer, final Provider<InternalModelServiceFactory> internalModelServiceFactoryProviderArg) {

		this.transformer = transformer;
		script = null;
		outputDataModel = Optional.absent();
		internalModelServiceFactoryProvider = internalModelServiceFactoryProviderArg;
	}

	public TransformationFlow(final Metamorph transformer, final String scriptArg,
			final Provider<InternalModelServiceFactory> internalModelServiceFactoryProviderArg) {

		this.transformer = transformer;
		script = scriptArg;
		outputDataModel = Optional.absent();
		internalModelServiceFactoryProvider = internalModelServiceFactoryProviderArg;
	}

	public TransformationFlow(final Metamorph transformer, final String scriptArg, final DataModel outputDataModelArg,
			final Provider<InternalModelServiceFactory> internalModelServiceFactoryProviderArg) {

		this.transformer = transformer;
		script = scriptArg;
		outputDataModel = Optional.of(outputDataModelArg);
		internalModelServiceFactoryProvider = internalModelServiceFactoryProviderArg;
	}

	// public String applyRecord(final String record) {
	//
	// final StringReader opener = new StringReader();
	//
	// return apply(record, opener);
	// }

	public String getScript() {

		return script;
	}

	public String applyRecordDemo(final String record) {

		final StringReader opener = new StringReader();

		return applyDemo(record, opener);
	}

	// public String applyResource(final String resourcePath) {
	//
	// // public String applyResource(final String resourcePath) {
	// //
	// // final ResourceOpener opener = new ResourceOpener();
	// //
	// // return apply(resourcePath, opener);
	// // }
	//
	// String applyResourceDemo(final String resourcePath) {
	//
	// final ResourceOpener opener = new ResourceOpener();
	//
	// return applyDemo(resourcePath, opener);
	// }

	public String applyResourceDemo(final String resourcePath) {

		final ResourceOpener opener = new ResourceOpener();

		return applyDemo(resourcePath, opener);
	}

	public String apply(final Iterator<Tuple<String, JsonNode>> tuples, final ObjectPipe<Iterator<Tuple<String, JsonNode>>, StreamReceiver> opener,
			final boolean writeResultToDatahub) throws DMPConverterException {

		// final String recordDummy = "record";

		final StreamUnflattener unflattener = new StreamUnflattener("", DMPStatics.ATTRIBUTE_DELIMITER);
		final StreamJsonCollapser collapser = new StreamJsonCollapser();
		final TripleEncoder converter = new TripleEncoder(outputDataModel);
		final RDFModelReceiver writer = new RDFModelReceiver();

		opener.setReceiver(transformer).setReceiver(unflattener).setReceiver(collapser).setReceiver(converter).setReceiver(writer);

		opener.process(tuples);
		opener.closeStream();
		// objectReceiver.closeStream();

		// return stringWriter.toString();

		final Optional<ImmutableList<RDFModel>> optionalRDFModels = Optional.of(writer.getCollection());

		if (!optionalRDFModels.isPresent()) {

			return null;
		}

		final Model model = ModelFactory.createDefaultModel();
		String recordClassUri = null;

		// transform to FE friendly JSON => or use Model#toJSON() ;)

		final ObjectMapper objectMapper = DMPPersistenceUtil.getJSONObjectMapper();
		// final ArrayNode result = objectMapper.createArrayNode();
		final Set<String> recordURIs = Sets.newHashSet();

		for (final RDFModel rdfModel : optionalRDFModels.get()) {

			// result.add(rdfModel.toRawJSON());

			if (rdfModel.getModel() == null) {

				continue;
			}<commit_msg>[TASK] remove unnecessary usage of Optional<commit_after>package de.avgl.dmp.converter.flow;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.io.StringWriter;
import java.io.UnsupportedEncodingException;
import java.util.Iterator;
import java.util.Set;

import org.culturegraph.mf.framework.DefaultObjectPipe;
import org.culturegraph.mf.framework.ObjectPipe;
import org.culturegraph.mf.framework.ObjectReceiver;
import org.culturegraph.mf.framework.StreamReceiver;
import org.culturegraph.mf.morph.Metamorph;
import org.culturegraph.mf.stream.converter.JsonEncoder;
import org.culturegraph.mf.stream.sink.ObjectJavaIoWriter;
import org.culturegraph.mf.stream.source.ResourceOpener;
import org.culturegraph.mf.stream.source.StringReader;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Optional;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Sets;
import com.google.inject.Provider;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.ModelFactory;

import de.avgl.dmp.converter.DMPConverterException;
import de.avgl.dmp.converter.mf.stream.RDFModelReceiver;
import de.avgl.dmp.converter.mf.stream.TripleEncoder;
import de.avgl.dmp.converter.mf.stream.reader.JsonNodeReader;
import de.avgl.dmp.converter.morph.MorphScriptBuilder;
import de.avgl.dmp.converter.pipe.StreamJsonCollapser;
import de.avgl.dmp.converter.pipe.StreamUnflattener;
import de.avgl.dmp.converter.reader.QucosaReader;
import de.avgl.dmp.init.util.DMPStatics;
import de.avgl.dmp.persistence.DMPPersistenceException;
import de.avgl.dmp.persistence.model.internal.rdf.RDFModel;
import de.avgl.dmp.persistence.model.job.Task;
import de.avgl.dmp.persistence.model.resource.DataModel;
import de.avgl.dmp.persistence.model.schema.Clasz;
import de.avgl.dmp.persistence.model.schema.Schema;
import de.avgl.dmp.persistence.model.types.Tuple;
import de.avgl.dmp.persistence.service.InternalModelService;
import de.avgl.dmp.persistence.service.InternalModelServiceFactory;
import de.avgl.dmp.persistence.util.DMPPersistenceUtil;

/**
 * Flow that executes a given set of transformations on data of a given data model.
 *
 * @author phorn
 * @author tgaengler
 */
public class TransformationFlow {

	private static final org.apache.log4j.Logger		LOG						= org.apache.log4j.Logger.getLogger(TransformationFlow.class);

	public static final String							DEFAULT_RESOURCE_PATH	= "qucosa_record.xml";

	private final Metamorph								transformer;

	private final String								script;

	private final Optional<DataModel>					outputDataModel;

	private final Provider<InternalModelServiceFactory>	internalModelServiceFactoryProvider;

	public TransformationFlow(final Metamorph transformer, final Provider<InternalModelServiceFactory> internalModelServiceFactoryProviderArg) {

		this.transformer = transformer;
		script = null;
		outputDataModel = Optional.absent();
		internalModelServiceFactoryProvider = internalModelServiceFactoryProviderArg;
	}

	public TransformationFlow(final Metamorph transformer, final String scriptArg,
			final Provider<InternalModelServiceFactory> internalModelServiceFactoryProviderArg) {

		this.transformer = transformer;
		script = scriptArg;
		outputDataModel = Optional.absent();
		internalModelServiceFactoryProvider = internalModelServiceFactoryProviderArg;
	}

	public TransformationFlow(final Metamorph transformer, final String scriptArg, final DataModel outputDataModelArg,
			final Provider<InternalModelServiceFactory> internalModelServiceFactoryProviderArg) {

		this.transformer = transformer;
		script = scriptArg;
		outputDataModel = Optional.of(outputDataModelArg);
		internalModelServiceFactoryProvider = internalModelServiceFactoryProviderArg;
	}

	// public String applyRecord(final String record) {
	//
	// final StringReader opener = new StringReader();
	//
	// return apply(record, opener);
	// }

	public String getScript() {

		return script;
	}

	public String applyRecordDemo(final String record) {

		final StringReader opener = new StringReader();

		return applyDemo(record, opener);
	}

	// public String applyResource(final String resourcePath) {
	//
	// // public String applyResource(final String resourcePath) {
	// //
	// // final ResourceOpener opener = new ResourceOpener();
	// //
	// // return apply(resourcePath, opener);
	// // }
	//
	// String applyResourceDemo(final String resourcePath) {
	//
	// final ResourceOpener opener = new ResourceOpener();
	//
	// return applyDemo(resourcePath, opener);
	// }

	public String applyResourceDemo(final String resourcePath) {

		final ResourceOpener opener = new ResourceOpener();

		return applyDemo(resourcePath, opener);
	}

	public String apply(final Iterator<Tuple<String, JsonNode>> tuples, final ObjectPipe<Iterator<Tuple<String, JsonNode>>, StreamReceiver> opener,
			final boolean writeResultToDatahub) throws DMPConverterException {

		// final String recordDummy = "record";

		final StreamUnflattener unflattener = new StreamUnflattener("", DMPStatics.ATTRIBUTE_DELIMITER);
		final StreamJsonCollapser collapser = new StreamJsonCollapser();
		final TripleEncoder converter = new TripleEncoder(outputDataModel);
		final RDFModelReceiver writer = new RDFModelReceiver();

		opener.setReceiver(transformer).setReceiver(unflattener).setReceiver(collapser).setReceiver(converter).setReceiver(writer);

		opener.process(tuples);
		opener.closeStream();
		// objectReceiver.closeStream();

		// return stringWriter.toString();

		final ImmutableList<RDFModel> rdfModels = writer.getCollection();

		final Model model = ModelFactory.createDefaultModel();
		String recordClassUri = null;

		// transform to FE friendly JSON => or use Model#toJSON() ;)

		final ObjectMapper objectMapper = DMPPersistenceUtil.getJSONObjectMapper();
		// final ArrayNode result = objectMapper.createArrayNode();
		final Set<String> recordURIs = Sets.newHashSet();

		for (final RDFModel rdfModel : rdfModels) {

			// result.add(rdfModel.toRawJSON());

			if (rdfModel.getModel() == null) {

				continue;
			}
{'fraction_non_alphanumeric': 0.09978801915678731, 'fraction_numerical': 0.00031404569364842585, 'mean_word_length': 5.426841574167508, 'pattern_counts': {'":': 0, '<': 28, '<?xml version=': 0, '>': 30, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 8}, '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': '4942208', 'n_tokens_mistral': 4038, 'n_tokens_neox': 3729, 'n_words': 737}