D-Net

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

	<parent>

    		<groupId>eu.dnetlib</groupId>

	        <artifactId>dnet-hadoop-parent</artifactId>

            <version>1.0.0-SNAPSHOT</version>

	</parent>

	<modelVersion>4.0.0</modelVersion>

	<artifactId>icm-iis-common</artifactId>

	<packaging>jar</packaging>

	<version>1.0.1-SNAPSHOT</version>

	<scm>

	  <developerConnection>

	    scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet40/modules/icm-iis-common/trunk

	  </developerConnection>

	</scm>

	<properties>

		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

	</properties>

	<dependencies>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>icm-iis-core</artifactId>

			<version>[1.0.0,2.0.0)</version>

		</dependency>

		<dependency>

            <groupId>eu.dnetlib</groupId>

            <artifactId>icm-iis-schemas</artifactId>

            <version>[1.0.0,2.0.0)</version>

        </dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-openaire-data-protos</artifactId>

			<version>[3.0.0, 4.0.0)</version>

		</dependency>

		<dependency>

			<groupId>org.apache.oozie</groupId>

			<artifactId>oozie-core</artifactId>

			<version>${iis.oozie.version}</version>

			<scope>provided</scope>

		</dependency>

        <dependency>

            <groupId>org.apache.hadoop</groupId>

            <artifactId>hadoop-core</artifactId>

            <version>${iis.hadoop.core.version}</version>

            <scope>provided</scope>

        </dependency>

        <dependency>

            <groupId>org.apache.hadoop</groupId>

            <artifactId>hadoop-common</artifactId>

            <version>${iis.hadoop.common.version}</version>

            <scope>provided</scope>

        </dependency>

        <dependency>

            <groupId>org.apache.avro</groupId>

            <artifactId>avro</artifactId>

            <version>${iis.avro.version}</version>

        </dependency>

        <dependency>

            <groupId>org.apache.avro</groupId>

            <artifactId>avro-mapred</artifactId>

            <version>${iis.avro.version}</version>

            <classifier>hadoop2</classifier>

        </dependency>

        <dependency>

        	<!-- required by caching mechanism -->

            <groupId>com.google.code.gson</groupId>

            <artifactId>gson</artifactId>

            <version>${google.gson.version}</version>

        </dependency>

        <dependency>

        	<!-- required by caching mechanism for setting chmod -->

            <groupId>org.springframework</groupId>

            <artifactId>spring-beans</artifactId>

            <version>${spring.version}</version>

        </dependency>

    	<dependency>

			<groupId>com.thoughtworks.xstream</groupId>

			<artifactId>xstream</artifactId>

			<version>1.4.7</version>

		</dependency>

	</dependencies>

	<build>

	</build>

	<repositories>

		<repository>

			<id>cloudera</id>

			<url>https://repository.cloudera.com/artifactory/cloudera-repos</url>

			<releases>

				<enabled>true</enabled>

			</releases>

			<snapshots>

				<enabled>false</enabled>

			</snapshots>

		</repository>

  </repositories>

</project>

package eu.dnetlib.iis.common.model.extrainfo.citations;

import static org.junit.Assert.assertFalse;

import static org.junit.Assert.assertTrue;

import java.util.ArrayList;

import java.util.Iterator;

import java.util.SortedSet;

import java.util.TreeSet;

import org.junit.Test;

public class AlphaNumericCitationComparatorTest {

	@Test

	public void testSorting() throws Exception {

		SortedSet<BlobCitationEntry> citations = new TreeSet<BlobCitationEntry>();

		BlobCitationEntry c1 = new BlobCitationEntry("[1] test");

		BlobCitationEntry c2 = new BlobCitationEntry("[10] test");

		BlobCitationEntry c3 = new BlobCitationEntry("[2] test");

		BlobCitationEntry c4 = new BlobCitationEntry("[1] test");

		c4.setIdentifiers(new ArrayList<TypedId>());

		c4.getIdentifiers().add(new TypedId("1", "openaire", 0.9f));

		citations.add(c4);

		citations.add(c3);

		citations.add(c2);

		citations.add(c1);

		Iterator<BlobCitationEntry> citationsIt = citations.iterator();

		assertTrue(c1==citationsIt.next());

		assertTrue(c4==citationsIt.next());

		assertTrue(c3==citationsIt.next());

		assertTrue(c2==citationsIt.next());

		assertFalse(citationsIt.hasNext());

	}

	@Test

	public void testSortingWithNulls() throws Exception {

		SortedSet<BlobCitationEntry> citations = new TreeSet<BlobCitationEntry>();

		BlobCitationEntry c2 = new BlobCitationEntry("[10] test");

		BlobCitationEntry c3 = new BlobCitationEntry(null);

		citations.add(c3);

		citations.add(c2);

		Iterator<BlobCitationEntry> citationsIt = citations.iterator();

		assertTrue(c2==citationsIt.next());

		assertTrue(c3==citationsIt.next());

	}

	@Test

