Revive occurrence tracking as a post-processing task, behind a per-project opt-in flag

ami/base/permissions.py

@@ -172,6 +172,24 @@ def has_object_permission(self, request, view, obj: BaseModel):
         return obj.check_permission(request.user, view.action)
 
 
+TRACKING_NOT_ENABLED_MESSAGE = "Tracking is not enabled for this project."
+
+
+class TrackingEnabled(permissions.BasePermission):
+    """
+    Refuse an action on an object whose project has not opted into tracking.
+
+    Pair it after ObjectPermission, so a user without rights on the object gets the
+    usual refusal rather than a hint about the project's settings.
+    """
+
+    message = TRACKING_NOT_ENABLED_MESSAGE
+
+    def has_object_permission(self, request, view, obj: BaseModel):
+        project = obj.get_project() if hasattr(obj, "get_project") else None
+        return bool(project and project.feature_flags.tracking)
+
+
 class ProjectPipelineConfigPermission(ObjectPermission):
     """
     Permission for the nested project pipelines route (/projects/{pk}/pipelines/).

ami/exports/base.py

@@ -11,6 +11,8 @@ class BaseExporter(ABC):
     """Base class for all data export handlers."""
 
     file_format = ""  # To be defined in child classes
+    # A project feature flag that must be on before this format can be requested.
+    required_feature_flag: str | None = None
     serializer_class = None
     filter_backends = []
 

ami/exports/format_types.py

@@ -15,10 +15,19 @@
 
 
 def get_export_serializer():
-    from ami.main.api.serializers import OccurrenceSerializer
+    from ami.main.api.serializers import DetectionNestedSerializer, OccurrenceSerializer
+
+    # The detail response pages its detections; an export carries every one of them.
+    detail_only_fields = ("grouping_summary", "first_detection", "last_detection")
 
     class OccurrenceExportSerializer(OccurrenceSerializer):
         detection_images = serializers.SerializerMethodField()
+        detections = DetectionNestedSerializer(many=True, read_only=True)
+
+        class Meta(OccurrenceSerializer.Meta):
+            # The grouping summary describes the detections as they stand at read
+            # time; it is not part of the occurrence record and is never exported.
+            fields = [name for name in OccurrenceSerializer.Meta.fields if name not in detail_only_fields]
 
         def get_detection_images(self, obj: Occurrence):
             """Convert the generator field to a list before serialization"""
@@ -110,6 +119,10 @@ class OccurrenceTabularSerializer(serializers.ModelSerializer):
     agreed_with_user = serializers.SerializerMethodField()
     determination_matches_machine_prediction = serializers.SerializerMethodField()
 
+    # Track grouping confirmation; who confirmed it is user data and is not exported.
+    grouping_verified = serializers.BooleanField(read_only=True)
+    grouping_verified_at = serializers.DateTimeField(allow_null=True, read_only=True)
+
     # Detection fields
     best_detection_url = serializers.SerializerMethodField()
     best_detection_bbox = serializers.SerializerMethodField()
@@ -139,6 +152,8 @@ class Meta:
             "agreed_with_algorithm",
             "agreed_with_user",
             "determination_matches_machine_prediction",
+            "grouping_verified",
+            "grouping_verified_at",
             "detections_count",
             "first_appearance_timestamp",
             "last_appearance_timestamp",
@@ -252,3 +267,25 @@ def export(self):
                 self.update_job_progress(records_exported)
         self.update_export_stats(file_temp_path=temp_file.name)
         return temp_file.name  # Return the file path
+
+
+class TracksCSVExporter(BaseExporter):
+    """One row per detection of every occurrence in scope; see ami/exports/tracks.py for the columns."""
+
+    file_format = "csv"
+    required_feature_flag = "tracking"
+
+    def get_queryset(self):
+        return Occurrence.objects.with_real_detections().filter(project=self.project)  # type: ignore[union-attr]
+
+    def export(self):
+        from ami.exports.tracks import write_tracks_csv
+
+        temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".csv", mode="w", newline="", encoding="utf-8")
+        with open(temp_file.name, "w", newline="", encoding="utf-8") as csvfile:
+            rows = write_tracks_csv(self.queryset, csvfile, on_chunk=self.update_job_progress)
+        self.update_export_stats(file_temp_path=temp_file.name)
+        # A tracks file holds one record per detection, not per occurrence.
+        self.data_export.record_count = rows
+        self.data_export.save(update_fields=["record_count"])
+        return temp_file.name

ami/exports/migrations/0002_add_tracks_csv_format.py

@@ -0,0 +1,24 @@
+# Generated by Django 4.2.10 on 2026-09-22 17:00
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("exports", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="dataexport",
+            name="format",
+            field=models.CharField(
+                choices=[
+                    ("occurrences_api_json", "occurrences_api_json"),
+                    ("occurrences_simple_csv", "occurrences_simple_csv"),
+                    ("tracks_csv", "tracks_csv"),
+                ],
+                max_length=255,
+            ),
+        ),
+    ]

ami/exports/registry.py

@@ -27,3 +27,4 @@ def get_supported_formats(cls):
 
 ExportRegistry.register("occurrences_api_json")(format_types.JSONExporter)
 ExportRegistry.register("occurrences_simple_csv")(format_types.CSVExporter)
+ExportRegistry.register("tracks_csv")(format_types.TracksCSVExporter)

ami/exports/serializers.py

@@ -62,6 +62,17 @@ def validate_format(self, value):
             raise serializers.ValidationError(f"Invalid format. Supported formats are: {supported_formats}")
         return value
 
+    def validate(self, attrs):
+        attrs = super().validate(attrs)
+        exporter = ExportRegistry.get_exporter(attrs.get("format"))
+        flag = getattr(exporter, "required_feature_flag", None)
+        project = attrs.get("project")
+        if flag and project is not None and not getattr(project.feature_flags, flag, False):
+            raise serializers.ValidationError(
+                {"format": f"The {attrs['format']} export needs the {flag} feature, which is off for this project."}
+            )
+        return attrs
+
     def get_file_url(self, obj):
         return obj.get_absolute_url(request=self.context.get("request"))