D-Net

package eu.dnetlib.data.transform;

import java.io.*;

import java.util.*;

import java.util.Map.Entry;

import java.util.zip.GZIPInputStream;

import javax.xml.transform.TransformerConfigurationException;

import javax.xml.transform.TransformerFactoryConfigurationError;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

import com.google.protobuf.InvalidProtocolBufferException;

import eu.dnetlib.common.rmi.UnimplementedException;

import eu.dnetlib.data.mapreduce.hbase.index.config.*;

import eu.dnetlib.data.mapreduce.util.*;

import eu.dnetlib.data.proto.KindProtos.Kind;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import eu.dnetlib.data.proto.TypeProtos.Type;

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

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.io.SAXReader;

import org.json.JSONObject;

import org.junit.Before;

import org.junit.Test;

import static org.junit.Assert.*;

public class XsltRowTransformerFactoryTest {

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

	private static String basePathProfiles = "/eu/dnetlib/test/profiles/TransformationRuleDSResources/TransformationRuleDSResourceType/";

	private XsltRowTransformerFactory factory;

	private EntityConfigTable entityConfigTable;

	@Before

	public void setUp() throws Exception {

		factory = new XsltRowTransformerFactory();

		entityConfigTable = IndexConfig.load(IndexConfigTest.config).getConfigMap();

	}

	@Test

	public void testParseManyAuthors() throws Exception {

		final Map<String, Object> xslParams = Maps.newHashMap();

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

		m.put("od______2367", "true"); // Puma

		m.put("od______2294", "true"); // UNIBI

		m.put("od________18", "false"); // Arxiv

		xslParams.put("mergeIdForHomonymsMap", m);

		final List<Row> rows = asRows(loadFromTransformationProfile("oaf_person2hbase.xml"), xslParams, load("recordManyAuthors.xml"));

		int authorCount = 0;

		for (final Row row : rows) {

			for (final Column<String, byte[]> col : row.getColumns()) {

				authorCount++;

				final OafDecoder d = OafDecoder.decode(col.getValue());

				assertNotNull(d);

				log.debug(d.getEntity().getPerson().getCoauthorList().size());

			}

		}

		log.info("authors' count: ---> " + authorCount);

	}

	@Test

	public void testParseAuthors() throws Exception {

		final Map<String, Object> xslParams = Maps.newHashMap();

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

		m.put("od______2367", "true"); // Puma

		m.put("od______2294", "true"); // UNIBI

		m.put("od________18", "false"); // Arxiv

		xslParams.put("mergeIdForHomonymsMap", m);

		final List<Row> rows = asRows(loadFromTransformationProfile("oaf_person2hbase.xml"), xslParams, load("recordArxiv.xml"));

		for (final Row row : rows) {

			for (final Column<String, byte[]> col : row.getColumns()) {

				final OafDecoder d = OafDecoder.decode(col.getValue());

				log.debug(d.getOaf());

			}

		}

	}

	@Test

