D-Net

package eu.dnetlib.msro.workflows.nodes.transform;

import java.time.Duration;

import java.time.LocalDateTime;

import org.junit.Test;

/**

 * Created by Alessia Bardi on 13/04/2017.

 *

 * @author Alessia Bardi

 */

public class X3MTransformJobNodeTest {

	private X3MTransformJobNode transformJob = new X3MTransformJobNode();

	private String header = "<oai:header xmlns:dri=\"http://www.driver-repository.eu/namespace/dri\">\n"

			+ "        <dri:objIdentifier>ariadne_mock::0000023f507999464aa2b78875b7e5d6</dri:objIdentifier>\n"

			+ "        <dri:recordIdentifier>2420500</dri:recordIdentifier>\n"

			+ "        <dri:dateOfCollection>2017-04-10T18:44:46.85+02:00</dri:dateOfCollection>\n"

			+ "        <dri:datasourceprefix>ariadne_mock</dri:datasourceprefix>\n"

			+ "        <dri:datasourcename>Ariadne Mock</dri:datasourcename>\n"

			+ "        <dri:dateOfTransformation>2017-04-12T16:31:45.766</dri:dateOfTransformation>\n"

			+ "        <dri:invalid value=\"true\">\n"

			+ "            <dri:error vocabularies=\"dnet:languages\" xpath=\"//*[local-name()='P72_has_language']\"\n"

			+ "                term=\"en\"/>\n"

			+ "        </dri:invalid>\n"

			+ "    </oai:header>";

	private String footer = "<oai:about xmlns:dri=\"http://www.driver-repository.eu/namespace/dri\">\n"

			+ "        <provenance xmlns=\"http://www.openarchives.org/OAI/2.0/provenance\"\n"

			+ "            xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\"\n"

			+ "            xsi:schemaLocation=\"http://www.openarchives.org/OAI/2.0/provenance http://www.openarchives.org/OAI/2.0/provenance.xsd\">\n"

			+ "            <originDescription xmlns=\"\" altered=\"true\" harvestDate=\"2017-04-10T18:44:46.85+02:00\">\n"

			+ "                <baseURL>sftp%3A%2F%2Fariadne2.isti.cnr.it%2F..%2F..%2Fdata%2Ftransform%2Facdm_correct</baseURL>\n"

			+ "                <identifier/>\n"

			+ "                <datestamp/>\n"

			+ "                <metadataNamespace/>\n"

			+ "            </originDescription>\n"

			+ "        </provenance>\n"

			+ "    </oai:about>";

	private String metadata="        <rdf:RDF xmlns:rdf=\"http://www.w3.org/1999/02/22-rdf-syntax-ns#\"\n"

			+ "            xmlns:dbpedia-owl=\"http://dbpedia.org/ontology/\"\n"

			+ "            xmlns:acdm=\"http://registry.ariadne-infrastructure.eu/\"\n"

			+ "            xmlns:xsd=\"http://www.w3.org/2001/XMLSchema#\"\n"

			+ "            xmlns:skos=\"http://www.w3.org/2004/02/skos/core#\"\n"

			+ "            xmlns:rdfs=\"http://www.w3.org/2000/01/rdf-schema#\"\n"

			+ "            xmlns:frbr=\"http://www.cidoc-crm.org/frbroo/\" xmlns:dcterms=\"http://purl.org/dc/terms/\"\n"

			+ "            xmlns:dcat=\"http://www.w3.org/ns/dcat#\" xmlns:foaf=\"http://xmlns.com/foaf/0.1/\"\n"

			+ "            xmlns:crm=\"http://www.cidoc-crm.org/cidoc-crm/\"\n"

			+ "            xmlns:dc=\"http://purl.org/dc/elements/1.1/\">\n"

			+ "            <frbr:F30_Publication_Event rdf:about=\"uuid:AAAA\"/>\n"

			+ "            <crm:E65_Creation rdf:about=\"uuid:AAAB\"/>\n"

			+ "            <crm:E73_Information_Object\n"

			+ "                rdf:about=\"http://registry.ariadne-infrastructure.eu/dataset/2420500\">\n"

			+ "                <crm:P129_is_about>\n"

			+ "                    <crm:E73_Information_Object\n"

			+ "                        rdf:about=\"http://registry.ariadne-infrastructure.eu/subject/CAIRN\">\n"

			+ "                        <crm:P129_is_about>CAIRN</crm:P129_is_about>\n"