	public void testSortingWithDuplicates() throws Exception {

		SortedSet<BlobCitationEntry> citations = new TreeSet<BlobCitationEntry>();

		BlobCitationEntry c1 = new BlobCitationEntry("[1] test");

		BlobCitationEntry c2 = new BlobCitationEntry("[1] test");

		BlobCitationEntry c3 = new BlobCitationEntry("[2] test");

		BlobCitationEntry c4 = new BlobCitationEntry("[2] test");

		BlobCitationEntry c5 = new BlobCitationEntry("[2] test");

		c3.setIdentifiers(new ArrayList<TypedId>());

		c3.getIdentifiers().add(new TypedId("1", "openaire", 0.9f));

		c4.setIdentifiers(new ArrayList<TypedId>());

		c4.getIdentifiers().add(new TypedId("1", "openaire", 0.9f));

		c5.setIdentifiers(new ArrayList<TypedId>());

		c5.getIdentifiers().add(new TypedId("2", "openaire", 0.9f));

		citations.add(c5);

		citations.add(c4);

		citations.add(c3);

		citations.add(c2);

		citations.add(c1);

		Iterator<BlobCitationEntry> citationsIt = citations.iterator();

		assertTrue(c2==citationsIt.next());

		assertTrue(c5==citationsIt.next());

		assertTrue(c4==citationsIt.next());

		assertFalse(citationsIt.hasNext());

	}

}

package eu.dnetlib.iis.common.model.extrainfo.converter;

import java.util.SortedSet;

import eu.dnetlib.iis.common.model.extrainfo.citations.BlobCitationEntry;

/**

 * {@link BlobCitationEntry} based avro to xml converter.

 * @author mhorst

 *

 */

public class CitationsExtraInfoConverter extends AbstractExtraInfoConverter<SortedSet<BlobCitationEntry>> {

	public CitationsExtraInfoConverter() {

		xstream.processAnnotations(BlobCitationEntry.class);

		xstream.alias("citations", SortedSet.class);

	}

	@SuppressWarnings("unchecked")

	@Override

	public SortedSet<BlobCitationEntry> deserialize(String source) throws UnsupportedOperationException {

		return (SortedSet<BlobCitationEntry>) xstream.fromXML(source);

	}

}

package eu.dnetlib.iis.common.model.extrainfo.converter;

/**

 * Avro to XML converter module.

 * @author mhorst

 *

 * @param <T>

 */

public interface ExtraInfoConverter<T> {

	/**

	 * Serializes object to its XML representation.

	 * @param object

	 * @return XML representation of avro object

	 */

	String serialize(T object);

	/**

	 * Deserializes extra info object.

	 * @param source

	 * @return deserialized object

	 * @throws UnsupportedOperationException

	 */

	T deserialize(String source) throws UnsupportedOperationException;

}

package eu.dnetlib.iis.common.model.extrainfo.converter;

import java.io.StringWriter;

import com.thoughtworks.xstream.XStream;

import com.thoughtworks.xstream.converters.extended.NamedMapConverter;

import com.thoughtworks.xstream.io.xml.DomDriver;

import com.thoughtworks.xstream.io.xml.PrettyPrintWriter;

/**

 * Abstract xml converter.

 * @author mhorst

 *

 */

public abstract class AbstractExtraInfoConverter<T> implements ExtraInfoConverter<T> {

	protected final XStream xstream;

	public AbstractExtraInfoConverter() {

		xstream = new XStream(new DomDriver());

		xstream.setMode(XStream.NO_REFERENCES);

//		removing class attribute because no unmarshalling is required

		xstream.aliasSystemAttribute(null, "class");

//		changing the way maps are generated

		NamedMapConverter namedMapConverter = new NamedMapConverter(

				xstream.getMapper(),"entry","key",String.class,"value",Integer.class,

				true, true, xstream.getConverterLookup());

		xstream.registerConverter(namedMapConverter);

	}

	@Override

	public String serialize(T object) {

		StringWriter sw = new StringWriter();

//		xstream.marshal(object, new CompactWriter(sw));

		xstream.marshal(object, new PrettyPrintWriter(sw));

		return sw.toString();

	}

	@Override

	public T deserialize(String source) throws UnsupportedOperationException {

		throw new UnsupportedOperationException("deserialization is unsupported");

	}

}

package eu.dnetlib.iis.common.model.extrainfo;

/**

 * ExtraInfo InformationSpace model constants.

 * @author mhorst

 *

 */

public class ExtraInfoConstants {

	public static final String NAME_PROJECT_STATISTICS = "project statistics";

	public static final String NAME_RESULT_STATISTICS = "result statistics";

	public static final String NAME_AUTHOR_STATISTICS = "author statistics";

	public static final String NAME_CITATIONS = "result citations";

	public static final String TYPOLOGY_CITATIONS = "citations";

	public static final String TYPOLOGY_STATISTICS = "statistics";

	public static final String CITATION_TYPE_OPENAIRE = "openaire";

}

package eu.dnetlib.iis.common.model.extrainfo.citations;

import com.thoughtworks.xstream.annotations.XStreamAlias;

import com.thoughtworks.xstream.annotations.XStreamAsAttribute;

/**

 * Typed identifier.

 * @author mhorst

 *

 */

@XStreamAlias("id")

public class TypedId {

	@XStreamAsAttribute

	String value;

	@XStreamAsAttribute

	String type;

	@XStreamAsAttribute

	float confidenceLevel;

	public TypedId(String value, String type,

			float confidenceLevel) {

		this.value = value;

		this.type = type;

		this.confidenceLevel = confidenceLevel;

	}

	public TypedId() {

		super();

	}

