feat(ingest, backend): improve curation notification by adding diff to message

Author: anna-parker

backend/src/main/kotlin/org/loculus/backend/controller/SubmissionController.kt

@@ -417,6 +417,10 @@ open class SubmissionController(
         @Parameter(
             description = "The metadata fields that should be returned. If not provided, all fields are returned.",
         ) @RequestParam(required = false) fields: List<String>?,
+        @Parameter(
+            description = "Filter by accession versions in 'accession.version' format. " +
+                "If not provided, all accession versions are considered.",
+        ) @RequestParam(required = false) accessionVersionsFilter: List<String>?,
         @Parameter(
             description = "Filter by group ids. If not provided, all groups are considered.",
         ) @RequestParam(required = false) groupIdsFilter: List<Int>?,
@@ -432,11 +436,22 @@ open class SubmissionController(
             headers.add(HttpHeaders.CONTENT_ENCODING, compression.compressionName)
         }
 
+        val parsedAccessionVersions = accessionVersionsFilter?.takeIf { it.isNotEmpty() }?.map {
+            val lastDot = it.lastIndexOf('.')
+            if (lastDot <= 0 || lastDot == it.length - 1) {
+                throw BadRequestException("Invalid accession version format '$it', expected 'accession.version'")
+            }
+            val version = it.substring(lastDot + 1).toLongOrNull()
+                ?: throw BadRequestException("Invalid version in accession version '$it', expected a number")
+            AccessionVersion(it.substring(0, lastDot), version)
+        }
+
         val totalRecords = submissionDatabaseService.countUnprocessedMetadata(
             authenticatedUser,
             organism,
             groupIdsFilter?.takeIf { it.isNotEmpty() },
             statusesFilter?.takeIf { it.isNotEmpty() },
+            parsedAccessionVersions,
         )
         headers.add(X_TOTAL_RECORDS, totalRecords.toString())
         // TODO(https://github.com/loculus-project/loculus/issues/2778)
@@ -452,6 +467,7 @@ open class SubmissionController(
                 groupIdsFilter?.takeIf { it.isNotEmpty() },
                 statusesFilter?.takeIf { it.isNotEmpty() },
                 fields?.takeIf { it.isNotEmpty() },
+                parsedAccessionVersions,
             )
         }
 

backend/src/main/kotlin/org/loculus/backend/service/submission/SubmissionDatabaseService.kt

@@ -1228,6 +1228,7 @@ class SubmissionDatabaseService(
         organism: Organism,
         groupIdsFilter: List<Int>?,
         statusesFilter: List<Status>?,
+        accessionVersionsFilter: List<AccessionVersion>?,
     ): Op<Boolean> {
         val organismCondition = SequenceEntriesView.organismIs(organism)
         val groupCondition = getGroupCondition(groupIdsFilter, authenticatedUser)
@@ -1236,7 +1237,12 @@ class SubmissionDatabaseService(
         } else {
             Op.TRUE
         }
-        val conditions = organismCondition and groupCondition and statusCondition
+        val accessionVersionCondition = if (accessionVersionsFilter != null) {
+            SequenceEntriesView.accessionVersionIsIn(accessionVersionsFilter)
+        } else {
+            Op.TRUE
+        }
+        val conditions = organismCondition and groupCondition and statusCondition and accessionVersionCondition
 
         return conditions
     }
@@ -1246,6 +1252,7 @@ class SubmissionDatabaseService(
         organism: Organism,
         groupIdsFilter: List<Int>?,
         statusesFilter: List<Status>?,
+        accessionVersionsFilter: List<AccessionVersion>?,
     ): Long = SequenceEntriesView
         .selectAll()
         .where(
@@ -1254,6 +1261,7 @@ class SubmissionDatabaseService(
                 organism,
                 groupIdsFilter,
                 statusesFilter,
+                accessionVersionsFilter,
             ),
         )
         .count()
@@ -1264,6 +1272,7 @@ class SubmissionDatabaseService(
         groupIdsFilter: List<Int>?,
         statusesFilter: List<Status>?,
         fields: List<String>?,
+        accessionVersionsFilter: List<AccessionVersion>?,
     ): Sequence<AccessionVersionUnprocessedMetadata> {
         val unprocessedMetadata = SequenceEntriesView.unprocessedDataColumn
             // It's actually <Map<String, String>?> but exposed does not support nullable types here
@@ -1284,6 +1293,7 @@ class SubmissionDatabaseService(
                     organism,
                     groupIdsFilter,
                     statusesFilter,
+                    accessionVersionsFilter,
                 ),
             )
             .fetchSize(streamBatchSize)

backend/src/test/kotlin/org/loculus/backend/controller/submission/GetUnprocessedMetadataEndpointTest.kt

@@ -190,6 +190,29 @@ class GetUnprocessedMetadataEndpointTest(
             .andExpect(status().isOk)
     }
 