			+ "                    </crm:E73_Information_Object>\n"

			+ "                </crm:P129_is_about>\n"

			+ "                <crm:P129_is_about>\n"

			+ "                    <crm:E73_Information_Object\n"

			+ "                        rdf:about=\"http://registry.ariadne-infrastructure.eu/subject/HUT%20CIRCLE\">\n"

			+ "                        <crm:P129_is_about>HUT CIRCLE</crm:P129_is_about>\n"

			+ "                    </crm:E73_Information_Object>\n"

			+ "                </crm:P129_is_about>\n"

			+ "                <crm:P104_is_subject_to>ADS Terms and Conditions</crm:P104_is_subject_to>\n"

			+ "                <crm:P129_is_about>\n"

			+ "                    <crm:E73_Information_Object\n"

			+ "                        rdf:about=\"http://registry.ariadne-infrastructure.eu/subject/SHEEPFOLD\">\n"

			+ "                        <crm:P129_is_about>SHEEPFOLD</crm:P129_is_about>\n"

			+ "                    </crm:E73_Information_Object>\n"

			+ "                </crm:P129_is_about>\n"

			+ "                <crm:P129_is_about>\n"

			+ "                    <crm:E73_Information_Object\n"

			+ "                        rdf:about=\"http://registry.ariadne-infrastructure.eu/subject/FIELD%20SYSTEM\">\n"

			+ "                        <crm:P129_is_about>FIELD SYSTEM</crm:P129_is_about>\n"

			+ "                    </crm:E73_Information_Object>\n"

			+ "                </crm:P129_is_about>\n"

			+ "                <crm:P102_has_title>MID GLEN CROE</crm:P102_has_title>\n"

			+ "                <crm:P165_incorporates>\n"

			+ "                    <crm:E33_Linguistic_Object rdf:about=\"uuid:AAAG\">\n"

			+ "                        <crm:P72_has_language>en</crm:P72_has_language>\n"

			+ "                    </crm:E33_Linguistic_Object>\n"

			+ "                </crm:P165_incorporates>\n"

			+ "                <crm:P67_refers_to>\n"

			+ "                    <crm:E1_CRM_Entity rdf:about=\"uuid:AAAH\">\n"

			+ "                        <crm:P2_has_type>Sites and monuments databases or\n"

			+ "                            inventories</crm:P2_has_type>\n"

			+ "                    </crm:E1_CRM_Entity>\n"

			+ "                </crm:P67_refers_to>\n"

			+ "                <crm:P93i_was_taken_out_of_existence_by>\n"

			+ "                    <crm:E6_Destruction rdf:about=\"uuid:AAAE\">\n"

			+ "                        <crm:P4_has_time-span>\n"

			+ "                            <crm:E52_Time-Span rdf:about=\"uuid:AAAF\">\n"

			+ "                                <crm:P81_ongoing_throughout>2013-12-09\n"

			+ "                                    00:00:00.0</crm:P81_ongoing_throughout>\n"

			+ "                            </crm:E52_Time-Span>\n"

			+ "                        </crm:P4_has_time-span>\n"

			+ "                    </crm:E6_Destruction>\n"

			+ "                </crm:P93i_was_taken_out_of_existence_by>\n"

			+ "                <crm:P94i_was_created_by>\n"

			+ "                    <frbr:F30_Publication_Event rdf:about=\"uuid:AAAC\">\n"

			+ "                        <crm:P4_has_time-span>\n"

			+ "                            <crm:E52_Time-Span rdf:about=\"uuid:AAAD\">\n"

			+ "                                <crm:P81_ongoing_throughout>2013-12-09\n"

			+ "                                    00:00:00.0</crm:P81_ongoing_throughout>\n"

			+ "                            </crm:E52_Time-Span>\n"

			+ "                        </crm:P4_has_time-span>\n"

			+ "                    </frbr:F30_Publication_Event>\n"

			+ "                </crm:P94i_was_created_by>\n"

			+ "                <crm:P129_is_about>\n"

			+ "                    <crm:E73_Information_Object\n"

			+ "                        rdf:about=\"http://registry.ariadne-infrastructure.eu/subject/BUILDING\">\n"

			+ "                        <crm:P129_is_about>BUILDING</crm:P129_is_about>\n"

			+ "                    </crm:E73_Information_Object>\n"

			+ "                </crm:P129_is_about>\n"

			+ "                <crm:P1_is_identified_by>2420500</crm:P1_is_identified_by>\n"

