feat: P4ADEV-1659 wf treasury opi ingestion error management

build.gradle.kts

@@ -60,6 +60,7 @@ val activationVersion = "2.1.3"
 val jaxbVersion = "4.0.5"
 val jaxbApiVersion = "4.0.2"
 val openApiToolsVersion = "0.2.6"
+val openCsvVersion = "5.9"
 
 
 dependencies {
@@ -90,6 +91,9 @@ dependencies {
 	compileOnly("org.projectlombok:lombok")
 	annotationProcessor("org.projectlombok:lombok")
 
+	//openCsv
+	implementation("com.opencsv:opencsv:$openCsvVersion")
+
 	//jaxb
 	runtimeOnly("org.glassfish.jaxb:jaxb-runtime:$jaxbVersion")
 	implementation("com.sun.xml.bind:jaxb-xjc:$jaxbVersion")
@@ -100,6 +104,15 @@ dependencies {
 
 	// openApi
 	implementation("org.openapitools:jackson-databind-nullable:$openApiToolsVersion")
+
+    jaxbext("com.github.jaxb-xew-plugin:jaxb-xew-plugin:2.1")
+    jaxbext("org.jvnet.jaxb:jaxb-plugins:4.0.0")
+
+
+	jaxbext("com.github.jaxb-xew-plugin:jaxb-xew-plugin:2.1")
+	jaxbext("org.jvnet.jaxb:jaxb-plugins:4.0.0")
+
+
 }
 
 
@@ -125,6 +138,20 @@ jaxb {
 			schema = file("src/main/resources/xsd/FlussoRiversamento.xsd")
 			bindings = layout.files("src/main/resources/xsd/FlussoRiversamento.xjb")
 		}
+		register("Opi14TresauryFlow") {
+			extension = true
+			args = listOf("-xmlschema","-Xsimplify")
+			outputDir = file("$projectDir/build/generated/jaxb/java")
+			schema = file("src/main/resources/xsd/OPI_GIORNALE_DI_CASSA_V_1_4.xsd")
+			bindings = layout.files("src/main/resources/xsd/OPI_GIORNALE_DI_CASSA_V_1_4.xjb")
+		}
+		register("Opi161TresauryFlow") {
+			extension = true
+			args = listOf("-xmlschema","-Xsimplify")
+			outputDir = file("$projectDir/build/generated/jaxb/java")
+			schema = file("src/main/resources/xsd/OPI_GIORNALE_DI_CASSA_V_1_6_1.xsd")
+			bindings = layout.files("src/main/resources/xsd/OPI_GIORNALE_DI_CASSA_V_1_6_1.xjb")
+		}
 	}
 }
 

gradle.lockfile

@@ -11,18 +11,23 @@ com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.17.2=compileClasspath
 com.fasterxml.jackson.module:jackson-module-parameter-names:2.18.1=compileClasspath
 com.fasterxml.jackson:jackson-bom:2.18.1=compileClasspath
 com.fasterxml:classmate:1.7.0=compileClasspath
+com.opencsv:opencsv:5.9=compileClasspath
 com.sun.xml.bind:jaxb-core:4.0.5=compileClasspath
 com.sun.xml.bind:jaxb-impl:4.0.5=compileClasspath
 com.sun.xml.bind:jaxb-jxc:4.0.5=compileClasspath
 com.sun.xml.bind:jaxb-xjc:4.0.5=compileClasspath
+commons-beanutils:commons-beanutils:1.9.4=compileClasspath
 commons-codec:commons-codec:1.16.1=compileClasspath
+commons-collections:commons-collections:3.2.2=compileClasspath
 commons-io:commons-io:2.16.1=compileClasspath
+commons-logging:commons-logging:1.2=compileClasspath
 io.micrometer:micrometer-commons:1.13.6=compileClasspath
 io.micrometer:micrometer-observation:1.13.6=compileClasspath
 jakarta.activation:jakarta.activation-api:2.1.3=compileClasspath
 jakarta.annotation:jakarta.annotation-api:2.1.1=compileClasspath
 jakarta.validation:jakarta.validation-api:3.0.2=compileClasspath
 jakarta.xml.bind:jakarta.xml.bind-api:4.0.2=compileClasspath
+org.apache.commons:commons-collections4:4.4=compileClasspath
 org.apache.commons:commons-compress:1.27.1=compileClasspath
 org.apache.commons:commons-lang3:3.17.0=compileClasspath
 org.apache.commons:commons-text:1.12.0=compileClasspath

