D-Net

package eu.dnetlib.data.collective.transformation.engine;

import java.io.StringReader;

import java.util.*;

import javax.xml.transform.TransformerFactory;

import javax.xml.xpath.XPath;

import javax.xml.xpath.XPathConstants;

import javax.xml.xpath.XPathExpressionException;

import javax.xml.xpath.XPathFactory;

import eu.dnetlib.common.profile.ResourceDao;

import eu.dnetlib.data.collective.transformation.IDatabaseConnector;

import eu.dnetlib.data.collective.transformation.TransformationException;

import eu.dnetlib.data.collective.transformation.VocabularyRegistry;

import eu.dnetlib.data.collective.transformation.core.xsl.ext.TransformationFunctionProxy;

import eu.dnetlib.data.collective.transformation.engine.core.ITransformation;

import eu.dnetlib.data.collective.transformation.engine.functions.*;

import eu.dnetlib.data.collective.transformation.engine.functions.RetrieveValue.FUNCTION;

import eu.dnetlib.data.collective.transformation.rulelanguage.Argument;

import eu.dnetlib.data.collective.transformation.rulelanguage.Argument.Type;

import eu.dnetlib.data.collective.transformation.rulelanguage.util.FunctionCall;

import eu.dnetlib.data.collective.transformation.utils.BlacklistConsumer;

import net.sf.saxon.expr.instruct.TerminationException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.core.io.Resource;

import org.w3c.dom.Node;

import org.xml.sax.InputSource;

// import eu.dnetlib.data.collective.transformation.engine.functions.Dblookup;

/**

 * @author jochen

 */

public class SimpleTransformationEngine {

	private static Log log = LogFactory.getLog(SimpleTransformationEngine.class);

	private final List<String> mdRecords = new LinkedList<String>();

	private ITransformation transformation;

	private VocabularyRegistry vocabularyRegistry;

	private IDatabaseConnector databaseConnector;

	private ResourceDao resourceDao;

	private IFeatureExtraction featureExtraction;

	private long totalTransformedRecords = 0;

	private long totalIgnoredRecords = 0;

	private String mappingFile;

	private boolean stylesheetParamsCalculated = false;

	private boolean preprocessingDone = false;

	private Map<String, String> stylesheetParams = new LinkedHashMap<String, String>();

	private Resource blacklistApi;

	private List<String> blacklistedRecords = new LinkedList<String>();

	private TransformerFactory transformerFactory;

	public SimpleTransformationEngine(final TransformerFactory transformerFactory){

		this.transformerFactory = transformerFactory;

	}

	/**

	 * execute any preprocessings declared in the transformation script prior starting the transformation of records

	 */

	public void preprocess(String dataSourceId) {

		for (Map<String, String> preprocMap : this.transformation.getRuleLanguageParser().getPreprocessings()) {

			Iterator<String> it = preprocMap.keySet().iterator();

			while (it.hasNext()) {

				String function = it.next();

				//				if (function.equals("dblookup")) {

				//					Dblookup fun = new Dblookup();

				//					fun.setDbConnector(databaseConnector);

				//					try {

				//						log.debug("preprocessingMap value: " + preprocMap.get(function));

				//						TransformationFunctionProxy.getInstance().setLookupRecord(fun.getResults(preprocMap.get(function)));

				//					} catch (Exception e) {

				//						log.debug(e.getMessage());

				//						throw new IllegalStateException(e);

				//					}

				//				}

				if (function.equals("blacklist")) {

					BlacklistConsumer bc = new BlacklistConsumer();

					try {

						blacklistedRecords = bc.getBlackList(blacklistApi.getURL() + dataSourceId);

					} catch (Exception e) {

						throw new IllegalStateException("error in preprocess: " + e.getMessage());

					}

				}

			}

		}

		log.debug("preprocessing done.");

	}

	/**

	 * check if blacklistedRecords exist and if so check if the current record is blacklisted by its objIdentifier

	 *

	 * @param aRecord

	 * @return

	 * @throws XPathExpressionException

	 * @throws ProcessingException

	 */