	public void testParseClaimUpdate() throws Exception {

		final List<Row> rows = Lists.newArrayList();

		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("recordClaim.xml")));

		rows.addAll(asRows(loadFromTransformationProfile("claimUpdates_2_hbase.xsl"), load("recordClaimUpdate.xml")));

		printAll(mapAll(buildTable(rows)));

	}

	@Test

	public void testParseDatasetPUB() throws Exception {

		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("recordDatasetPUB.xml"));

	}

	@Test

	public void testParseClaim() throws Exception {

		doTest(loadFromTransformationProfile("oaf2hbase.xml"), load("recordClaim.xml"));

	}

	@Test

	public void testParseACM() throws Exception {

		doTest(loadFromTransformationProfile("oaf2hbase.xml"), load("recordACM.xml"));

	}

	@Test

	public void testParseASB() throws Exception {

		doTest(loadFromTransformationProfile("oaf2hbase.xml"), load("recordASB.xml"));

	}

	@Test

	public void testParseProjectCorda() throws Exception {

		doTest(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordCorda.xml"));

	}

	@Test

	public void testParseProjectFCT() throws Exception {

		doTest(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordFCT.xml"));

	}

	@Test

	public void testParseOaf() throws Exception {

		doTest(loadFromTransformationProfile("oaf2hbase.xml"), load("record.xml"));

	}

	@Test

	public void testParseOafPublication() throws Exception {

		doTest(loadFromTransformationProfile("oaf_publication2hbase.xml"), load("record.xml"));

	}

	@Test

	public void testParseDatacite() throws Exception {

		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("recordDatacite.xml"));

	}

	@Test

	public void testParseDatacite2() throws Exception {

		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("recordDatacite2.xml"));

	}

	@Test

	public void testLinkPangaea() throws Exception {

		final List<Row> rows = Lists.newArrayList();

		rows.addAll(asRows(loadFromTransformationProfile("odf2hbase.xml"), load("pangaeODF1.xml")));

		rows.addAll(asRows(loadFromTransformationProfile("odf2hbase.xml"), load("pangaeODF2.xml")));

		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("pangaeOAF.xml")));

		rows.addAll(asRows(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordCordaPangaea.xml")));

		printAll(mapAll(buildTable(rows)));

	}

	@Test

	public void testLinkCorda() throws Exception {

		final List<Row> rows = Lists.newArrayList();

		rows.addAll(asRows(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordCorda.xml")));

		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("recordCorda.xml")));

		printAll(mapAll(buildTable(rows)));

	}

	@Test

	public void testLinkFCT() throws Exception {

		final List<Row> rows = Lists.newArrayList();

		rows.addAll(asRows(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordFCT.xml")));

		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("recordFCT.xml")));

		printAll(mapAll(buildTable(rows)));

	}

	@Test

	public void testLinkARC() throws Exception {

		final List<Row> rows = Lists.newArrayList();

		rows.addAll(asRows(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordARC.xml")));

		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("recordARC.xml")));

		printAll(mapAll(buildTable(rows)));

	}

	@Test

	public void testLinkWT() throws Exception {

		final List<Row> rows = Lists.newArrayList();

		rows.addAll(asRows(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordWT.xml")));

		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("recordWT.xml")));

		printAll(mapAll(buildTable(rows)));

	}

	@Test

	public void testLinkOrganization() throws Exception {

		final List<Row> rows = Lists.newArrayList();

		rows.addAll(asRows(loadFromTransformationProfile("organizations_2_hbase.xsl"), load("organization.xml")));

		rows.addAll(asRows(loadFromTransformationProfile("projectorganization_2_hbase.xsl"), load("project_organization.xml")));

		rows.addAll(asRows(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordCorda.xml")));

		printAll(mapAll(buildTable(rows)));

	}

	@Test

	public void testUnpackAuthors() throws Exception {

		final Map<String, Object> xslParams = Maps.newHashMap();

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

		m.put("od______2367", "true"); // Puma

		m.put("od______2294", "true"); // UNIBI

		m.put("od________18", "false"); // Arxiv

		xslParams.put("mergeIdForHomonymsMap", m);

		final List<Row> rows = Lists.newArrayList();

		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), xslParams, load("recordArxiv.xml")));

		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), xslParams, load("recordArxiv2.xml")));

		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), xslParams, load("recordPuma1.xml")));

		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), xslParams, load("recordPuma2.xml")));

		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), xslParams, load("recordUNIBI.xml")));

		printPersonFullnames(mapAll(buildTable(rows)));

	}

	@Test

	public void testParseDoajOAF() throws Exception {

		doTest(loadFromTransformationProfile("oaf2hbase.xml"), load("doajUniversityRecord.xml"));

	}

	@Test

	public void testParseDatasource() throws Exception {

		doTest(loadFromTransformationProfile("datasources_2_hbase.xsl"), load("datasourceNative.xml"));

	}

	@Test

	public void testParseDataDatasource() throws Exception {

		doTest(loadFromTransformationProfile("datasources_2_hbase.xsl"), load("dataDatasource.xml"));

	}

	@Test

	public void testFromMongodbCompressedDump() throws Exception {

		doTestJsonGz(loadFromTransformationProfile("oaf2hbase.xml"), load("mdstore_cleaned.json.gz"));

	}

	@Test

	public void testLoadFromTransformationProfile() throws IOException {

		InputStream in = loadFromTransformationProfile("oaf2hbase.xml");

		log.info(IOUtils.toString(in));

	}

	@Test

	public void test_template() throws Exception {

		final String xslt = IOUtils.toString(loadFromTransformationProfile("oaf2hbase.xml"));

		final XsltRowTransformer transformer = factory.getTransformer(xslt);

		assertNotNull(transformer);

		final String record = IOUtils.toString(load("record.xml"));

		final List<Row> rows = transformer.apply(record);

		System.out.println(rows);

	}

	@Test

	public void testWrongCharsOrganization() throws Exception {

		final List<Row> rows = Lists.newArrayList();

		rows.addAll(asRows(loadFromTransformationProfile("organizations_2_hbase.xsl"), load("organizationWrongChars.xml")));

		printAll(mapAll(buildTable(rows)));

	}

	private void doTest(final InputStream xsltStream, final InputStream recordStream) throws Exception {

		try {

			final List<Row> rows = asRows(xsltStream, recordStream);

			log.info(rows);

			final Map<String, Map<String, Map<String, byte[]>>> table = buildTable(rows);

			// System.out.println("\n" + table.toString());

			final Map<String, XmlRecordFactory> builders = mapAll(table);

			printAll(builders);

		} catch (final InvalidProtocolBufferException e) {

			throw new Exception(e);

		} catch (final TransformerConfigurationException e) {

			throw new Exception(e);

		} catch (final TransformerFactoryConfigurationError e) {

			throw new Exception(e);

		} catch (final DocumentException e) {

			throw new Exception(e);

		}

	}

	private void doTestJsonGz(final InputStream xsltStream, final InputStream recordStream) throws Exception {

		final Iterator<List<Row>> rowsIterator = asRowsJsonGzip(xsltStream, recordStream);

		int i = 0;

		while (rowsIterator.hasNext()) {

			final List<Row> rows = rowsIterator.next();

			i++;

			if ((i % 10000) == 0) {

				System.out.println(i);

			}

			final Map<String, Map<String, Map<String, byte[]>>> table = buildTableDoaj(rows);

			for (final Map<String, Map<String, byte[]>> m : table.values()) {

				for (final Map<String, byte[]> mv : m.values()) {

					for (final byte[] v : mv.values()) {

						final OafDecoder d = OafDecoder.decode(v);

						assertNotNull(d);

						assertNotNull(d.getOaf());

						switch (d.getKind()) {

						case entity:

							assertNotNull(d.getMetadata());

							if (d.getOaf().getEntity().getType().equals(Type.result)) {

								System.out.println(d.getOaf());

							}

							break;

						case relation:

							assertNotNull(d.getRel());

							break;

						default:

							break;

						}

					}

				}

			}

		}

	}

	private List<Row> asRows(final InputStream xsltStream, final InputStream recordStream) throws Exception {

		return asRows(xsltStream, new HashMap<String, Object>(), recordStream);

	}

	private List<Row> asRows(final InputStream xsltStream, final Map<String, Object> params, final InputStream recordStream) throws Exception {

		final String xslt = IOUtils.toString(xsltStream);

		final XsltRowTransformer transformer = factory.getTransformer(xslt, params);

		assertNotNull(transformer);

		final String record = IOUtils.toString(recordStream);

		final List<Row> rows = transformer.apply(record);

		assertNotNull(rows);

		assertFalse(rows.isEmpty());

		return rows;

	}

	private Iterator<List<Row>> asRowsJsonGzip(final InputStream xsltStream, final InputStream recordStreamJsonGzip) throws Exception {

		final String xslt = IOUtils.toString(xsltStream);

		final XsltRowTransformer transformer = factory.getTransformer(xslt);

		assertNotNull(transformer);

		assertNotNull(recordStreamJsonGzip);

		final GZIPInputStream stream = new GZIPInputStream(recordStreamJsonGzip);

		assertNotNull(stream);

		final BufferedReader inStream = new BufferedReader(new InputStreamReader(stream));

		assertNotNull(inStream);

		return new Iterator<List<Row>>() {

			String jsonRecord = null;

			@Override

			public boolean hasNext() {

				try {

					return (jsonRecord = inStream.readLine()) != null;

				} catch (final IOException e) {

					throw new RuntimeException(e);

				}

			}

			@Override

			public List<Row> next() {

				final JSONObject jsonObj = new JSONObject(jsonRecord);

				final String body = jsonObj.getString("body");

				try {

					assertTrue(StringUtils.isNotBlank(body));

					// System.out.println(body);

					final List<Row> rows = transformer.apply(body);

					assertNotNull(rows);

					assertFalse(rows.isEmpty());

					return rows;

				} catch (final Throwable e) {

					System.err.println("error transforming document: " + body);

					throw new RuntimeException(e);

				}

			}

			@Override

			public void remove() {

				throw new UnimplementedException();

			}

		};

	}

	private Map<String, Map<String, Map<String, byte[]>>> buildTableDoaj(final List<Row> rows) throws UnsupportedEncodingException {

		final Map<String, Map<String, Map<String, byte[]>>> table = Maps.newHashMap();

		for (final Row row : rows) {

			final String rowKey = row.getKey();

			final String cf = row.getColumnFamily();

			if (!table.containsKey(rowKey)) {

				table.put(rowKey, new HashMap<String, Map<String, byte[]>>());

			}

			if (!table.get(rowKey).containsKey(cf)) {

				table.get(rowKey).put(row.getColumnFamily(), new HashMap<String, byte[]>());

			}

			for (final Column<String, byte[]> c : row.getColumns()) {

				// System.out.println(String.format("ADDING K:%s CF:%s Q:%s", rowKey, cf, c.getName()));

				table.get(rowKey).get(cf).put(c.getName(), c.getValue());

				if (cf.equals("result") && c.getName().equals("body")) {

					// System.out.println(String.format("ADDING K:%s CF:%s Q:%s", rowKey, cf, c.getName()));

					assertTrue(StringUtils.isNotBlank(new String(c.getValue(), "UTF-8")));

				}

			}

		}

		return table;

	}

	private Map<String, Map<String, Map<String, byte[]>>> buildTable(final List<Row> rows) throws UnsupportedEncodingException {

		final Map<String, Map<String, Map<String, byte[]>>> table = Maps.newHashMap();

		for (final Row row : rows) {

			final String rowKey = row.getKey();

			final String cf = row.getColumnFamily();

			if (!table.containsKey(rowKey)) {

				table.put(rowKey, new HashMap<String, Map<String, byte[]>>());

			}

			if (!table.get(rowKey).containsKey(cf)) {

				table.get(rowKey).put(row.getColumnFamily(), new HashMap<String, byte[]>());

			}

			for (final Column<String, byte[]> c : row.getColumns()) {

				System.out.println(String.format("ADDING K:%s CF:%s Q:%s", rowKey, cf, c.getName()));

				table.get(rowKey).get(cf).put(c.getName(), c.getValue());

				if (c.getName().equals("body")) {

					final String theBody = new String(c.getValue(), "UTF-8");

					assertTrue(StringUtils.isNotBlank(theBody));

					// System.out.println(theBody);

				}

			}

		}

		return table;

	}

	private Map<String, XmlRecordFactory> mapAll(final Map<String, Map<String, Map<String, byte[]>>> table) throws Exception {

		final Map<String, XmlRecordFactory> builders = Maps.newHashMap();

		for (final Entry<String, Map<String, Map<String, byte[]>>> e : table.entrySet()) {

			map(builders, e.getKey(), e.getValue());

		}

		return builders;

	}

	// private Map<String, XmlRecordFactory> mapResultsOnly(final Map<String, Map<String, Map<String, byte[]>>> table) throws Exception {

	//

	// final Map<String, XmlRecordFactory> builders = Maps.newHashMap();

	// for (final Entry<String, Map<String, Map<String, byte[]>>> e : table.entrySet()) {

	// final Type type = OafRowKeyDecoder.decode(e.getKey()).getType();

	// if (type == Type.result) {

	// map(builders, e.getKey(), e.getValue());

	// }

	// }

	// return builders;

	// }

	private void map(final Map<String, XmlRecordFactory> builders, final String rowKey, final Map<String, Map<String, byte[]>> row) throws Exception {

		final Type type = OafRowKeyDecoder.decode(rowKey).getType();

		final Map<String, byte[]> familyMap = row.get(type.toString());

		if (familyMap == null) return;

		final byte[] bodyB = familyMap.get("body");

		if (bodyB != null) {

			ensureBuilder(builders, rowKey);

			final Oaf oaf = UpdateMerger.mergeBodyUpdates(familyMap);

			final OafDecoder mainEntity = OafDecoder.decode(oaf);

			builders.get(rowKey).setMainEntity(mainEntity);

			for (final LinkDescriptor ld : entityConfigTable.getDescriptors(type)) {

				final String it = ld.getRelDescriptor().getIt();

				final Map<String, byte[]> cols = row.get(it);

				if ((cols != null) && !cols.isEmpty()) {

					for (final byte[] oafB : cols.values()) {

						final Oaf.Builder relBuilder = Oaf.newBuilder(Oaf.parseFrom(oafB));

						if (ld.isSymmetric()) {

							final RelDescriptor rd = ld.getRelDescriptor();

							relBuilder.getRelBuilder().setCachedTarget(mainEntity.getEntity()).setRelType(rd.getRelType()).setSubRelType(rd.getSubRelType());

						}

						relBuilder.getRelBuilder().setChild(ld.isChild());

						final Oaf.Builder oafBuilder = Oaf.newBuilder().setKind(Kind.relation).setLastupdatetimestamp(System.currentTimeMillis());

						oafBuilder.mergeFrom(relBuilder.build());

						final String targetId = ld.isSymmetric() ? oafBuilder.getRel().getTarget() : oafBuilder.getRel().getSource();

						ensureBuilder(builders, targetId);

						final OafDecoder decoder = OafDecoder.decode(oafBuilder.build());

						if (ld.isChild()) {

							builders.get(targetId).addChild(type, decoder);

						} else {

							builders.get(targetId).addRelation(type, decoder);

						}

					}

				}

			}

		}

	}

	private void printAll(final Map<String, XmlRecordFactory> builders) throws DocumentException {

		print(Sets.newHashSet(Type.values()), builders, null);

	}

	private void printPerson(final Map<String, XmlRecordFactory> builders) throws DocumentException {

		print(Sets.newHashSet(Type.person), builders, null);

	}

	private void printPersonFullnames(final Map<String, XmlRecordFactory> builders) throws DocumentException {

		final Map<Type, Set<String>> xpaths = Maps.newHashMap();

		final Set<String> personPaths = Sets.newHashSet();

		personPaths.add("//fullname");

		xpaths.put(Type.person, personPaths);

		print(Sets.newHashSet(Type.person), builders, xpaths);

	}

	private void print(final Set<Type> types, final Map<String, XmlRecordFactory> builders, final Map<Type, Set<String>> xpaths) throws DocumentException {

		final SAXReader r = new SAXReader();

		for (final Entry<String, XmlRecordFactory> e : builders.entrySet()) {

			final OafRowKeyDecoder kd = OafRowKeyDecoder.decode(e.getKey());

			if (!e.getValue().isValid()) throw new IllegalArgumentException("invalid builder: " + e.getKey());

			if (types.contains(kd.getType())) {

				final String val = IndentXmlString.apply(e.getValue().build());

				if ((xpaths != null) && !xpaths.isEmpty() && (xpaths.get(kd.getType()) != null)) {

					final Document doc = r.read(new StringReader(val));

					log.debug("\n" + e.getKey());

					for (final String xpath : xpaths.get(kd.getType())) {

						log.debug(doc.valueOf(xpath));

					}

				} else {

					log.info(val);

				}

			}

		}

	}

	private void printNoIndent(final Map<String, XmlRecordFactory> builders) {

		for (final Entry<String, XmlRecordFactory> e : builders.entrySet()) {

			if (e.getValue().isValid()) {

				log.debug(e.getValue().build());

			} else {

				log.debug("invalid builder: " + e.getKey());

			}

		}

	}

	private void ensureBuilder(final Map<String, XmlRecordFactory> builders, final String rowKey) throws Exception {

		if (!builders.containsKey(rowKey)) {

			builders.put(rowKey, newBuilder());

		}

	}

	private XmlRecordFactory newBuilder() throws TransformerConfigurationException, TransformerFactoryConfigurationError, DocumentException {

		return new XmlRecordFactory(entityConfigTable, ContextMapper.fromXml(eu.dnetlib.data.mapreduce.hbase.index.config.Context.xml),

				RelClasses.fromJSon(RelClassesTest.relClassesJson), XmlRecordFactoryTest.SCHEMA_LOCATION, true, true, false);

	}

	private InputStream load(final String fileName) {

		return getClass().getResourceAsStream(fileName);

	}

	private InputStream loadFromTransformationProfile(final String profilePath) {

		log.info("Loading xslt from: " + basePathProfiles + profilePath);

		InputStream profile = getClass().getResourceAsStream(basePathProfiles + profilePath);

		SAXReader saxReader = new SAXReader();

		Document doc = null;

		try {

			doc = saxReader.read(profile);

		} catch (DocumentException e) {

			e.printStackTrace();

			throw new RuntimeException(e);

		}

		String xslt = doc.selectSingleNode("//SCRIPT/CODE/*[local-name()='stylesheet']").asXML();

		return IOUtils.toInputStream(xslt);

	}

}