	public String getValue() {

		return value;

	}

	public void setValue(String value) {

		this.value = value;

	}

	public String getType() {

		return type;

	}

	public void setType(String type) {

		this.type = type;

	}

	public float getConfidenceLevel() {

		return confidenceLevel;

	}

	public void setConfidenceLevel(float confidenceLevel) {

		this.confidenceLevel = confidenceLevel;

	}

	@Override

	public int hashCode() {

		final int prime = 31;

		int result = 1;

		result = prime * result + Float.floatToIntBits(confidenceLevel);

		result = prime * result + ((type == null) ? 0 : type.hashCode());

		result = prime * result + ((value == null) ? 0 : value.hashCode());

		return result;

	}

	@Override

	public boolean equals(Object obj) {

		if (this == obj)

			return true;

		if (obj == null)

			return false;

		if (getClass() != obj.getClass())

			return false;

		TypedId other = (TypedId) obj;

		if (Float.floatToIntBits(confidenceLevel) != Float

				.floatToIntBits(other.confidenceLevel))

			return false;

		if (type == null) {

			if (other.type != null)

				return false;

		} else if (!type.equals(other.type))

			return false;

		if (value == null) {

			if (other.value != null)

				return false;

		} else if (!value.equals(other.value))

			return false;

		return true;

	}

}

package eu.dnetlib.iis.common.model.extrainfo.citations;

import java.text.DecimalFormatSymbols;

import java.util.List;

import java.util.Locale;

import java.util.regex.Matcher;

import java.util.regex.Pattern;

import com.thoughtworks.xstream.annotations.XStreamAlias;

import com.thoughtworks.xstream.annotations.XStreamAsAttribute;

import com.thoughtworks.xstream.annotations.XStreamImplicit;

import com.thoughtworks.xstream.annotations.XStreamOmitField;

/**

 * Comparable citation entry.

 * @author mhorst

 *

 */

@XStreamAlias("citation")

public class BlobCitationEntry implements Comparable<BlobCitationEntry> {

	/**

	 * Citation position.

	 */

	@XStreamAsAttribute

	protected int position;

	/**

	 * Raw citation text.

	 */

	protected String rawText;

	/**

	 * Matched publications identifiers.

	 */

	@XStreamImplicit

	protected List<TypedId> identifiers;

	@XStreamOmitField

	private final static Pattern alphaNumChunkPattern = Pattern.compile("(\\d+\\" + 

			new DecimalFormatSymbols(Locale.getDefault()).getDecimalSeparator() + "\\d+)|(\\d+)|(\\D+)");;

	public BlobCitationEntry() {

		super();

	}

	public BlobCitationEntry(String rawText) {

		this.rawText = rawText;

	}

	public String getRawText() {

		return rawText;

	}

	public void setRawText(String rawText) {

		this.rawText = rawText;

	}

	public List<TypedId> getIdentifiers() {

		return identifiers;

	}

	public void setIdentifiers(List<TypedId> identifiers) {

		this.identifiers = identifiers;

	}

	@Override

	public int hashCode() {

		final int prime = 31;

		int result = 1;

		result = prime * result

				+ ((identifiers == null) ? 0 : identifiers.hashCode());

		result = prime * result + position;

		result = prime * result + ((rawText == null) ? 0 : rawText.hashCode());

		return result;

	}

	@Override

	public boolean equals(Object obj) {

		if (this == obj)

			return true;

		if (obj == null)

			return false;

		if (getClass() != obj.getClass())

			return false;

		BlobCitationEntry other = (BlobCitationEntry) obj;

		if (identifiers == null) {

			if (other.identifiers != null)

				return false;

		} else if (!identifiers.equals(other.identifiers))

			return false;

		if (position != other.position)

			return false;

		if (rawText == null) {

			if (other.rawText != null)

				return false;

		} else if (!rawText.equals(other.rawText))

			return false;

		return true;

	}

	@Override

	public int compareTo(BlobCitationEntry c2) {

		if (c2!=null) {

			if (this.position>c2.position) {

				return 1;

			} else if (this.position<c2.position) {

				return -1;

			} else {

				   if (this.getRawText()!=null) {

						if (c2.getRawText()!=null) {

							int textCompareResult = compareText(this.getRawText(), c2.getRawText());

							if (textCompareResult==0) {

								return compareIdentifiers(this.getIdentifiers(), c2.getIdentifiers());

							} else {

								return textCompareResult;	

							}

						} else {

							return -1;

						}

					} else {

						if (c2.getRawText()!=null) {

							return 1;

						} else {

							return compareIdentifiers(this.getIdentifiers(), c2.getIdentifiers());

						}

					}   

			}

		} else {

			if (this.getRawText()!=null) {

				return -1;	

			} else {

				if (this.getIdentifiers()!=null) {

					return -1;

				} else {

//					should we check position in any way?

					return 0;

				}

			}

		}

	}

	private int compareIdentifiers(List<TypedId> ids1, List<TypedId> ids2) {

		   if (ids2!=null) {

			   if (ids1!=null) {

				   if (ids1.equals(ids2)) {

					   return 0;

				   } else {

//					   no matter what value, order by ids is irrelevant

//					   we have to return non-zero value to prevent treating objects as the same

					   return 1;

				   }

			   } else {

				   return -1;

			   }

			} else {

				if (ids1!=null) {

					return 1;

				} else {

					return 0;

				}

			}

	   }

