dnet-applications/libs/dnet-broker-apps-common/src/main/java/eu/dnetlib/broker/common/subscriptions/MapCondition.java

package eu.dnetlib.broker.common.subscriptions;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.Operator;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.search.MatchQuery.ZeroTermsQuery;

import eu.dnetlib.broker.common.elasticsearch.Event;
import eu.dnetlib.broker.common.utils.DateParser;
import eu.dnetlib.broker.common.utils.MapValueType;

public class MapCondition {

	private static final Log log = LogFactory.getLog(MapCondition.class);

	private String field;
	private MapValueType fieldType;
	private ConditionOperator operator;
	private List<ConditionParams> listParams = new ArrayList<>();

	public MapCondition() {}

	public MapCondition(final String field, final MapValueType fieldType, final ConditionOperator operator, final List<ConditionParams> listParams) {
		this.field = field;
		this.fieldType = fieldType;
		this.operator = operator;
		this.listParams = listParams;
	}

	public String getField() {
		return field;
	}

	public void setField(final String field) {
		this.field = field;
	}

	public MapValueType getFieldType() {
		return fieldType;
	}

	public void setFieldType(final MapValueType fieldType) {
		this.fieldType = fieldType;
	}

	public ConditionOperator getOperator() {
		return operator;
	}

	public void setOperator(final ConditionOperator operator) {
		this.operator = operator;
	}

	public List<ConditionParams> getListParams() {
		return listParams;
	}

	public void setListParams(final List<ConditionParams> listParams) {
		this.listParams = listParams;
	}

	public boolean verifyEvent(final Event event) {
		final Object val = event.getMap().get(field);

		return getListParams().stream().anyMatch(cp -> {
			try {
				switch (fieldType) {
				case STRING:
					return cp.verify(val.toString(), operator);
				case INTEGER:
					return cp.verify(val instanceof Long ? (long) val : NumberUtils.toInt(val.toString()), operator);
				case FLOAT:
					return cp.verify(val instanceof Double ? (double) val : NumberUtils.toDouble(val.toString()), operator);
				case BOOLEAN:
					return cp.verify(val instanceof Boolean ? (boolean) val : val.toString().equalsIgnoreCase("true"), operator);
				case DATE:
					return cp.verify(val instanceof Date ? (Date) val : DateParser.parse(val.toString()), operator);
				case LIST_STRING:
					return ((List<?>) val).stream().map(Object::toString).anyMatch(s -> cp.verify(s, operator));
				case LIST_DATE:
					return ((List<?>) val).stream().map(o -> o instanceof Date ? (Date) o : DateParser.parse(o.toString()))
							.anyMatch(s -> cp.verify(s, operator));
				case LIST_INTEGER:
					return ((List<?>) val).stream().map(Object::toString).map(NumberUtils::toInt).anyMatch(n -> cp.verify(n, operator));
				case LIST_FLOAT:
					return ((List<?>) val).stream().map(Object::toString).map(NumberUtils::toDouble).anyMatch(n -> cp.verify(n, operator));
				default:
					break;
				}
			} catch (final Throwable e) {
				log.error("Error verifying condition " + this + " on event: " + event, e);
			}
			return false;
		});
	}

	public QueryBuilder asQueryBuilder() {
		if (listParams == null || listParams.isEmpty()) { return null; }

		if (operator == ConditionOperator.RANGE) {
			if (listParams.size() == 1) {
				return createSimpleRangeOperator(listParams.get(0));
			} else {
				return createListRangeOperator(listParams);
			}
		} else {
			if (listParams.size() == 1) {
				return createSimpleOperator(listParams.get(0));
			} else {
				return createListOperator(listParams);
			}
		}
	}

	private QueryBuilder createSimpleOperator(final ConditionParams p) {
		if (StringUtils.isNotBlank(p.getValue())) {
			return QueryBuilders.matchQuery("map." + field, convertToType(p.getValue())).operator(Operator.AND)
					.zeroTermsQuery(ZeroTermsQuery.ALL);
		} else {
			return null;
		}
	}

	private QueryBuilder createListOperator(final List<ConditionParams> list) {
		final BoolQueryBuilder query = QueryBuilders.boolQuery();
		for (final ConditionParams p : list) {
			query.should(QueryBuilders.matchQuery("map." + field, convertToType(p.getValue())).operator(Operator.AND)
					.zeroTermsQuery(ZeroTermsQuery.ALL));
		}
		return query;
	}

	private QueryBuilder createSimpleRangeOperator(final ConditionParams p) {
		if (StringUtils.isNotBlank(p.getValue()) || StringUtils.isNotBlank(p.getOtherValue())) {
			return QueryBuilders.rangeQuery("map." + field)
					.from(convertToType(p.getValue()))
					.to(convertToType(p.getOtherValue()));
		} else {
			return null;
		}
	}

	private QueryBuilder createListRangeOperator(final List<ConditionParams> list) {
		final BoolQueryBuilder query = QueryBuilders.boolQuery();

		for (final ConditionParams p : list) {
			query.should(QueryBuilders
					.rangeQuery("map." + field)
					.from(convertToType(p.getValue()))
					.to(convertToType(p.getOtherValue())));
		}
		return query;
	}

	private Object convertToType(final String s) {
		switch (fieldType) {
		case STRING:
		case LIST_STRING:
			return s;
		case INTEGER:
		case LIST_INTEGER:
			return NumberUtils.toLong(s, 0);
		case FLOAT:
		case LIST_FLOAT:
			return NumberUtils.toDouble(s, 0);
		case DATE:
		case LIST_DATE:
			return DateParser.parse(s).getTime();
		case BOOLEAN:
		case LIST_BOOLEAN:
			return "true".equalsIgnoreCase(s);
		}
		return null;
	}

	@Override
	public String toString() {
		return String.format("[ %s %s %s ]", field, operator, listParams);
	}
}