D-Net

package eu.dnetlib.data.mapreduce.hbase.propagation.projecttoresult;

import com.google.common.collect.Sets;

import eu.dnetlib.data.mapreduce.hbase.propagation.NotValidResultSequenceException;

import eu.dnetlib.data.mapreduce.hbase.propagation.ResultIterator;

import eu.dnetlib.data.mapreduce.hbase.propagation.Value;

import eu.dnetlib.data.proto.KindProtos;

import eu.dnetlib.data.proto.OafProtos;

import eu.dnetlib.data.proto.RelMetadataProtos;

import eu.dnetlib.data.proto.ResultProjectProtos;

import org.apache.commons.lang3.StringUtils;

import org.apache.hadoop.io.Text;

import java.util.*;

import static eu.dnetlib.data.mapreduce.hbase.propagation.PropagationConstants.*;

import static eu.dnetlib.data.mapreduce.hbase.propagation.Utils.getDataInfo;

import static eu.dnetlib.data.mapreduce.hbase.propagation.Utils.getQualifier;

public class ResultProjectIterator extends ResultIterator {

    private Iterator<String> pli ;

    public ResultProjectIterator(final Iterable<Text> values, final String key) throws NotValidResultSequenceException{

        super(values, key);

    }

    @Override

    protected void checkSequence() throws NotValidResultSequenceException {

        if(!it.hasNext()){

            throw new NotValidResultSequenceException("Empty information for key");

        }

        pli = loadProjectList().iterator();

        getNext();

    }

    private void getNext(){

        if (pli.hasNext())

            resultId = pli.next();

        else

            resultId = TERMINATOR;

    }

    private Set<String> loadProjectList() throws NotValidResultSequenceException {

        final Set<String> fromResult = new HashSet<>();

        final Set<String> fromSemRel = new HashSet<>();

        while (it.hasNext()) {

            final Value v = Value.fromJson(it.next().toString());

            if(trust == null) {

                trust = v.getTrust();

            }

            switch (v.getType()) {

                case fromsemrel:

                    fromSemRel.addAll(Arrays.asList(StringUtils.split(v.getValue(), ",")));

                    break;

                case fromresult:

                    if(StringUtils.isNotBlank(v.getValue()))

                        fromResult.addAll(Arrays.asList(StringUtils.split(v.getValue(), ",")));

                    break;

                case valid:

                case notvalid:

                    throw new NotValidResultSequenceException("invalid type of result: " + v.getType());

            }

        }

       //takes the projects to be associated to the result that have not already been associated to it

        return Sets.difference(fromSemRel, fromResult);

    }

    @Override

    public List<OafProtos.Oaf> next() {

        ArrayList<OafProtos.Oaf> ret = new ArrayList<OafProtos.Oaf> (Arrays.asList(

                getOafRel(keyb, resultId, REL_RESULT_PROJECT),

                getOafRel(resultId, keyb, REL_PROJECT_RESULT)

        ));

        getNext();

        return ret;

    }

    private OafProtos.Oaf getOafRel(String source, String target, String semantics){

        final ResultProjectProtos.ResultProject.Builder rpb = ResultProjectProtos.ResultProject.newBuilder()

                .setOutcome(

                        ResultProjectProtos.ResultProject.Outcome.newBuilder()

                                .setRelMetadata(

                                        RelMetadataProtos.RelMetadata.newBuilder()

                                                .setSemantics(

                                                        getQualifier(semantics,DNET_RELATION_SCHEMA,semantics,DNET_RELATION_SCHEMA)

                                                )

                                )

                );

        final OafProtos.OafRel.Builder relation = OafProtos.OafRel.newBuilder()

                .setChild(false)

                .setSubRelType(SUBREL_TYPE)

                .setRelType(REL_TYPE)

                .setRelClass(semantics)

                .setTarget(target)

                .setSource(source)

                .setResultProject(rpb);

        return OafProtos.Oaf.newBuilder().setKind(KindProtos.Kind.relation)

                .setRel(relation)

                .setDataInfo(getDataInfo(trust, CLASS_PROJECT_ID,SCHEMA_ID,SCHEMA_NAME,DATA_INFO_TYPE,CLASS_PROJECT_NAME))

                .build();

    }

}