Validate legacy GRB snapshot evidence

backend/app/services/temporal_analysis_service.py

@@ -500,7 +500,10 @@ class TemporalAnalysisService:
             dataset.source_name != "grb"
             or not str(dataset.temporal_series_key or "").startswith("grb:")
             or source_metadata.get("authority_level") != "authoritative"
-            or source_metadata.get("geometry_clipped_to_area") is not True
+            or not TemporalAnalysisService._has_governed_grb_area_contract(
+                source_metadata,
+                provenance,
+            )
             or provenance.get("operator_tool") not in TemporalAnalysisService.GOVERNED_GRB_IDENTITY_OPERATORS
             or provenance.get("reference_truncated") is not False
         ):
@@ -515,6 +518,44 @@ class TemporalAnalysisService:
             prefixes = tuple(sorted(f"{str(collection)}:{str(collection)}." for collection in collections))
         return "grb_ogc_feature_id", prefixes
 
+    @staticmethod
+    def _has_governed_grb_area_contract(
+        source_metadata: dict[str, Any],
+        provenance: dict[str, Any],
+    ) -> bool:
+        if source_metadata.get("geometry_clipped_to_area") is True:
+            return True
+
+        partition_checksums = provenance.get("partition_checksums")
+        artifact_checksum = str(provenance.get("artifact_sha256") or "")
+        has_valid_checksum = len(artifact_checksum) == 64 and all(
+            character in "0123456789abcdefABCDEF" for character in artifact_checksum
+        )
+        has_valid_partition_checksums = (
+            isinstance(partition_checksums, dict)
+            and len(partition_checksums) == 28
+            and all(
+                len(str(checksum)) == 64
+                and all(character in "0123456789abcdefABCDEF" for character in str(checksum))
+                for checksum in partition_checksums.values()
+            )
+        )
+        return (
+            source_metadata.get("coverage_scope") == "kempen-transport-region"
+            and source_metadata.get("scope_type") == "transport_region"
+            and source_metadata.get("member_count") == 28
+            and source_metadata.get("partition_count") == 28
+            and source_metadata.get("partition_strategy")
+            in {
+                "municipality_bbox_maximum_boundary_intersection",
+                "municipality_bbox_maximum_same_dimension_intersection",
+            }
+            and bool(provenance.get("manifest_path"))
+            and bool(provenance.get("source_url") or provenance.get("source_urls"))
+            and has_valid_checksum
+            and has_valid_partition_checksums
+        )
+
     @staticmethod
     def _rows_match_identity_contract(rows: list[VectorFeature], prefixes: tuple[str, ...]) -> bool:
         identities = [str(row.source_feature_id or "").strip() for row in rows]