D-Net

package eu.dnetlib.data.mapreduce.hbase.broker;

import java.io.IOException;

import java.util.List;

import java.util.Map;

import java.util.Set;

import com.google.common.base.Function;

import com.google.common.base.Joiner;

import com.google.common.base.Predicate;

import com.google.common.base.Splitter;

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 com.googlecode.protobuf.format.JsonFormat;

import eu.dnetlib.data.proto.FieldTypeProtos.KeyValue;

import eu.dnetlib.data.proto.FieldTypeProtos.StringField;

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.ResultProtos.Result.Instance;

import eu.dnetlib.data.proto.ResultProtos.Result.Metadata;

import org.apache.commons.lang.StringUtils;

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;

/**

 * Created by claudio on 08/07/16.

 */

public class EnrichmentReducer extends Reducer<ImmutableBytesWritable, ImmutableBytesWritable, Text, Text> {

	private static final int LIMIT = 1000;

	private Set<String> pidType;

	private final static String PRODUCER_ID = "OpenAIRE";

	public enum Topic {

		PID("/ENRICH/PID"),

		OA_STATUS("/ENRICH/OA_STATUS"),

		ABSTRACT("/ENRICH/ABSTRACT"),

		PUBLICATION_DATE("/ENRICH/PUBLICATION_DATE");

		Topic(final String value){

			this.value = value;

		}

		protected String value;

		public String getValue() {

			return this.value;

		}

	}

	private Text tKey;

	private Text tValue;

	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> oaDsList = Sets.newHashSet();

	private Set<String> dsTypeWhitelist = Sets.newHashSet();

	@Override

	protected void setup(final Context context) throws IOException, InterruptedException {

		super.setup(context);

		System.out.println("LIMIT: " + LIMIT);

		tKey = new Text("");

		tValue = new Text();

		pidType = Sets.newHashSet("doi", "pmc", "pmid", "urn", "arxiv");

		dsWhitelist.addAll(Lists.newArrayList(Splitter.on(",").omitEmptyStrings().split(context.getConfiguration().get("broker.datasource.id.whitelist", ""))));

		dsBlacklist.addAll(Lists.newArrayList(Splitter.on(",").omitEmptyStrings().split(context.getConfiguration().get("broker.datasource.id.blacklist", ""))));

		dsTypeWhitelist.addAll(Lists.newArrayList(Splitter.on(",").omitEmptyStrings().split(context.getConfiguration().get("broker.datasource.type.whitelist", ""))));

		oaDsList.addAll(Lists.newArrayList(Splitter.on(",").omitEmptyStrings().split(context.getConfiguration().get("broker.datasource.oa.list", ""))));

		dsTypeMap = getDsTypeMap(context, dsTypeWhitelist);

		System.out.println("datasource whitelist: " + dsWhitelist);

		System.out.println("datasource blacklist: " + dsBlacklist);

		System.out.println("datasource OA list: " + oaDsList);

		System.out.println("datasource type whitelist: " + dsTypeWhitelist);

	}

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

		final HTable table = new HTable(context.getConfiguration(), tableName);

