D-Net

package eu.dnetlib.data.transform;

import static org.junit.Assert.assertFalse;

import static org.junit.Assert.assertNotNull;

import java.io.InputStream;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

import javax.xml.transform.TransformerConfigurationException;

import javax.xml.transform.TransformerFactoryConfigurationError;

import org.apache.commons.io.IOUtils;

import org.dom4j.DocumentException;

import org.junit.Before;

import org.junit.Test;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.google.protobuf.InvalidProtocolBufferException;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

public class XsltRowTransformerFactoryTest {

	private XsltRowTransformerFactory factory;

	private EntityConfigTable entityConfigTable;

	// private final InputStream datacite2insertActionsInputStream =

	// getClass().getResourceAsStream("/eu/dnetlib/actionmanager/xslt/datacite2insertActions.xslt");

	private final InputStream dmf2hbaseInputStream = getClass().getResourceAsStream("dmf_2_hbase.xsl");

	private final InputStream datacite2hbaseInputStream = getClass().getResourceAsStream("datacite_2_hbase.xsl");

	private final InputStream project2hbaseInputStream = getClass().getResourceAsStream("projects_2_hbase.xsl");

	private final InputStream recordInputStream = getClass().getResourceAsStream("record.xml");

	private final InputStream recordDataciteInputStream = getClass().getResourceAsStream("recordDatacite.xml");

	private final InputStream projectRecordInputStream = getClass().getResourceAsStream("projectRecord.xml");

	private final InputStream recordClaimInputStream = getClass().getResourceAsStream("recordClaim.xml");

	private final InputStream recordPangaeOafInputStream = getClass().getResourceAsStream("pangaeOAF.xml");

	private final InputStream recordPangaeOdfInputStream = getClass().getResourceAsStream("pangaeODF.xml");

	@Before

	public void setUp() throws Exception {

		factory = new XsltRowTransformerFactory();

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

	}

	@Test

	public void testParseClaim() throws Exception {

		doTest(dmf2hbaseInputStream, recordClaimInputStream);

	}

	@Test

	public void testParseProjectWithStats() throws Exception {

		doTest(project2hbaseInputStream, projectRecordInputStream);

	}

	@Test

	public void testParseDmf() throws Exception {

		doTest(dmf2hbaseInputStream, recordInputStream);

	}

	@Test

	public void testParseDatacite() throws Exception {

		doTest(datacite2hbaseInputStream, recordDataciteInputStream);

	}

	@Test

	public void testParsePangaeOAF() throws Exception {

		doTest(dmf2hbaseInputStream, recordPangaeOafInputStream);

	}

	@Test

	public void testParsePangaeODF() throws Exception {

		doTest(datacite2hbaseInputStream, recordPangaeOdfInputStream);

	}

	@Test

	public void testLinkPangae() throws Exception {

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

		rows.addAll(asRows(datacite2hbaseInputStream, recordPangaeOdfInputStream));

		rows.addAll(asRows(dmf2hbaseInputStream, recordPangaeOafInputStream));

		print(mapAll(buildTable(rows)));

	}

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

		try {

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

			// System.out.println(rows);

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

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

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

			print(builders);

		} catch (InvalidProtocolBufferException e) {

			throw new Exception(e);

		} catch (TransformerConfigurationException e) {

			throw new Exception(e);

		} catch (TransformerFactoryConfigurationError e) {

			throw new Exception(e);

		} catch (DocumentException e) {

			throw new Exception(e);

		}

	}

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

		String xslt = IOUtils.toString(xsltStream);

		XsltRowTransformer transformer = factory.getTransformer(xslt);

		assertNotNull(transformer);

		String record = IOUtils.toString(recordStream);

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

		assertNotNull(rows);

		assertFalse(rows.isEmpty());

		return rows;

	}

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

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

		for (Row row : rows) {

			String rowKey = row.getKey();

			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 (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());

			}

		}

		return table;

	}

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

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

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

			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();

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

		if (colEntity == null) return;

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

		if (bodyB != null) {

			ensureBuilder(builders, rowKey);

			OafDecoder mainEntity = OafDecoder.decode(Oaf.parseFrom(bodyB));

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

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

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

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

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

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

						Oaf oaf = Oaf.parseFrom(oafB);

						OafRel.Builder relBuilder = OafRel.newBuilder(oaf.getRel());

						if (ld.isSymmetric()) {

							RelDescriptor rd = ld.getRelDescriptor();

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

						}

						OafRel oafRel = relBuilder.setChild(ld.isChild()).build();

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

						oafBuilder.getRelBuilder().mergeFrom(oafRel);

						String targetId = ld.isSymmetric() ? oafRel.getTarget() : oafRel.getSource();

						ensureBuilder(builders, targetId);

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

						if (ld.isChild()) {

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

						} else {

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

						}

					}

				}

			}

		}

	}

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

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

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

				System.out.println(IndentXmlString.apply(e.getValue().build()));

			} else {

				System.out.println("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, false, false);

	}

	@Test

	public void test_template() throws Exception {

		String xslt = IOUtils.toString(dmf2hbaseInputStream);

		XsltRowTransformer transformer = factory.getTransformer(xslt);

		assertNotNull(transformer);

		String record = IOUtils.toString(recordInputStream);

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

		System.out.println(rows);

	}

}