			+ "                <crm:P106i_forms_part_of>http://registry.ariadne-infrastructure.eu/collection/22721290</crm:P106i_forms_part_of>\n"

			+ "                <crm:P3_has_note>Multiple instances of: SHEEPFOLD&lt;br /&gt;&lt;br /&gt;Multiple\n"

			+ "                    instances of: BUILDING<br />Multiple instances of: FIELD SYSTEM<br\n"

			+ "                    />Possible instance of: CAIRN<br />Multiple instances of: HUT\n"

			+ "                    CIRCLE&lt;br /&gt;Possible instance of: HUT CIRCLE</crm:P3_has_note>\n"

			+ "            </crm:E73_Information_Object>\n"

			+ "        </rdf:RDF>";

	@Test

	public void testBuildXML(){

		LocalDateTime now = LocalDateTime.now();

		String res = transformJob.buildXML(header, now.toString(), metadata, footer);

		LocalDateTime end = LocalDateTime.now();

		System.out.println("Building XML took:"+Duration.between(now, end).toMillis());

	}

}

package eu.dnetlib.msro.workflows.nodes.transform;

import java.time.Duration;

import java.time.Instant;

import java.time.LocalDateTime;

import java.util.List;

import java.util.Map;

import javax.xml.parsers.DocumentBuilder;

import javax.xml.parsers.DocumentBuilderFactory;

import com.google.common.collect.Maps;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.enabling.resultset.factory.ResultSetFactory;

import eu.dnetlib.miscutils.functional.xml.SaxonHelper;

import eu.dnetlib.miscutils.functional.xml.XMLIndenter;

import eu.dnetlib.msro.workflows.graph.Arc;

import eu.dnetlib.msro.workflows.nodes.AsyncJobNode;

import eu.dnetlib.msro.workflows.procs.Env;

import eu.dnetlib.rmi.common.ResultSet;

import eu.dnetlib.rmi.enabling.ISLookUpException;

import eu.dnetlib.rmi.enabling.ISLookUpService;

import eu.dnetlib.rmi.manager.MSROException;

import net.sf.saxon.s9api.SaxonApiException;

import net.sf.saxon.s9api.Serializer.Property;

import net.sf.saxon.s9api.XPathSelector;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.w3c.dom.Document;

import org.w3c.dom.Element;

import org.w3c.dom.Node;

public class X3MTransformJobNode extends AsyncJobNode {

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

	private static final String OAI_NAMESPACE_URI = "http://www.openarchives.org/OAI/2.0/";

	private static final String DRI_NAMESPACE_URI = "http://www.driver-repository.eu/namespace/dri";

	private String inputEprParam;

	private String outputEprParam;

	private String mappingPolicyProfileId;

	private String mappingProfileIds;

	private boolean verboseLogging;

	private XPathSelector xpathSelectorMetadata;

	private XPathSelector xpathSelectorHeader;

	private XPathSelector xpathSelectorFooter;

	private XPathSelector xpathSelectorObjIdentifier;

	/**

	 * true to pass the full record to X3m-engine. False to pass only what's in the metadata section.

	 **/

	private boolean passFullRecord;

	@Autowired

	private ResultSetFactory resultSetFactory;

	@Autowired

	private UniqueServiceLocator serviceLocator;

	@Autowired

	private SaxonHelper saxonHelper;

	@Override