	private boolean isBlacklistRecord(String aRecord) {

		if (blacklistedRecords.size() == 0) return false;

		XPath xpath = XPathFactory.newInstance().newXPath();

		try {

			Node root = (Node) xpath.evaluate("/", new InputSource(new StringReader(aRecord)), XPathConstants.NODE);

			String objId = xpath.evaluate("//*[local-name()='objIdentifier']", root);

			if (blacklistedRecords.contains(objId)) return true;

		} catch (Exception e) {

			throw new IllegalStateException("error in isBlacklistRecord: " + e.getMessage());

		}

		return false;

	}

	/**

	 * transforms a source record

	 *

	 * @param sourceRecord the record to transform

	 * @return transformed record

	 */

	public String transform(final String sourceRecord) {

		List<String> objectRecords = new LinkedList<String>();

		objectRecords.add(sourceRecord);

		int index = 0;

		mdRecords.clear();

		initTransformationFunction();

		if (!stylesheetParamsCalculated) {

			try {

				calculateStylesheetParams(sourceRecord);

			} catch (Exception e) {

				throw new IllegalStateException("error in calculateStylesheetParams" + e.getMessage());

			}

		}

		if (!preprocessingDone) {

			// xpath sourceRecord dataSourceid

			preprocess(stylesheetParams.get("varBlacklistDataSourceId"));

			preprocessingDone = true;

		}

		if (isBlacklistRecord(sourceRecord)) {

			try {

				mdRecords.add(transformation.transformRecord(sourceRecord, ITransformation.XSLSyntaxcheckfailed));

			} catch (Exception e) {

				log.fatal(sourceRecord);

				throw new IllegalStateException(e);

			}

		} else if (!transformation.getRuleLanguageParser().isXslStylesheet()) {

			// iterate over all rules which are functionCalls

			log.debug("functionCalls size: " + transformation.getRuleLanguageParser().getFunctionCalls().size());

			for (FunctionCall functionCall : transformation.getRuleLanguageParser().getFunctionCalls()) {

				preprocess(objectRecords, functionCall);

			}

			for (String record : objectRecords) {

				// log.debug(record);

				try {

					log.debug("now run transformation for record with index: " + index);

					try {

						String transformedRecord = transformation.transformRecord(record, index);

						mdRecords.add(transformedRecord);

					} catch (TerminationException e) {

						log.debug("record transformation terminated.");

						String failedRecord = transformation.transformRecord(record, ITransformation.XSLSyntaxcheckfailed);

						log.debug(failedRecord);

						totalIgnoredRecords++;

						mdRecords.add(failedRecord);

					}

				} catch (TransformationException e) {

					log.error(sourceRecord);

					throw new IllegalStateException(e);

				}

				index++;

			}

		} else {

			for (String record : objectRecords) {

				// test for init params and assign values

				try {

					log.debug("now run transformation for record with index: " + index);

					try {

						String transformedRecord = transformation.transformRecord(record, stylesheetParams);

						mdRecords.add(transformedRecord);

					} catch (TerminationException e) {

						String failedRecord = transformation.transformRecord(record, ITransformation.XSLSyntaxcheckfailed);

						totalIgnoredRecords++;

						log.debug(failedRecord);

						mdRecords.add(failedRecord);

					}

				} catch (TransformationException e) {

					log.error(sourceRecord);

					throw new IllegalStateException(e);

				}

				index++;

			}

		}

		totalTransformedRecords = totalTransformedRecords + mdRecords.size();

		log.debug("objRecordSize: " + objectRecords.size() + ", mdRecordSize: " + mdRecords.size() + ", ignoredRecordSize: " + totalIgnoredRecords);

		return mdRecords.get(0);

	}

