aswi2022code-of-duty-gitlab/backend/src/main/java/cz/zcu/kiv/backendapi/external/ExternalCatalogServiceImpl.java @ e21b7441

package cz.zcu.kiv.backendapi.external;

import com.zaxxer.hikari.HikariDataSource;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;
import org.apache.tomcat.util.http.fileupload.FileUtils;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.persistence.Table;
import java.io.*;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.zip.GZIPInputStream;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;

/**
 * External catalog service implementation
 */
@Service
@Transactional
@RequiredArgsConstructor
@Slf4j
public class ExternalCatalogServiceImpl implements IExternalCatalogService {
    /**
     * Buffer size
     */
    public static final int BUFFER_SIZE = 1024;
    /**
     * Directory where files for external directory will be stored
     */
    private static final String DIRECTORY_FOR_EXTERNAL_FILES = "sources";
    /**
     * URL for Pleiades file
     */
    private static final String PLEIADES_FILE_URL = "https://atlantides.org/downloads/pleiades/dumps/pleiades-names-latest.csv.gz";
    /**
     * Name of Pleiades file
     */
    private static final String PLEIADES_FILE_NAME = "pleiades-names-latest.csv";
    /**
     * URL for Pleiades file – needs to be formatted (more sources)
     */
    private static final String GEONAMES_FILE_URL = "https://download.geonames.org/export/dump/%s.zip";
    /**
     * Name of GeoNames file – needs to be formatted (more sources)
     */
    private static final String GEONAMES_FILE_NAME = "%s.txt";
    /**
     * URL for CIGS file
     */
    private static final String CIGS_FILE_URL = "https://zenodo.org/record/5642899/files/CIGS_v1_4_20211101.csv";
    /**
     * Name of CIGS file
     */
    private static final String CIGS_FILE_NAME = "CIGS_v1_4_20211101.csv";
    /**
     * Batch size for saving items
     */
    private static final int BATCH_SIZE = 5000;

    /**
     * External catalog repository
     */
    private final ExternalCatalogRepository externalCatalogRepository;

    /**
     * Hikari data source
     */
    private final HikariDataSource hikariDataSource;


    @Override
    public void updateCatalog() {
        log.info("Updating external catalog");
        try {
            Files.createDirectories(Paths.get(DIRECTORY_FOR_EXTERNAL_FILES)); // creates directory if not exists
            FileUtils.cleanDirectory(new File(DIRECTORY_FOR_EXTERNAL_FILES)); // cleans the directory
            externalCatalogRepository.deleteAll(); // clears database – updated list will be stored later
            addPleiadesSource();
            addGeonamesSources();
            addCigsSources();
        } catch (IOException e) {
            e.printStackTrace();
        }
        log.info("External catalog updated");
    }

    /**
     * Downloads, extracts and reads Pleiades sources and saves them to database
     */
    private void addPleiadesSource() {
        List<ExternalCatalogItem> externalCatalogItems = new ArrayList<>();
        byte[] buffer = new byte[BUFFER_SIZE];
        File pleiadesFile = new File(new File(DIRECTORY_FOR_EXTERNAL_FILES), PLEIADES_FILE_NAME);
        try (InputStream fileIn = new URL(PLEIADES_FILE_URL).openStream();
             GZIPInputStream gZIPInputStream = new GZIPInputStream(fileIn);
             FileOutputStream fileOutputStream = new FileOutputStream(pleiadesFile)) {

            int bytes_read;

            while ((bytes_read = gZIPInputStream.read(buffer)) > 0) {
                fileOutputStream.write(buffer, 0, bytes_read);
            }

            log.info("The Pleiades file was decompressed successfully");

        } catch (IOException ex) {
            ex.printStackTrace();
        }

        try (InputStream csvData = new FileInputStream(pleiadesFile)) {
            CSVParser parser = CSVParser.parse(csvData, StandardCharsets.UTF_8, CSVFormat.Builder.create(CSVFormat.DEFAULT)
                    .setHeader()
                    .setSkipHeaderRecord(true)
                    .build());
            for (CSVRecord csvRecord : parser) {
                externalCatalogItems.add(new ExternalCatalogItem(csvRecord.toList(), ExternalSource.PLEIADES));
            }

        } catch (IOException ex) {
            ex.printStackTrace();
        }
        saveAllWithThreads(externalCatalogItems);
    }

