package com.walker.di.univocity;

import com.univocity.parsers.common.processor.BatchedColumnProcessor;
import com.univocity.parsers.common.processor.ObjectRowListProcessor;
import com.univocity.parsers.csv.CsvParser;
import com.univocity.parsers.csv.CsvParserSettings;
import com.walker.di.BatchLoadListener;
import com.walker.di.Constants;
import com.walker.di.DataImportException;
import com.walker.di.ErrorWriter;
import com.walker.di.support.InputStreamDataImportor;
import com.walker.di.univocity.util.BatchLoadUtils;
import com.walker.infrastructure.ApplicationRuntimeException;
import com.walker.infrastructure.utils.StringUtils;

import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * univocity实现的CSV文件导入。
 * @author 时克英
 * @date 2023-01-31
 */
public abstract class CsvDataImportor extends InputStreamDataImportor {

    private String batchError = null;

    @Override
    public String getBatchError(){
        return this.batchError;
    }

    @Override
    public void setBatchEnabled() {
        this.setBatchSize(MAX_BATCH_SIZE);
    }

    @Override
    public void setBatchSize(long batchSize) {
        super.setBatchSize(batchSize);
        // 2023-02-01 设置默认的批量加载监听器实现。
        this.setBatchLoadListener(new InternalBatchListenerImpl());
    }

    @Override
    public String getImportFileSuffix(){
        return Constants.IMPORT_ERROR_SUFFIX_CSV;
    }

    @Override
    protected List<Object[]> acquireImportDataList(Object source) throws DataImportException {
        this.checkSource(source);
        try{
            ObjectRowListProcessor rowListProcessor = new ObjectRowListProcessor();
            CsvParserSettings csvParserSettings = new CsvParserSettings();
            csvParserSettings.setHeaderExtractionEnabled(true);
//            csvParserSettings.setCommentCollectionEnabled(true);
            csvParserSettings.setMaxColumns(64);
            // 忽略第一行
            if(this.getIgnoreRows() > 0){
                csvParserSettings.setNumberOfRowsToSkip(this.getIgnoreRows());
            }
            csvParserSettings.setProcessor(rowListProcessor);
            CsvParser csvParser = new CsvParser(csvParserSettings);
            csvParser.parse((InputStream) source);

            String[] headers = rowListProcessor.getHeaders();
            if(headers == null || headers.length == 0){
                throw new DataImportException("导入文件未找到列名称行", null);
            }
            this.setFieldNames(Arrays.asList(headers));

            return rowListProcessor.getRows();

        } catch (Exception ex){
            throw new DataImportException("CsvParser解析异常:" + ex.getMessage(), ex);
        } finally {
            this.releaseSource(source);
        }
    }

    @Override
    protected void acquireImportBatch(BatchLoadListener batchLoadListener, Object source) throws DataImportException {
        this.checkSource(source);
        long sleepMillSeconds = this.getSleepMillSeconds();

        CsvParserSettings csvParserSettings = new CsvParserSettings();
        csvParserSettings.setMaxColumns(64);
        // 忽略第一行
        if(this.getIgnoreRows() > 0){
            csvParserSettings.setNumberOfRowsToSkip(this.getIgnoreRows());
        }
        csvParserSettings.setHeaderExtractionEnabled(true);
        BatchedColumnProcessor batchedColumnProcessor = new BatchedColumnProcessor((int)CsvDataImportor.this.getBatchSize()) {
            @Override
            public void batchProcessed(int dataSize) {
                // 设置列信息。
                CsvDataImportor.this.setFieldNames(Arrays.asList(this.getHeaders()));

                Map<String, List<String>> columnsByName = this.getColumnValuesAsMapOfNames();
                List<Object[]> rows = batchLoadListener.onBatchLoad(columnsByName, this.getHeaders(), dataSize);
                if(StringUtils.isEmptyList(rows)){
                    logger.error("未批量解析到任何数据:batchLoadListener.onBatchLoad, dataSize=" + dataSize);
                    return;
                }
                logger.debug("批量解析到数据量:" + rows.size());

                // 校验没批数据，出现问题直接记录到文件中。
                // 校验放到 this.doExecuteImport()方法中了 2023-02-02
//                CsvDataImportor.this.doValidate(rows);

                try {
                    CsvDataImportor.this.doExecuteImport(rows, getFieldNames());
                } catch (DataImportException e) {
                    batchError = e.getMessage();
                    throw new ApplicationRuntimeException("批量导入监听错误:" + e.getMessage(), e);
                } finally {
                    releaseSource(source);
                }

                if(sleepMillSeconds > 0){
                    try {
                        TimeUnit.MILLISECONDS.sleep(sleepMillSeconds);
                        logger.debug("-----> sleep: " + sleepMillSeconds);
                    } catch (InterruptedException e) {
                        logger.error("间隔等待(防止CUP过高)，出现异常:" + e.getMessage());
                    }
                }

//                for(Map.Entry<String, List<String>> entry : columnsByName.entrySet()){
//                    System.out.println("column = " + entry.getKey());
//                    System.out.println("values = " + entry.getValue());
//                    System.out.println("------------------------------");
//                }
            }
        };
        csvParserSettings.setProcessor(batchedColumnProcessor);
        CsvParser csvParser = new CsvParser(csvParserSettings);
        try{
            csvParser.parse((InputStream) source);
        } catch (Exception ex){
            throw new DataImportException(ex.getMessage(), ex);
        } finally {
            this.releaseSource(source);
        }
    }

//    @Override
//    protected void recordUnValidateData(Map<String, String> data) {
//    }

//    private void checkSource(Object source){
//        if(!(source instanceof InputStream)){
//            throw new IllegalArgumentException("source必须是:InputStream");
//        }
//    }

    private class InternalBatchListenerImpl implements BatchLoadListener{
        @Override
        public List<Object[]> onBatchLoad(Map<String, List<String>> columnsByName, String[] fieldNames, int dataSize) {
            return BatchLoadUtils.translateTo(columnsByName, fieldNames, dataSize);
        }
    }

    @Override
    protected ErrorWriter acquireErrorWriter(String id, List<String> fieldNames){
        InputStream templateFileStream = null;
        OutputStream errorFileStream = null;
        File errorFile = null;
        try {
//            templateFile = ResourceUtils.getFile(Constants.IMPORT_ERROR_FILE_TEMPLATE);
//            templateFileStream = this.getClass().getClassLoader().getResourceAsStream(Constants.IMPORT_ERROR_FILE_TEMPLATE);
            errorFile = new File(this.getErrorFile());
            errorFileStream = new BufferedOutputStream(new FileOutputStream(errorFile));
            // 这里不能关闭输出流，否则无法写入数据，后续会一直写入。
//            FileCopyUtils.copy(templateFileStream, errorFileStream, false);
            return new CsvErrorWriter(errorFileStream, fieldNames);
//            return new TsvErrorWriter(errorFileStream, fieldNames);

        } catch (FileNotFoundException e) {
            logger.error("未找到'错误模板文件':" + Constants.IMPORT_ERROR_FILE_TEMPLATE, e);
            return null;
        }
//        catch (IOException ex){
//            logger.error(errorFile.getAbsolutePath());
//            logger.error("通过模板拷贝错误文件失败:" + Constants.IMPORT_ERROR_FILE_TEMPLATE, ex);
//            return null;
//        }
        finally {
            if(templateFileStream != null){
                try {
                    templateFileStream.close();
                } catch (IOException e) {}
            }
//            if(errorFileStream != null){
//                try {
//                    errorFileStream.close();
//                } catch (IOException e) {}
//            }
        }
    }

}