Index: modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/PrepareEnrichementMapper.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/PrepareEnrichementMapper.java	(revision 48144)
+++ modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/PrepareEnrichementMapper.java	(nonexistent)
@@ -1,132 +0,0 @@
-package eu.dnetlib.data.mapreduce.hbase.broker.enrich;
-
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.Map;
-
-import org.apache.commons.collections.MapUtils;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.mapreduce.TableMapper;
-import org.apache.hadoop.hbase.util.Bytes;
-
-import com.google.common.base.Function;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-import com.google.protobuf.InvalidProtocolBufferException;
-
-import eu.dnetlib.data.mapreduce.util.DedupUtils;
-import eu.dnetlib.data.mapreduce.util.OafDecoder;
-import eu.dnetlib.data.mapreduce.util.OafRowKeyDecoder;
-import eu.dnetlib.data.proto.FieldTypeProtos.KeyValue;
-import eu.dnetlib.data.proto.OafProtos.Oaf;
-import eu.dnetlib.data.proto.OafProtos.OafRel;
-import eu.dnetlib.data.proto.TypeProtos.Type;
-
-public class PrepareEnrichementMapper extends TableMapper<ImmutableBytesWritable, ImmutableBytesWritable> {
-
-	@Override
-	protected void map(final ImmutableBytesWritable key, final Result value, final Context context) throws IOException, InterruptedException {
-
-		final Type type = getEntityType(key);
-
-		final byte[] body = value.getValue(Bytes.toBytes(type.toString()), Bytes.toBytes("body"));
-
-		if (body == null) {
-			context.getCounter("body", "empty oaf").increment(1);
-			return;
-		}
-
-		switch (type) {
-		case project:
-			for (final ImmutableBytesWritable resultId : listRelatedIds(value, "resultProject_outcome_produces")) {
-				context.write(resultId, new ImmutableBytesWritable(body));
-			}
-			break;
-		case result:
-			final Map<byte[], byte[]> mergedIn = value.getFamilyMap(DedupUtils.getDedupCF_mergedInBytes(Type.result));
-
-			if (MapUtils.isNotEmpty(mergedIn)) { // This is to handle the Dedup Based Enrichemnt Events
-				context.write(getRootId(mergedIn), new ImmutableBytesWritable(body));
-			} else if (!DedupUtils.isRoot(key)) {  // This is to handle the projects related Events
-				final Oaf.Builder oafBuilder = Oaf.newBuilder(OafDecoder.decode(body).getOaf());
-				for (final String relName : Arrays.asList("resultProject_outcome_isProducedBy")) { //  TODO add dataset rels
-					handleRels(oafBuilder, value, body, relName);
-				}
-				context.write(new ImmutableBytesWritable(key.copyBytes()), new ImmutableBytesWritable(oafBuilder.build().toByteArray()));
-			}
-			break;
-		default:
-			context.getCounter("invalid entity", type.toString()).increment(1);
-			break;
-		}
-	}
-
-	public void handleRels(final Oaf.Builder oafBuilder, final Result value, final byte[] body, final String relName) throws IOException, InterruptedException {
-		final Iterable<Oaf> rels = listRelations(value, relName);
-		if (!Iterables.isEmpty(rels)) {
-			for (final Oaf oaf : rels) {
-				final String provenance = oaf.getDataInfo().getProvenanceaction().getClassid();
-				final OafRel.Builder relBuilder = OafRel.newBuilder(oaf.getRel());
-
-				relBuilder.addCollectedfrom(KeyValue.newBuilder().setKey(provenance));
-				oafBuilder.getEntityBuilder().addCachedRel(relBuilder.build());
-			}
-		}
-	}
-
-	private Iterable<Oaf> listRelations(final Result value, final String relType) {
-		final Map<byte[], byte[]> map = value.getFamilyMap(Bytes.toBytes(relType));
-		if (map != null) {
-			return Iterables.transform(map.keySet(), new Function<byte[], Oaf>() {
-
-				@Override
-				public Oaf apply(final byte[] input) {
-					try {
-						return Oaf.parseFrom(input);
-					} catch (final InvalidProtocolBufferException e) {
-						throw new RuntimeException(e);
-					}
-				}
-			});
-		} else {
-			return Lists.newArrayList();
-		}
-	}
-
-	private Iterable<ImmutableBytesWritable> listRelatedIds(final Result value, final String relType) {
-		final Map<byte[], byte[]> map = value.getFamilyMap(Bytes.toBytes(relType));
-		if (map != null) {
-			return Iterables.transform(map.keySet(), new Function<byte[], ImmutableBytesWritable>() {
-
-				@Override
-				public ImmutableBytesWritable apply(final byte[] input) {
-					return new ImmutableBytesWritable(input);
-				}
-			});
-		} else {
-			return Lists.newArrayList();
-		}
-	}
-
-	private Type getEntityType(final ImmutableBytesWritable key) {
-		return OafRowKeyDecoder.decode(key.copyBytes()).getType();
-	}
-
-	private ImmutableBytesWritable getRootId(final Map<byte[], byte[]> mergedIn) {
-		final Iterable<ImmutableBytesWritable> ids = Iterables.transform(mergedIn.keySet(), new Function<byte[], ImmutableBytesWritable>() {
-
-			@Override
-			public ImmutableBytesWritable apply(final byte[] input) {
-				return new ImmutableBytesWritable(input);
-			}
-		});
-		try {
-			return Iterables.getOnlyElement(ids);
-		} catch (final IllegalArgumentException e) {
-			System.err.println(ids);
-			throw e;
-		}
-	}
-
-}
Index: modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/PrepareEnrichementReducer.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/PrepareEnrichementReducer.java	(revision 48144)
+++ modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/PrepareEnrichementReducer.java	(nonexistent)
@@ -1,199 +0,0 @@
-package eu.dnetlib.data.mapreduce.hbase.broker.enrich;
-
-import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getKey;
-import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getPropertyValues;
-
-import java.io.IOException;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.commons.lang.StringUtils;
-import org.apache.commons.lang.math.RandomUtils;
-import org.apache.hadoop.hbase.client.HTable;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.filter.FilterList;
-import org.apache.hadoop.hbase.filter.FilterList.Operator;
-import org.apache.hadoop.hbase.filter.PrefixFilter;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapreduce.Reducer;
-import org.dom4j.DocumentException;
-
-import com.google.common.base.Function;
-import com.google.common.collect.Iterables;
-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.data.mapreduce.hbase.broker.AbstractEventFactory;
-import eu.dnetlib.data.mapreduce.hbase.broker.OAVersionEventFactory;
-import eu.dnetlib.data.mapreduce.hbase.broker.PIDEventFactory;
-import eu.dnetlib.data.mapreduce.hbase.broker.ProjectEventFactory;
-import eu.dnetlib.data.mapreduce.hbase.broker.PublicationDateEventFactory;
-import eu.dnetlib.data.proto.OafProtos.Oaf;
-import eu.dnetlib.data.proto.OafProtos.OafRel;
-import eu.dnetlib.data.proto.TypeProtos.Type;
-
-public class PrepareEnrichementReducer extends Reducer<ImmutableBytesWritable, ImmutableBytesWritable, Text, Text> {
-
-	private static final int LIMIT = 1000;
-
-	private Map<String, String> dsTypeMap = Maps.newHashMap();
-
-	private Set<String> dsWhitelist = Sets.newHashSet();
-
-	private Set<String> dsBlacklist = Sets.newHashSet();
-
-	// This is for EuropePMC. They expose OA abstracts, but we want to identify real OA publications. WTF.
-	private Set<String> untrustedOaDsList = Sets.newHashSet();
-
-	// White list for datasource typologies.
-	private Set<String> dsTypeWhitelist = Sets.newHashSet();
-
-	@Override
-	protected void setup(final Context context) throws IOException, InterruptedException {
-		super.setup(context);
-
-		System.out.println("LIMIT: " + LIMIT);
-
-		dsWhitelist.addAll(getPropertyValues(context, "broker.datasource.id.whitelist"));
-		dsBlacklist.addAll(getPropertyValues(context, "broker.datasource.id.blacklist"));
-		dsTypeWhitelist.addAll(getPropertyValues(context, "broker.datasource.type.whitelist"));
-		untrustedOaDsList.addAll(getPropertyValues(context, "broker.datasource.untrusted.oa.list"));
-
-		dsTypeMap = getDsTypeMap(context, dsTypeWhitelist);
-
-		System.out.println("datasource whitelist: " + dsWhitelist);
-		System.out.println("datasource blacklist: " + dsBlacklist);
-		System.out.println("datasource OA list: " + untrustedOaDsList);
-
-		System.out.println("datasource type whitelist: " + dsTypeWhitelist);
-	}
-
-	@Override
-	protected void reduce(final ImmutableBytesWritable key, final Iterable<ImmutableBytesWritable> values, final Context context)
-			throws IOException, InterruptedException {
-
-		final Map<String, Oaf> projects = Maps.newHashMap();
-		final List<Oaf> pubs = Lists.newArrayList();
-
-		for (final Oaf oaf : Iterables.transform(Iterables.limit(values, LIMIT), oafDeserialiser())) {
-			if (oaf.getEntity().getType() == Type.project) {
-				projects.put(oaf.getEntity().getId(), oaf);
-			} else {
-				pubs.add(oaf);
-			}
-		}
-		try {
-			generateEvents(pubs, projects, context);
-		} catch (final DocumentException e) {
-			throw new RuntimeException(e);
-		}
-
-	}
-
-	private Map<String, String> getDsTypeMap(final Context context, final Set<String> dsTypeWhitelist) throws IOException {
-		System.out.println("loading datasource typology mapping");
-
-		final Map<String, String> dsTypeMap = Maps.newHashMap();
-
-		final Scan scan = new Scan();
-		final FilterList fl = new FilterList(Operator.MUST_PASS_ALL);
-		fl.addFilter(new PrefixFilter(Bytes.toBytes("10")));
-		scan.setFilter(fl);
-		scan.addFamily(Bytes.toBytes("datasource"));
-
-		final String tableName = context.getConfiguration().get("hbase.mapred.inputtable");
-
-		System.out.println(String.format("table name: '%s'", tableName));
-
-		try (final HTable table = new HTable(context.getConfiguration(), tableName); final ResultScanner res = table.getScanner(scan)) {
-
-			for (final Result r : res) {
-				final byte[] b = r.getValue(Bytes.toBytes("datasource"), Bytes.toBytes("body"));
-				if (b != null) {
-					final Oaf oaf = Oaf.parseFrom(b);
-					final String dsId = StringUtils.substringAfter(oaf.getEntity().getId(), "|");
-					final String dsType = oaf.getEntity().getDatasource().getMetadata().getDatasourcetype().getClassid();
-
-					if (dsTypeWhitelist.contains(dsType)) {
-						System.out.println(String.format("dsId '%s', dsType '%s'", dsId, dsType));
-						dsTypeMap.put(dsId, dsType);
-					}
-				}
-			}
-
-		}
-		System.out.println("datasource type map size: " + dsTypeMap.size());
-		return dsTypeMap;
-	}
-
-	private void generateEvents(final List<Oaf> pubs, final Map<String, Oaf> projects, final Context context)
-			throws IOException, InterruptedException, DocumentException {
-
-		for (final Oaf current : pubs) {
-
-			final String currentId = current.getEntity().getId();
-
-			final String currentDsId = StringUtils.substringAfter(getKey(current.getEntity().getCollectedfromList()), "|");
-			final String currentDsType = dsTypeMap.get(currentDsId);
-
-			// System.out.println(String.format("'%s' -> '%s'", currentDsId, currentDsType));
-
-			if (StringUtils.isBlank(currentDsType) && !dsWhitelist.contains(currentDsId)) {
-				context.getCounter("events skipped", "datasource type excluded").increment(1);
-			} else if (dsBlacklist.contains(currentDsId)) {
-				context.getCounter("events skipped", "datasource blacklisted").increment(1);
-			} else {
-				for (final Oaf other : pubs) {
-					final String otherId = other.getEntity().getId();
-					if (!currentId.equals(otherId)) {
-						final Oaf enrichedCurrent = addProjects(current, projects, Sets.newHashSet("sysimport:crosswalk:repository", "iis"));
-						PIDEventFactory.process(context, enrichedCurrent, other, RandomUtils.nextFloat());
-						OAVersionEventFactory.process(context, enrichedCurrent, other, RandomUtils.nextFloat(), untrustedOaDsList);
-						AbstractEventFactory.process(context, enrichedCurrent, other, RandomUtils.nextFloat());
-						PublicationDateEventFactory.process(context, enrichedCurrent, other, RandomUtils.nextFloat());
-
-						final Oaf enrichedOther = addProjects(other, projects, Sets.newHashSet("sysimport:crosswalk:repository", "iis"));
-						ProjectEventFactory.process(context, enrichedCurrent, enrichedOther, RandomUtils.nextFloat());
-					}
-				}
-				final Oaf enrichedCurrent = addProjects(current, projects, Sets.newHashSet("iis"));
-				ProjectEventFactory.process(context, enrichedCurrent, null, RandomUtils.nextFloat());
-			}
-		}
-	}
-
-	private Oaf addProjects(final Oaf current, final Map<String, Oaf> projects, final Set<String> allowedProvenance) {
-
-		final Oaf.Builder oafBuilder = Oaf.newBuilder(current);
-		for (final OafRel.Builder rel : oafBuilder.getEntityBuilder().getCachedRelBuilderList()) {
-			final String provenance = Iterables.getOnlyElement(rel.getCollectedfromBuilderList()).getKey();
-			if (projects.containsKey(rel.getTarget()) & allowedProvenance.contains(provenance)) {
-				rel.setCachedTarget(projects.get(rel.getTarget()).getEntity());
-			}
-		}
-
-		return oafBuilder.build();
-	}
-
-	private Function<ImmutableBytesWritable, Oaf> oafDeserialiser() {
-		return new Function<ImmutableBytesWritable, Oaf>() {
-
-			@Override
-			public Oaf apply(final ImmutableBytesWritable input) {
-				try {
-					return Oaf.parseFrom(input.copyBytes());
-				} catch (final InvalidProtocolBufferException e) {
-					throw new IllegalArgumentException(e);
-				}
-			}
-		};
-	}
-
-}
Index: modules/dnet-mapreduce-jobs/trunk/src/test/java/eu/dnetlib/data/mapreduce/hbase/index/config/RelClassesTest.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/test/java/eu/dnetlib/data/mapreduce/hbase/index/config/RelClassesTest.java	(revision 48144)
+++ modules/dnet-mapreduce-jobs/trunk/src/test/java/eu/dnetlib/data/mapreduce/hbase/index/config/RelClassesTest.java	(revision 48145)
@@ -8,13 +8,31 @@
 public class RelClassesTest {
 
 	public static final String relClassesJson =
-			"[\"isProvidedBy:provides\",\r\n\"provides:isProvidedBy\",\r\n\"merges:isMergedIn\",\r\n\"isMergedIn:merges\",\r\n\"isSimilarTo:isSimilarTo\",\r\n\"isCoAuthorOf:isCoAuthorOf\",\r\n\"isAuthorOf:hasAuthor\",\r\n\"hasAuthor:isAuthorOf\",\r\n\"isParticipant:hasParticipant\",\r\n\"hasParticipant:isParticipant\",\r\n\"isProducedBy:produces\",\r\n\"produces:isProducedBy\",\r\n\"hasAmongTopNSimilarDocuments:isAmongTopNSimilarDocuments\",\r\n\"isAmongTopNSimilarDocuments:hasAmongTopNSimilarDocuments\",\r\n\"isRelatedTo:isRelatedTo\",\r\n\"isContact:hasContact\",\r\n\"hasContact:isContact\",\n"
+			"[\"isProvidedBy:provides\","
+					+ "\"provides:isProvidedBy\","
+					+ "\"merges:isMergedIn\","
+					+ "\"isMergedIn:merges\","
+					+ "\"isSimilarTo:isSimilarTo\","
+					+ "\"isCoAuthorOf:isCoAuthorOf\","
+					+ "\"isAuthorOf:hasAuthor\","
+					+ "\"hasAuthor:isAuthorOf\","
+					+ "\"isParticipant:hasParticipant\","
+					+ "\"hasParticipant:isParticipant\","
+					+ "\"isProducedBy:produces\","
+					+ "\"produces:isProducedBy\","
+					+ "\"hasAmongTopNSimilarDocuments:isAmongTopNSimilarDocuments\","
+					+ "\"isAmongTopNSimilarDocuments:hasAmongTopNSimilarDocuments\","
+					+ "\"isRelatedTo:isRelatedTo\","
+					+ "\"isContact:hasContact\","
+					+ "\"hasContact:isContact\","
 					+ "\"isContributorOf:hasContributor\","
 					+ "\"hasContributor:isContributorOf\","
-					+ "\"isPartOf:hasPart\",\n"
-					+ "\"hasPart:isPartOf\",\n"
+					+ "\"isPartOf:hasPart\","
+					+ "\"hasPart:isPartOf\","
 					+ "\"isSupplementedBy:isSupplementTo\","
-					+ "\"isSupplementTo:isSupplementedBy\"]";
+					+ "\"isSupplementTo:isSupplementedBy\","
+					+ "\"hasAuthorInstitution:isAuthorInstitutionOf\","
+					+ "\"isAuthorInstitutionOf:hasAuthorInstitution\"]";
 
 	@Test
 	public void testRelClasses() {
Index: modules/dnet-mapreduce-jobs/trunk/src/test/resources/eu/dnetlib/data/transform/result_organization.xml
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/test/resources/eu/dnetlib/data/transform/result_organization.xml	(nonexistent)
+++ modules/dnet-mapreduce-jobs/trunk/src/test/resources/eu/dnetlib/data/transform/result_organization.xml	(revision 48145)
@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<ROW>
+	<FIELD name="organization">corda__h2020::937348734</FIELD>
+	<FIELD name="result">50|od_______296::74732b873f881d9bff67c2a11b6a66d6</FIELD>
+	<FIELD name="provenanceaction">sysimport:crosswalk:entityregistry@@@sysimport:crosswalk:entityregistry@@@dnet:provenanceActions@@@dnet:provenanceActions</FIELD>
+	<FIELD name="semantics">coordinator@@@coordinator@@@dnet:project_organization_relations@@@dnet:project_organization_relations</FIELD>
+	<FIELD name="trust">0.9</FIELD>
+	<FIELD name="inferenceprovenance" isNull="true"/>
+	<FIELD name="deletedbyinference">false</FIELD>
+	<FIELD name="inferred">false</FIELD>
+</ROW>
Index: modules/dnet-mapreduce-jobs/trunk/src/test/resources/eu/dnetlib/data/transform/record.xml
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/test/resources/eu/dnetlib/data/transform/record.xml	(revision 48144)
+++ modules/dnet-mapreduce-jobs/trunk/src/test/resources/eu/dnetlib/data/transform/record.xml	(revision 48145)
@@ -54,5 +54,6 @@
     	type="dataset">https://www.ebi.ac.uk/chembldb/assay/inspect/CHEMBL1658379</oaf:reference>
     <oaf:hostedBy id="opendoar____::296" name="Tesis Doctorals en Xarxa"/>
     <oaf:collectedFrom id="opendoar____::296" name="Tesis Doctorals en Xarxa"/>
+    <oaf:journal eissn="" ep="238" iss="February" issn="" sp="231" vol="1II">Constatin Brancusi University of Targu Jiu Annals - Economy Series</oaf:journal>
   </metadata>
 </record>
Index: modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/ProjectEventFactory.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/ProjectEventFactory.java	(revision 48144)
+++ modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/ProjectEventFactory.java	(revision 48145)
@@ -1,73 +1,159 @@
 package eu.dnetlib.data.mapreduce.hbase.broker;
 
-import static eu.dnetlib.data.mapreduce.hbase.broker.mapping.EventFactory.asEvent;
-
 import java.io.IOException;
+import java.util.Map;
+import java.util.Set;
 
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapreduce.Reducer.Context;
-import org.dom4j.DocumentException;
-
+import com.google.common.base.Predicate;
 import com.google.common.collect.Iterables;
-
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import eu.dnetlib.broker.objects.Instance;
 import eu.dnetlib.broker.objects.OpenAireEventPayload;
+import eu.dnetlib.broker.objects.Project;
+import eu.dnetlib.broker.objects.Provenance;
 import eu.dnetlib.data.mapreduce.hbase.broker.mapping.HighlightFactory;
 import eu.dnetlib.data.mapreduce.hbase.broker.mapping.OpenAireEventPayloadFactory;
+import eu.dnetlib.data.mapreduce.hbase.broker.mapping.ProtoMapping;
 import eu.dnetlib.data.mapreduce.hbase.broker.model.EventMessage;
+import eu.dnetlib.data.mapreduce.util.DedupUtils;
 import eu.dnetlib.data.proto.OafProtos.Oaf;
 import eu.dnetlib.data.proto.OafProtos.OafEntity;
-import eu.dnetlib.data.proto.OafProtos.OafRel;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Reducer.Context;
+import org.dom4j.DocumentException;
 
+import static eu.dnetlib.data.mapreduce.hbase.broker.mapping.EventFactory.asEvent;
+import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getValue;
+
 @SuppressWarnings("rawtypes")
-public class ProjectEventFactory {
+public class ProjectEventFactory extends ProtoMapping {
 
+	private static String COUNTER_GROUP = "Broker Enrichment projects";
+
+	private Set<String> inferenceProvenance = Sets.newHashSet("sysimport:mining:repository", "iis");
+	private Set<String> aggregationProvenance = Sets.newHashSet("sysimport:crosswalk:repository");
+
 	protected Text tKey = new Text("");
 
-	public static void process(final Context context, final Oaf current, final Oaf other, final float trust)
+	public static void process(final Context context, final Oaf current, final Oaf other, final Float trust)
 			throws IOException, InterruptedException, DocumentException {
+
+		/*
+		if (!current.getRel().hasCachedOafTarget() || (other != null && !other.getRel().hasCachedOafTarget())) {
+			context.getCounter(COUNTER_GROUP, "events skipped: missing project 2nd step").increment(1);
+			return;
+		}
+		*/
+
 		new ProjectEventFactory().processProject(context, current, other, trust);
 	}
 
-	private void processProject(final Context context, final Oaf current, final Oaf other, final float trust)
+	public static void process(final Context context, final Oaf current)
 			throws IOException, InterruptedException, DocumentException {
+		process(context, current, null, null);
+	}
 
+	private void processProject(final Context context, final Oaf current, final Oaf other, final Float trust)
+			throws IOException, InterruptedException, DocumentException {
+
 		if (other == null) {
-			for (final OafRel rel : current.getEntity().getCachedRelList()) {
-				doProcessProject(context, current, rel.getCachedTarget(), getProvenance(rel), Topic.ENRICH_MISSING_PROJECT, trust);
+			for (final Oaf oafRel : current.getEntity().getCachedOafRelList()) {
+
+				final String provenance = oafRel.getDataInfo().getProvenanceaction().getClassid();
+				if (inferenceProvenance.contains(provenance)) {
+					final OafEntity project = oafRel.getRel().getCachedOafTarget().getEntity();
+					doProcessProject(context, current, current, project, provenance, Topic.ENRICH_MISSING_PROJECT, trust(trust, oafRel));
+				}
 			}
 		} else {
-			for (final OafRel currentRel : current.getEntity().getCachedRelList()) {
-				for (final OafRel otherRel : other.getEntity().getCachedRelList()) {
-					if (!currentRel.getTarget().equals(otherRel.getTarget())) {
-						doProcessProject(context, current, other.getEntity(), getProvenance(otherRel), Topic.ENRICH_MISSING_PROJECT, trust);
+			for (final Oaf currentOafRel : current.getEntity().getCachedOafRelList()) {
+				for (final Oaf otherOafRel : other.getEntity().getCachedOafRelList()) {
+
+					final String currentTarget = currentOafRel.getRel().getTarget();
+					final String otherTarget = otherOafRel.getRel().getTarget();
+
+					if (!currentTarget.equals(otherTarget) && !DedupUtils.isRoot(current.getEntity().getId())) {
+
+						final String provenance = otherOafRel.getDataInfo().getProvenanceaction().getClassid();
+
+						final OafEntity project = otherOafRel.getRel().getCachedOafTarget().getEntity();
+
+						boolean currentHasProject = Iterables.tryFind(current.getEntity().getCachedOafRelList(), new Predicate<Oaf>() {
+							@Override
+							public boolean apply(final Oaf oaf) {
+								final String currentProjectId = oaf.getRel().getCachedOafTarget().getEntity().getId();
+								//System.out.println(String.format("%s = %s ? %s", currentProjectId, project.getId(), currentProjectId.equals(project.getId())));
+								return currentProjectId.equals(project.getId());
+							}
+						}).isPresent();
+
+						if (!currentHasProject) {
+							//System.out.println(String.format("generating event for other = %s\n\nproject = %s", other, project));
+							doProcessProject(context, current, other, project, provenance, Topic.ENRICH_MISSING_PROJECT, trust(trust, currentOafRel));
+						}
 					}
-
 				}
 			}
 		}
 	}
 
-	private String getProvenance(final OafRel rel) {
-		return Iterables.getOnlyElement(rel.getCollectedfromList()).getKey();
-	}
-
 	private void doProcessProject(final Context context,
 			final Oaf current,
+			final Oaf other,
 			final OafEntity project,
 			final String provenance,
 			final Topic topic,
-			final float trust)
+			final Float trust)
 			throws IOException, InterruptedException, DocumentException {
 
-		// final OafEntity oaf, final Topic topic, final OpenAireEventPayload payload, final OafEntity source, final float trust
+		final OafEntity currentEntity = current.getEntity();
+		final OafEntity otherEntity = other.getEntity();
 
-		final OpenAireEventPayload payload = OpenAireEventPayloadFactory.fromOAF(current.getEntity(), current.getEntity(), trust);
-		final EventMessage event = asEvent(current.getEntity(), topic, payload, current.getEntity(), trust);
+		//System.out.println("ProjectEventFactory.doProcessProject = " + entity);
 
+		final Provenance prov = getProvenance(otherEntity, provenance);
+
+		final OpenAireEventPayload payload = addProject(OpenAireEventPayloadFactory.fromOAF(currentEntity, trust, prov), project);
+
+		final EventMessage event = asEvent(currentEntity, topic, payload, otherEntity, trust);
 		event.setPayload(HighlightFactory.highlightEnrichProject(payload, project, provenance).toJSON());
 
 		context.write(tKey, new Text(event.toString()));
-		context.getCounter("event", topic.getValue()).increment(1);
+		context.getCounter(COUNTER_GROUP, "Event:  " + topic.getValue()).increment(1);
 	}
 
+	private OpenAireEventPayload addProject(final OpenAireEventPayload payload, final OafEntity project) {
+		final Map<String, Project> projects = Maps.newHashMap();
+		for(Project prj : payload.getPublication().getProjects()) {
+			projects.put(prj.getCode() + prj.getTitle(), prj);
+		}
+		final Project hlProject = mapRelatedProject(project.getProject());
+		projects.put(hlProject.getCode() + hlProject.getTitle(), hlProject);
+
+		payload.getPublication().setProjects(Lists.newArrayList(projects.values()));
+
+		return payload;
+	}
+
+	private Provenance getProvenance(final OafEntity entity, final String provenance) {
+		if (inferenceProvenance.contains(provenance)) {
+			return new Provenance()
+					.setRepositoryName("OpenAIRE")
+					.setUrl("https://beta.openaire.eu/search/publication?articleId=" + StringUtils.substringAfter(entity.getId(), "|"))
+					.setId(Iterables.getFirst(entity.getOriginalIdList(), ""));
+		}
+		return new Provenance()
+				.setRepositoryName(getValue(entity.getCollectedfromList()))
+				.setUrl(Iterables.getFirst(mapInstances(entity.getResult().getInstanceList()), new Instance()).getUrl())
+				.setId(getValue(entity.getOriginalIdList()));
+	}
+
+	private Float trust(final Float trust, final Oaf oaf) {
+		final Float provenanceTrust = Float.valueOf(oaf.getDataInfo().getTrust());
+		return trust != null ? trust * provenanceTrust : provenanceTrust;
+	}
+
 }
Index: modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/mapping/HighlightFactory.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/mapping/HighlightFactory.java	(revision 48144)
+++ modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/mapping/HighlightFactory.java	(revision 48145)
@@ -1,22 +1,17 @@
 package eu.dnetlib.data.mapreduce.hbase.broker.mapping;
 
-import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getValue;
-import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.listValues;
-
 import java.util.List;
 
-import com.google.common.base.Function;
-import com.google.common.base.Joiner;
-import com.google.common.collect.Iterables;
-
+import com.google.common.collect.Lists;
 import eu.dnetlib.broker.objects.OpenAireEventPayload;
 import eu.dnetlib.data.proto.FieldTypeProtos.StringField;
 import eu.dnetlib.data.proto.FieldTypeProtos.StructuredProperty;
 import eu.dnetlib.data.proto.OafProtos.OafEntity;
-import eu.dnetlib.data.proto.ProjectProtos.Project;
-import eu.dnetlib.data.proto.ProjectProtos.Project.Metadata;
 import eu.dnetlib.data.proto.ResultProtos.Result.Instance;
 
+import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getValue;
+import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.listValues;
+
 /**
  * Created by alessia on 21/07/16.
  */
@@ -48,24 +43,12 @@
 	}
 
 	// TODO: highlight for the following reasons
-	public static OpenAireEventPayload highlightEnrichProject(final OpenAireEventPayload p, final OafEntity projectEntity, final String provenance) {
+	public static OpenAireEventPayload highlightEnrichProject(final OpenAireEventPayload p, final OafEntity project, final String provenance) {
 		// TODO: this can wait. Think about generating the openaire string for project links: it will be easier for subscribers to integrate
 		// it back to their records!
-		final Project project = projectEntity.getProject();
-		final Metadata projectMetadata = project.getMetadata();
-		final StringField acro = projectMetadata.getAcronym();
-		final StringField title = projectMetadata.getTitle();
-		final StringField code = projectMetadata.getCode();
-		final List<StringField> fundingtree = projectMetadata.getFundingtreeList();
-		final String fundingTrees = "<fundings>" + Joiner.on(" ").join(Iterables.transform(fundingtree, new Function<StringField, String>() {
 
-			@Override
-			public String apply(final StringField v) {
-				return v.getValue();
-			}
-		}
+		p.getHighlight().setProjects(Lists.newArrayList(mapRelatedProject(project.getProject())));
 
-		)) + "</fundings>";
 		return p;
 	}
 
@@ -82,9 +65,9 @@
 	// Commenting highlights for addition events: the record is new for the subscriber!
 	/*
 	 * public static OpenAireEventPayload highlightAddByAffiliation(final OpenAireEventPayload p){ return p; }
-	 * 
+	 *
 	 * public static OpenAireEventPayload highlightAddByReference(final OpenAireEventPayload p){ return p; }
-	 * 
+	 *
 	 * public static OpenAireEventPayload highlightAddByProject(final OpenAireEventPayload p){ //TODO: this cannot wait return p; }
 	 */
 
Index: modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/mapping/ProtoMapping.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/mapping/ProtoMapping.java	(revision 48144)
+++ modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/mapping/ProtoMapping.java	(revision 48145)
@@ -1,29 +1,30 @@
 package eu.dnetlib.data.mapreduce.hbase.broker.mapping;
 
-import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getKey;
-import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getValue;
-
 import java.io.StringReader;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
-import org.apache.commons.lang.StringUtils;
-import org.dom4j.Document;
-import org.dom4j.DocumentException;
-import org.dom4j.io.SAXReader;
-
 import com.google.common.base.Function;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
-
-import eu.dnetlib.broker.objects.ExternalReference;
-import eu.dnetlib.broker.objects.Instance;
-import eu.dnetlib.broker.objects.Pid;
-import eu.dnetlib.broker.objects.Project;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import eu.dnetlib.broker.objects.*;
 import eu.dnetlib.data.proto.FieldTypeProtos.StructuredProperty;
+import eu.dnetlib.data.proto.OafProtos.Oaf;
 import eu.dnetlib.data.proto.OafProtos.OafEntity;
-import eu.dnetlib.data.proto.OafProtos.OafRel;
+import eu.dnetlib.data.proto.ProjectProtos;
+import eu.dnetlib.data.proto.ResultProtos;
 import eu.dnetlib.data.proto.ResultProtos.Result;
+import org.apache.commons.lang.StringUtils;
+import org.dom4j.Document;
+import org.dom4j.DocumentException;
+import org.dom4j.io.SAXReader;
 
+import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getKey;
+import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getValue;
+
 /**
  * Created by claudio on 22/07/16.
  */
@@ -30,7 +31,7 @@
 public abstract class ProtoMapping {
 
 	protected static List<Instance> mapInstances(final List<Result.Instance> in) {
-		return Lists.newArrayList(Iterables.transform(in, new Function<Result.Instance, Instance>() {
+		final Set<Instance> instances = Sets.newHashSet(Iterables.transform(in, new Function<Result.Instance, Instance>() {
 
 			@Override
 			public Instance apply(final Result.Instance i) {
@@ -41,6 +42,7 @@
 						.setUrl(Iterables.getFirst(i.getUrlList(), ""));
 			}
 		}));
+		return Lists.newArrayList(instances);
 	}
 
 	protected static List<Pid> mapPids(final List<StructuredProperty> sp) {
@@ -53,6 +55,14 @@
 		}));
 	}
 
+	protected static Journal mapJournal(final ResultProtos.Result.Journal j) {
+		return new eu.dnetlib.broker.objects.Journal()
+				.setName(j.getName())
+				.setIssn(j.getIssnPrinted())
+				.setEissn(j.getIssnOnline())
+				.setLissn(j.getIssnLinking());
+	}
+
 	protected static List<ExternalReference> mapExternalRefs(final List<Result.ExternalReference> ext) {
 		return Lists.newArrayList(Iterables.transform(ext, new Function<Result.ExternalReference, ExternalReference>() {
 
@@ -67,43 +77,48 @@
 		}));
 	}
 
-	protected static final List<Project> mapRelatedProjects(final OafEntity oaf) {
+	protected static final List<Project> mapRelatedProjects(final OafEntity entity) {
 
-		// TODO: per ogni REL di tipo progetto creare un oggetto Project (sfruttando gli elementi nella seconda lista)
-		final Function<OafRel, Project> toProject = new Function<OafRel, Project>() {
+		final Map<String, Oaf> projectMap = Maps.newHashMap();
+		for(Oaf rel : entity.getCachedOafRelList()) {
+			final OafEntity p = rel.getRel().getCachedOafTarget().getEntity();
+			projectMap.put(p.getId(), Oaf.newBuilder(rel).build());
+		}
 
-			@Override
-			public Project apply(final OafRel rel) {
-				final Project project = new Project();
-				switch (rel.getRelType()) {
-				case resultProject:
-					final eu.dnetlib.data.proto.ProjectProtos.Project.Metadata mp = rel.getCachedTarget().getProject().getMetadata();
-					final String ftree = getValue(mp.getFundingtreeList());
+		return Lists.transform(Lists.newArrayList(projectMap.values()), getProjectMappingFunction());
+	}
 
-					project.setCode(getValue(mp.getCode()));
-					project.setAcronym(getValue(mp.getAcronym()));
-					project.setTitle(getValue(mp.getTitle()));
+	protected static final Project mapRelatedProject(final ProjectProtos.Project project) {
+		final Project p = new Project();
+		final ProjectProtos.Project.Metadata mp = project.getMetadata();
 
-					if (StringUtils.isNotBlank(ftree)) {
-						try {
-							final Document fdoc = new SAXReader().read(new StringReader(ftree));
-							project.setFunder(fdoc.valueOf("/funder/shortname"));
-							project.setJurisdiction(fdoc.valueOf("/funder/jurisdiction"));
-							project.setFundingProgram(fdoc.valueOf("//funding_level_0/name"));
-						} catch (final DocumentException e) {
-							throw new RuntimeException(e);
-						}
-					}
-					break;
-				default:
-					break;
-				}
-				return project;
+		p.setCode(getValue(mp.getCode()));
+		p.setAcronym(getValue(mp.getAcronym()));
+		p.setTitle(getValue(mp.getTitle()));
+
+		final String ftree = getValue(mp.getFundingtreeList());
+		if (StringUtils.isNotBlank(ftree)) {
+			try {
+				final Document fdoc = new SAXReader().read(new StringReader(ftree));
+				p.setFunder(fdoc.valueOf("/fundingtree/funder/shortname"));
+				p.setJurisdiction(fdoc.valueOf("/fundingtree/funder/jurisdiction"));
+				p.setFundingProgram(fdoc.valueOf("//funding_level_0/name"));
+			} catch (final DocumentException e) {
+				throw new RuntimeException(e);
 			}
+		}
+		//System.out.println("ProtoMapping.mapRelatedProjects project = " + project.toJSON());
+		return p;
+	}
+
+	private static Function<Oaf, Project> getProjectMappingFunction() {
+		return new Function<Oaf, Project>() {
+
+			@Override
+			public Project apply(final Oaf oafRel) {
+				return mapRelatedProject(oafRel.getRel().getCachedOafTarget().getEntity().getProject());
+			}
 		};
-		if (oaf.getCachedRelList() != null) { return Lists.transform(oaf.getCachedRelList(), toProject); }
-		return Lists.newArrayList();
-
 	}
 
 }
Index: modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/mapping/OpenAireEventPayloadFactory.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/mapping/OpenAireEventPayloadFactory.java	(revision 48144)
+++ modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/mapping/OpenAireEventPayloadFactory.java	(revision 48145)
@@ -1,21 +1,15 @@
 package eu.dnetlib.data.mapreduce.hbase.broker.mapping;
 
-import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getKey;
-import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getValue;
-import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.listValues;
-
-import org.dom4j.DocumentException;
-
 import com.google.common.collect.Iterables;
-
-import eu.dnetlib.broker.objects.Journal;
+import eu.dnetlib.broker.objects.Instance;
 import eu.dnetlib.broker.objects.OpenAireEventPayload;
 import eu.dnetlib.broker.objects.Provenance;
 import eu.dnetlib.broker.objects.Publication;
 import eu.dnetlib.data.proto.OafProtos.OafEntity;
 import eu.dnetlib.data.proto.ResultProtos.Result;
-import eu.dnetlib.data.proto.ResultProtos.Result.Instance;
 
+import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.*;
+
 /**
  * Created by claudio on 11/07/16.
  */
@@ -25,6 +19,15 @@
 			final OafEntity source,
 			final float trust) {
 
+		return fromOAF(oaf, trust, new Provenance()
+				.setRepositoryName(getValue(source.getCollectedfromList()))
+				.setUrl(Iterables.getFirst(mapInstances(oaf.getResult().getInstanceList()), new Instance()).getUrl())
+				.setId(getValue(source.getOriginalIdList())));
+	}
+
+	public static OpenAireEventPayload fromOAF(final OafEntity oaf,
+			final float trust, final Provenance provenance) {
+
 		final Result.Metadata m = oaf.getResult().getMetadata();
 		final Publication p = new Publication()
 				.setOriginalId(getValue(oaf.getOriginalIdList()))
@@ -37,11 +40,7 @@
 				.setPublisher(getValue(m.getPublisher()))
 				.setEmbargoenddate(getValue(m.getEmbargoenddate()))
 				.setContributor(listValues(m.getContributorList()))
-				.setJournal(new Journal()
-						.setName(m.getJournal().getName())
-						.setIssn(m.getJournal().getIssnPrinted())
-						.setEissn(m.getJournal().getIssnOnline())
-						.setLissn(m.getJournal().getIssnLinking()))
+				.setJournal(mapJournal(m.getJournal()))
 				.setCollectedFrom(listValues(oaf.getCollectedfromList()))
 				.setPids(mapPids(oaf.getPidList()))
 				.setInstances(mapInstances(oaf.getResult().getInstanceList()))
@@ -48,14 +47,11 @@
 				.setExternalReferences(mapExternalRefs(oaf.getResult().getExternalReferenceList()))
 				.setProjects(mapRelatedProjects(oaf));
 
-		final OpenAireEventPayload ep = new OpenAireEventPayload();
-
-		final Instance instance = Iterables.getFirst(source.getResult().getInstanceList(), Instance.newBuilder().build());
-
-		ep.setPublication(p).setHighlight(new Publication()).setProvenance(new Provenance().setRepositoryName(getValue(source.getCollectedfromList()))
-				.setUrl(getValue(instance.getUrlList())).setId(getValue(source.getOriginalIdList()))).setTrust(trust);
-
-		return ep;
+		return new OpenAireEventPayload()
+				.setPublication(p)
+				.setHighlight(new Publication())
+				.setProvenance(provenance)
+				.setTrust(trust);
 	}
 
 }
Index: modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/mapping/EventFactory.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/mapping/EventFactory.java	(revision 48144)
+++ modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/mapping/EventFactory.java	(revision 48145)
@@ -1,16 +1,10 @@
 package eu.dnetlib.data.mapreduce.hbase.broker.mapping;
 
-import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getKey;
-import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getValue;
-import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.listValues;
-
+import java.text.ParseException;
 import java.util.List;
 import java.util.Map;
 
-import org.apache.commons.lang.StringUtils;
-
 import com.google.common.collect.Maps;
-
 import eu.dnetlib.broker.objects.OpenAireEventPayload;
 import eu.dnetlib.data.mapreduce.hbase.broker.Topic;
 import eu.dnetlib.data.mapreduce.hbase.broker.model.EventMessage;
@@ -17,7 +11,12 @@
 import eu.dnetlib.data.mapreduce.hbase.broker.model.MapValue;
 import eu.dnetlib.data.mapreduce.hbase.broker.model.MapValueType;
 import eu.dnetlib.data.proto.OafProtos.OafEntity;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.lang.time.DateUtils;
 
+import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.*;
+
 /**
  * Created by claudio on 20/07/16.
  */
@@ -25,6 +24,8 @@
 
 	private final static String PRODUCER_ID = "OpenAIRE";
 
+	private final static String[] datePatterns = { "yyyy-MM-dd", "yyyy-MM-dd" };
+
 	public static EventMessage asEvent(final OafEntity oaf, final Topic topic, final OpenAireEventPayload payload, final OafEntity source, final float trust) {
 		return asEvent(oaf, topic, payload, getKey(source.getCollectedfromList()), getValue(source.getCollectedfromList()), trust,
 				listValues(source.getOriginalIdList()));
@@ -65,17 +66,17 @@
 		}
 
 		final String date = oaf.getResult().getMetadata().getDateofacceptance().getValue();
-		if (StringUtils.isNotBlank(date)) {
+		if (isValidDate(date)) {
 			map.put("target_dateofacceptance", new MapValue(MapValueType.DATE, date));
 		}
 
 		final List<String> subjects = listValues(oaf.getResult().getMetadata().getSubjectList());
-		if ((subjects != null) && !subjects.isEmpty()) {
+		if (CollectionUtils.isNotEmpty(subjects)) {
 			map.put("target_publication_subject_list", new MapValue(MapValueType.LIST_STRING, subjects));
 		}
 
 		final List<String> authors = listValues(oaf.getResult().getAuthorList());
-		if ((authors != null) && !authors.isEmpty()) {
+		if (CollectionUtils.isNotEmpty(authors)) {
 			map.put("target_publication_author_list", new MapValue(MapValueType.LIST_STRING, authors));
 		}
 
@@ -100,4 +101,16 @@
 		return new EventMessage(PRODUCER_ID, topic, payload.toJSON(), EventMessage.TTH_INFINITE, map);
 	}
 
+	private static boolean isValidDate(final String date) {
+		if (StringUtils.isBlank(date)) {
+			return false;
+		}
+		try {
+			DateUtils.parseDate(date, datePatterns);
+			return true;
+		} catch (ParseException e) {
+			return false;
+		}
+	}
+
 }
Index: modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/Topic.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/Topic.java	(revision 48144)
+++ modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/Topic.java	(revision 48145)
@@ -5,19 +5,12 @@
  */
 public enum Topic {
 
-	/*
-      <int name="mesheuropmc">1185961</int>
-      <int name="arxiv">842888</int>
-      <int name="jel">485877</int>
-      <int name="ddc">323948</int>
-      <int name="acm">311403</int>
-      <int name="rvk">11889</int>
-
-	 */
-
 	// ENRICHMENT MISSING
-	ENRICH_MISSING_OA_VERSION("ENRICH/MISSING/OPENACCESS_VERSION"), ENRICH_MISSING_ABSTRACT("ENRICH/MISSING/ABSTRACT"), ENRICH_MISSING_PUBLICATION_DATE(
-			"ENRICH/MISSING/PUBLICATION_DATE"), ENRICH_MISSING_PID("ENRICH/MISSING/PID"), ENRICH_MISSING_PROJECT("ENRICH/MISSING/PROJECT"),
+	ENRICH_MISSING_OA_VERSION("ENRICH/MISSING/OPENACCESS_VERSION"),
+	ENRICH_MISSING_ABSTRACT("ENRICH/MISSING/ABSTRACT"),
+	ENRICH_MISSING_PUBLICATION_DATE("ENRICH/MISSING/PUBLICATION_DATE"),
+	ENRICH_MISSING_PID("ENRICH/MISSING/PID"),
+	ENRICH_MISSING_PROJECT("ENRICH/MISSING/PROJECT"),
 	ENRICH_MISSING_SUBJECT_MESHEUROPMC("ENRICH/MISSING/SUBJECT/MESHEUROPMC"),
 	ENRICH_MISSING_SUBJECT_ARXIV("ENRICH/MISSING/SUBJECT/ARXIV"),
 	ENRICH_MISSING_SUBJECT_JEL("ENRICH/MISSING/SUBJECT/JEL"),
@@ -25,10 +18,12 @@
 	ENRICH_MISSING_SUBJECT_ACM("ENRICH/MISSING/SUBJECT/ACM"),
 	ENRICH_MISSING_SUBJECT_RVK("ENRICH/MISSING/SUBJECT/RVK"),
 
-
 	// ENRICHMENT MORE
-	ENRICH_MORE_PID("ENRICH/MORE/PID"), ENRICH_MORE_OA_VERSION("ENRICH/MORE/OPENACCESS_VERSION"), ENRICH_MORE_ABSTRACT(
-			"ENRICH/MORE/ABSTRACT"), ENRICH_MORE_PUBLICATION_DATE("ENRICH/MORE/PUBLICATION_DATE"), ENRICH_MORE_PROJECT("ENRICH/MORE/PROJECT"),
+	ENRICH_MORE_PID("ENRICH/MORE/PID"),
+	ENRICH_MORE_OA_VERSION("ENRICH/MORE/OPENACCESS_VERSION"),
+	ENRICH_MORE_ABSTRACT("ENRICH/MORE/ABSTRACT"),
+	ENRICH_MORE_PUBLICATION_DATE("ENRICH/MORE/PUBLICATION_DATE"),
+	ENRICH_MORE_PROJECT("ENRICH/MORE/PROJECT"),
 	ENRICH_MORE_SUBJECT_MESHEUROPMC("ENRICH/MORE/SUBJECT/MESHEUROPMC"),
 	ENRICH_MORE_SUBJECT_ARXIV("ENRICH/MORE/SUBJECT/ARXIV"),
 	ENRICH_MORE_SUBJECT_JEL("ENRICH/MORE/SUBJECT/JEL"),
Index: modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/EnrichmentMapper.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/EnrichmentMapper.java	(revision 48144)
+++ modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/EnrichmentMapper.java	(revision 48145)
@@ -1,20 +1,12 @@
 package eu.dnetlib.data.mapreduce.hbase.broker.enrich;
 
 import java.io.IOException;
-import java.util.Collection;
-import java.util.HashSet;
 import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
 
-import com.google.common.base.Function;
-import com.google.common.base.Predicate;
 import com.google.common.collect.Iterables;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
 import eu.dnetlib.data.mapreduce.util.DedupUtils;
+import eu.dnetlib.data.mapreduce.util.OafRowKeyDecoder;
 import eu.dnetlib.data.mapreduce.util.UpdateMerger;
-import eu.dnetlib.data.proto.DedupProtos.Dedup;
 import eu.dnetlib.data.proto.DedupProtos.Dedup.RelName;
 import eu.dnetlib.data.proto.OafProtos.Oaf;
 import eu.dnetlib.data.proto.TypeProtos.Type;
@@ -21,30 +13,26 @@
 import org.apache.commons.collections.MapUtils;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.mapreduce.TableMapper;
 import org.apache.hadoop.hbase.util.Bytes;
 
 /**
  * Created by claudio on 08/07/16.
  */
-public class EnrichmentMapper extends TableMapper<ImmutableBytesWritable, ImmutableBytesWritable> {
+public class EnrichmentMapper extends AbstractEnrichmentMapper {
 
-	private ImmutableBytesWritable outValue;
-
-	private ImmutableBytesWritable outKey;
-
 	@Override
-	protected void setup(final Context context) {
-		outKey = new ImmutableBytesWritable();
-		outValue = new ImmutableBytesWritable();
+	protected String counterGroup() {
+		return "Broker Enrichment";
 	}
 
 	@Override
 	protected void map(final ImmutableBytesWritable key, final Result value, final Context context) throws IOException, InterruptedException {
 
+		final String type = OafRowKeyDecoder.decode(key.copyBytes()).getType().toString();
+
 		final Map<byte[], byte[]> result = value.getFamilyMap(Bytes.toBytes(Type.result.name()));
 		if (MapUtils.isEmpty(result)) {
-			context.getCounter(Type.result.name(), "empty family map").increment(1);
+			context.getCounter(counterGroup(), type + ": empty family map").increment(1);
 			return;
 		}
 
@@ -51,7 +39,7 @@
 		final Oaf body = UpdateMerger.mergeBodyUpdates(context, result);
 
 		if (body == null) {
-			context.getCounter(Type.result.name(), "body null").increment(1);
+			context.getCounter(counterGroup(), type + ": body null").increment(1);
 			return;
 		}
 
@@ -58,44 +46,18 @@
 		final String mergedInCF = DedupUtils.getDedupCF_mergedIn(Type.result);
 		final Map<byte[], byte[]> mergedIn = value.getFamilyMap(Bytes.toBytes(mergedInCF));
 
-		final String outKey = getEmitKey(context, key, mergedIn);
+		final byte[] outKey = getEmitKey(context, key, mergedIn);
 
-		emit(context, outKey, body);
+		emit(context, outKey, body.toByteArray(), type);
 	}
 
-	private String getEmitKey(final Context context, final ImmutableBytesWritable key, final Map<byte[], byte[]> mergedIn) {
+	private byte[] getEmitKey(final Context context, final ImmutableBytesWritable key, final Map<byte[], byte[]> mergedIn) {
 		if (MapUtils.isNotEmpty(mergedIn)) {
 			context.getCounter(Type.result.name(), RelName.isMergedIn.name()).increment(1);
-			return getRootId(mergedIn, context);
+			return Iterables.getOnlyElement(mergedIn.keySet());
 		} else {
-			return new String(key.copyBytes());
+			return key.copyBytes();
 		}
 	}
 
-	private void emit(final Context context, final String key, final Oaf body) throws IOException, InterruptedException {
-		outKey.set(Bytes.toBytes(key));
-		outValue.set(body.toByteArray());
-
-		context.write(outKey, outValue);
-	}
-
-	private String getRootId(final Map<byte[], byte[]> mergedIn, Context context) {
-		final HashSet<String> ids = Sets.newHashSet(Iterables.transform(mergedIn.keySet(), new Function<byte[], String>() {
-
-			@Override
-			public String apply(final byte[] input) {
-				return new String(input);
-			}
-		}));
-
-		//context.getCounter("duplicate group size", String.valueOf(ids.size())).increment(1);
-
-		try {
-			return Iterables.getOnlyElement(ids);
-		} catch(IllegalArgumentException e) {
-			System.err.println(ids);
-			throw e;
-		}
-	}
-
 }
Index: modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/AbstractEnrichmentMapper.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/AbstractEnrichmentMapper.java	(nonexistent)
+++ modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/AbstractEnrichmentMapper.java	(revision 48145)
@@ -0,0 +1,34 @@
+package eu.dnetlib.data.mapreduce.hbase.broker.enrich;
+
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.mapreduce.TableMapper;
+
+/**
+ * Created by claudio on 21/02/2017.
+ */
+public abstract class AbstractEnrichmentMapper extends TableMapper<ImmutableBytesWritable, ImmutableBytesWritable> {
+
+	protected ImmutableBytesWritable outValue;
+
+	protected ImmutableBytesWritable outKey;
+
+	protected abstract String counterGroup();
+
+	@Override
+	protected void setup(final Context context) {
+		outKey = new ImmutableBytesWritable();
+		outValue = new ImmutableBytesWritable();
+	}
+
+	protected void emit(final Context context, final byte[] key, final byte[] value, final String entityType) {
+		outKey.set(key);
+		outValue.set(value);
+		try {
+			context.write(outKey, outValue);
+			context.getCounter(counterGroup(), "entity type: " + entityType).increment(1);
+		} catch (Exception e) {
+			throw new IllegalArgumentException(e);
+		}
+	}
+
+}
Index: modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/EnrichmentReducer.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/EnrichmentReducer.java	(revision 48144)
+++ modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/EnrichmentReducer.java	(revision 48145)
@@ -1,141 +1,29 @@
 package eu.dnetlib.data.mapreduce.hbase.broker.enrich;
 
-import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getKey;
-import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getPropertyValues;
-
 import java.io.IOException;
 import java.util.List;
-import java.util.Map;
-import java.util.Set;
 
-import eu.dnetlib.data.mapreduce.Algorithms;
-import eu.dnetlib.data.mapreduce.JobParams;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
 import eu.dnetlib.data.mapreduce.hbase.broker.*;
 import eu.dnetlib.data.mapreduce.util.DedupUtils;
-import eu.dnetlib.pace.config.DedupConfig;
-import eu.dnetlib.pace.distance.PaceDocumentDistance;
-import eu.dnetlib.pace.distance.eval.ScoreResult;
-import eu.dnetlib.pace.model.MapDocument;
-import eu.dnetlib.pace.model.ProtoDocumentBuilder;
+import eu.dnetlib.data.proto.OafProtos.Oaf;
 import org.apache.commons.lang.StringUtils;
-import org.apache.commons.lang.math.RandomUtils;
-import org.apache.commons.math.util.MathUtils;
-import org.apache.hadoop.hbase.client.HTable;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.filter.FilterList;
-import org.apache.hadoop.hbase.filter.FilterList.Operator;
-import org.apache.hadoop.hbase.filter.PrefixFilter;
 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapreduce.Reducer;
 import org.dom4j.DocumentException;
 
-import com.google.common.base.Function;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-import com.google.protobuf.InvalidProtocolBufferException;
+import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getKey;
 
-import eu.dnetlib.data.proto.OafProtos.Oaf;
-
 /**
  * Created by claudio on 08/07/16.
  */
-public class EnrichmentReducer extends Reducer<ImmutableBytesWritable, ImmutableBytesWritable, Text, Text> {
+public class EnrichmentReducer extends AbstractEnrichmentReducer {
 
-	private static final int LIMIT = 1000;
-	private static final int SCORE_DECIMALS = 2;
-
-	private Map<String, String> dsTypeMap = Maps.newHashMap();
-
-	private Set<String> dsWhitelist = Sets.newHashSet();
-
-	private Set<String> dsBlacklist = Sets.newHashSet();
-
-	// This is for EuropePMC. They expose OA abstracts, but we want to identify real OA publications. WTF.
-	private Set<String> untrustedOaDsList = Sets.newHashSet();
-
-	// White list for datasource typologies.
-	private Set<String> dsTypeWhitelist = Sets.newHashSet();
-
-	private DedupConfig dedupConf;
-
-	/**
-	 * lower bound for trust scaling
-	 */
-	private double scaleLB;
-
 	@Override
-	protected void setup(final Context context) throws IOException, InterruptedException {
-		super.setup(context);
-
-		System.out.println("LIMIT: " + LIMIT);
-
-		dsWhitelist.addAll(getPropertyValues(context, "broker.datasource.id.whitelist"));
-		dsBlacklist.addAll(getPropertyValues(context, "broker.datasource.id.blacklist"));
-		dsTypeWhitelist.addAll(getPropertyValues(context, "broker.datasource.type.whitelist"));
-		untrustedOaDsList.addAll(getPropertyValues(context, "broker.datasource.untrusted.oa.list"));
-
-		dsTypeMap = getDsTypeMap(context, dsTypeWhitelist);
-
-		System.out.println("datasource whitelist: " + dsWhitelist);
-		System.out.println("datasource blacklist: " + dsBlacklist);
-		System.out.println("datasource OA list: " + untrustedOaDsList);
-
-		System.out.println("datasource type whitelist: " + dsTypeWhitelist);
-
-		final String dedupConfJson = context.getConfiguration().get(JobParams.DEDUP_CONF);
-
-		System.out.println("got dedup conf: " + dedupConfJson);
-
-		dedupConf = DedupConfig.load(dedupConfJson);
-
-		System.out.println("parsed dedup conf: " + dedupConf.toString());
-
-		scaleLB = dedupConf.getWf().getThreshold() - 0.01;
+	protected String counterGroup() {
+		return "Broker Enrichment";
 	}
 
-	private Map<String, String> getDsTypeMap(final Context context, final Set<String> dsTypeWhitelist) throws IOException {
-		System.out.println("loading datasource typology mapping");
-
-		final Map<String, String> dsTypeMap = Maps.newHashMap();
-
-		final Scan scan = new Scan();
-		final FilterList fl = new FilterList(Operator.MUST_PASS_ALL);
-		fl.addFilter(new PrefixFilter(Bytes.toBytes("10")));
-		scan.setFilter(fl);
-		scan.addFamily(Bytes.toBytes("datasource"));
-
-		final String tableName = context.getConfiguration().get("hbase.mapred.inputtable");
-
-		System.out.println(String.format("table name: '%s'", tableName));
-
-		try (final HTable table = new HTable(context.getConfiguration(), tableName);
-				final ResultScanner res = table.getScanner(scan)) {
-
-			for (final Result r : res) {
-				final byte[] b = r.getValue(Bytes.toBytes("datasource"), Bytes.toBytes("body"));
-				if (b != null) {
-					final Oaf oaf = Oaf.parseFrom(b);
-					final String dsId = StringUtils.substringAfter(oaf.getEntity().getId(), "|");
-					final String dsType = oaf.getEntity().getDatasource().getMetadata().getDatasourcetype().getClassid();
-
-					if (dsTypeWhitelist.contains(dsType)) {
-						System.out.println(String.format("dsId '%s', dsType '%s'", dsId, dsType));
-						dsTypeMap.put(dsId, dsType);
-					}
-				}
-			}
-
-		}
-		System.out.println("datasource type map size: " + dsTypeMap.size());
-		return dsTypeMap;
-	}
-
 	@Override
 	protected void reduce(final ImmutableBytesWritable key, final Iterable<ImmutableBytesWritable> values, final Context context) throws IOException,
 			InterruptedException {
@@ -190,29 +78,4 @@
 		}
 	}
 
-	private float similarity(final Oaf oa, final Oaf ob) {
-
-		final MapDocument a = ProtoDocumentBuilder.newInstance(oa.getEntity().getId(), oa.getEntity(), dedupConf.getPace().getModel());
-		final MapDocument b = ProtoDocumentBuilder.newInstance(ob.getEntity().getId(), ob.getEntity(), dedupConf.getPace().getModel());
-
-		final ScoreResult sr =  new PaceDocumentDistance().between(a, b, dedupConf);
-		final float score = (float) Algorithms.scale(sr.getScore(), scaleLB, 1, 0, 1);
-
-		return MathUtils.round(score, SCORE_DECIMALS);
-	}
-
-	private Function<ImmutableBytesWritable, Oaf> oafDeserialiser() {
-		return new Function<ImmutableBytesWritable, Oaf>() {
-
-			@Override
-			public Oaf apply(final ImmutableBytesWritable input) {
-				try {
-					return Oaf.parseFrom(input.copyBytes());
-				} catch (final InvalidProtocolBufferException e) {
-					throw new IllegalArgumentException(e);
-				}
-			}
-		};
-	}
-
 }
Index: modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/ProjectEnrichmentMapper.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/ProjectEnrichmentMapper.java	(nonexistent)
+++ modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/ProjectEnrichmentMapper.java	(revision 48145)
@@ -0,0 +1,118 @@
+package eu.dnetlib.data.mapreduce.hbase.broker.enrich;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Map;
+
+import com.google.common.base.Function;
+import com.google.common.base.Predicate;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import eu.dnetlib.data.mapreduce.util.DedupUtils;
+import eu.dnetlib.data.mapreduce.util.OafDecoder;
+import eu.dnetlib.data.mapreduce.util.OafRowKeyDecoder;
+import eu.dnetlib.data.proto.OafProtos.Oaf;
+import eu.dnetlib.data.proto.TypeProtos.Type;
+import org.apache.commons.collections.MapUtils;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * Created by claudio on 08/07/16.
+ */
+public class ProjectEnrichmentMapper extends AbstractEnrichmentMapper {
+
+	@Override
+	protected String counterGroup() {
+		return "Broker Enrichment projects";
+	}
+
+	@Override
+	protected void map(final ImmutableBytesWritable key, final Result value, final Context context) throws IOException, InterruptedException {
+
+		final Type type = getEntityType(key);
+
+		final byte[] body = value.getValue(Bytes.toBytes(type.toString()), Bytes.toBytes("body"));
+
+		if (body == null) {
+			context.getCounter(counterGroup(), "missing metadata").increment(1);
+			return;
+		}
+
+		switch (type) {
+		case project:
+			for (final byte[] resultId : listRelatedIds(value, "resultProject_outcome_produces")) {
+				emit(context, resultId, body, "project");
+			}
+			break;
+		case result:
+
+			final Oaf.Builder oafBuilder = Oaf.newBuilder(OafDecoder.decode(body).getOaf());
+			for (final String relName : Arrays.asList("resultProject_outcome_isProducedBy")) { // Â TODO add dataset rels
+				for (final Oaf rel : listRelations(value, relName)) {
+					oafBuilder.getEntityBuilder().addCachedOafRel(rel);
+					context.getCounter(counterGroup(), "rel: " + relName).increment(1);
+				}
+			}
+
+			final Map<byte[], byte[]> mergedIn = value.getFamilyMap(DedupUtils.getDedupCF_mergedInBytes(Type.result));
+
+			if (MapUtils.isEmpty(mergedIn) & !DedupUtils.isRoot(key)) {
+				emit(context, key.copyBytes(), oafBuilder.build().toByteArray(), "result not deduped");
+			} else if (DedupUtils.isRoot(key)) {
+				emit(context, key.copyBytes(), oafBuilder.build().toByteArray(), "result merges");
+			} else {
+				emit(context, getRootId(mergedIn), oafBuilder.build().toByteArray(), "result mergedIn");
+			}
+
+			break;
+		default:
+			throw new IllegalArgumentException("invalid type: " + type);
+		}
+	}
+
+	private Iterable<Oaf> listRelations(final Result value, final String relType) {
+
+		//TODO consider only relationshipt not deletedbyinference
+
+		final Map<byte[], byte[]> map = value.getFamilyMap(Bytes.toBytes(relType));
+		if (MapUtils.isNotEmpty(map)) {
+			return Iterables.filter(Iterables.transform(map.values(), new Function<byte[], Oaf>() {
+
+				@Override
+				public Oaf apply(final byte[] input) {
+					return OafDecoder.decode(input).getOaf();
+				}
+			}), new Predicate<Oaf>() {
+				@Override
+				public boolean apply(final Oaf rel) {
+					return !rel.getRel().getTarget().contains("unidentified");
+				}
+			});
+		} else {
+			return Lists.newArrayList();
+		}
+	}
+
+	private Iterable<byte[]> listRelatedIds(final Result value, final String relType) {
+
+		//TODO consider only relationshipt not deletedbyinference
+
+		final Map<byte[], byte[]> map = value.getFamilyMap(Bytes.toBytes(relType));
+		if (MapUtils.isNotEmpty(map)) {
+			return map.keySet();
+		} else {
+			return Lists.newArrayList();
+		}
+	}
+
+	private Type getEntityType(final ImmutableBytesWritable key) {
+		return OafRowKeyDecoder.decode(key.copyBytes()).getType();
+	}
+
+	private byte[] getRootId(final Map<byte[], byte[]> mergedIn) {
+		return Iterables.getOnlyElement(mergedIn.keySet());
+	}
+
+}
Index: modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/AbstractEnrichmentReducer.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/AbstractEnrichmentReducer.java	(nonexistent)
+++ modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/AbstractEnrichmentReducer.java	(revision 48145)
@@ -0,0 +1,155 @@
+package eu.dnetlib.data.mapreduce.hbase.broker.enrich;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import com.google.protobuf.InvalidProtocolBufferException;
+import eu.dnetlib.data.mapreduce.Algorithms;
+import eu.dnetlib.data.mapreduce.JobParams;
+import eu.dnetlib.data.proto.OafProtos.Oaf;
+import eu.dnetlib.pace.config.DedupConfig;
+import eu.dnetlib.pace.distance.PaceDocumentDistance;
+import eu.dnetlib.pace.distance.eval.ScoreResult;
+import eu.dnetlib.pace.model.MapDocument;
+import eu.dnetlib.pace.model.ProtoDocumentBuilder;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.math.util.MathUtils;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.filter.FilterList;
+import org.apache.hadoop.hbase.filter.FilterList.Operator;
+import org.apache.hadoop.hbase.filter.PrefixFilter;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Reducer;
+
+import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getPropertyValues;
+
+/**
+ * Created by claudio on 20/02/2017.
+ */
+public abstract class AbstractEnrichmentReducer extends Reducer<ImmutableBytesWritable, ImmutableBytesWritable, Text, Text> {
+
+	protected DedupConfig dedupConf;
+
+	protected static final int LIMIT = 1000;
+
+	protected static final int SCORE_DECIMALS = 2;
+
+	protected Map<String, String> dsTypeMap = Maps.newHashMap();
+
+	protected Set<String> dsWhitelist = Sets.newHashSet();
+
+	protected Set<String> dsBlacklist = Sets.newHashSet();
+
+	// This is for EuropePMC. They expose OA abstracts, but we want to identify real OA publications. WTF.
+	protected Set<String> untrustedOaDsList = Sets.newHashSet();
+
+	// White list for datasource typologies.
+	protected Set<String> dsTypeWhitelist = Sets.newHashSet();
+
+	/**
+	 * lower bound for trust scaling
+	 */
+	protected double scaleLB;
+
+	protected abstract String counterGroup();
+
+	@Override
+	protected void setup(final Context context) throws IOException, InterruptedException {
+		super.setup(context);
+
+		System.out.println("LIMIT: " + LIMIT);
+
+		dsWhitelist.addAll(getPropertyValues(context, "broker.datasource.id.whitelist"));
+		dsBlacklist.addAll(getPropertyValues(context, "broker.datasource.id.blacklist"));
+		dsTypeWhitelist.addAll(getPropertyValues(context, "broker.datasource.type.whitelist"));
+		untrustedOaDsList.addAll(getPropertyValues(context, "broker.datasource.untrusted.oa.list"));
+
+		dsTypeMap = getDsTypeMap(context, dsTypeWhitelist);
+
+		System.out.println("datasource whitelist: " + dsWhitelist);
+		System.out.println("datasource blacklist: " + dsBlacklist);
+		System.out.println("datasource OA list: " + untrustedOaDsList);
+
+		System.out.println("datasource type whitelist: " + dsTypeWhitelist);
+
+		final String dedupConfJson = context.getConfiguration().get(JobParams.DEDUP_CONF);
+
+		System.out.println("got dedup conf: " + dedupConfJson);
+
+		dedupConf = DedupConfig.load(dedupConfJson);
+
+		System.out.println("parsed dedup conf: " + dedupConf.toString());
+
+		scaleLB = dedupConf.getWf().getThreshold() - 0.01;
+	}
+
+	protected Map<String, String> getDsTypeMap(final Context context, final Set<String> dsTypeWhitelist) throws IOException {
+		System.out.println("loading datasource typology mapping");
+
+		final Map<String, String> dsTypeMap = Maps.newHashMap();
+
+		final Scan scan = new Scan();
+		final FilterList fl = new FilterList(Operator.MUST_PASS_ALL);
+		fl.addFilter(new PrefixFilter(Bytes.toBytes("10")));
+		scan.setFilter(fl);
+		scan.addFamily(Bytes.toBytes("datasource"));
+
+		final String tableName = context.getConfiguration().get("hbase.mapred.inputtable");
+
+		System.out.println(String.format("table name: '%s'", tableName));
+
+		try (final HTable table = new HTable(context.getConfiguration(), tableName); final ResultScanner res = table.getScanner(scan)) {
+
+			for (final Result r : res) {
+				final byte[] b = r.getValue(Bytes.toBytes("datasource"), Bytes.toBytes("body"));
+				if (b != null) {
+					final Oaf oaf = Oaf.parseFrom(b);
+					final String dsId = StringUtils.substringAfter(oaf.getEntity().getId(), "|");
+					final String dsType = oaf.getEntity().getDatasource().getMetadata().getDatasourcetype().getClassid();
+
+					if (dsTypeWhitelist.contains(dsType)) {
+						//System.out.println(String.format("dsId '%s', dsType '%s'", dsId, dsType));
+						dsTypeMap.put(dsId, dsType);
+					}
+				}
+			}
+		}
+		System.out.println("datasource type map size: " + dsTypeMap.size());
+		return dsTypeMap;
+	}
+
+	protected float similarity(final Oaf oa, final Oaf ob) {
+
+		final MapDocument a = ProtoDocumentBuilder.newInstance(oa.getEntity().getId(), oa.getEntity(), dedupConf.getPace().getModel());
+		final MapDocument b = ProtoDocumentBuilder.newInstance(ob.getEntity().getId(), ob.getEntity(), dedupConf.getPace().getModel());
+
+		final ScoreResult sr =  new PaceDocumentDistance().between(a, b, dedupConf);
+		final float score = (float) Algorithms.scale(sr.getScore(), scaleLB, 1, 0, 1);
+
+		return MathUtils.round(score, SCORE_DECIMALS, BigDecimal.ROUND_HALF_DOWN);
+	}
+
+	protected Function<ImmutableBytesWritable, Oaf> oafDeserialiser() {
+		return new Function<ImmutableBytesWritable, Oaf>() {
+
+			@Override
+			public Oaf apply(final ImmutableBytesWritable input) {
+				try {
+					return Oaf.parseFrom(input.copyBytes());
+				} catch (final InvalidProtocolBufferException e) {
+					throw new IllegalArgumentException(e);
+				}
+			}
+		};
+	}
+}
Index: modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/ProjectEnrichmentReducer.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/ProjectEnrichmentReducer.java	(nonexistent)
+++ modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/enrich/ProjectEnrichmentReducer.java	(revision 48145)
@@ -0,0 +1,137 @@
+package eu.dnetlib.data.mapreduce.hbase.broker.enrich;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import com.google.common.base.Function;
+import com.google.common.base.Predicate;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import eu.dnetlib.data.mapreduce.hbase.broker.ProjectEventFactory;
+import eu.dnetlib.data.proto.OafProtos.Oaf;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.dom4j.DocumentException;
+
+import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getKey;
+
+/**
+ * Created by claudio on 08/07/16.
+ */
+public class ProjectEnrichmentReducer extends AbstractEnrichmentReducer {
+
+	@Override
+	protected String counterGroup() {
+		return "Broker Enrichment projects";
+	}
+
+	@Override
+	protected void reduce(final ImmutableBytesWritable key, final Iterable<ImmutableBytesWritable> values, final Context context)
+			throws IOException, InterruptedException {
+
+		final Map<String, Oaf> projects = Maps.newHashMap();
+		final List<Oaf> pubs = Lists.newArrayList();
+
+		for (final Oaf oaf : Iterables.transform(Iterables.limit(values, LIMIT), oafDeserialiser())) {
+
+			switch (oaf.getEntity().getType()) {
+			case result:
+				pubs.add(oaf);
+				break;
+			case project:
+				projects.put(oaf.getEntity().getId(), oaf);
+				break;
+			default:
+				throw new IllegalArgumentException("invalid type: " + oaf.getEntity().getType());
+			}
+		}
+		if (projects.isEmpty() || pubs.isEmpty()) {
+			context.getCounter(counterGroup(), "Events Skipped - Nothing to do").increment(1);
+			return;
+		}
+
+		try {
+			generateEvents(pubs, projects, context);
+		} catch (final DocumentException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	private void generateEvents(final List<Oaf> pubs, final Map<String, Oaf> projects, final Context context)
+			throws IOException, InterruptedException, DocumentException {
+
+		final List<Oaf> enrichedOaf = Lists.newArrayList(Iterables.transform(pubs, new Function<Oaf, Oaf>() {
+			@Override
+			public Oaf apply(final Oaf pub) {
+				return addProjects(pub, projects);
+			}
+		}));
+
+		final List<Oaf> valid = Lists.newArrayList(Iterables.filter(enrichedOaf, new Predicate<Oaf>() {
+			@Override
+			public boolean apply(final Oaf result) {
+				return !result.getEntity().getCachedOafRelList().isEmpty();
+			}
+		}));
+
+		if (valid.isEmpty()) {
+			context.getCounter(counterGroup(), "Events Skipped - Missing project").increment(1);
+			return;
+		}
+
+		for (final Oaf current : valid) {
+
+			final String currentId = current.getEntity().getId();
+
+			final String currentDsId = StringUtils.substringAfter(getKey(current.getEntity().getCollectedfromList()), "|");
+			final String currentDsType = dsTypeMap.get(currentDsId);
+
+			// System.out.println(String.format("'%s' -> '%s'", currentDsId, currentDsType));
+
+			if (StringUtils.isBlank(currentDsType) && !dsWhitelist.contains(currentDsId)) {
+				context.getCounter(counterGroup(), "Events Skipped - Datasource type excluded").increment(1);
+			} else if (dsBlacklist.contains(currentDsId)) {
+				context.getCounter(counterGroup(), "Events Skipped - Datasource blacklisted").increment(1);
+			} else {
+				if (valid.size() == 1) {
+					ProjectEventFactory.process(context, valid.get(0));
+					return;
+				}
+
+				for (final Oaf other : valid) {
+					final String otherId = other.getEntity().getId();
+					if (!currentId.equals(otherId)) {
+
+						final float trust = similarity(current, other);
+						ProjectEventFactory.process(context, current, other, trust);
+					}
+				}
+			}
+		}
+
+	}
+
+	private Oaf addProjects(final Oaf current, final Map<String, Oaf> projects) {
+
+		final Oaf.Builder oafBuilder = Oaf.newBuilder(current);
+		final List<Oaf> cachedRels = Lists.newArrayList();
+
+		for (final Oaf.Builder cachedOafRel : oafBuilder.getEntityBuilder().getCachedOafRelBuilderList()) {
+			final String projectId = cachedOafRel.getRel().getTarget();
+
+			if (projects.containsKey(projectId)) {
+				final Oaf project = projects.get(projectId);
+
+				cachedOafRel.getRelBuilder().setCachedOafTarget(project);
+				cachedRels.add(cachedOafRel.build());
+			}
+		}
+
+		oafBuilder.getEntityBuilder().clearCachedOafRel().addAllCachedOafRel(cachedRels);
+
+		return oafBuilder.build();
+	}
+
+}
Index: modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/SubjectEventFactory.java
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/SubjectEventFactory.java	(revision 48144)
+++ modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/broker/SubjectEventFactory.java	(revision 48145)
@@ -1,7 +1,6 @@
 package eu.dnetlib.data.mapreduce.hbase.broker;
 
 import java.io.IOException;
-import java.util.List;
 
 import com.google.common.base.Predicate;
 import com.google.common.collect.Iterables;
@@ -14,6 +13,7 @@
 import eu.dnetlib.data.proto.OafProtos.Oaf;
 import eu.dnetlib.data.proto.OafProtos.OafEntity;
 import eu.dnetlib.data.proto.ResultProtos.Result.Metadata;
+import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Reducer.Context;
 import org.dom4j.DocumentException;
@@ -77,10 +77,12 @@
 			final Float trust) throws IOException, InterruptedException {
 
 		if (topic == null) {
-			context.getCounter("event", "SUBJECT not mapped").increment(1);
+			context.getCounter("events skipped", "SUBJECT not mapped").increment(1);
 			return;
 		}
-
+		// we got java.lang.NumberFormatException: empty String, because of empty trust values. Default to 1.0 in such cases
+		final String t = subjectOther.getDataInfo().getTrust();
+		final float adjustedTrust = trust * Float.valueOf(StringUtils.isBlank(t) ? "1.0" : t);
 		final Oaf.Builder prototype = Oaf.newBuilder(current);
 		prototype.getEntityBuilder().getResultBuilder().getMetadataBuilder().addSubject(subjectOther);
 		final Oaf oaf = prototype.build();
@@ -88,9 +90,9 @@
 		final OafEntity eventEntity = other != null ? other.getEntity() : current.getEntity();
 
 		final OpenAireEventPayload payload =
-				HighlightFactory.highlightEnrichSubject(OpenAireEventPayloadFactory.fromOAF(oaf.getEntity(), eventEntity, trust),
+				HighlightFactory.highlightEnrichSubject(OpenAireEventPayloadFactory.fromOAF(oaf.getEntity(), eventEntity, adjustedTrust),
 						Lists.newArrayList(subjectOther));
-		final EventMessage event = asEvent(oaf.getEntity(), topic, payload, eventEntity, trust);
+		final EventMessage event = asEvent(oaf.getEntity(), topic, payload, eventEntity, adjustedTrust);
 
 		context.write(tKey, new Text(event.toString()));
 		context.getCounter("event", topic.getValue()).increment(1);
Index: modules/dnet-mapreduce-jobs/trunk/pom.xml
===================================================================
--- modules/dnet-mapreduce-jobs/trunk/pom.xml	(revision 48144)
+++ modules/dnet-mapreduce-jobs/trunk/pom.xml	(revision 48145)
@@ -210,8 +210,8 @@
 		</dependency>
 		<dependency>
 			<groupId>org.mockito</groupId>
-			<artifactId>mockito-all</artifactId>
-			<version>1.8.5</version>
+			<artifactId>mockito-core</artifactId>
+			<version>${mockito.version}</version>
 			<scope>test</scope>
 		</dependency>
 		<dependency>
@@ -223,7 +223,7 @@
 		<dependency>
 			<groupId>com.google.guava</groupId>
 			<artifactId>guava</artifactId>
-			<version>RELEASE</version>
+			<version>${google.guava.version}</version>
 		</dependency>
 
 	</dependencies>
