dhp-graph-dump/dump/src/main/java/eu/dnetlib/dhp/oa/graph/dump/skgif/DumpGrant.java

package eu.dnetlib.dhp.oa.graph.dump.skgif;

import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;

import java.io.Serializable;
import java.io.StringReader;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

import org.apache.avro.generic.GenericData;
import org.apache.commons.io.IOUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FilterFunction;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.api.java.function.MapGroupsFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.io.SAXReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import eu.dnetlib.dhp.application.ArgumentApplicationParser;
import eu.dnetlib.dhp.oa.graph.dump.skgif.beans.EncloseMinElement;
import eu.dnetlib.dhp.schema.oaf.Project;
import eu.dnetlib.dhp.schema.oaf.Relation;
import eu.dnetlib.dhp.skgif.model.*;
import scala.Tuple2;

/**
 * @author miriam.baglioni
 * @Date 22/02/24
 */
public class DumpGrant implements Serializable {
	private static final Logger log = LoggerFactory.getLogger(DumpGrant.class);

	public static void main(String[] args) throws Exception {
		String jsonConfiguration = IOUtils
			.toString(
				DumpGrant.class
					.getResourceAsStream(
						"/eu/dnetlib/dhp/oa/graph/dump/dump_grant_parameters.json"));

		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
		parser.parseArgument(args);

		Boolean isSparkSessionManaged = Optional
			.ofNullable(parser.get("isSparkSessionManaged"))
			.map(Boolean::valueOf)
			.orElse(Boolean.TRUE);

		log.info("isSparkSessionManaged: {}", isSparkSessionManaged);

		final String inputPath = parser.get("sourcePath");
		log.info("inputPath: {}", inputPath);

		final String workingDir = parser.get("workingDir");
		log.info("workingDir: {}", workingDir);

		final String outputPath = parser.get("outputPath");
		log.info("outputPath: {}", outputPath);

		SparkConf conf = new SparkConf();

		runWithSparkSession(
			conf,
			isSparkSessionManaged,
			spark -> {
				Utils.removeOutputDir(spark, outputPath + "Grant");

				mapGrants(spark, inputPath, outputPath, workingDir);
			});
	}

	private static void mapGrants(SparkSession spark, String inputPath, String outputPath, String workingDir) {
		Dataset<Project> projects = Utils
			.readPath(spark, inputPath + "project", Project.class)
			.filter(
				(FilterFunction<Project>) p -> !p.getDataInfo().getDeletedbyinference() &&
					!p.getDataInfo().getInvisible());
		Dataset<Relation> relations = Utils
			.readPath(spark, inputPath + "relation", Relation.class)
			.filter(
				(FilterFunction<Relation>) r -> !r.getDataInfo().getDeletedbyinference() &&
					!r.getDataInfo().getInvisible() &&
					r.getRelClass().equalsIgnoreCase(RelationType.ORGANIZATION_PARTICIPANT_IN_PROJECT.label));
		Dataset<EncloseMinElement> eme = Utils
			.readPath(spark, workingDir + "minEntity", EncloseMinElement.class)
			.filter((FilterFunction<EncloseMinElement>) e -> e.getMinOrganization() != null);

		Dataset<Tuple2<String, EncloseMinElement>> partecipantOrganization = relations
			.joinWith(eme, relations.col("source").equalTo(eme.col("enclosedEntityId")))
			.map(
				(MapFunction<Tuple2<Relation, EncloseMinElement>, Tuple2<String, EncloseMinElement>>) t2 -> new Tuple2<>(
					t2._1().getTarget(), t2._2()),
				Encoders.tuple(Encoders.STRING(), Encoders.bean(EncloseMinElement.class)));

		projects
			.joinWith(partecipantOrganization, projects.col("id").equalTo(partecipantOrganization.col("_1")), "left")
			.groupByKey(
				(MapFunction<Tuple2<Project, Tuple2<String, EncloseMinElement>>, String>) t2 -> t2._1().getId(),
				Encoders.STRING())
			.mapGroups(
				(MapGroupsFunction<String, Tuple2<Project, Tuple2<String, EncloseMinElement>>, Grant>) (k, v) -> {
					Grant g = new Grant();
					Tuple2<Project, Tuple2<String, EncloseMinElement>> first = v.next();
					g.setLocal_identifier( k);
					g.setGrantCode(first._1().getCode().getValue());
					g.setIdentifiers(getProjectIdentifier(first._1()));
					g.setTitle(first._1().getTitle().getValue());
					g
						.setSummary(
							Optional
								.ofNullable(first._1().getSummary())
								.map(value -> value.getValue())
								.orElse(new String()));
					g
						.setAcronym(
							Optional
								.ofNullable(first._1().getAcronym())
								.map(value -> value.getValue())
								.orElse(new String()));
					g.setFunder(Utils.getFunderName(first._1().getFundingtree().get(0).getValue()));
					// * private String funding_stream;// fundingtree to be used the xpath //funding_level_[n]
					g.setFunding_stream(getFundingStream(first._1().getFundingtree().get(0).getValue()));
					g
						.setCurrency(
							Optional
								.ofNullable(first._1().getCurrency())
								.map(value -> value.getValue())
								.orElse(new String()));
					g
						.setFunded_amount(
							Optional
								.ofNullable(first._1().getFundedamount())
								.orElse(null));
					g
						.setKeywords(
							first
								._1()
								.getSubjects()
								.stream()
								.map(s -> s.getValue())
								.collect(Collectors.toList()));
					g
						.setStart_date(
							Optional
								.ofNullable(first._1().getStartdate())
								.map(value -> value.getValue())
								.orElse(new String()));
					g
						.setEnd_date(
							Optional
								.ofNullable(first._1().getEnddate())
								.map(value -> value.getValue())
								.orElse(new String()));
					g
						.setWebsite(
							Optional
								.ofNullable(first._1().getWebsiteurl())
								.map(value -> value.getValue())
								.orElse(new String()));
					if (Optional.ofNullable(first._2()).isPresent()) {
						List<MinOrganization> relevantOrganizatios = new ArrayList<>();
						relevantOrganizatios.add(first._2()._2().getMinOrganization());
						v
							.forEachRemaining(
								t2 -> relevantOrganizatios
									.add(t2._2()._2().getMinOrganization()));
						g.setBeneficiaries(relevantOrganizatios);
					}
					return g;
				}, Encoders.bean(Grant.class))
			.write()
			.mode(SaveMode.Overwrite)
			.option("compression", "gzip")
			.json(outputPath + "Grant");
	}

	private static String getFundingStream(String fundingtree) throws DocumentException {
		final Document doc;

		doc = new SAXReader().read(new StringReader(fundingtree));
		if (Optional.ofNullable(doc.selectNodes("//funding_level_0")).isPresent() &&
			doc.selectNodes("//funding_level_0").size() > 0)
			return ((org.dom4j.Node) (doc.selectNodes("//funding_level_0").get(0))).getText();
		return new String();

	}

	private static List<Identifier> getProjectIdentifier(Project project) throws DocumentException {
		List<Identifier> identifiers = new ArrayList<>();
		if (project.getPid().size() > 0)
			project
				.getPid()
				.stream()
				.forEach(p -> identifiers.add(Identifier.newInstance(p.getQualifier().getClassid(), p.getValue())));
		identifiers
			.add(
				Identifier
					.newInstance(
						Utils.getFunderName(project.getFundingtree().get(0).getValue()), project.getCode().getValue()));
		return identifiers;

	}
}