	private void calculateStylesheetParams(final String aRecord) throws XPathExpressionException, ProcessingException {

		stylesheetParamsCalculated = true;

		XPath xpath = XPathFactory.newInstance().newXPath();

		Node root = (Node) xpath.evaluate("/", new InputSource(new StringReader(aRecord)), XPathConstants.NODE);

		String datasourcePrefix = xpath.evaluate("//*[local-name()='datasourceprefix']", root);

		String profileXquery = "collection('/db/DRIVER/RepositoryServiceResources')//RESOURCE_PROFILE[.//EXTRA_FIELDS/FIELD[key=\"NamespacePrefix\"][value=\""

				+ datasourcePrefix + "\"]]";

		//String repositoryId = xpath.evaluate("//*[local-name()='repositoryId']", root);

		log.debug("profileXquery: " + profileXquery);

		// static $varDatasourceid = getValue(PROFILEFIELD, [xpath:"//dri:repositoryId",

		// xpath:"//EXTRA_FIELDS/FIELD[key='OpenAireDataSourceId']/value"]);

		RetrieveValue retrieveValue = new RetrieveValue();

		retrieveValue.setResourceDao(resourceDao);

		List<Argument> argList = new LinkedList<Argument>();

		argList.add(new Argument(Type.VALUE, profileXquery));

		Argument argXpath = new Argument(Type.INPUTFIELD, "//OFFICIAL_NAME");

		argList.add(argXpath);

		String varOfficialName = retrieveValue.executeSingleValue(FUNCTION.PROFILEFIELD.toString(), argList, null, new HashMap<String, String>());

		stylesheetParams.put("varOfficialName", varOfficialName);

		argList.remove(argXpath);

		argXpath = new Argument(Type.INPUTFIELD, "//EXTRA_FIELDS/FIELD[key='OpenAireDataSourceId']/value");

		argList.add(argXpath);

		String varDataSourceId = retrieveValue.executeSingleValue(FUNCTION.PROFILEFIELD.toString(), argList, null, new HashMap<String, String>());

		stylesheetParams.put("varDataSourceId", varDataSourceId);

		argList.remove(argXpath);

		argXpath = new Argument(Type.INPUTFIELD, "//CONFIGURATION/DATASOURCE_TYPE");

		argList.add(argXpath);

		String varDsType = retrieveValue.executeSingleValue(FUNCTION.PROFILEFIELD.toString(), argList, null, new HashMap<String, String>());

		stylesheetParams.put("varDsType", varDsType);

		argList.remove(argXpath);

		// if blacklist

		for (Map<String, String> preprocMap : this.transformation.getRuleLanguageParser().getPreprocessings()) {

			Iterator<String> it = preprocMap.keySet().iterator();

			while (it.hasNext()) {

				String function = it.next();

				if (function.equals("blacklist")) {

					argXpath = new Argument(Type.INPUTFIELD, preprocMap.get(function)); // blacklistDataSourceIdXpath

					argList.add(argXpath);

					String varBlacklistDataSourceId =

							retrieveValue.executeSingleValue(FUNCTION.PROFILEFIELD.toString(), argList, null, new HashMap<String, String>());

					stylesheetParams.put("varBlacklistDataSourceId", varBlacklistDataSourceId);

					argList.remove(argXpath);

				}

			}

		}

	}

	private void initTransformationFunction() {

		if (this.vocabularyRegistry == null) { throw new IllegalStateException("vocabularyReg is null"); }

		Convert convertFunction = new Convert();

		convertFunction.setVocabularyRegistry(this.vocabularyRegistry);

		TransformationFunctionProxy.getInstance(transformerFactory).setConvertFunction(convertFunction);

	}

	/**

	 * preprocesses function if function is configured resp.

	 *

	 * @param records       list of object records

	 * @param aFunctionCall

	 */

