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

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 org.springframework.web.multipart.MultipartFile;

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 item service implementation
 */
@Service
@Transactional
@RequiredArgsConstructor
@Slf4j
public class ExternalCatalogItemItemServiceImpl implements IExternalCatalogItemService {
    /**
     * 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";

    /**
     * Name of ANE file
     */
    private static final String ANE_FILE = "ANE.csv";

    /**
     * Batch size for saving items
     */
    private static final int BATCH_SIZE = 5000;

    /**
     * Regex for one arbitrary character in string
     */
    private static final String WILDCARD_CHARACTER_REGEX = "\\?";

    /**
     * Regex for one arbitrary character in string used in SQL
     */
    private static final String WILDCARD_CHARACTER_REGEX_SQL = "_";

    /**
     * Regex for any number of arbitrary characters in string
     */
    private static final String WILDCARD_CHARACTERS_REGEX = "\\*";

    /**
     * Regex for any number of arbitrary characters in string used in SQL
     */
    private static final String WILDCARD_CHARACTERS_REGEX_SQL = "%";

    /**
     * External catalog repository
     */
    private final ExternalCatalogItemRepository externalCatalogItemRepository;

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

    /**
     * Set to string convertor
     */
    private final SetToStringConverter setToStringConverter;


    @Override
    public void updateCatalog(MultipartFile file) {
        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
        } catch (Exception e) {
            e.printStackTrace();
        }

        externalCatalogItemRepository.deleteAll(); // clears database – updated list will be stored later

        try {
            addPleiadesRecords();
        } catch (Exception e) {
            e.printStackTrace();
        }

        try {
            addGeonamesRecords();
        } catch (Exception e) {
            e.printStackTrace();
        }

        try {
            if (file.getOriginalFilename() != null) {
                File cigsFile = new File(new File(DIRECTORY_FOR_EXTERNAL_FILES), file.getOriginalFilename());
                Files.copy(file.getInputStream(), cigsFile.toPath());
                addCigsRecords(cigsFile);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        try {
            addAneRecords();
        } catch (Exception e) {
            e.printStackTrace();
        }

        log.info("External catalog updated");
    }

    @Override
    public List<ExternalCatalogItem> getCatalog(String name, ExternalSource source) {
        name = WILDCARD_CHARACTERS_REGEX_SQL +
                name.replaceAll(WILDCARD_CHARACTER_REGEX, WILDCARD_CHARACTER_REGEX_SQL).replaceAll(WILDCARD_CHARACTERS_REGEX, WILDCARD_CHARACTERS_REGEX_SQL) +
                WILDCARD_CHARACTERS_REGEX_SQL;
        return externalCatalogItemRepository.getFilteredExternalCatalog(name, source.name());
    }

    /**
     * Downloads, extracts and reads records from Pleiades and saves them to database
     */
    private void addPleiadesRecords() {
        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) {
                ExternalCatalogItem e = new ExternalCatalogItem(csvRecord.toList(), ExternalSource.PLEIADES);
                externalCatalogItems.add(e);
            }

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

        log.info("Records from Pleiades added");
    }

    /**
     * Downloads, extracts and reads records from GeoNames and saves them to database
     */
    private void addGeonamesRecords() {
        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) {
                    ExternalCatalogItem e = new ExternalCatalogItem(List.of(line.split("\t")), ExternalSource.GEONAMES);
                    externalCatalogItems.add(e);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            saveAllWithThreads(externalCatalogItems);
        }

        log.info("Records from GeoNames added");
    }

    /**
     * Saves records from ANE to database
     */
    private void addAneRecords() {
        List<ExternalCatalogItem> externalCatalogItems = new ArrayList<>();
        // Reads file and adds catalog items to list
        try (InputStream csvData = getClass().getClassLoader().getResourceAsStream(ANE_FILE)) {
            if (csvData == null) {
                return;
            }
            CSVParser parser = CSVParser.parse(csvData, StandardCharsets.UTF_8, CSVFormat.Builder.create(CSVFormat.DEFAULT)
                    .setHeader()
                    .setSkipHeaderRecord(true)
                    .build());
            for (CSVRecord csvRecord : parser) {
                ExternalCatalogItem e = new ExternalCatalogItem(csvRecord.toList(), ExternalSource.ANE);
                externalCatalogItems.add(e);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        saveAllWithThreads(externalCatalogItems);

        log.info("Records from ANE added");
    }


    /**
     * Reads records from CIGS and saves them to database
     *
     * @param cigsFile CIGS file
     */
    private void addCigsRecords(File cigsFile) {
        List<ExternalCatalogItem> externalCatalogItems = new ArrayList<>();

        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) {
                ExternalCatalogItem e = new ExternalCatalogItem(csvRecord.toList(), ExternalSource.CIGS);
                externalCatalogItems.add(e);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        saveAllWithThreads(externalCatalogItems);

        log.info("Records from CIGS added");
    }

    /**
     * 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 sqlExternalCatalogItem = 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 statementExternalCatalogItem = connection.prepareStatement(sqlExternalCatalogItem)) {
            int counter = 0;
            for (ExternalCatalogItem item : externalCatalogItems) {
                statementExternalCatalogItem.clearParameters();
                statementExternalCatalogItem.setObject(1, item.getId());
                statementExternalCatalogItem.setString(2, item.getExternalSource().name());
                statementExternalCatalogItem.setObject(3, item.getLatitude());
                statementExternalCatalogItem.setObject(4, item.getLongitude());
                statementExternalCatalogItem.setString(5, item.getLocationPrecision());
                statementExternalCatalogItem.setObject(6, item.getMaxDate());
                statementExternalCatalogItem.setObject(7, item.getMinDate());
                statementExternalCatalogItem.setString(8, item.getTimePeriodKeys());
                statementExternalCatalogItem.setString(9, item.getPid());
                statementExternalCatalogItem.setString(10, setToStringConverter.convertToDatabaseColumn(item.getNames()));
                statementExternalCatalogItem.setString(11, item.getFeatureCode());
                statementExternalCatalogItem.setString(12, item.getCountry());
                statementExternalCatalogItem.setObject(13, item.getAccuracy());
                statementExternalCatalogItem.setObject(14, item.getGeonameId());
                statementExternalCatalogItem.setObject(15, item.getPleiadesId());
                statementExternalCatalogItem.setObject(16, item.getOsmId());
                statementExternalCatalogItem.addBatch();

                counter++;
                if (counter % BATCH_SIZE == 0 || counter == externalCatalogItems.size()) {
                    statementExternalCatalogItem.executeBatch();
                    statementExternalCatalogItem.clearBatch();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}