    /**
     * Downloads, extracts and reads GeoNames sources and saves them to database
     */
    private void addGeonamesSources() {
        byte[] buffer = new byte[BUFFER_SIZE];
        for (String countryCode : ExternalCatalogItem.COUNTRY_CODES.keySet()) {
            List<ExternalCatalogItem> externalCatalogItems = new ArrayList<>();
            // Downloads file from URL and extracts it
            String url = String.format(GEONAMES_FILE_URL, countryCode);
            try (ZipInputStream zis = new ZipInputStream(new URL(url).openStream())) {
                ZipEntry zipEntry = zis.getNextEntry();
                while (zipEntry != null) {
                    FileOutputStream fileOutputStream = new FileOutputStream(new File(new File(DIRECTORY_FOR_EXTERNAL_FILES), zipEntry.getName()));
                    int bytes_read;
                    while ((bytes_read = zis.read(buffer)) > 0) {
                        fileOutputStream.write(buffer, 0, bytes_read);
                    }
                    zipEntry = zis.getNextEntry();
                    fileOutputStream.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }

            log.info("The Geonames file with country code " + countryCode + " was decompressed successfully");

            // Reads file and adds catalog items to list
            File geoNamesFile = new File(new File(DIRECTORY_FOR_EXTERNAL_FILES), String.format(GEONAMES_FILE_NAME, countryCode));
            try (BufferedReader reader = new BufferedReader(new FileReader(geoNamesFile))) {
                String line;
                while ((line = reader.readLine()) != null) {
                    externalCatalogItems.add(new ExternalCatalogItem(List.of(line.split("\t")), ExternalSource.GEONAMES));
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            saveAllWithThreads(externalCatalogItems);
        }
    }

    /**
     * Downloads and reads CIGS sources and saves them to database
     */
    private void addCigsSources() {
        List<ExternalCatalogItem> externalCatalogItems = new ArrayList<>();
        byte[] buffer = new byte[BUFFER_SIZE];
        File cigsFile = new File(new File(DIRECTORY_FOR_EXTERNAL_FILES), CIGS_FILE_NAME);

        try (InputStream inputStream = new URL(CIGS_FILE_URL).openStream();
             FileOutputStream fileOutputStream = new FileOutputStream(cigsFile)) {
            int bytes_read;

            while ((bytes_read = inputStream.read(buffer)) > 0) {

                fileOutputStream.write(buffer, 0, bytes_read);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        log.info("The CIGS file was downloaded successfully");

        try (InputStream csvData = new FileInputStream(cigsFile)) {
            CSVParser parser = CSVParser.parse(csvData, StandardCharsets.UTF_8, CSVFormat.Builder.create(CSVFormat.DEFAULT)
                    .setHeader()
                    .setSkipHeaderRecord(true)
                    .build());
            for (CSVRecord csvRecord : parser) {
                externalCatalogItems.add(new ExternalCatalogItem(csvRecord.toList(), ExternalSource.CIGS));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        saveAllWithThreads(externalCatalogItems);
    }

    /**
     * Creates list of lists of external catalog items divided by batch size
     *
     * @param externalCatalogItems list of external catalog items
     * @return divided list of lists of external catalog items
     */
    private List<List<ExternalCatalogItem>> createSublist(List<ExternalCatalogItem> externalCatalogItems) {
        List<List<ExternalCatalogItem>> listOfSubList = new ArrayList<>();
        for (int i = 0; i < externalCatalogItems.size(); i += BATCH_SIZE) {
            if (i + BATCH_SIZE <= externalCatalogItems.size()) {
                listOfSubList.add(externalCatalogItems.subList(i, i + BATCH_SIZE));
            } else {
                listOfSubList.add(externalCatalogItems.subList(i, externalCatalogItems.size()));
            }
        }
        return listOfSubList;
    }

    /**
     * Divides list of external catalog items to sublist, creates threads (for saving sublists in batch) and executes them
     *
     * @param externalCatalogItems list of external catalog items
     */
    private void saveAllWithThreads(List<ExternalCatalogItem> externalCatalogItems) {
        ExecutorService executorService = Executors.newFixedThreadPool(hikariDataSource.getMaximumPoolSize());
        List<List<ExternalCatalogItem>> subList = createSublist(externalCatalogItems);
        List<Callable<Void>> callables = subList.stream().map(sublist ->
                (Callable<Void>) () -> {
                    saveAllInBatch(sublist);
                    return null;
                }).collect(Collectors.toList());
        try {
            executorService.invokeAll(callables);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * Saves external catalog items in batch
     *
     * @param externalCatalogItems list of external catalog items
     */
    private void saveAllInBatch(List<ExternalCatalogItem> externalCatalogItems) {
        String sql = String.format("INSERT INTO %s (id, external_source, latitude, longitude, location_precision, max_date, " +
                "min_date, time_period_keys, pid, names, feature_code, country, accuracy, geoname_id, pleiades_id, osm_id) " +
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ExternalCatalogItem.class.getAnnotation(Table.class).name());
        try (Connection connection = hikariDataSource.getConnection();
             PreparedStatement statement = connection.prepareStatement(sql)) {
            int counter = 0;
            for (ExternalCatalogItem item : externalCatalogItems) {
                statement.clearParameters();
                statement.setObject(1, item.getId());
                statement.setString(2, item.getExternalSource().name());
                statement.setObject(3, item.getLatitude());
                statement.setObject(4, item.getLongitude());
                statement.setString(5, item.getLocationPrecision());
                statement.setObject(6, item.getMaxDate());
                statement.setObject(7, item.getMinDate());
                statement.setString(8, item.getTimePeriodKeys());
                statement.setString(9, item.getPid());
                statement.setString(10, String.join(",", item.getNames()));
                statement.setString(11, item.getFeatureCode());
                statement.setString(12, item.getCountry());
                statement.setObject(13, item.getAccuracy());
                statement.setObject(14, item.getGeonameId());
                statement.setObject(15, item.getPleiadesId());
                statement.setObject(16, item.getOsmId());
                statement.addBatch();
                counter++;
                if (counter % BATCH_SIZE == 0 || counter == externalCatalogItems.size()) {
                    statement.executeBatch();
                    statement.clearBatch();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}