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

import java.io.Serializable;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.http.HttpStatus;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import eu.dnetlib.dhp.application.ArgumentApplicationParser;
import eu.dnetlib.dhp.oa.graph.dump.exceptions.NoAvailableEntityTypeException;
import eu.dnetlib.dhp.oa.zenodoapi.MissingConceptDoiException;
import eu.dnetlib.dhp.oa.zenodoapi.ZenodoAPIClient;

public class SendToZenodoHDFS implements Serializable {

	private static final String NEW = "new"; // to be used for a brand new deposition in zenodo
	private static final String VERSION = "version"; // to be used to upload a new version of a published deposition
	private static final String UPDATE = "update"; // to upload content to an open deposition not published
	private static final Integer NUMBER_OF_RETRIES = 5;
	private static final Integer DELAY = 10;
	private static final Integer MULTIPLIER = 5;

	private static final Logger log = LoggerFactory.getLogger(SendToZenodoHDFS.class);

	public static void main(final String[] args) throws Exception, MissingConceptDoiException {
		final ArgumentApplicationParser parser = new ArgumentApplicationParser(
			IOUtils
				.toString(
					SendToZenodoHDFS.class
						.getResourceAsStream(
							"/eu/dnetlib/dhp/oa/graph/dump/upload_zenodo.json")));

		parser.parseArgument(args);

		final String hdfsPath = parser.get("hdfsPath");
		final String hdfsNameNode = parser.get("nameNode");
		final String access_token = parser.get("accessToken");
		final String connection_url = parser.get("connectionUrl");
		final String metadata = parser.get("metadata");
		final String depositionType = parser.get("depositionType");
		final String concept_rec_id = Optional
			.ofNullable(parser.get("conceptRecordId"))
			.orElse(null);
		final Boolean publish = Optional
			.ofNullable(parser.get("publish"))
			.map(Boolean::valueOf)
			.orElse(false);

		final String depositionId = Optional.ofNullable(parser.get("depositionId")).orElse(null);

		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", hdfsNameNode);

		FileSystem fileSystem = FileSystem.get(conf);

		RemoteIterator<LocatedFileStatus> fileStatusListIterator = fileSystem
			.listFiles(
				new Path(hdfsPath), true);
		ZenodoAPIClient zenodoApiClient = new ZenodoAPIClient(connection_url, access_token);
		switch (depositionType) {
			case NEW:
				zenodoApiClient.newDeposition();
				break;
			case VERSION:
				if (concept_rec_id == null) {
					throw new MissingConceptDoiException("No concept record id has been provided");
				}
				zenodoApiClient.newVersion(concept_rec_id);
				break;
			case UPDATE:
				if (depositionId == null) {
					throw new MissingConceptDoiException("No deposition id has been provided");
				}
				zenodoApiClient.uploadOpenDeposition(depositionId);
				break;
			default:
				throw new NoAvailableEntityTypeException();
		}

		while (fileStatusListIterator.hasNext()) {
			LocatedFileStatus fileStatus = fileStatusListIterator.next();

			Path p = fileStatus.getPath();
			String pString = p.toString();
			boolean retry = true;
			int numberOfRetries = 0;

			if (!pString.endsWith("_SUCCESS")) {
				String name = pString.substring(pString.lastIndexOf("/") + 1);
				log.info("Upoloading: {}", name);
				FSDataInputStream inputStream = fileSystem.open(p);
				while (retry && numberOfRetries < NUMBER_OF_RETRIES) {
					int response_code = -1;

					try {
						response_code = zenodoApiClient
							.uploadIS3(inputStream, name, fileSystem.getFileStatus(p).getLen());
					} catch (Exception e) {
						log.info(e.getMessage());
						throw new RuntimeException("Error while uploading on Zenodo");
					}
					log.info("response code: {}", response_code);
					if (HttpStatus.SC_OK == response_code || HttpStatus.SC_CREATED == response_code) {
						retry = false;
					} else {
						numberOfRetries += 1;
						TimeUnit.SECONDS.sleep(DELAY * MULTIPLIER ^ numberOfRetries);
					}
				}
				if (numberOfRetries == NUMBER_OF_RETRIES) {
					throw new RuntimeException("reached the maximun number or retries to upload on Zenodo");
				}

			}
			// log.info(DateTime.now().toDateTimeISO().toString());
			TimeUnit.SECONDS.sleep(DELAY);
			// log.info("Delayed: {}", DateTime.now().toDateTimeISO().toString());
		}
		if (!metadata.equals("")) {
			zenodoApiClient.sendMretadata(metadata);
		}

	}

}
