D-Net

package eu.dnetlib.data.mapreduce.util;

import java.util.List;

import java.util.Map;

import java.util.Set;

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 com.google.protobuf.Descriptors.FieldDescriptor;

import com.google.protobuf.Message.Builder;

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.KindProtos.Kind;

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

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

import eu.dnetlib.data.proto.PersonProtos.Person;

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

import eu.dnetlib.data.proto.SpecialTrustProtos.SpecialTrust;

public class OafEntityMerger {

	private final Predicate<StringField> skipEmptyStringField = new Predicate<StringField>() {

		@Override

		public boolean apply(StringField s) {

			return s != null && s.getValue() != null && !s.getValue().isEmpty();

		}

	};

	private final Predicate<String> skipEmptyString = new Predicate<String>() {

		@Override

		public boolean apply(String s) {

			return s != null && !s.isEmpty();

		}

	};

	public static Oaf.Builder merge(String id, Iterable<Oaf> entities) {

		return new OafEntityMerger().mergeEntities(id, entities);

	}

	public static Oaf.Builder merge(Oaf.Builder builder) {

		return new OafEntityMerger().doMergeEntities(builder);

	}

	public Oaf.Builder mergeEntities(String id, Iterable<Oaf> entities) {

		Oaf.Builder builder = Oaf.newBuilder();

		String trust = "0.0";

		for (Oaf oaf : TrustOrdering.sort(entities)) {

			// doublecheck we're dealing only with main entities

			if (!oaf.getKind().equals(Kind.entity)) { throw new IllegalArgumentException("expected OafEntity!"); }

			String currentTrust = oaf.getDataInfo().getTrust();

			if (!currentTrust.equals(SpecialTrust.NEUTRAL.toString())) {

				trust = currentTrust;

			}

			builder.mergeFrom(oaf);

		}

		builder = doMergeEntities(builder);

		builder.getEntityBuilder().setId(id);

		builder.getDataInfoBuilder().setInferred(true).setDeletedbyinference(false).setTrust(trust);

		return builder;

	}

	public Oaf.Builder doMergeEntities(Oaf.Builder builder) {

		switch (builder.getEntity().getType()) {

		case datasource:

			break;

		case organization:

			break;

		case person:

			Person.Metadata.Builder person = builder.getEntityBuilder().getPersonBuilder().getMetadataBuilder();

			for (String field : Lists.newArrayList("secondnames")) {

				setSingleString(person, field);

			}

			break;

		case project:

			break;

		case result:

			Result.Metadata.Builder result = builder.getEntityBuilder().getResultBuilder().getMetadataBuilder();

			setTitle(result);

			// for (String field : Lists.newArrayList("subject", "relevantdate")) {

			for (String field : OafUtils.getFieldNames(Result.Metadata.getDescriptor(), Result.Metadata.SUBJECT_FIELD_NUMBER,

					Result.Metadata.RELEVANTDATE_FIELD_NUMBER)) {

				setStructuredProperty(result, field);

			}

			for (String field : OafUtils.getFieldNames(Result.Metadata.getDescriptor(), Result.Metadata.DESCRIPTION_FIELD_NUMBER)) {

				setLongestStringField(result, field);

			}

			for (String field : OafUtils.getFieldNames(Result.Metadata.getDescriptor(), Result.Metadata.SOURCE_FIELD_NUMBER)) {

				setUniqueStringField(result, field);

			}

			for (String field : OafUtils.getFieldNames(OafEntity.getDescriptor(), OafEntity.COLLECTEDFROM_FIELD_NUMBER)) {

				setKeyValues(builder.getEntityBuilder(), field);

			}

			for (String field : OafUtils.getFieldNames(OafEntity.getDescriptor(), OafEntity.PID_FIELD_NUMBER)) {

				setStructuredProperty(builder.getEntityBuilder(), field);

			}

			for (String field : OafUtils.getFieldNames(OafEntity.getDescriptor(), OafEntity.ORIGINALID_FIELD_NUMBER)) {

				setUniqueString(builder.getEntityBuilder(), field);

			}

			break;

		default:

			break;

		}

		return builder;

	}