+    @Test
+    fun `WHEN I filter by accessionVersions THEN should return only those entries`() {
+        val entries = convenienceClient.prepareDefaultSequenceEntriesToApprovedForRelease()
+        val target = entries.first()
+        val accessionVersion = target.displayAccessionVersion()
+
+        val response = submissionControllerClient.getSubmittedMetadata(
+            accessionVersionsFilter = listOf(accessionVersion),
+        )
+        response.andExpect(status().isOk)
+            .andExpect(header().string("x-total-records", `is`("1")))
+        val responseBody = response.expectNdjsonAndGetContent<AccessionVersionSubmittedMetadata>()
+        assertThat(responseBody, hasSize(1))
+        assertThat(responseBody[0].displayAccessionVersion(), `is`(accessionVersion))
+    }
+
+    @Test
+    fun `WHEN I filter by accessionVersions with invalid format THEN returns 400`() {
+        submissionControllerClient.getSubmittedMetadata(
+            accessionVersionsFilter = listOf("not-a-valid-accession-version"),
+        ).andExpect(status().isBadRequest)
+    }
+
     // Regression test for https://github.com/loculus-project/loculus/issues/4036
     @Test
     fun `GIVEN revoked sequences exist THEN endpoint does not throw exception`() {

backend/src/test/kotlin/org/loculus/backend/controller/submission/SubmissionControllerClient.kt

@@ -307,6 +307,7 @@ class SubmissionControllerClient(private val mockMvc: MockMvc, private val objec
         jwt: String? = jwtForDefaultUser,
         groupIdsFilter: List<Int>? = null,
         statusesFilter: List<Status>? = null,
+        accessionVersionsFilter: List<String>? = null,
         fields: List<String>? = null,
         compression: String? = null,
     ): ResultActions = mockMvc.perform(
@@ -320,6 +321,7 @@ class SubmissionControllerClient(private val mockMvc: MockMvc, private val objec
             }
             .param("groupIdsFilter", groupIdsFilter?.joinToString(",") { it.toString() })
             .param("statusesFilter", statusesFilter?.joinToString(",") { it.name })
+            .param("accessionVersionsFilter", accessionVersionsFilter?.joinToString(","))
             .param("fields", fields?.joinToString(",")),
     )
 

ingest/scripts/call_loculus.py

@@ -106,7 +106,15 @@ def record_factory(*args, **kwargs):
 
     if mode == "get-submitted":
         logger.info("Getting submitted sequences")
-        get_submitted(config, output)
+        if config.segmented:
+            insdc_key = [
+                "insdcAccessionBase" + "_" + segment for segment in config.nucleotide_sequences
+            ]
+        else:
+            insdc_key = ["insdcAccessionBase"]
+
+        fields = ["hash", *insdc_key]
+        get_submitted(config, output, fields)
 
 
 if __name__ == "__main__":

ingest/scripts/compare_hashes.py

@@ -9,6 +9,7 @@
 import orjsonl
 import requests
 import yaml
+from loculus_client import get_submitted
 
 logger = logging.getLogger(__name__)
 logging.basicConfig(
@@ -22,8 +23,10 @@
 @dataclass
 class Config:
     organism: str
-    segmented: str
+    segmented: bool
     nucleotide_sequences: list[str]
+    group_name: str
+    batch_chunk_size: int
     slack_hook: str = ""
 
 
@@ -86,16 +89,63 @@ def sample_out_hashed_records(
     return not keep
 
 
+def calculate_metadata_diff(
+    config: Config, new_metadata: dict[str, Any], previous_entry: LatestLoculusVersion
+) -> dict[str, Any]:
+    previous_metadata_list = get_submitted(
+        config,  # type: ignore
+        None,
+        fields=None,
+        accessionVersionsFilter=[
+            f"{previous_entry.loculus_accession}.{previous_entry.latest_version}"
+        ],
+    )
+    if not previous_metadata_list or len(previous_metadata_list) != 1:
+        logger.warning(
+            f"Could not retrieve previous metadata for {previous_entry.loculus_accession} "
+            f"version {previous_entry.latest_version} to calculate metadata diff"
+        )
+        return {}
+    previous_metadata = previous_metadata_list[0]
+
+    diff = {}
+
+    new_keys = set(new_metadata.keys())
+    old_keys = set(previous_metadata.keys())
+
+    for key in new_keys | old_keys:
+        key_in_new = key in new_metadata
+        key_in_old = key in previous_metadata
+
+        if not key_in_new:
+            print(f"Key only exists in previous metadata: {key}")
+            diff[key] = {"new": None, "old": previous_metadata[key]}
+
+        elif not key_in_old:
+            print(f"Key only exists in new metadata: {key}")
+            diff[key] = {"new": new_metadata[key], "old": None}
+
+        else:
+            new_value = new_metadata[key]
+            old_value = previous_metadata[key]
+
+            if new_value != old_value:
+                diff[key] = {"new": new_value, "old": old_value}
+
+    return diff
+
+
 def process_hashes(
     ingested_insdc_accession: InsdcAccession,
     metadata_id: SubmissionId,
-    newly_ingested_hash: float | None,
+    new_metadata: dict[str, Any],
     submitted: dict[InsdcAccession, LatestLoculusVersion],
     update_manager: SequenceUpdateManager,
 ):
     """
     Decide if metadata_id should be submitted, revised, or noop
     """
+    newly_ingested_hash: float | None = new_metadata.get("hash")
 
     if ingested_insdc_accession not in submitted:
         update_manager.submit.append(metadata_id)
@@ -114,12 +164,15 @@ def process_hashes(
         return update_manager
 
     if previously_submitted_entry.curated:
+        metadata_diff = calculate_metadata_diff(
+            update_manager.config, new_metadata, previously_submitted_entry
+        )
         # Sequence has been curated before - special case
         notification = (
             f"Ingest: Sequence {corresponding_loculus_accession} with INSDC "
             f"accession {ingested_insdc_accession} has been curated before "
             f"- do not know how to proceed. New hash: {newly_ingested_hash}, "
-            f"old hash: {previously_submitted_entry.hash}."
+            f"old hash: {previously_submitted_entry.hash}. Metadata diff: {metadata_diff}"
         )
         logger.warning(notification)
         notify(update_manager.config, notification)
@@ -328,7 +381,6 @@ def main(
     for field in orjsonl.stream(metadata):
         metadata_id: SubmissionId = field["id"]
         record: dict[str, Any] = field["metadata"]
-        ingested_hash: float | None = record.get("hash")
         if not config.segmented:
             insdc_accession_base = record["insdcAccessionBase"]
             if not insdc_accession_base:
@@ -337,9 +389,7 @@ def main(
             if sample_out_hashed_records(insdc_accession_base, subsample_fraction):
                 update_manager.sampled_out.append(insdc_accession_base)
                 continue
-            process_hashes(
-                insdc_accession_base, metadata_id, ingested_hash, submitted, update_manager
-            )
+            process_hashes(insdc_accession_base, metadata_id, record, submitted, update_manager)
             current_ingested_accessions.add(insdc_accession_base)
             continue
 
@@ -366,7 +416,7 @@ def main(
         ):
             # grouping is the same, can just look at first segment in group
             accession = insdc_accession_base_list[0]
-            process_hashes(accession, metadata_id, ingested_hash, submitted, update_manager)
+            process_hashes(accession, metadata_id, record, submitted, update_manager)
             continue
         # old group is subset of new group, new group has new segments
         old_submitted = [
@@ -381,7 +431,7 @@ def main(
         ):
             # has a new segment, must be revised
             accession = old_submitted[0]
-            process_hashes(accession, metadata_id, ingested_hash, submitted, update_manager)
+            process_hashes(accession, metadata_id, record, submitted, update_manager)
             continue
         old_accessions: dict[LoculusAccession, JointInsdcAccession] = {
             submitted[a].loculus_accession: submitted[a].jointAccession

ingest/scripts/loculus_client.py

@@ -447,28 +447,22 @@ def get_sequence_status(config: Config):
     return result
 
 
-def get_submitted(config: Config, output: str):
+def get_submitted(config: Config, output: str | None, fields: list[str] | None = None, accessionVersionsFilter: list[str] | None = None):
     """Get previously submitted sequences as ndjson
     This way we can avoid submitting the same sequences again
     Adds status to the output (as this is not returned by get-unprocessed-metadata)
     """
 
     url = f"{organism_url(config)}/get-unprocessed-metadata"
 
-    if config.segmented:
-        insdc_key = [
-            "insdcAccessionBase" + "_" + segment for segment in config.nucleotide_sequences
-        ]
-    else:
-        insdc_key = ["insdcAccessionBase"]
-
-    fields = ["hash", *insdc_key]
-
     params = {
-        "fields": fields,
         "groupIdsFilter": [],
         "statusesFilter": [],
     }
+    if fields:
+        params["fields"] = fields
+    if accessionVersionsFilter:
+        params["accessionVersionsFilter"] = accessionVersionsFilter
 
     while True:
         logger.info("Getting previously submitted sequences")
@@ -499,6 +493,9 @@ def get_submitted(config: Config, output: str):
     statuses: dict[str, dict[int, str]] = get_sequence_status(config)
     logger.info(f"Got info on {len(statuses.keys())} previously submitted sequences/accessions")
 
+    if not output:
+        return entries
+
     for entry in entries:
         status = statuses.get(entry["accession"], {}).get(entry["version"], "UNKNOWN")
         entry_with_status = entry.copy()
@@ -508,3 +505,4 @@ def get_submitted(config: Config, output: str):
     if len(entries) == 0:
         with open(output, "w", encoding="utf-8"):
             pass
+    return None