	protected String execute(final Env env) throws Exception {

		log.info("Mapping profile ids read from node configuration: " + mappingProfileIds);

		log.info("Mapping Policy profile id read from node configuration: " + mappingPolicyProfileId);

		final String[] mappings = getMappingsCode(mappingProfileIds.split(","));

		final String policy = getProfileCode(mappingPolicyProfileId);

		LocalDateTime now = LocalDateTime.now();

		final ResultSet<?> rsIn = env.getAttribute(this.inputEprParam, ResultSet.class);

		if ((rsIn == null)) { throw new MSROException("InputEprParam (" + this.inputEprParam + ") not found in ENV"); }

		prepareXpathSelectors();

		final ResultSet<String> rsOut = this.resultSetFactory.map(rsIn, String.class, record -> {

			//JUST FOR DEBUGGING THE TIMEOUT OF THE MONGO CURSOR: is there a metadata record that it is really slow to transform?

			if(log.isDebugEnabled()) {

				String objIdentifier = extractFromRecord(record, xpathSelectorObjIdentifier);

				log.debug("Transforming record objIdentifier: " + objIdentifier);

			}

			ApplyX3Mapping mappingFunction = new ApplyX3Mapping(mappings, policy, verboseLogging);

			String toTransform = record;

			Instant startExtraction = Instant.now();

			if(!isPassFullRecord()) {

				log.debug("Extracting XML from the metadata block");

				toTransform = extractFromRecord(record, xpathSelectorMetadata);

			}

			String header = extractFromRecord(record, xpathSelectorHeader);

			String provenanceFooter = extractFromRecord(record, xpathSelectorFooter);

			Instant endExtraction = Instant.now();

			Instant startTransform = Instant.now();

			String transformed = mappingFunction.apply(toTransform);

			Instant endTransform = Instant.now();

			if(log.isDebugEnabled()){

				log.debug("Extraction took "+ Duration.between(startExtraction, endExtraction).toMillis()+" ms");

				log.debug("Transformation took "+ Duration.between(startTransform, endTransform).toMillis()+" ms");

				log.debug("Total mapping time: "+Duration.between(startExtraction, endTransform).toMillis()+" ms");

			}

			String res = buildXML(header, now.toString(), transformed, provenanceFooter);

			if(log.isDebugEnabled()) {

				log.debug("SOURCE:\n"+toTransform);

				log.debug("TRANFORMED:\n"+res);

			}

			return res;

		});

		env.setAttribute(this.outputEprParam, rsOut);

		return Arc.DEFAULT_ARC;

	}

	private String[] getMappingsCode(String[] mappingIds) throws ISLookUpException {

		String[] mappings = new String[mappingIds.length];

		for(int i =0; i < mappingIds.length; i++){

			mappings[i] = getProfileCode(mappingIds[i]);

		}

		return mappings;

	}

	protected String buildXML(final String header, final String transformationDate, final String metadata, final String provenance) {

		Instant start = Instant.now();

		try {

			XMLIndenter xmlHelper = new XMLIndenter();

			DocumentBuilderFactory docFactory = DocumentBuilderFactory.newInstance();

			DocumentBuilder docBuilder = docFactory.newDocumentBuilder();

			// root elements

			Document doc = docBuilder.newDocument();

			Element rootElement = doc.createElementNS(OAI_NAMESPACE_URI, "oai:record");

			Element headerElem = docBuilder.parse(IOUtils.toInputStream(header, "UTF-8")).getDocumentElement();

			Node headerNode = doc.importNode(headerElem, true);

			rootElement.appendChild(headerNode);

			Element transDate = doc.createElementNS(DRI_NAMESPACE_URI, "dri:dateOfTransformation");

			transDate.setTextContent(transformationDate);

			headerNode.appendChild(transDate);

			Element metadataElement = doc.createElementNS(OAI_NAMESPACE_URI, "oai:metadata");

			Element contentElem = docBuilder.parse(IOUtils.toInputStream(metadata, "UTF-8")).getDocumentElement();

			Node contentNode = doc.importNode(contentElem, true);

			metadataElement.appendChild(contentNode);

			rootElement.appendChild(metadataElement);

			Element aboutElem = docBuilder.parse(IOUtils.toInputStream(provenance, "UTF-8")).getDocumentElement();

			Node aboutNode = doc.importNode(aboutElem, true);

			rootElement.appendChild(aboutNode);

			doc.appendChild(rootElement);

			Instant startIndent = Instant.now();

			String res = xmlHelper.indent(doc);

			Instant end = Instant.now();

			if(log.isDebugEnabled()){

				log.debug("XML built in "+ Duration.between(start, end).toMillis()+" ms");

				log.debug("Serialization with indent took "+ Duration.between(startIndent, end).toMillis()+" ms");

			}

			return res;

		} catch (Exception e) {

			throw new RuntimeException("Cannot build the transformed xml file", e);

		}

	}

	private void prepareXpathSelectors() throws SaxonApiException {

		Map<String, String> namespaces = Maps.newHashMap();

		namespaces.put("oai", OAI_NAMESPACE_URI);

		namespaces.put("dri", DRI_NAMESPACE_URI);

		xpathSelectorHeader = this.saxonHelper.help().prepareXPathSelector("//oai:header", namespaces);

		xpathSelectorMetadata = this.saxonHelper.help().prepareXPathSelector("//oai:metadata/*", namespaces);

		xpathSelectorFooter = this.saxonHelper.help().prepareXPathSelector("//oai:about", namespaces);

		xpathSelectorObjIdentifier = this.saxonHelper.help().prepareXPathSelector("//oai:header/*[local-name()='objIdentifier']/text()", namespaces);

	}

