feat: add endpoints necessary for automized processing

loichenninger · loichenninger · commit 00d942a17e0c · 2025-12-05T15:07:27.000+01:00
diff --git a/src/main/java/fr/insee/genesis/controller/rest/responses/RawResponseController.java b/src/main/java/fr/insee/genesis/controller/rest/responses/RawResponseController.java
@@ -191,7 +191,7 @@ public ResponseEntity<String> saveRawResponsesFromRawResponseDto(
     }
 
     //PROCESS
-    @Operation(summary = "Process raw data of a campaign")
+    @Operation(summary = "Process raw data for a list of interrogations")
     @PostMapping(path = "/raw-responses/process")
     @PreAuthorize("hasRole('SCHEDULER')")
     public ResponseEntity<String> processRawResponses(
@@ -202,9 +202,8 @@ public ResponseEntity<String> processRawResponses(
             @RequestParam("collectionInstrumentId") String collectionInstrumentId,
             @RequestBody List<String> interrogationIdList
     ) {
-        log.info("Try to process raw responses for questionnaireId {} and {} interrogationIds", collectionInstrumentId, interrogationIdList.size());
+        log.info("Try to process raw responses for collectionInstrumentId {} and {} interrogationIds", collectionInstrumentId, interrogationIdList.size());
         List<GenesisError> errors = new ArrayList<>();
-
         try {
             DataProcessResult result = rawResponseApiPort.processRawResponses(collectionInstrumentId, interrogationIdList, errors);
             return result.formattedDataCount() == 0 ?
@@ -215,6 +214,29 @@ public ResponseEntity<String> processRawResponses(
             return ResponseEntity.status(e.getStatus()).body(e.getMessage());
         }
     }
+
+    @Operation(summary = "Process raw data for all data of an collection instrument")
+    @PostMapping(path = "/raw-responses/{collectionInstrumentId}/process")
+    @PreAuthorize("hasRole('SCHEDULER')")
+    public ResponseEntity<String> processRawResponsesByCollectionInstrumentId(
+            @Parameter(
+                    description = "Id of the collection instrument (old questionnaireId)",
+                    example = "ENQTEST2025X00"
+            )
+            @PathVariable("collectionInstrumentId") String collectionInstrumentId
+    ) {
+        log.info("Try to process raw responses for collectionInstrumentId {}", collectionInstrumentId);
+        try {
+            DataProcessResult result = rawResponseApiPort.processRawResponses(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 the list of collection instruments containing unprocessed interrogations")
     @GetMapping(path = "/raw-responses/unprocessed/collection-intrument-ids")
     @PreAuthorize("hasRole('SCHEDULER')")
@@ -269,7 +291,7 @@ public ResponseEntity<String> processJsonRawData(
     }
 
     @Operation(summary = "Process raw data of a questionnaire")
-    @PostMapping(path = "/{collectionInstrumentId}/process")
+    @PostMapping(path = "/responses/raw/lunatic-json/{collectionInstrumentId}/process")
     @PreAuthorize("hasRole('SCHEDULER')")
     public ResponseEntity<String> processJsonRawData(
             @PathVariable String collectionInstrumentId
diff --git a/src/main/java/fr/insee/genesis/domain/ports/api/RawResponseApiPort.java b/src/main/java/fr/insee/genesis/domain/ports/api/RawResponseApiPort.java
@@ -14,7 +14,9 @@ public interface RawResponseApiPort {
 
     List<RawResponse> getRawResponses(String collectionInstrumentId, Mode mode, List<String> interrogationIdList);
     DataProcessResult processRawResponses(String collectionInstrumentId, List<String> interrogationIdList, List<GenesisError> errors) throws GenesisException;
+    DataProcessResult processRawResponses(String collectionInstrumentId) throws GenesisException;
     List<SurveyUnitModel> convertRawResponse(List<RawResponse> rawResponses, VariablesMap variablesMap);
     List<String> getUnprocessedCollectionInstrumentIds();
     void updateProcessDates(List<SurveyUnitModel> surveyUnitModels);
+
 }
diff --git a/src/main/java/fr/insee/genesis/domain/ports/spi/RawResponsePersistencePort.java b/src/main/java/fr/insee/genesis/domain/ports/spi/RawResponsePersistencePort.java
@@ -11,4 +11,5 @@ public interface RawResponsePersistencePort {
     List<RawResponse> findRawResponses(String collectionInstrumentId, Mode mode, List<String> interrogationIdList);
     void updateProcessDates(String collectionInstrumentId, Set<String> interrogationIds);
     List<String> getUnprocessedCollectionIds();
-}
+    Set<String> findUnprocessedInterrogationIdsByCollectionInstrumentId(String collectionInstrumentId);
+ }
diff --git a/src/main/java/fr/insee/genesis/domain/service/rawdata/RawResponseService.java b/src/main/java/fr/insee/genesis/domain/service/rawdata/RawResponseService.java
@@ -79,14 +79,7 @@ public DataProcessResult processRawResponses(String collectionInstrumentId, List
         List<Mode> modesList = controllerUtils.getModesList(collectionInstrumentId, null);
         for (Mode mode : modesList) {
             //Load and save metadata into database, throw exception if none
-            VariablesMap variablesMap = metadataService.loadAndSaveIfNotExists(collectionInstrumentId, collectionInstrumentId, 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(collectionInstrumentId,mode,errors);
             int totalBatchs = Math.ceilDiv(interrogationIdList.size() , config.getRawDataProcessingBatchSize());
             int batchNumber = 1;
             List<String> interrogationIdListForMode = new ArrayList<>(interrogationIdList);
@@ -127,7 +120,84 @@ public DataProcessResult processRawResponses(String collectionInstrumentId, List
                 batchNumber++;
             }
         }
-        return new DataProcessResult(dataCount, formattedDataCount, List.of());
+        return new DataProcessResult(dataCount, formattedDataCount, errors);
+    }
+
+    @Override
+    public DataProcessResult processRawResponses(String collectionInstrumentId) throws GenesisException {
+        int dataCount=0;
+        int formattedDataCount=0;
+        DataProcessingContextModel dataProcessingContext =
+                dataProcessingContextService.getContextByCollectionInstrumentId(collectionInstrumentId);
+        List<GenesisError> errors = new ArrayList<>();
+
+        List<Mode> modesList = controllerUtils.getModesList(collectionInstrumentId, null);
+        for (Mode mode : modesList) {
+            //Load and save metadata into database, throw exception if none
+            VariablesMap variablesMap = getVariablesMap(collectionInstrumentId,mode,errors);
+            Set<String> interrogationIds =
+                    rawResponsePersistencePort.findUnprocessedInterrogationIdsByCollectionInstrumentId(collectionInstrumentId);
+
+            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(
+                        collectionInstrumentId,
+                        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 List<SurveyUnitModel> getConvertedSurveyUnits(String collectionInstrumentId, Mode mode, List<String> interrogationIdListForMode, int maxIndex, VariablesMap variablesMap) {
+        List<String> interrogationIdToProcess = interrogationIdListForMode.subList(0, maxIndex);
+        List<RawResponse> rawResponses = getRawResponses(collectionInstrumentId, mode, interrogationIdToProcess);
+        return convertRawResponse(
+                rawResponses,
+                variablesMap
+        );
+    }
+
+    private VariablesMap getVariablesMap(String collectionInstrumentId, Mode mode, List<GenesisError> errors) throws GenesisException {
+        VariablesMap variablesMap = metadataService.loadAndSaveIfNotExists(collectionInstrumentId, collectionInstrumentId, 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;
     }
 
     @Override
@@ -179,7 +249,6 @@ public List<SurveyUnitModel> convertRawResponse(List<RawResponse> rawResponses,
             }
         }
         return surveyUnitModels;
-        //return List.of();
     }
 
     @Override
diff --git a/src/main/java/fr/insee/genesis/infrastructure/adapter/RawResponseMongoAdapter.java b/src/main/java/fr/insee/genesis/infrastructure/adapter/RawResponseMongoAdapter.java
@@ -16,6 +16,7 @@
 import org.springframework.stereotype.Service;
 
 import java.time.LocalDateTime;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 
@@ -51,4 +52,10 @@ public void updateProcessDates(String collectionInstrumentId, Set<String> interr
     public List<String> getUnprocessedCollectionIds() {
         return repository.findDistinctCollectionInstrumentIdByProcessDateIsNull();
     }
+
+    @Override
+    public Set<String> findUnprocessedInterrogationIdsByCollectionInstrumentId(String collectionInstrumentId) {
+        // We remove duplicate ids
+        return new HashSet<>(repository.findInterrogationIdByCollectionInstrumentId(collectionInstrumentId));
+    }
 }
diff --git a/src/main/java/fr/insee/genesis/infrastructure/repository/RawResponseRepository.java b/src/main/java/fr/insee/genesis/infrastructure/repository/RawResponseRepository.java
@@ -19,4 +19,11 @@ public interface RawResponseRepository extends MongoRepository<RawResponseDocume
             "{ $project: { _id: 0, collectionInstrumentId: '$_id' } }"
     })
     List<String> findDistinctCollectionInstrumentIdByProcessDateIsNull();
+
+    @Aggregation(pipeline = {
+            "{ $match: { collectionInstrumentId: ?0 } }",
+            "{ $project: { _id: 0, interrogationId: '$interrogationId' } }"
+    })
+    List<String> findInterrogationIdByCollectionInstrumentId(String collectionInstrumentId);
+
 }
diff --git a/src/test/java/cucumber/functional_tests/RawDataDefinitions.java b/src/test/java/cucumber/functional_tests/RawDataDefinitions.java
@@ -115,6 +115,11 @@ public DataProcessResult processRawResponses(String questionnaireId, List<String
             return null;
         }
 
+        @Override
+        public DataProcessResult processRawResponses(String collectionInstrumentId) throws GenesisException {
+            return null;
+        }
+
         @Override
         public List<SurveyUnitModel> convertRawResponse(List<RawResponse> rawResponses, VariablesMap variablesMap) {
             return List.of();
diff --git a/src/test/java/fr/insee/genesis/controller/rest/responses/RawResponseControllerTest.java b/src/test/java/fr/insee/genesis/controller/rest/responses/RawResponseControllerTest.java
@@ -88,6 +88,11 @@ public DataProcessResult processRawResponses(String questionnaireId, List<String
             return null;
         }
 
+        @Override
+        public DataProcessResult processRawResponses(String collectionInstrumentId) throws GenesisException {
+            return null;
+        }
+
         @Override
         public List<SurveyUnitModel> convertRawResponse(List<RawResponse> rawResponses, VariablesMap variablesMap) {
             return List.of();

Original file line number	Diff line number	Diff line change
`@@ -14,7 +14,9 @@ public interface RawResponseApiPort {`
`14`	`14`
`15`	`15`	`List<RawResponse> getRawResponses(String collectionInstrumentId, Mode mode, List<String> interrogationIdList);`
`16`	`16`	`DataProcessResult processRawResponses(String collectionInstrumentId, List<String> interrogationIdList, List<GenesisError> errors) throws GenesisException;`
	`17`	`+ DataProcessResult processRawResponses(String collectionInstrumentId) throws GenesisException;`
`17`	`18`	`List<SurveyUnitModel> convertRawResponse(List<RawResponse> rawResponses, VariablesMap variablesMap);`
`18`	`19`	`List<String> getUnprocessedCollectionInstrumentIds();`
`19`	`20`	`void updateProcessDates(List<SurveyUnitModel> surveyUnitModels);`
	`21`	`+`
`20`	`22`	`}`