chore: update branch from main

Author: loichenninger

CHANGELOG.md

@@ -1,4 +1,8 @@
 # Changelog
+## 1.13.0 [2025-12-04]
+### Changed
+- New raw data process endpoint
+
 ## 1.12.2 [2025-11-06]
 ### Fixed
 - Scientific notation for Doubles during raw data processing

pom.xml

@@ -4,7 +4,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>fr.insee.genesis</groupId>
     <artifactId>genesis-api</artifactId>
-    <version>1.12.4</version>
+    <version>1.13.0</version>
     <packaging>jar</packaging>
     <name>genesis-api</name>
 
@@ -36,7 +36,7 @@
         <pitest.version>1.22.0</pitest.version>
         <pitest.junit.version>1.2.3</pitest.junit.version>
         <jackson.version>2.19.0</jackson.version>
-        <bpm.version>1.0.19</bpm.version>
+        <bpm.version>1.0.20</bpm.version>
     </properties>
     <dependencies>
         <dependency>

src/main/java/fr/insee/genesis/controller/rest/responses/RawResponseController.java

@@ -241,6 +241,7 @@ public ResponseEntity<LunaticJsonRawDataModel> getJsonRawData(
     @Operation(summary = "Process raw data of a campaign")
     @PostMapping(path = "/responses/raw/lunatic-json/process")
     @PreAuthorize("hasRole('SCHEDULER')")
+    @Deprecated(since = "1.13.0")
     public ResponseEntity<String> processJsonRawData(
             @RequestParam("campaignName") String campaignName,
             @RequestParam("questionnaireId") String questionnaireId,
@@ -260,6 +261,24 @@ public ResponseEntity<String> processJsonRawData(
         }
     }
 
+    @Operation(summary = "Process raw data of a questionnaire")
+    @PostMapping(path = "/{collectionInstrumentId}/process")
+    @PreAuthorize("hasRole('SCHEDULER')")
+    public ResponseEntity<String> processJsonRawData(
+            @PathVariable String collectionInstrumentId
+    ) {
+        log.info("Try to process raw JSON datas for questionnaire {}",collectionInstrumentId);
+        try {
+            DataProcessResult result = lunaticJsonRawDataApiPort.processRawData(collectionInstrumentId);
+            return result.formattedDataCount() == 0 ?
+                    ResponseEntity.ok("%d document(s) processed".formatted(result.dataCount()))
+                    : ResponseEntity.ok("%d document(s) processed, including %d FORMATTED after data verification"
+                    .formatted(result.dataCount(), result.formattedDataCount()));
+        } catch (GenesisException e) {
+            return ResponseEntity.status(e.getStatus()).body(e.getMessage());
+        }
+    }
+
     @Operation(summary = "Get processed data ids from last n hours (default 24h)")
     @GetMapping(path = "/responses/raw/lunatic-json/processed/ids")
     @PreAuthorize("hasRole('ADMIN')")

src/main/java/fr/insee/genesis/controller/rest/responses/ResponseController.java

@@ -503,7 +503,7 @@ private void processCampaignWithMode(String campaignName, Mode mode, String root
         }
 
         //Create context if not exist
-        if(contextService.getContextByPartitionId(campaignName) == null){
+        if(contextService.getContextByCollectionInstrumentId(campaignName) == null){
             contextService.saveContext(campaignName, false);
         }
 

src/main/java/fr/insee/genesis/domain/model/surveyunit/rawdata/DataProcessResult.java

@@ -1,4 +1,8 @@
 package fr.insee.genesis.domain.model.surveyunit.rawdata;
 
-public record DataProcessResult(int dataCount, int formattedDataCount) {
+import fr.insee.genesis.exceptions.GenesisError;
+
+import java.util.List;
+
+public record DataProcessResult(int dataCount, int formattedDataCount, List<GenesisError> errors) {
 }

src/main/java/fr/insee/genesis/domain/ports/api/DataProcessingContextApiPort.java

@@ -31,7 +31,6 @@ void saveKraftwerkExecutionSchedule(String partitionId,
     long countSchedules();
 
     DataProcessingContextModel getContext(String interrogationId) throws GenesisException;
-    DataProcessingContextModel getContextByPartitionId(String partitionId) throws GenesisException;
 
     DataProcessingContextModel getContextByCollectionInstrumentId(String collectionInstrumentId);
 

src/main/java/fr/insee/genesis/domain/ports/api/LunaticJsonRawDataApiPort.java

@@ -28,6 +28,10 @@ public interface LunaticJsonRawDataApiPort {
     long countResponsesByQuestionnaireId(String campaignId);
     Page<LunaticJsonRawDataModel> findRawDataByCampaignIdAndDate(String campaignId, Instant  startDt, Instant endDt, Pageable pageable);
 
+    @Deprecated(since = "1.13.0")
     DataProcessResult processRawData(String campaignName, List<String> interrogationIdList, List<GenesisError> errors) throws GenesisException;
+
+    DataProcessResult processRawData(String collectionInstrumentId) throws GenesisException;
+
     Map<String, List<String>> findProcessedIdsgroupedByQuestionnaireSince(LocalDateTime since);
 }

src/main/java/fr/insee/genesis/domain/ports/spi/LunaticJsonRawDataPersistencePort.java

@@ -22,4 +22,5 @@ public interface LunaticJsonRawDataPersistencePort {
     long countResponsesByQuestionnaireId(String questionnaireId);
     List<GroupedInterrogation> findProcessedIdsGroupedByQuestionnaireSince(LocalDateTime since);
     List<GroupedInterrogation> findUnprocessedIds();
+    Set<String> findUnprocessedInterrogationIdsByCollectionInstrumentId(String collectionInstrumentId);
 }

src/main/java/fr/insee/genesis/domain/service/context/DataProcessingContextService.java

@@ -167,13 +167,6 @@ public DataProcessingContextModel getContext(String interrogationId) throws Gene
         );
     }
 
-    @Override
-    public DataProcessingContextModel getContextByPartitionId(String partitionId){
-        return DataProcessingContextMapper.INSTANCE.documentToModel(
-                dataProcessingContextPersistancePort.findByPartitionId(partitionId)
-        );
-    }
-
     @Override
     public DataProcessingContextModel getContextByCollectionInstrumentId(String collectionInstrumentId){
         return DataProcessingContextMapper.INSTANCE.documentToModel(

src/main/java/fr/insee/genesis/domain/service/rawdata/LunaticJsonRawDataService.java

@@ -100,22 +100,16 @@ public List<LunaticJsonRawDataModel> getRawData(String campaignName, Mode mode,
     }
 
     @Override
+    @Deprecated(since = "1.13.0")
     public DataProcessResult processRawData(String campaignName, List<String> interrogationIdList, List<GenesisError> errors) throws GenesisException {
         int dataCount=0;
         int formattedDataCount=0;
         DataProcessingContextModel dataProcessingContext =
-                dataProcessingContextService.getContextByPartitionId(campaignName);
+                dataProcessingContextService.getContextByCollectionInstrumentId(campaignName);
         List<Mode> modesList = controllerUtils.getModesList(campaignName, null);
         for (Mode mode : modesList) {
             //Load and save metadata into database, throw exception if none
-            VariablesMap variablesMap = metadataService.loadAndSaveIfNotExists(campaignName, campaignName, mode, fileUtils,
-                    errors).getVariables();
-            if (variablesMap == null) {
-                throw new GenesisException(400,
-                        "Error during metadata parsing for mode %s :%n%s"
-                                .formatted(mode, errors.getLast().getMessage())
-                );
-            }
+            VariablesMap variablesMap = getVariablesMap(campaignName, mode, errors);
             int totalBatchs = Math.ceilDiv(interrogationIdList.size() , config.getRawDataProcessingBatchSize());
             int batchNumber = 1;
             List<String> interrogationIdListForMode = new ArrayList<>(interrogationIdList);
@@ -156,7 +150,85 @@ public DataProcessResult processRawData(String campaignName, List<String> interr
                 batchNumber++;
             }
         }
-        return new DataProcessResult(dataCount, formattedDataCount);
+        return new DataProcessResult(dataCount, formattedDataCount, errors);
+    }
+
+    @Override
+    public DataProcessResult processRawData(String questionnaireId) throws GenesisException {
+        int dataCount=0;
+        int formattedDataCount=0;
+        DataProcessingContextModel dataProcessingContext =
+                dataProcessingContextService.getContextByCollectionInstrumentId(questionnaireId);
+        List<GenesisError> errors = new ArrayList<>();
+
+        List<Mode> modesList = controllerUtils.getModesList(questionnaireId, null);
+        for (Mode mode : modesList) {
+            //Load and save metadata into database, throw exception if none
+            VariablesMap variablesMap = getVariablesMap(questionnaireId, mode, errors);
+            Set<String> interrogationIds =
+                    lunaticJsonRawDataPersistencePort.findUnprocessedInterrogationIdsByCollectionInstrumentId(questionnaireId);
+
+            int totalBatchs = Math.ceilDiv(interrogationIds.size() , config.getRawDataProcessingBatchSize());
+            int batchNumber = 1;
+            List<String> interrogationIdListForMode = new ArrayList<>(interrogationIds);
+            while(!interrogationIdListForMode.isEmpty()){
+                log.info("Processing raw data batch {}/{}", batchNumber, totalBatchs);
+
+                int maxIndex = Math.min(interrogationIdListForMode.size(), config.getRawDataProcessingBatchSize());
+                List<SurveyUnitModel> surveyUnitModels = getConvertedSurveyUnits(
+                        questionnaireId,
+                        mode,
+                        interrogationIdListForMode,
+                        maxIndex,
+                        variablesMap
+                );
+
+                //Save converted data
+                surveyUnitQualityService.verifySurveyUnits(surveyUnitModels, variablesMap);
+                surveyUnitService.saveSurveyUnits(surveyUnitModels);
+
+                //Update process dates
+                updateProcessDates(surveyUnitModels);
+
+                //Increment data count
+                dataCount += surveyUnitModels.size();
+                formattedDataCount += surveyUnitModels.stream()
+                        .filter(surveyUnitModel -> surveyUnitModel.getState().equals(DataState.FORMATTED))
+                        .toList()
+                        .size();
+
+                //Send processed ids grouped by questionnaire (if review activated)
+                if(dataProcessingContext != null && dataProcessingContext.isWithReview()) {
+                    sendProcessedIdsToQualityTool(surveyUnitModels);
+                }
+
+                //Remove processed ids from list
+                interrogationIdListForMode = interrogationIdListForMode.subList(maxIndex, interrogationIdListForMode.size());
+                batchNumber++;
+            }
+        }
+        return new DataProcessResult(dataCount, formattedDataCount, errors);
+    }
+
+    private VariablesMap getVariablesMap(String questionnaireId, Mode mode, List<GenesisError> errors) throws GenesisException {
+        VariablesMap variablesMap = metadataService.loadAndSaveIfNotExists(questionnaireId, questionnaireId, mode, fileUtils,
+                errors).getVariables();
+        if (variablesMap == null) {
+            throw new GenesisException(400,
+                    "Error during metadata parsing for mode %s :%n%s"
+                            .formatted(mode, errors.getLast().getMessage())
+            );
+        }
+        return variablesMap;
+    }
+
+    private List<SurveyUnitModel> getConvertedSurveyUnits(String questionnaireId, Mode mode, List<String> interrogationIdListForMode, int maxIndex, VariablesMap variablesMap) {
+        List<String> interrogationIdToProcess = interrogationIdListForMode.subList(0, maxIndex);
+        List<LunaticJsonRawDataModel> rawData = getRawData(questionnaireId, mode, interrogationIdToProcess);
+        return convertRawData(
+                rawData,
+                variablesMap
+        );
     }
 
     private void sendProcessedIdsToQualityTool(List<SurveyUnitModel> surveyUnitModels) {