		final ResultScanner res = table.getScanner(scan);

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

				}

			}

		}

		res.close();

		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 {

		final List<Oaf> oafList = Lists.newArrayList(Iterables.transform(Iterables.limit(values, LIMIT), oafDeserialiser()));

		generateEvents(oafList, context);

	}

	private void generateEvents(final List<Oaf> oafList, final Context context) throws IOException, InterruptedException {

		for(Oaf current : oafList) {

			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)) {

				context.getCounter("target datasource", "events skipped").increment(1);

			} else {

				//if (dsWhitelist.contains(currentDsId) | !dsBlacklist.contains(currentDsId)) {

				for (Oaf other : oafList) {

					final String otherId = other.getEntity().getId();

					if (!currentId.equals(otherId)) {

						//PIDS

						for (final String type : pidType) {

							if (!hasPid(current, type) && hasPid(other, type)) {

								final Oaf.Builder prototype = Oaf.newBuilder(current);

								final Iterable<StructuredProperty> pids = Iterables.filter(other.getEntity().getPidList(), new Predicate<StructuredProperty>() {

									@Override

									public boolean apply(final StructuredProperty pid) {

										return pid.getQualifier().getClassid().equalsIgnoreCase(type);

									}

								});

								prototype.getEntityBuilder().addAllPid(pids);

								emit(asEvent(prototype.build(), Topic.PID.getValue(), other), context);

								context.getCounter("event", Topic.PID.getValue()).increment(1);

							}

						}

						final String otherDsId = StringUtils.substringAfter(getKey(other.getEntity().getCollectedfromList()), "|");

						//OPEN ACCESS STATUS

						if ((oaDsList.contains(currentDsId) || !hasAccess(current, "OPEN", false)) && (hasAccess(other, "OPEN", false) && !oaDsList.contains(otherDsId))) {

							final Oaf.Builder prototype = Oaf.newBuilder(current);

							final Iterable<Instance> i = Iterables.filter(other.getEntity().getResult().getInstanceList(), new Predicate<Instance>() {

								@Override

								public boolean apply(final Instance i) {

									return "OPEN".equalsIgnoreCase(i.getLicence().getClassid());

								}

							});

							prototype.getEntityBuilder().getResultBuilder().addAllInstance(i);

							emit(asEvent(prototype.build(), Topic.OA_STATUS.getValue(), other), context);

							context.getCounter("event", Topic.OA_STATUS.getValue()).increment(1);

						}

						//ABSTRACT

						if (!hasAbstract(current) && hasAbstract(other)) {

							final Oaf.Builder prototype = Oaf.newBuilder(current);

							final List<StringField> descriptionList = other.getEntity().getResult().getMetadata().getDescriptionList();

							prototype.getEntityBuilder().getResultBuilder().getMetadataBuilder().addAllDescription(descriptionList);

							emit(asEvent(prototype.build(), Topic.ABSTRACT.getValue(), other), context);

							context.getCounter("event", Topic.ABSTRACT.getValue()).increment(1);

						}

						//PUBLICATION DATE

						if (!hasPubDate(current) && hasPubDate(other)) {

							final Oaf.Builder prototype = Oaf.newBuilder(current);

							final StringField date = other.getEntity().getResult().getMetadata().getDateofacceptance();

							prototype.getEntityBuilder().getResultBuilder().getMetadataBuilder().setDateofacceptance(date);

							emit(asEvent(prototype.build(), Topic.PUBLICATION_DATE.getValue(), other), context);

							context.getCounter("event", Topic.PUBLICATION_DATE.getValue()).increment(1);

						}

					}

				}

			}

			/*} else {

				context.getCounter("target datasource", "events skipped").increment(1);

			}*/

		}

	}

	private void emit(final EventMessage e, final Context context) throws IOException, InterruptedException {

		//tKey.set(e.getMap().get("id"));

		tValue.set(e.toString());

		context.write(tKey, tValue);

	}

	private EventMessage asEvent(final Oaf oaf, final String topic, final Oaf source) {

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

		//TARGET INFO

		final String targetDatasourceName = getValue(oaf.getEntity().getCollectedfromList());

		if (StringUtils.isNotBlank(targetDatasourceName)) {

			map.put("target.datasource.name", targetDatasourceName);

		}

		final String title = getValue(oaf.getEntity().getResult().getMetadata().getTitleList());

		if (StringUtils.isNotBlank(title)) {

			map.put("target.publication.title", title);

		}

		map.put("target.publication.id", getValue(oaf.getEntity().getOriginalIdList()));

		map.put("trust", "1");

		//PROVENANCE INFO

		final String sourceDatasourceName = getValue(source.getEntity().getCollectedfromList());

		if (StringUtils.isNotBlank(sourceDatasourceName)) {

			map.put("provenance.datasource.name", sourceDatasourceName);

		}

		final String sourceOriginalIds = Joiner.on(",").join(getValues(source.getEntity().getOriginalIdList()));

		if (StringUtils.isNotBlank(sourceOriginalIds)) {

			map.put("provenance.publication.id.csv", sourceOriginalIds);

		}

		//final OpenAireEventPayload p = new OpenAireEventPayload(oaf.getEntity().getId());

		return new EventMessage(PRODUCER_ID, topic, JsonFormat.printToString(oaf), EventMessage.TTH_INFINITE, map);

	}

	private boolean hasPubDate(final Oaf current) {

		final Metadata m = current.getEntity().getResult().getMetadata();

		return StringUtils.isNotBlank(m.getDateofacceptance().getValue());

	}

	private boolean hasAbstract(final Oaf oaf) {

		return Iterables.all(oaf.getEntity().getResult().getMetadata().getDescriptionList(), new Predicate<StringField>() {

			@Override

			public boolean apply(final StringField s) {

				return StringUtils.isBlank(s.getValue());

			}

		});

	}

	private boolean hasAccess(final Oaf oaf, final String access, final boolean strict) {

		return Iterables.all(oaf.getEntity().getChildrenList(), new Predicate<OafEntity>() {

			@Override

			public boolean apply(final OafEntity entity) {

				final Predicate<Instance> p = new Predicate<Instance>() {

					@Override

					public boolean apply(final Instance i) {

						return access.equalsIgnoreCase(i.getLicence().getClassid());

					}

				};

				return strict ? Iterables.all(entity.getResult().getInstanceList(), p) :  Iterables.any(entity.getResult().getInstanceList(), p);

			}

		});

	}

	private boolean hasPid(final Oaf oaf, final String type) {

		return Iterables.any(oaf.getEntity().getPidList(), new Predicate<StructuredProperty>() {

			@Override

			public boolean apply(final StructuredProperty pid) {

				return pid.getQualifier().getClassid().equalsIgnoreCase(type);

			}

		});

	}

	private Function<ImmutableBytesWritable, Oaf> oafDeserialiser() {

		return new Function<ImmutableBytesWritable, Oaf>() {

			@Override

			public Oaf apply(final ImmutableBytesWritable input) {

				try {

					return Oaf.parseFrom(input.copyBytes());

				} catch (InvalidProtocolBufferException e) {

					throw new IllegalArgumentException(e);

				}

			}

		};

	}

	private <T> String getValue(Iterable<T> ts) {

		return Iterables.getFirst(getValues(ts), "");

	}

	private <T> List<String> getValues(Iterable<T> ts) {

		return Lists.newArrayList(Iterables.transform(ts, new Function<T, String>() {

			@Override

			public String apply(final T t) {

				if (t instanceof StructuredProperty) return ((StructuredProperty) t).getValue();

				if (t instanceof KeyValue) return ((KeyValue) t).getValue();

				if (t instanceof String) return (String) t;

				return "";

			}

		}));

	}

	private <T> String getKey(Iterable<T> ts) {

		return Iterables.getFirst(getKeys(ts), "");

	}

	private <T> List<String> getKeys(Iterable<T> ts) {

		return Lists.newArrayList(Iterables.transform(ts, new Function<T, String>() {

			@Override

			public String apply(final T t) {

				if (t instanceof KeyValue) return ((KeyValue) t).getKey();

				if (t instanceof String) return (String) t;

				return "";

			}

		}));

	}

}