	private void preprocess(final List<String> records, final FunctionCall aFunctionCall) {

		try {

			log.debug("preprocess");

			if (transformation.getRuleLanguageParser() == null) { throw new IllegalStateException("rulelanguageparser not initialised"); }

			if (transformation.getRuleLanguageParser().getNamespaceDeclarations() == null) { throw new IllegalStateException("nsDecl is null"); }

			PreProcessor preProc = new PreProcessor();

			preProc.setConvertFunction(TransformationFunctionProxy.getInstance(transformerFactory).getConvertFunction());

			RetrieveValue retrieveValue = new RetrieveValue();

			retrieveValue.setResourceDao(resourceDao);

			preProc.setRetrieveFunction(retrieveValue);

			RegularExpression regExpr = new RegularExpression();

			preProc.setRegExprFunction(regExpr);

			TransformationFunctionProxy functionProxy = TransformationFunctionProxy.getInstance(transformerFactory);

			preProc.setFunctionProxy(functionProxy);

			Extract extractFunction = new Extract();

			extractFunction.setFeatureExtraction(featureExtraction);

			preProc.setExtractFunction(extractFunction);

			if (aFunctionCall.doPreprocess() || aFunctionCall.isStatic()) {

				// log.debug("now call preprocess with: " + aFunctionCall.getExternalFunctionName() + " " + aFunctionCall.getUuid());

				preProc.preprocess(

						aFunctionCall,

						records,

						transformation.getRuleLanguageParser().getNamespaceDeclarations(),

						transformation.getStaticTransformationResults(),

						transformation.getJobProperties(),

						transformation.getRuleLanguageParser().getVariableMappingRules());

				// log.debug("preprocess end");

			} else {

				log.debug("skip preprocessing for function: " + aFunctionCall.getExternalFunctionName());

			}

		} catch (Exception e) {

			throw new IllegalStateException(e);

		}

	}

	/**

	 * @return the transformation

	 */

	public ITransformation getTransformation() {

		return transformation;

	}

	/**

	 * @param transformation the transformation to set

	 */

	public void setTransformation(final ITransformation transformation) {

		this.transformation = transformation;

	}

	/**

	 * @return the vocabularyRegistry

	 */

	public VocabularyRegistry getVocabularyRegistry() {

		return vocabularyRegistry;

	}

	/**

	 * @param vocabularyRegistry the vocabularyRegistry to set

	 */

	public void setVocabularyRegistry(final VocabularyRegistry vocabularyRegistry) {

		this.vocabularyRegistry = vocabularyRegistry;

	}

	/**

	 * @return the resourceDao

	 */

	public ResourceDao getResourceDao() {

		return resourceDao;

	}

	/**

	 * @param resourceDao the resourceDao to set

	 */

	public void setResourceDao(final ResourceDao resourceDao) {

		this.resourceDao = resourceDao;

	}

	/**

	 * @return the featureExtraction

	 */

	public IFeatureExtraction getFeatureExtraction() {

		return featureExtraction;

	}

	/**

	 * @param featureExtraction the featureExtraction to set

	 */

	public void setFeatureExtraction(final IFeatureExtraction featureExtraction) {

		this.featureExtraction = featureExtraction;

	}

	/**

	 * @return the databaseConnector

	 */

	public IDatabaseConnector getDatabaseConnector() {

		return databaseConnector;

	}

	/**

	 * @param databaseConnector the databaseConnector to set

	 */

	public void setDatabaseConnector(final IDatabaseConnector databaseConnector) {

		this.databaseConnector = databaseConnector;

	}

	public long getTotalTransformedRecords() {

		return this.totalTransformedRecords;

	}

	public long getTotalIgnoredRecords() {

		return this.totalIgnoredRecords;

	}

	/**

	 * @return the mappingFile

	 */

	public String getMappingFile() {

		return mappingFile;

	}

	/**

	 * @param mappingFile the mappingFile to set

	 */

	public void setMappingFile(final String mappingFile) {

		this.mappingFile = mappingFile;

	}

	public Resource getBlacklistApi() {

		return blacklistApi;

	}

	public void setBlacklistApi(Resource blacklistApi) {

		this.blacklistApi = blacklistApi;

	}

}