	   private int compareText(String s1, String s2) {

	      int compareValue = 0;

	      Matcher s1ChunkMatcher = alphaNumChunkPattern.matcher(s1);

	      Matcher s2ChunkMatcher = alphaNumChunkPattern.matcher(s2);

	      String s1ChunkValue = null;

	      String s2ChunkValue = null;

	      while (s1ChunkMatcher.find() && s2ChunkMatcher.find() && compareValue == 0) {

	         s1ChunkValue = s1ChunkMatcher.group();

	         s2ChunkValue = s2ChunkMatcher.group();

	         try {

	            // compare double values - ints get converted to doubles. Eg. 100 = 100.0

	            Double s1Double = Double.valueOf(s1ChunkValue);

	            Double s2Double = Double.valueOf(s2ChunkValue);

	            compareValue = s1Double.compareTo(s2Double);

	         } catch (NumberFormatException e) {

	            // not a number, use string comparison.

	            compareValue = s1ChunkValue.compareTo(s2ChunkValue);

	         }

	         // if they are equal thus far, but one has more left, it should come after the one that doesn't.

	         if (compareValue == 0) {

	            if (s1ChunkMatcher.hitEnd() && !s2ChunkMatcher.hitEnd()) {

	               compareValue = -1;

	            } else if (!s1ChunkMatcher.hitEnd() && s2ChunkMatcher.hitEnd()) {

	               compareValue = 1;

	            }

	         }

	      }

	      return compareValue;

	   }

	/**

	 * @return the position

	 */

	public int getPosition() {

		return position;

	}

	/**

	 * @param position the position to set

	 */

	public void setPosition(int position) {

		this.position = position;

	}

}

package eu.dnetlib.iis.common.protobuf;

import com.google.protobuf.Message;

import org.apache.avro.generic.IndexedRecord;

/**

 * @author Mateusz Fedoryszak (m.fedoryszak@icm.edu.pl)

 */

public interface AvroToProtoBufConverter<IN extends IndexedRecord, OUT extends Message> {

    public String convertIntoKey(IN datum);

    public OUT convertIntoValue(IN datum);

}

package eu.dnetlib.iis.common.protobuf;

import com.google.protobuf.Message;

import org.apache.avro.generic.IndexedRecord;

import org.apache.avro.mapred.AvroKey;

import org.apache.hadoop.io.BytesWritable;

import org.apache.hadoop.io.NullWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.log4j.Logger;

import java.io.IOException;

/**

 * @author Mateusz Fedoryszak (m.fedoryszak@icm.edu.pl)

 */

public class AvroToProtoBufOneToOneMapper<IN extends IndexedRecord, OUT extends Message>

        extends Mapper<AvroKey<IN>, NullWritable, Text, BytesWritable> {

    private static final String CONVERTER_CLASS_PROPERTY = "converter_class";

    private final Logger log = Logger.getLogger(AvroToProtoBufOneToOneMapper.class);

    private final Text keyWritable = new Text();

    private final BytesWritable valueWritable = new BytesWritable();

    private AvroToProtoBufConverter<IN, OUT> converter;

    @SuppressWarnings("unchecked")

    @Override

    public void setup(Context context) throws IOException, InterruptedException {

        Class<?> converterClass = context.getConfiguration().getClass(CONVERTER_CLASS_PROPERTY, null);

        if (converterClass == null) {

            throw new IOException("Please specify " + CONVERTER_CLASS_PROPERTY);

        }

        try {

            converter = (AvroToProtoBufConverter<IN, OUT>) converterClass.newInstance();

        } catch (ClassCastException e) {

            throw new IOException(

                    "Class specified in " + CONVERTER_CLASS_PROPERTY + " doesn't implement AvroToProtoBufConverter");

        } catch (Exception e) {

            throw new IOException(

                    "Could not instantiate specified AvroToProtoBufConverter class, " + converterClass, e);

        }

    }

    @Override

    public void map(AvroKey<IN> avro, NullWritable ignore, Context context)

            throws IOException, InterruptedException {

        String key = null;

        try {

            key = converter.convertIntoKey(avro.datum());

            keyWritable.set(key);

            byte[] value = converter.convertIntoValue(avro.datum()).toByteArray();

            valueWritable.set(value, 0, value.length);

            context.write(keyWritable, valueWritable);

        } catch (Exception e) {

            log.error("Error" + (key != null ? " while processing  " + key : ""), e);

        }

    }

}

package eu.dnetlib.iis.common.utils;

import java.io.File;

import java.io.FileOutputStream;

import java.io.OutputStream;

import java.util.Collections;

import java.util.HashMap;

import java.util.Map;

import java.util.Properties;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import eu.dnetlib.iis.core.java.PortBindings;

import eu.dnetlib.iis.core.java.Ports;

import eu.dnetlib.iis.core.java.Process;

import eu.dnetlib.iis.core.java.io.CloseableIterator;

import eu.dnetlib.iis.core.java.io.DataStore;

import eu.dnetlib.iis.core.java.io.FileSystemPath;

import eu.dnetlib.iis.core.java.porttype.AnyPortType;

import eu.dnetlib.iis.core.java.porttype.PortType;

/**

 * Simple process verifying whether given datastore is empty.

 * @author mhorst

 *

 */

