D-Net

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

import com.google.gson.Gson;

import com.jayway.jsonpath.DocumentContext;

import com.jayway.jsonpath.JsonPath;

import eu.dnetlib.data.bulktag.CommunityConfiguration;

import eu.dnetlib.data.bulktag.CommunityConfigurationFactory;

import eu.dnetlib.data.proto.FieldTypeProtos;

import eu.dnetlib.data.proto.OafProtos;

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

import org.apache.commons.lang.StringUtils;

import org.apache.hadoop.hbase.client.Put;

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 org.apache.hadoop.io.Writable;

import java.io.IOException;

import java.util.ArrayList;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

public class BulkTaggingMapper extends TableMapper<ImmutableBytesWritable, Writable> {

	private CommunityConfiguration cc;

	private ResultTagger tagger;

	private boolean enabled;

//	private Map<String,String> protoMappingParams;

//

//	private ProtoMappingParamEncapsulator param;

	private ProtoMap protoMappingParams;

	@Override

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

		super.setup(context);

		final String conf = context.getConfiguration().get("tagging.conf");

		enabled = context.getConfiguration().getBoolean("tagging.enabled",false);

		if (StringUtils.isBlank(conf)) {

			throw new IllegalArgumentException("missing bulk tagging configuration");

		}

		System.out.println("conf = " + conf);

		cc = CommunityConfigurationFactory.fromJson(conf);

		tagger = new ResultTagger();

		tagger.setTrust(context.getConfiguration().get("bulktagging.trust", "0.85"));

		String mappingProto = context.getConfiguration().get("mapping.proto", "{}");

		System.out.println(String.format("got mapping.proto: '%s'", mappingProto));

		protoMappingParams = new Gson().fromJson(mappingProto,ProtoMap.class);

		/*param = new Gson().fromJson(mappingProto, ProtoMappingParamEncapsulator.class);

		System.out.println("Sto usando la nuova implementazione");

		protoMappingParams = param.getP();*/

	}

	@Override

	protected void map(final ImmutableBytesWritable key, final Result value, final Context context) throws IOException, InterruptedException {

		final Map<byte[], byte[]> resultMap = value.getFamilyMap(Bytes.toBytes("result"));

		final byte[] body = resultMap.get(Bytes.toBytes("body"));

		if (body != null) {

			context.getCounter("Bulk Tagging", "not null body ").increment(1);

			//final Oaf tagged = tagger.enrichContext(Oaf.parseFrom(body), cc, context);

			final Oaf oaf = Oaf.parseFrom(body);

			final Oaf tagged = tagger.enrichContextCriteria(oaf,cc,context, protoMappingParams);// (Oaf.parseFrom(body), cc, context,protoMappingParams);

			if (tagged == null) {

				return;

			}

			long nTagged = tagged.getEntity().getResult().getMetadata().getContextList().stream()

					.flatMap(c -> c.getDataInfoList().stream())

					.map(FieldTypeProtos.DataInfo::getInferenceprovenance)

					.filter(infProv -> "bulktagging".equals(infProv))

					.count();

			context.getCounter("Bulk Tagging", " bulktagged ").increment(nTagged);

			final Put put = new Put(key.copyBytes()).add(Bytes.toBytes("result"), Bytes.toBytes("body"), tagged.toByteArray());

			if (enabled) {

				context.write(key, put);

				context.getCounter("Bulk Tagging", " write op ").increment(1);

			}

		}

		else{

			context.getCounter("Bulk Tagging", " null body ").increment(1);

		}

	}

	private Map<String,List<String>> getParamMap(byte[] body){

		Map<String,List<String>> param = new HashMap<>();

		DocumentContext jsonContext = JsonPath.parse(body);

		for(String key: protoMappingParams.keySet()) {

			try {

				param.put(key, jsonContext.read(protoMappingParams.get(key)));

			} catch (Exception e) {

				param.put(key, new ArrayList<>());

			}

		}

		return param;

	}

}