	/**

	 * Helper method, avoid duplicated StructuredProperties in the given builder for the given fieldName

	 * 

	 * @param builder

	 * @param fieldName

	 */

	@SuppressWarnings("unchecked")

	private void setStructuredProperty(Builder builder, String fieldName) {

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

		final FieldDescriptor fd = builder.getDescriptorForType().findFieldByName(fieldName);

		final List<StructuredProperty> sps = (List<StructuredProperty>) builder.getField(fd);

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

			for (StructuredProperty sp : sps) {

				map.put(sp.getValue(), sp);

			}

			if (!map.isEmpty()) {

				builder.clearField(fd).setField(fd, Lists.newArrayList(map.values()));

			}

		}

	}

	/**

	 * Helper method, avoid duplicated KeyValues in the given builder for the given fieldName

	 * 

	 * @param builder

	 * @param fieldName

	 */

	@SuppressWarnings("unchecked")

	private void setKeyValues(Builder builder, String fieldName) {

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

		final FieldDescriptor fd = builder.getDescriptorForType().findFieldByName(fieldName);

		final List<KeyValue> kvs = (List<KeyValue>) builder.getField(fd);

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

			for (KeyValue sp : kvs) {

				map.put(sp.getKey(), sp);

			}

			if (!map.isEmpty()) {

				builder.clearField(fd).setField(fd, Lists.newArrayList(map.values()));

			}

		}

	}

	@SuppressWarnings("unchecked")

	private void setSingleString(Builder builder, String fieldName) {

		final FieldDescriptor fd = builder.getDescriptorForType().findFieldByName(fieldName);

		final List<StringField> field = (List<StringField>) builder.getField(fd);

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

			final StringField s = (StringField) Iterables.getLast(Iterables.filter(field, skipEmptyStringField), "");

			if (s != null && s.getValue() != null && !s.getValue().isEmpty()) {

				builder.clearField(fd).setField(fd, Lists.newArrayList(s));

			}

		}

	}

	@SuppressWarnings("unchecked")

	private void setLongestStringField(Builder builder, String fieldName) {

		final FieldDescriptor fd = builder.getDescriptorForType().findFieldByName(fieldName);

		final List<StringField> field = (List<StringField>) builder.getField(fd);

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

			StringField.Builder max = StringField.newBuilder().setValue("");

			int maxLength = 0;

			for (StringField sf : field) {

				if (sf.getValue().length() > maxLength) {

					maxLength = sf.getValue().length();

					max.clear();

					max.mergeFrom(sf);

				}

			}

			builder.clearField(fd).setField(fd, Lists.newArrayList(max.build()));

		}

	}

	@SuppressWarnings("unchecked")

	private void setUniqueStringField(Builder builder, String fieldName) {

		final FieldDescriptor fd = builder.getDescriptorForType().findFieldByName(fieldName);

		final List<StringField> field = (List<StringField>) builder.getField(fd);

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

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

			for (StringField s : Iterables.filter(field, skipEmptyStringField)) {

				map.put(s.getValue(), s);

			}

			builder.clearField(fd).setField(fd, Lists.newArrayList(map.values()));

		}

	}

	@SuppressWarnings("unchecked")

	private void setUniqueString(Builder builder, String fieldName) {

		final FieldDescriptor fd = builder.getDescriptorForType().findFieldByName(fieldName);

		final List<String> field = (List<String>) builder.getField(fd);

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

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

			for (String s : Iterables.filter(field, skipEmptyString)) {

				set.add(s);

			}

			builder.clearField(fd).setField(fd, Lists.newArrayList(set));

		}

	}

	private void setTitle(Result.Metadata.Builder metadata) {

		Iterable<StructuredProperty> filtered = Iterables.filter(metadata.getTitleList(), OafUtils.mainTitleFilter());

		if (!Iterables.isEmpty(filtered)) {

			metadata.clearTitle().addTitle(Iterables.getLast(filtered));

		}

	}

}