public class EmptyDatastoreVerifierProcess implements Process {

	public static final String INPUT_PORT_NAME = "input";

	public static final String OOZIE_ACTION_OUTPUT_FILENAME = "oozie.action.output.properties";

	public static final String DEFAULT_ENCODING = "UTF-8";

	public static final String OUTPUT_PROPERTY_IS_EMPTY = "isEmpty";

	/**

	 * Ports handled by this module.

	 */

	private static final Ports ports;

	static {

//		preparing ports

		Map<String, PortType> input = new HashMap<String, PortType>();

		input.put(INPUT_PORT_NAME, new AnyPortType());

		Map<String, PortType> output = Collections.emptyMap();

		ports = new Ports(input, output);

	}

	/**

	 * Returns ports in a static way.

	 * @return

	 */

	static protected Ports getStaticPorts() {

		return ports;

	}

	@Override

	public Map<String, PortType> getInputPorts() {

		return getStaticPorts().getInput();

	}

	@Override

	public Map<String, PortType> getOutputPorts() {

		return getStaticPorts().getOutput();

	}

	@Override

	public void run(PortBindings portBindings, Configuration conf,

			Map<String, String> parameters) throws Exception {

		CloseableIterator<?> closeableIt = DataStore.getReader(

				new FileSystemPath(FileSystem.get(conf), 

						portBindings.getInput().get(INPUT_PORT_NAME)));

		try {

			File file = new File(System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME));

	        Properties props = new Properties();

			props.setProperty(OUTPUT_PROPERTY_IS_EMPTY, 

	        		Boolean.toString(!closeableIt.hasNext()));	

			OutputStream os = new FileOutputStream(file);

	        try {

		        	props.store(os, "");	

	        } finally {

	        	os.close();	

	        }	

		} finally {

			closeableIt.close();

		}

	}

}

package eu.dnetlib.iis.common;

public abstract class WorkflowRuntimeParameters {

	private WorkflowRuntimeParameters() {}

	public static final char DEFAULT_CSV_DELIMITER = ',';

	public static final String UNDEFINED_NONEMPTY_VALUE = "$UNDEFINED$";

}

package eu.dnetlib.iis.common.hbase;

import java.io.UnsupportedEncodingException;

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

/**

 * HBase constants

 * @author mhorst

 *

 */

public final class HBaseConstants {

	public static final float CONFIDENCE_TO_TRUST_LEVEL_FACTOR = 0.9f;

	public static final String STATIC_FIELDS_ENCODING_UTF8 = "utf-8";

	public static final char ROW_PREFIX_SEPARATOR = '|';

	public static final String ID_NAMESPACE_SEPARATOR = "::";

	public static final String CLASSIFICATION_HIERARCHY_SEPARATOR = ID_NAMESPACE_SEPARATOR;

	public static final String INFERENCE_PROVENANCE_SEPARATOR = ID_NAMESPACE_SEPARATOR;

	public static final byte[] ROW_PREFIX_RESULT;

	public static final byte[] ROW_PREFIX_PROJECT;

	public static final byte[] ROW_PREFIX_PERSON;

	public static final byte[] QUALIFIER_BODY;

	public static final String SEMANTIC_CLASS_MAIN_TITLE = "main title";

	public static final String SEMANTIC_CLASS_PUBLICATION = "publication";

	public static final String SEMANTIC_CLASS_UNKNOWN = "UNKNOWN";

	public static final String SEMANTIC_CLASS_IIS = "iis";

	public static final String SEMANTIC_SCHEME_DNET_PERSON_ROLES = "dnet:personroles";

	public static final String SEMANTIC_SCHEME_DNET_RELATIONS_RESULT_RESULT = "dnet:result_result_relations";

	public static final String SEMANTIC_SCHEME_DNET_RELATIONS_RESULT_PROJECT = "dnet:result_project_relations";

	public static final String SEMANTIC_SCHEME_DNET_TITLE_TYPOLOGIES = "dnet:title_typologies";

	public static final String SEMANTIC_SCHEME_DNET_RESULT_TYPOLOGIES = "dnet:result_typologies";

	public static final String SEMANTIC_SCHEME_DNET_PROVENANCE_ACTIONS = "dnet:provenanceActions";

	public static final String SEMANTIC_SCHEME_DNET_LANGUAGES = "dnet:languages";

	public static final String SEMANTIC_SCHEME_DNET_PID_TYPES = "dnet:pid_types";

	public static final String SEMANTIC_SCHEME_DNET_CLASSIFICATION_TAXONOMIES = "dnet:subject_classification_typologies";

//	resultResult citation and similarity related

	public static final String SEMANTIC_SCHEME_DNET_DATASET_PUBLICATION_RELS = "dnet:dataset_publication_rels";

	public static final String SEMANTIC_CLASS_TAXONOMIES_ARXIV = "arxiv";

	public static final String SEMANTIC_CLASS_TAXONOMIES_WOS = "wos";

	public static final String SEMANTIC_CLASS_TAXONOMIES_DDC = "ddc";

	public static final String SEMANTIC_CLASS_TAXONOMIES_MESHEUROPMC = "mesheuropmc";

	public static final String SEMANTIC_CLASS_TAXONOMIES_ACM = "acm";