src/main/java/it/gov/pagopa/payhub/activities/activity/paymentsreporting/PaymentsReportingIngestionFlowFileActivityImpl.java

@@ -6,6 +6,7 @@
 import it.gov.pagopa.payhub.activities.dto.IngestionFlowFileDTO;
 import it.gov.pagopa.payhub.activities.dto.paymentsreporting.PaymentsReportingDTO;
 import it.gov.pagopa.payhub.activities.dto.paymentsreporting.PaymentsReportingIngestionFlowFileActivityResult;
+import it.gov.pagopa.payhub.activities.enums.IngestionFlowFileType;
 import it.gov.pagopa.payhub.activities.exception.ActivitiesException;
 import it.gov.pagopa.payhub.activities.exception.IngestionFlowFileNotFoundException;
 import it.gov.pagopa.payhub.activities.service.ingestionflow.IngestionFlowFileArchiverService;
@@ -31,8 +32,8 @@
 @Lazy
 @Component
 public class PaymentsReportingIngestionFlowFileActivityImpl implements PaymentsReportingIngestionFlowFileActivity {
-	private final String ingestionflowFileType;
-	private final String archiveDirectory;
+	private final IngestionFlowFileType ingestionflowFileType;
+    private final String archiveDirectory;
 	private final IngestionFlowFileDao ingestionFlowFileDao;
 	private final IngestionFlowFileRetrieverService ingestionFlowFileRetrieverService;
 	private final FlussoRiversamentoUnmarshallerService flussoRiversamentoUnmarshallerService;
@@ -41,16 +42,15 @@ public class PaymentsReportingIngestionFlowFileActivityImpl implements PaymentsR
 	private final PaymentsReportingDao paymentsReportingDao;
 	private final IngestionFlowFileArchiverService ingestionFlowFileArchiverService;
 
-	public PaymentsReportingIngestionFlowFileActivityImpl(@Value("${ingestion-flow-file-type:R}") String ingestionflowFileType,
-	                                                      @Value("${archive-relative-path:processed/}") String archiveRelativePathDirectory,
+	public PaymentsReportingIngestionFlowFileActivityImpl(@Value("${archive-relative-path:processed/}") String archiveRelativePathDirectory,
 	                                                      IngestionFlowFileDao ingestionFlowFileDao,
 	                                                      IngestionFlowFileRetrieverService ingestionFlowFileRetrieverService,
 	                                                      FlussoRiversamentoUnmarshallerService flussoRiversamentoUnmarshallerService,
 	                                                      PaymentsReportingIngestionFlowFileValidatorService paymentsReportingIngestionFlowFileValidatorService,
 	                                                      PaymentsReportingMapperService paymentsReportingMapperService,
 	                                                      PaymentsReportingDao paymentsReportingDao,
 	                                                      IngestionFlowFileArchiverService ingestionFlowFileArchiverService) {
-		this.ingestionflowFileType = ingestionflowFileType;
+        this.ingestionflowFileType = IngestionFlowFileType.PAYMENTS_REPORTING;
 		this.archiveDirectory = archiveRelativePathDirectory;
 		this.ingestionFlowFileDao = ingestionFlowFileDao;
 		this.ingestionFlowFileRetrieverService = ingestionFlowFileRetrieverService;

src/main/java/it/gov/pagopa/payhub/activities/activity/treasury/TreasuryOpiIngestionActivity.java

@@ -14,5 +14,5 @@ public interface TreasuryOpiIngestionActivity {
      * @param ingestionFlowId the unique identifier related to the file to process.
      * @return {@link TreasuryIufResult} containing the list of IUFs and status.
      */
-    TreasuryIufResult processFile(String ingestionFlowId);
+    TreasuryIufResult processFile(Long ingestionFlowId);
 }

src/main/java/it/gov/pagopa/payhub/activities/activity/treasury/TreasuryOpiIngestionActivityImpl.java

@@ -0,0 +1,254 @@
+package it.gov.pagopa.payhub.activities.activity.treasury;
+
+import it.gov.pagopa.payhub.activities.dao.FlussoTesoreriaPIIDao;
+import it.gov.pagopa.payhub.activities.dao.IngestionFlowFileDao;
+import it.gov.pagopa.payhub.activities.dao.TreasuryDao;
+import it.gov.pagopa.payhub.activities.dto.IngestionFlowFileDTO;
+import it.gov.pagopa.payhub.activities.dto.treasury.*;
+import it.gov.pagopa.payhub.activities.enums.IngestionFlowFileType;
+import it.gov.pagopa.payhub.activities.exception.ActivitiesException;
+import it.gov.pagopa.payhub.activities.exception.IngestionFlowFileNotFoundException;
+
+import it.gov.pagopa.payhub.activities.service.ingestionflow.IngestionFlowFileArchiverService;
+import it.gov.pagopa.payhub.activities.service.ingestionflow.IngestionFlowFileRetrieverService;
+
+import it.gov.pagopa.payhub.activities.service.treasury.TreasuryOpi14MapperService;
+import it.gov.pagopa.payhub.activities.service.treasury.TreasuryOpi161MapperService;
+import it.gov.pagopa.payhub.activities.service.treasury.TreasuryUnmarshallerService;
+import it.gov.pagopa.payhub.activities.service.treasury.TreasuryValidatorService;
+import it.gov.pagopa.payhub.activities.util.CsvUtils;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.stereotype.Component;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+
+@Slf4j
+@Lazy
+@Component
+public class TreasuryOpiIngestionActivityImpl implements TreasuryOpiIngestionActivity {
+
+    private final IngestionFlowFileType ingestionflowFileType;
+    private final IngestionFlowFileDao ingestionFlowFileDao;
+    private final String archiveDirectory;
+    private final TreasuryDao treasuryDao;
+    private final FlussoTesoreriaPIIDao flussoTesoreriaPIIDao;
+    private final IngestionFlowFileRetrieverService ingestionFlowFileRetrieverService;
+    private final TreasuryUnmarshallerService treasuryUnmarshallerService;
+    private final TreasuryOpi14MapperService treasuryOpi14MapperService;
+    private final TreasuryOpi161MapperService treasuryOpi161MapperService;
+    private final TreasuryValidatorService treasuryValidatorService;
+    private final IngestionFlowFileArchiverService ingestionFlowFileArchiverService;
+
+    public TreasuryOpiIngestionActivityImpl(
+            IngestionFlowFileDao ingestionFlowFileDao, TreasuryDao treasuryDao, FlussoTesoreriaPIIDao flussoTesoreriaPIIDao,
+            IngestionFlowFileRetrieverService ingestionFlowFileRetrieverService,
+            TreasuryUnmarshallerService treasuryUnmarshallerService,
+            TreasuryOpi14MapperService treasuryOpi14MapperService, TreasuryOpi161MapperService treasuryOpi161MapperService,
+            TreasuryValidatorService treasuryValidatorService, IngestionFlowFileArchiverService ingestionFlowFileArchiverService,
+            String archiveDirectory) {
+        this.archiveDirectory = archiveDirectory;
+        this.ingestionFlowFileArchiverService = ingestionFlowFileArchiverService;
+        this.ingestionflowFileType = IngestionFlowFileType.OPI;
+        this.ingestionFlowFileDao = ingestionFlowFileDao;
+        this.treasuryDao = treasuryDao;
+        this.flussoTesoreriaPIIDao = flussoTesoreriaPIIDao;
+        this.ingestionFlowFileRetrieverService = ingestionFlowFileRetrieverService;
+        this.treasuryUnmarshallerService = treasuryUnmarshallerService;
+        this.treasuryOpi14MapperService = treasuryOpi14MapperService;
+        this.treasuryOpi161MapperService = treasuryOpi161MapperService;
+        this.treasuryValidatorService = treasuryValidatorService;
+    }
+
+
+    @Override
+    public TreasuryIufResult processFile(Long ingestionFlowFileId) {
+        List<String> iufIuvList = new ArrayList<>();
+        List<Path> ingestionFlowFiles = new ArrayList<>();
+        IngestionFlowFileDTO ingestionFlowFileDTO = null;
+        AtomicReference<TreasuryIufResult> treasuryIufResult = new AtomicReference<>();
+
+        try {
+            ingestionFlowFileDTO = findIngestionFlowFileRecord(ingestionFlowFileId);
+
+            ingestionFlowFiles = retrieveFiles(ingestionFlowFileDTO);
+
+
+            if (ingestionFlowFiles != null && !ingestionFlowFiles.isEmpty()) {
+
+                IngestionFlowFileDTO finalIngestionFlowFileDTO = ingestionFlowFileDTO;
+                List<Path> finalIngestionFlowFiles = ingestionFlowFiles;
+                ingestionFlowFiles.forEach(path -> {
+                    File ingestionFlowFile = path.toFile();
+                    log.debug("file from zip archive with name {} loaded successfully ", ingestionFlowFile.getName());
+
+                    treasuryIufResult.set(parseData(ingestionFlowFile, finalIngestionFlowFileDTO, finalIngestionFlowFiles.size()));
+                    try {
+                        archive(finalIngestionFlowFileDTO);
+                    } catch (IOException e) {
+                        throw new RuntimeException(e);
+                    }
+
+
+                });
+            }
+        } catch (Exception e) {
+            log.error("Error during TreasuryOpiIngestionActivity ingestionFlowFileId {}", ingestionFlowFileId, e);
+            return new TreasuryIufResult(Collections.emptyList(), false, e.getMessage());
+        }
+        return treasuryIufResult.get();
+    }
+
+    private IngestionFlowFileDTO findIngestionFlowFileRecord(Long ingestionFlowFileId) {
+        IngestionFlowFileDTO ingestionFlowFileDTO = ingestionFlowFileDao.findById(ingestionFlowFileId)
+                .orElseThrow(() -> new IngestionFlowFileNotFoundException("Cannot found ingestionFlow having id: " + ingestionFlowFileId));
+        if (!ingestionFlowFileDTO.getFlowFileType().equals(ingestionflowFileType)) {
+            throw new IllegalArgumentException("invalid ingestionFlow file type " + ingestionFlowFileDTO.getFlowFileType());
+        }
+        return ingestionFlowFileDTO;
+    }
+
+    private List<Path> retrieveFiles(IngestionFlowFileDTO ingestionFlowFileDTO) throws IOException {
+
+        return ingestionFlowFileRetrieverService
+                .retrieveAndUnzipFile(Path.of(ingestionFlowFileDTO.getFilePath()), ingestionFlowFileDTO.getFileName());
+    }
+
+    private TreasuryIufResult parseData(File ingestionFlowFile, IngestionFlowFileDTO finalIngestionFlowFileDTO, int zipFileSize) {
+        Map<String, List<Pair<TreasuryDTO, FlussoTesoreriaPIIDTO>>> treasuryDtoMap = null;
+        String versione = null;
+        Set<String> iufList = new HashSet<>();
+        boolean success = true;
+
+        it.gov.pagopa.payhub.activities.xsd.treasury.opi14.FlussoGiornaleDiCassa flussoGiornaleDiCassa14 = null;
+        it.gov.pagopa.payhub.activities.xsd.treasury.opi161.FlussoGiornaleDiCassa flussoGiornaleDiCassa161 = null;
+
+        try {
+            flussoGiornaleDiCassa161 = treasuryUnmarshallerService.unmarshalOpi161(ingestionFlowFile);
+            log.debug("file flussoGiornaleDiCassa with Id {} parsed successfully ", flussoGiornaleDiCassa161.getId());
+        } catch (Exception e) {
+            log.error("file flussoGiornaleDiCassa parsing error with opi 1.6.1 format {} ", e.getMessage());
+        }
+        if (flussoGiornaleDiCassa161 == null) {
+            try {
+                flussoGiornaleDiCassa14 = treasuryUnmarshallerService.unmarshalOpi14(ingestionFlowFile);
+                log.debug("file flussoGiornaleDiCassa with Id {} parsed successfully ", flussoGiornaleDiCassa14.getId());
+                versione = TreasuryValidatorService.v14;
+            } catch (Exception e) {
+                log.error("file flussoGiornaleDiCassa parsing error with opi 1.4 format {} ", e.getMessage());
+                success = false;
+            }
+        } else
+            versione = TreasuryValidatorService.v161;
+
+        assert versione != null;
+//        if (!treasuryValidatorService.validatePageSize(flussoGiornaleDiCassa14, flussoGiornaleDiCassa161, zipFileSize, versione)) {
+//          log.error("invalid total page number for ingestionFlowFile with name {}", ingestionFlowFile.getName());
+//          throw new RuntimeException("invalid total page number for ingestionFlowFile with name " + ingestionFlowFile.getName() + " versione " + versione);
+//        }
+
+
+        List<TreasuryErrorDTO> errorDTOList = treasuryValidatorService.validateData(flussoGiornaleDiCassa14, flussoGiornaleDiCassa161, ingestionFlowFile, versione);
+
+        String[] headerArray= new String[]{"FileName","Anno Bolletta", "Codice Bolletta", "Error Code", "Error Message"};
+        List<String[]> header = new ArrayList<>(List.of());
+        header.add(headerArray);
+
+        List<String[]> data = errorDTOList.stream()
+                .map(errorDTO -> new String[] {
+                        errorDTO.getNomeFile(),
+                        errorDTO.getDeAnnoBolletta(),
+                        errorDTO.getCodBolletta(),
+                        errorDTO.getErrorCode(),
+                        errorDTO.getErrorMessage()
+                })
+                .collect(Collectors.toList());
+
+
+        try {
+            CsvUtils.createCsv("ciao", header,data);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+
+
+        treasuryDtoMap = switch (versione) {
+            case TreasuryValidatorService.v14 ->
+                    treasuryOpi14MapperService.apply(flussoGiornaleDiCassa14, finalIngestionFlowFileDTO);
+            case TreasuryValidatorService.v161 ->
+                    treasuryOpi161MapperService.apply(flussoGiornaleDiCassa161, finalIngestionFlowFileDTO);
+            default -> treasuryDtoMap;
+        };
+
+        List<Pair<TreasuryDTO, FlussoTesoreriaPIIDTO>> pairs = treasuryDtoMap.get(StringUtils.firstNonBlank(TreasuryOpi161MapperService.insert, TreasuryOpi14MapperService.insert));
+        pairs.forEach(pair -> {
+            long idFlussoTesoreriaPiiId = flussoTesoreriaPIIDao.insert(pair.getRight());
+            TreasuryDTO treasuryDTO = pair.getLeft();
+            treasuryDTO.setPersonalDataId(idFlussoTesoreriaPiiId);
+            treasuryDao.insert(treasuryDTO);
+            iufList.add(treasuryDTO.getCodIdUnivocoFlusso());
+        });
+
+        return new TreasuryIufResult(iufList.stream().toList(), success, null);
+    }
+
+    /**
+     * Archives the file specified in the given {@link IngestionFlowFileDTO}. The file is moved to
+     * the archive directory located within the same file path.
+     *
+     * @param ingestionFlowFileDTO the DTO containing details of the file to be archived.
+     * @throws IOException if an error occurs during file movement or directory creation.
+     */
+    private void archive(IngestionFlowFileDTO ingestionFlowFileDTO) throws IOException {
+        Path originalFilePath = Paths.get(ingestionFlowFileDTO.getFilePath(), ingestionFlowFileDTO.getFileName());
+        Path targetDirectory = Paths.get(ingestionFlowFileDTO.getFilePath(), archiveDirectory);
+        ingestionFlowFileArchiverService.archive(List.of(originalFilePath), targetDirectory);
+    }
+
+    /**
+     * Archives an error file to a specified target directory.
+     * This method takes an error file and moves it to a target directory for archiving. It constructs
+     * the original file path and the target directory path, then invokes the {@link IngestionFlowFileArchiverService}
+     * to perform the archiving operation.
+     *
+     * @param errorFile the error file to be archived. This file is moved from its original location to the target directory.
+     * @param targetDir the directory where the error file should be archived. The target directory path is constructed relative
+     *                  to the parent directory of the error file.
+     * @throws IOException if an I/O error occurs while archiving the file, such as issues with reading, writing, or accessing file paths.
+     */
+    private void archiveErrorFile(File errorFile, String targetDir) throws IOException {
+        Path originalFilePath = Paths.get(errorFile.getParent(), errorFile.getName());
+        Path targetDirectory = Paths.get(errorFile.getParent(), targetDir);
+        ingestionFlowFileArchiverService.archive(List.of(originalFilePath), targetDirectory);
+    }
+
+
+    /**
+     * Delete the specified file if not null.
+     *
+     * @param file2Delete the file to delete.
+     * @throws IOException if an error occurs during deletion.
+     */
+    private void deletion(File file2Delete) {
+        if (file2Delete != null) {
+            try {
+                Files.delete(file2Delete.toPath());
+            } catch (IOException e) {
+                throw new ActivitiesException("Error occured while delete file: " + file2Delete);
+            }
+        }
+    }
+
+
+}

src/main/java/it/gov/pagopa/payhub/activities/dao/FlussoTesoreriaPIIDao.java

@@ -0,0 +1,7 @@
+package it.gov.pagopa.payhub.activities.dao;
+
+import it.gov.pagopa.payhub.activities.dto.treasury.FlussoTesoreriaPIIDTO;
+
+public interface FlussoTesoreriaPIIDao {
+  Long insert(FlussoTesoreriaPIIDTO flussoTesoreriaPIIDTO);
+}