	private String extractFromRecord(final String record, final XPathSelector xPathSelector) {

		try {

			return this.saxonHelper.help().setSerializerProperty(Property.OMIT_XML_DECLARATION, "yes").evaluateSingleAsString(record, xPathSelector);

		} catch (SaxonApiException e) {

			throw new RuntimeException("Cannot extract content ", e);

		}

	}

	private String getProfileCode(String profId) throws ISLookUpException {

		if (StringUtils.isBlank(profId)) return null;

		String xquery = "string(collection('/db/DRIVER/TransformationRuleDSResources')//RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value ='" +

				profId + "']//CODE)";

		List<String> res = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xquery);

		if (res.isEmpty() || StringUtils.isBlank(res.get(0))) {

			throw new RuntimeException("Can't find transformation rule CODE for " + profId);

		}

		return res.get(0);

	}

	public String getInputEprParam() {

		return this.inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getOutputEprParam() {

		return this.outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	public String getMappingPolicyProfileId() {

		return mappingPolicyProfileId;

	}

	public void setMappingPolicyProfileId(final String mappingPolicyProfileId) {

		this.mappingPolicyProfileId = mappingPolicyProfileId;

	}

	public String getMappingProfileIds() {

		return mappingProfileIds;

	}

	public void setMappingProfileIds(final String mappingProfileIds) {

		this.mappingProfileIds = mappingProfileIds;

	}

	public boolean isVerboseLogging() {

		return verboseLogging;

	}

	public void setVerboseLogging(final boolean verboseLogging) {

		this.verboseLogging = verboseLogging;

	}

	public ResultSetFactory getResultSetFactory() {

		return resultSetFactory;

	}

	public void setResultSetFactory(final ResultSetFactory resultSetFactory) {

		this.resultSetFactory = resultSetFactory;

	}

	public UniqueServiceLocator getServiceLocator() {

		return serviceLocator;

	}

	public void setServiceLocator(final UniqueServiceLocator serviceLocator) {

		this.serviceLocator = serviceLocator;

	}

	public boolean isPassFullRecord() {

		return passFullRecord;

	}

	public void setPassFullRecord(final boolean passFullRecord) {

		this.passFullRecord = passFullRecord;

	}

}

	private String[] mappings;

	public ApplyX3Mapping(final String[] mappings, final String generatorPolicy, final boolean verboseLogging) {

		this.mappings = mappings;

		this.generatorPolicy = generatorPolicy;

		this.verboseLogging = verboseLogging;

	}

		InputStream[] mappingStreams = null;

		if (mappingUrl==null) {

			mappingStreams = new InputStream[mappings.length];

			try {

				for (int i = 0; i < mappings.length; i++) {

					mappingStreams[i] = getStream(mappings[i]);

				}

			}catch(IOException e){

				log.error("Can't create mappingStreams for mappings");

				closeStreams(mappingStreams);

				throw new RuntimeException(e);

			}

		}

			X3MLEngineFactory x3mEngineFactory  = null;

			if (mappingUrl==null) {

				x3mEngineFactory = getConfiguredX3MEngineFactory(mappingStreams, policyStream);

			}

			else {

				x3mEngineFactory = getConfiguredX3MEngineFactory(mappingUrl, policyStream);

			}

		finally {

			if (mappingStreams!=null) {

				closeStreams(mappingStreams);

			}

		}

	private void closeStreams(InputStream[] streams){

		for(InputStream s: streams){

			try {

				if(s != null) s.close();

			} catch (IOException e) {

				log.error("Can't close stream"+e.getMessage());

			}

		}

	}

	private X3MLEngineFactory getConfiguredX3MEngineFactory(InputStream[] mappings, InputStream policy) {

		X3MLEngineFactory x3mEngineFactory = X3MLEngineFactory.create().withMappings(mappings);

		if (policy != null) {

			x3mEngineFactory.withGeneratorPolicy(policy);

		}

		if (verboseLogging)

			x3mEngineFactory.withVerboseLogging();

		//to enable real UUID

		x3mEngineFactory.withUuidSize(0);

		return x3mEngineFactory;

	}

	<bean id="wfNodeX3MTransform"

	      class="eu.dnetlib.msro.workflows.nodes.transform.X3MTransformJobNode"

	      scope="prototype" />