	public static final String EXTERNAL_ID_TYPE_INSTANCE_URL = "dnet:instance-url";

	public static final String EXTERNAL_ID_TYPE_UNKNOWN = "unknown";

//	publication types class ids

	public static final String SEMANTIC_CLASS_INSTANCE_TYPE_ARTICLE = "0001";

	public static final String SEMANTIC_CLASS_INSTANCE_TYPE_DATASET = "0021";

	static {

		try {

			ROW_PREFIX_RESULT = "50|".getBytes(STATIC_FIELDS_ENCODING_UTF8);

			ROW_PREFIX_PROJECT = "40|".getBytes(STATIC_FIELDS_ENCODING_UTF8);

			ROW_PREFIX_PERSON = "30|".getBytes(STATIC_FIELDS_ENCODING_UTF8);

			QUALIFIER_BODY = "body".getBytes(STATIC_FIELDS_ENCODING_UTF8);

		} catch (UnsupportedEncodingException e) {

			throw new RuntimeException(e);

		}

	}

	/**

	 * Returns collumn family name for given entity type.

	 * @param entityType

	 * @return collumn family name for given entity type

	 */

	public static byte[] getCollumnFamily(Type entityType) {

		try {

			return entityType.name().getBytes(STATIC_FIELDS_ENCODING_UTF8);

		} catch (UnsupportedEncodingException e) {

			throw new RuntimeException(e);

		}

	}

	private HBaseConstants() {}

}

package eu.dnetlib.iis.common;

/**

 * Byte array utility class.

 * @author mhorst

 *

 */

public class ByteArrayUtils {

	/**

	 * Does this byte array begin with match array content?

	 * @param source Byte array to examine

	 * @param match Byte array to locate in <code>source</code>

	 * @return true If the starting bytes are equal

	 */

	public static boolean startsWith(byte[] source, byte[] match) {

		return startsWith(source, 0, match);

	}

	/**

	 * Does this byte array begin with match array content?

	 * @param source Byte array to examine

	 * @param offset An offset into the <code>source</code> array

	 * @param match Byte array to locate in <code>source</code>

	 * @return true If the starting bytes are equal

	 */

	public static boolean startsWith(byte[] source, int offset, byte[] match) {

		if (match.length > (source.length - offset)) {

			return false;

		}

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

			if (source[offset + i] != match[i]) {

				return false;

			}

		}

		return true;

	}

}

package eu.dnetlib.iis.common.cache;

import java.io.File;

import java.io.FileOutputStream;

import java.io.IOException;

import java.io.InputStreamReader;

import java.io.OutputStream;

import java.io.OutputStreamWriter;

import java.util.Collections;

import java.util.Map;

import java.util.Properties;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FSDataInputStream;

import org.apache.hadoop.fs.FSDataOutputStream;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import com.google.gson.Gson;

import com.google.gson.stream.JsonWriter;

import eu.dnetlib.iis.common.FsShellPermissions;

import eu.dnetlib.iis.core.java.PortBindings;

import eu.dnetlib.iis.core.java.porttype.PortType;

/**

 * CacheMetadata managing process.

 * 

 * @author mhorst

 *

 */

public class CacheMetadataManagingProcess implements eu.dnetlib.iis.core.java.Process {

	public static final String OUTPUT_PROPERTY_CACHE_ID = "cache_id";

	public static final String PARAM_CACHE_DIR = "default_cache_location";

	public static final String PARAM_MODE = "mode";

	public static final String PARAM_ID = "id";

	public static final String MODE_READ_CURRENT_ID = "read_current_id";

	public static final String MODE_GENERATE_NEW_ID = "generate_new_id";

	public static final String MODE_WRITE_ID = "write_id";

	public static final String DEFAULT_METAFILE_NAME = "meta.json";

	public static final int CACHE_ID_PADDING_LENGTH = 6;

	public static final String NON_EXISTING_CACHE_ID = "$UNDEFINED$"; 

	public static final String OOZIE_ACTION_OUTPUT_FILENAME = "oozie.action.output.properties";

	public static final String DEFAULT_ENCODING = "UTF-8";

	public class CacheMeta {

		protected String currentCacheId;

//		TODO handle this property in a safe way!

//		protected boolean isCacheUnderConstruction;

		public String getCurrentCacheId() {

			return currentCacheId;

		}

		public void setCurrentCacheId(String currentCacheId) {

			this.currentCacheId = currentCacheId;

		}

	}

	@Override

	public Map<String, PortType> getInputPorts() {

		return Collections.emptyMap();

	}

	@Override

	public Map<String, PortType> getOutputPorts() {

		return Collections.emptyMap();

	}

	protected String getExistingCacheId(Configuration conf, Map<String, String> parameters) throws IOException {

		if (parameters.containsKey(PARAM_CACHE_DIR)) {

			FileSystem fs = FileSystem.get(conf);

			Path cacheFilePath = new Path(parameters.get(PARAM_CACHE_DIR), 

					DEFAULT_METAFILE_NAME);

			if (fs.exists(cacheFilePath)) {

				FSDataInputStream inputStream = fs.open(cacheFilePath);

				InputStreamReader reader = new InputStreamReader(

						inputStream, DEFAULT_ENCODING);

				try {

					Gson gson = new Gson();

					CacheMeta cacheMeta = gson.fromJson(reader, CacheMeta.class);

					return cacheMeta.getCurrentCacheId();

				} finally {

					reader.close();

					inputStream.close();

				}

			} else {

//				cache does not exist yet

				return NON_EXISTING_CACHE_ID;

			}

		} else {

			throw new RuntimeException("cache directory location not provided! "

					+ "'" + PARAM_CACHE_DIR + "' parameter is missing!");

		}

	}

