ITPGRFAStatusUpdater.java

/*
 * Copyright 2019 Global Crop Diversity Trust
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.genesys.server.service.worker;

import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;

import org.genesys.server.model.impl.Country;
import org.genesys.server.model.impl.ITPGRFAStatus;
import org.genesys.server.service.CountryService;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Component;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.StopWatch;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.poi.ss.usermodel.DataFormatter;
import org.apache.poi.ss.usermodel.Row;
import org.apache.poi.ss.usermodel.Sheet;
import org.apache.poi.ss.usermodel.Workbook;
import org.apache.poi.xssf.usermodel.XSSFWorkbook;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Update country ITPGRFA status by fetching data from
 * {@link ITPGRFAStatusUpdater#ITPGRFA_STATUS_URL}.
 *
 * @author Matija Obreza
 */
@Component
public class ITPGRFAStatusUpdater {

	public static String ITPGRFA_STATUS_URL = "https://www.fao.org/media/docs/planttreatylibraries/membership/list_cps.xlsx";

	public static final Logger LOG = LoggerFactory.getLogger(ITPGRFAStatusUpdater.class);

	@Autowired
	private CountryService countryService;

	@Autowired
	private TaskExecutor taskExecutor;

	private static final int BATCH_SIZE = 20;

	private static final String[] HEADER = { "Country", "ISO3", "Region", "Contracting Party", "Income", "Development", "Entry into Force" };

	protected static final int COLUMN_COUNTRY_CODE3 = 1;
	protected static final int COLUMN_CONTRACTING_PARTY = 3;
//	protected static final int COLUMN_MEMBERSHIP = 4;
//	protected static final int COLUMN_MEMBERSHIP_BY = 5;

	/**
	 * Update local {@link ITPGRFAStatus} entries with data from XLSX
	 *
	 * @throws IOException
	 */
	public void downloadAndUpdate() throws IOException {
		InputStream itpgrfaXLSXStream = null;

		final HttpGet httpget = new HttpGet(ITPGRFA_STATUS_URL);
		HttpResponse response = null;
		final CloseableHttpClient httpclient = HttpClientBuilder.create().build();
		try {
			response = httpclient.execute(httpget);

			// Get hold of the response entity
			final HttpEntity entity = response.getEntity();
			if (entity == null) {
				LOG.warn("No HttpEntity in response, bailing out");
				return;
			}
			LOG.debug("{} {}", entity.getContentType(), entity.getContentLength());

			// If the response does not enclose an entity, there is no
			// need to bother about connection release
			if (entity != null) {
				itpgrfaXLSXStream = new BufferedInputStream(entity.getContent());
			}

			updateFromStream(itpgrfaXLSXStream);

		} catch (final ClientProtocolException e) {
			LOG.error(e.getMessage(), e);
			throw new IOException(e);
		} catch (final IOException e) {
			LOG.error(e.getMessage(), e);
			throw e;
		} finally {
			IOUtils.closeQuietly(itpgrfaXLSXStream);
			IOUtils.closeQuietly(httpclient);
		}
	}

	private void updateFromStream(InputStream instream) throws IOException {
		Workbook workbook = null;
		try {
			workbook = new XSSFWorkbook(instream);
			final Sheet sheet = workbook.getSheetAt(0);
			final DataFormatter dataFormatter = new DataFormatter();

			final List<String[]> batch = new ArrayList<>(BATCH_SIZE);

			// Read and validate headers from the first row
			final Row headerRow = sheet.getRow(0);
			if (headerRow == null) {
				throw new IOException("XLSX header row missing");
			}
			final String[] headers = new String[HEADER.length];
			for (int i = 0; i < HEADER.length; i++) {
				String hv = dataFormatter.formatCellValue(headerRow.getCell(i));
				hv = StringUtils.isBlank(hv) ? null : hv.trim();
				headers[i] = hv;
			}
			LOG.warn("Got headers: {}", ArrayUtils.toString(headers));
			for (int i = HEADER.length - 1; i >= 0; i--) {
				if (headers[i] == null || !headers[i].equals(HEADER[i])) {
					throw new IOException("XLSX header mismatch, found '" + headers[i] + "' instead of '" + HEADER[i] + "'");
				}
			}

			// Timer
			final StopWatch stopWatch = new StopWatch();
			stopWatch.start();

			final int lastRow = sheet.getLastRowNum();
			for (int r = 1; r <= lastRow; r++) {
				final Row row = sheet.getRow(r);
				if (row == null) continue;
				final String[] line = new String[HEADER.length];
				for (int c = 0; c < HEADER.length; c++) {
					String val = dataFormatter.formatCellValue(row.getCell(c));
					if (StringUtils.isBlank(val) || "null".equalsIgnoreCase(val)) {
						line[c] = null;
					} else {
						line[c] = val.trim();
					}
				}

				if (LOG.isDebugEnabled()) {
					LOG.debug(">>> {}", ArrayUtils.toString(line, "NULL"));
				}

				batch.add(line);
				if (batch.size() >= BATCH_SIZE) {
					workIt(batch);
					batch.clear();
				}
			}

			if (batch.size() > 0) {
				LOG.debug("Have items in the batch after loop.");
				workIt(batch);
				batch.clear();
			}

			stopWatch.stop();
			LOG.info("Done importing ITPGRFA status in {}ms", stopWatch.getTime());
		} finally {
			IOUtils.closeQuietly(workbook);
		}
	}

	private void workIt(final List<String[]> batch) {

		// Need copy!
		final List<String[]> batchCopy = new ArrayList<String[]>(batch);

		taskExecutor.execute(() -> {
			for (final String[] line : batchCopy) {
				if (LOG.isDebugEnabled()) {
					LOG.debug("Working on {}", ArrayUtils.toString(line, "NULL"));
				}
				if (!StringUtils.isBlank(line[COLUMN_COUNTRY_CODE3])) {
					updateCountry(line[COLUMN_COUNTRY_CODE3], line[COLUMN_CONTRACTING_PARTY], null, null); //, line[COLUMN_MEMBERSHIP], line[COLUMN_MEMBERSHIP_BY]);
				}
			}
		});
	}

	protected void updateCountry(String countryCode, String contractingParty, String membership, String membershipBy) {
		final Country country = countryService.findCountry(countryCode);

		if (country == null) {
			LOG.error("No country with name={}", countryCode);
			return;
		}

		countryService.updateITPGRFA(country, contractingParty, membership, membershipBy);
	}
}