	protected String generateNewCacheId(Configuration conf, Map<String, String> parameters) throws IOException {

		if (parameters.containsKey(PARAM_CACHE_DIR)) {

			FileSystem fs = FileSystem.get(conf);

			Path cacheFilePath = new Path(parameters.get(PARAM_CACHE_DIR), 

					DEFAULT_METAFILE_NAME);

			CacheMeta cachedMeta = null;

			if (fs.exists(cacheFilePath)) {

				FSDataInputStream inputStream = fs.open(cacheFilePath);

				InputStreamReader reader = new InputStreamReader(

						inputStream, DEFAULT_ENCODING);

				try {

					Gson gson = new Gson();

					cachedMeta = gson.fromJson(reader, CacheMeta.class);

				} finally {

					reader.close();

					inputStream.close();

				}

			}

			if (cachedMeta!=null) {

				int currentIndex = convertCacheIdToInt(cachedMeta.getCurrentCacheId());

				return convertIntToCacheId(currentIndex+1);

			} else {

//				initializing cache meta

				return convertIntToCacheId(1);

			}

		} else {

			throw new RuntimeException("cache directory location not provided! "

					+ "'" + PARAM_CACHE_DIR + "' parameter is missing!");

		}

	}

	protected void writeCacheId(Configuration conf, Map<String, String> parameters) throws IOException {

		if (parameters.containsKey(PARAM_CACHE_DIR)) {

			if (parameters.containsKey(PARAM_ID)) {

				String cacheId = parameters.get(PARAM_ID);

				FileSystem fs = FileSystem.get(conf);

				Path cacheFilePath = new Path(parameters.get(PARAM_CACHE_DIR), 

						DEFAULT_METAFILE_NAME);

				CacheMeta cachedMeta = null;

				if (fs.exists(cacheFilePath)) {

					FSDataInputStream inputStream = fs.open(cacheFilePath);

					InputStreamReader reader = new InputStreamReader(

							inputStream, DEFAULT_ENCODING);

					try {

						Gson gson = new Gson();

						cachedMeta = gson.fromJson(reader, CacheMeta.class);

					} finally {

						reader.close();

						inputStream.close();

					}

				}

//				writing new id

				if (cachedMeta==null) {

					cachedMeta = new CacheMeta();

				}

				cachedMeta.setCurrentCacheId(cacheId);

				Gson gson = new Gson();

				FSDataOutputStream outputStream = fs.create(cacheFilePath, true);

				JsonWriter writer = new JsonWriter(

						new OutputStreamWriter(outputStream, DEFAULT_ENCODING));

				try {

					gson.toJson(cachedMeta, CacheMeta.class, writer);

				} finally {

					writer.close();

					outputStream.close();

//					changing file permission to +rw to allow writing for different users

					FsShellPermissions.changePermissions(fs, 

							conf, 

							FsShellPermissions.Op.CHMOD, 

							false, "0666", cacheFilePath.toString());

				}

			} else {

				throw new RuntimeException("unable to write new cache id in meta.json file, "

						+ "no '" + PARAM_ID + "' input parameter provied!");

			}

		} else {

			throw new RuntimeException("cache directory location not provided! "

					+ "'" + PARAM_CACHE_DIR + "' parameter is missing!");

		}

	}

	protected static int convertCacheIdToInt(String cacheId) {

		StringBuffer strBuff = new StringBuffer(cacheId);

		while (true) {

			if (strBuff.charAt(0)=='0') {

				strBuff.deleteCharAt(0);

			} else {

				break;

			}

		}

		return Integer.parseInt(strBuff.toString());

	}

	protected static String convertIntToCacheId(int cacheIndex) {

		StringBuffer strBuff = new StringBuffer(String.valueOf(cacheIndex));

		while(strBuff.length()<CACHE_ID_PADDING_LENGTH) {

			strBuff.insert(0, '0');

		}

		return strBuff.toString();

	}

	@Override

	public void run(PortBindings portBindings, Configuration conf,

			Map<String, String> parameters) throws Exception {

		String mode = parameters.get(PARAM_MODE);

		File file = new File(System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME));

        Properties props = new Properties();

		if (MODE_READ_CURRENT_ID.equals(mode)) {

			props.setProperty(OUTPUT_PROPERTY_CACHE_ID, 

	        		getExistingCacheId(conf, parameters));

		} else if (MODE_GENERATE_NEW_ID.equals(mode)) {

			props.setProperty(OUTPUT_PROPERTY_CACHE_ID, 

	        		generateNewCacheId(conf, parameters));

		} else if (MODE_WRITE_ID.equals(mode)) {

			writeCacheId(conf, parameters);

		} else {

			throw new RuntimeException("unsupported mode: " + mode);	

		}

        OutputStream os = new FileOutputStream(file);

        try {

        	props.store(os, "");	

        } finally {

        	os.close();	

        }	

	}

}

package eu.dnetlib.iis.common.affiliation;

import java.util.List;

import org.jdom.Element;

import eu.dnetlib.iis.metadataextraction.schemas.Affiliation;

/**

 * Affiliation builder.

 * @author mhorst

 *

 */

public class AffiliationBuilder {

	/**

	 * Creates affiliation object based on node.

	 * @param node

	 * @return affiliation object

	 */

	public static Affiliation build(Element node) {

		String affId = node.getAttributeValue("id");

        String country = node.getChildText("country");

        String countryCode = null;

        if (node.getChild("country") != null) {

            countryCode = node.getChild("country").getAttributeValue("country");

        }

        String address = node.getChildText("addr-line");

        StringBuilder sb = new StringBuilder();

        List<?> institutions = node.getChildren("institution");

        for (Object institution : institutions) {

            if (!sb.toString().isEmpty()) {

                sb.append(", ");

            }

            sb.append(((Element) institution).getTextTrim());

        }

        String institution = sb.toString().replaceFirst(", $", "");

        if (institution.isEmpty()) {

            institution = null;

        }

        return Affiliation.newBuilder()

                .setRawText(node.getValue().trim().replaceFirst(affId, "").trim())

                .setOrganization(institution!=null?institution.trim():null)

                .setAddress(address!=null?address.trim():null)

                .setCountryName(country!=null?country.trim():null)

                .setCountryCode(countryCode)

                .build();

	}

}

package eu.dnetlib.iis.common.oozie.property;

import java.io.File;

import java.io.FileOutputStream;

import java.io.OutputStream;

import java.util.Collections;

import java.util.Map;

import java.util.Properties;

import org.apache.hadoop.conf.Configuration;

import eu.dnetlib.iis.core.java.PortBindings;

import eu.dnetlib.iis.core.java.Process;

import eu.dnetlib.iis.core.java.porttype.PortType;

/**

 * This process is a solution for setting dynamic properties in oozie workflow definition.

 * 

 * Expects three parameters to be provided: the first 'condition' parameter is boolean value 

 * based on which either first 'inCaseOfTrue' or second 'elseCase' parameter value is set as 

 * the 'result' property.

 *  

 * This can be understood as the: 

 * 

 * condition ? inCaseOfTrue : elseCase

 * 

 * java syntax equivalent.

 * 

 * @author mhorst

 *

 */

public class ConditionalPropertySetter implements Process {

	public static final String PARAM_CONDITION = "condition";

	public static final String PARAM_INCASEOFTRUE = "inCaseOfTrue";

	public static final String PARAM_ELSECASE = "elseCase";

	public static final String OUTPUT_PROPERTY_RESULT = "result";

	public static final String OOZIE_ACTION_OUTPUT_FILENAME = "oozie.action.output.properties";

	@Override

	public Map<String, PortType> getInputPorts() {

		return Collections.emptyMap();

	}

	@Override

	public Map<String, PortType> getOutputPorts() {

		return Collections.emptyMap();

	}

	@Override

	public void run(PortBindings portBindings, Configuration conf,

			Map<String, String> parameters) throws Exception {

		String condition = parameters.get(PARAM_CONDITION);

		if (condition == null) {

			throw new RuntimeException("unable to make decision: " + 

					PARAM_CONDITION + " parameter was not set!");

		}

		Properties props = new Properties();

        props.setProperty(OUTPUT_PROPERTY_RESULT, 

        		Boolean.parseBoolean(condition)?

        				parameters.get(PARAM_INCASEOFTRUE):

        					parameters.get(PARAM_ELSECASE));

        OutputStream os = new FileOutputStream(

        		new File(System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)));

        try {

        	props.store(os, "");	

        } finally {

        	os.close();	

        }

	}

}

package eu.dnetlib.iis.common.oozie.property;

import java.util.Collections;

import java.util.HashMap;

import java.util.Map;

import java.util.Properties;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FSDataOutputStream;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import eu.dnetlib.iis.core.java.PortBindings;

import eu.dnetlib.iis.core.java.porttype.AnyPortType;

import eu.dnetlib.iis.core.java.porttype.PortType;

/**

 * Job properties dumper process. 

 * Writes all properties to job.properties file stored in output directory provided as input parameter.

 * 

 * @author mhorst

 *

 */

public class JobPropertiesDumperProcess implements eu.dnetlib.iis.core.java.Process {

	public static final String OUTPUT_PORT = "output";

	private static final Map<String, PortType> outputPorts = new HashMap<String, PortType>();

	{

		outputPorts.put(OUTPUT_PORT, new AnyPortType());

	}

	@Override

	public void run(PortBindings portBindings, Configuration conf,

			Map<String, String> parameters) throws Exception {

        Properties props = new Properties();

        for(Map.Entry<String,String> property : conf) {

        	props.setProperty(property.getKey(), 

        			property.getValue());

        }

		FileSystem fs = FileSystem.get(conf);

        FSDataOutputStream outputStream = fs.create(new Path(

        				portBindings.getOutput().get(OUTPUT_PORT),

        				"job.properties"), true);

        try {

        	props.store(outputStream, "");	

        } finally {

        	outputStream.close();	

        }	

	}

	@Override

	public Map<String, PortType> getInputPorts() {

		return Collections.emptyMap();

	}

	@Override

	public Map<String, PortType> getOutputPorts() {

		return outputPorts;