fix format of custom bool dynamic attrs posted by CS frontend; refactor CS... (9b7bb635) · Commits · trapper-project / trapper

trapper/trapper-project/trapper/apps/media_classification/migrations/0084_convert_custom_boolean_attrs.py

0 → 100644

+136 −0

Original line number	Diff line number	Diff line
		from django.db import migrations


		def convert_boolean_attrs(apps, schema_editor):
		# Get models from the migration state
		ClassificationProject = apps.get_model(
		"media_classification", "ClassificationProject"
		)
		UserClassificationDynamicAttrs = apps.get_model(
		"media_classification", "UserClassificationDynamicAttrs"
		)
		ClassificationDynamicAttrs = apps.get_model(
		"media_classification", "ClassificationDynamicAttrs"
		)

		# Get all CS-enabled classification projects
		cs_projects = ClassificationProject.objects.filter(
		citizen_science_status__in=[1, 2], # Public or Private
		classificator__isnull=False,
		).select_related("classificator")

		# Process each project
		for project in cs_projects:
		classificator = project.classificator

		# Skip if no custom attributes or no boolean custom attributes
		if not classificator.custom_attrs:
		continue

		# Get boolean custom attributes
		boolean_attrs = [
		attr_name
		for attr_name, attr_spec in classificator.custom_attrs.items()
		if attr_spec.get("field_type") == "B"
		]

		if not boolean_attrs:
		continue

		# Get all dynamic attrs for user classifications in this project
		dynamic_attrs = UserClassificationDynamicAttrs.objects.filter(
		userclassification__classification__project=project
		).exclude(attrs={})

		# Process dynamic attrs records in batches
		to_update = []
		batch_size = 1000

		for dyn_attr in dynamic_attrs:
		modified = False

		# Check each boolean attribute
		for attr_name in boolean_attrs:
		if attr_name not in dyn_attr.attrs:
		continue

		value = dyn_attr.attrs[attr_name]
		if isinstance(value, str):
		# Convert string boolean to Python boolean
		if value.lower() == "true":
		dyn_attr.attrs[attr_name] = True
		modified = True
		elif value.lower() == "false":
		dyn_attr.attrs[attr_name] = False
		modified = True

		if modified:
		to_update.append(dyn_attr)

		# Update in batches
		if len(to_update) >= batch_size:
		UserClassificationDynamicAttrs.objects.bulk_update(
		to_update, ["attrs"], batch_size=batch_size
		)
		to_update = []

		# Update remaining records
		if to_update:
		UserClassificationDynamicAttrs.objects.bulk_update(
		to_update, ["attrs"], batch_size=batch_size
		)

		# Process approved classifications derived from user classifications
		approved_dynamic_attrs = ClassificationDynamicAttrs.objects.filter(
		classification__project=project,
		classification__approved_source__isnull=False,
		).exclude(attrs={})

		# Process approved dynamic attrs records in batches
		to_update = []

		for dyn_attr in approved_dynamic_attrs:
		modified = False

		# Check each boolean attribute
		for attr_name in boolean_attrs:
		if attr_name not in dyn_attr.attrs:
		continue

		value = dyn_attr.attrs[attr_name]
		if isinstance(value, str):
		# Convert string boolean to Python boolean
		if value.lower() == "true":
		dyn_attr.attrs[attr_name] = True
		modified = True
		elif value.lower() == "false":
		dyn_attr.attrs[attr_name] = False
		modified = True

		if modified:
		to_update.append(dyn_attr)

		# Update in batches
		if len(to_update) >= batch_size:
		ClassificationDynamicAttrs.objects.bulk_update(
		to_update, ["attrs"], batch_size=batch_size
		)
		to_update = []

		# Update remaining records
		if to_update:
		ClassificationDynamicAttrs.objects.bulk_update(
		to_update, ["attrs"], batch_size=batch_size
		)


		class Migration(migrations.Migration):
		dependencies = [
		("media_classification", "0083_rebuild_sequences"),
		]

		operations = [
		migrations.RunPython(
		convert_boolean_attrs, reverse_code=migrations.RunPython.noop
		)
		]

trapper/trapper-project/trapper/apps/media_classification/serializers_rest.py

+71 −84

Original line number	Diff line number	Diff line
		@@ -101,14 +101,16 @@ class ClassifyDynamicVideoSerializer(ClassifyDynamicSerializer):

		class BaseClassifyMixin:
		"""
		Add user to ClassificationProjectRole.
		Add a user to ClassificationProjectRole.

		Set bboxes for user classification.

		Send mail and create a Trapper Message to each active admins about
		detection of tracked species in specific classification project
		Validate custom dynamic attributes for each individual observation.

		Tip: this class should be used by single and group classification
		Send an email and create a Trapper Message for each active admin about
		the detection of tracked species in a specific classification project.

		Tip: This class should be used for both single and group classification.
		"""

		@staticmethod
		@@ -141,6 +143,63 @@ class BaseClassifyMixin:
		else:
		celery_approve_user_classifications(**params)

		@staticmethod
		def _custom_dynamic_attrs_validation(dynamic_data: dict, custom_attrs_spec: dict):
		"""
		Validate custom dynamic attributes for each individual observation and convert
		string boolean values to proper Python booleans.

		Args:
		dynamic_data: List of dictionaries containing dynamic attributes
		custom_attrs_spec: Dictionary of custom attribute specifications from the classificator

		Raises:
		serializers.ValidationError: If validation fails for any attribute
		"""
		error_list = {"dynamic": [{} for _ in range(len(dynamic_data))]}

		for attr_name, attr_setup in custom_attrs_spec.items():
		for index, observation in enumerate(dynamic_data):
		custom_attrs = observation.get("attrs", {})

		# If custom attribute is required then can not be null or empty
		if attr_setup.get("required", False):
		if attr_name not in custom_attrs:
		error_list["dynamic"][index][attr_name] = [
		_("This field is required.")
		]
		else:
		value = custom_attrs.get(attr_name, None)

		if value is None:
		error_list["dynamic"][index][attr_name] = [
		_("This field may not be null.")
		]

		# Extra validation for field type string
		if isinstance(value, str):
		value = value.strip()
		if not value:
		error_list["dynamic"][index][attr_name] = [
		_("This field may not be empty.")
		]

		# Convert string booleans to Python booleans for boolean fields if needed
		if attr_setup["field_type"] == "B" and attr_name in custom_attrs:
		value = custom_attrs[attr_name]
		if isinstance(value, str):
		if value.lower() == "true":
		custom_attrs[attr_name] = True
		elif value.lower() == "false":
		custom_attrs[attr_name] = False
		else:
		error_list["dynamic"][index][attr_name] = [
		_("Boolean field must be 'true' or 'false'.")
		]

		if any(error_list["dynamic"]):
		raise serializers.ValidationError(error_list)

		@staticmethod
		def add_user_to_classification_project(
		user: User, classification_project: ClassificationProject
		@@ -226,53 +285,17 @@ class SingleClassifySerializer(BaseClassifyMixin, serializers.ModelSerializer):
		if dynamic_data[0]["observation_type"] == ObservationType.BLANK:
		attrs["dynamic"][0] = {"observation_type": ObservationType.BLANK}

		self._custom_dynamic_attrs_validation(dynamic_data)
		self._validate_dynamic_attrs(dynamic_data)

		self._check_classification()
		return attrs

		def _custom_dynamic_attrs_validation(self, dynamic_data: dict):
		"""
		Validate custom dynamic attributes for each individual observation
		"""

		def _validate_dynamic_attrs(self, dynamic_data: dict):
		"""Validate custom dynamic attributes using the base mixin method"""
		custom_attrs_spec = self.context[
		"classification"
		].project.classificator.custom_attrs

		# DO NOT USE [{}] * len(dynamic_data)
		# It would create list of copies of the same dict object
		error_list = {"dynamic": [{} for _ in range(len(dynamic_data))]}

		for attr_name, attr_setup in custom_attrs_spec.items():
		for index, observation in enumerate(dynamic_data):
		# If custom attribute is required then can not be null or empty
		if attr_setup.get("required", False):
		custom_attrs = observation.get("attrs", [])

		if attr_name not in custom_attrs:
		error_list["dynamic"][index][attr_name] = [
		_("This field is required.")
		]
		else:
		value = custom_attrs.get(attr_name, None)

		if value is None:
		error_list["dynamic"][index][attr_name] = [
		_("This field may not be null.")
		]

		# Extra validation for field type string
		if isinstance(value, str):
		value = value.strip()

		if not value:
		error_list["dynamic"][index][attr_name] = [
		_("This field may not be empty.")
		]

		if any(error_list["dynamic"]):
		raise serializers.ValidationError(error_list)
		self._custom_dynamic_attrs_validation(dynamic_data, custom_attrs_spec)

		def _check_classification(self):
		classification = self.context["classification"]
		@@ -566,53 +589,17 @@ class GroupClassifySerializer(BaseClassifyMixin, serializers.ModelSerializer):
		_("Sequence ID or Classification IDs is required.")
		)

		self._custom_dynamic_attrs_validation(dynamic_data)
		self._validate_dynamic_attrs(dynamic_data)

		self._check_classification()
		return super().validate(attrs)

		def _custom_dynamic_attrs_validation(self, dynamic_data: dict):
		"""
		Validate custom dynamic attributes for each individual observation
		"""

		def _validate_dynamic_attrs(self, dynamic_data: dict):
		"""Validate custom dynamic attributes using the base mixin method"""
		custom_attrs_spec = self.context[
		"classification_project"
		].classificator.custom_attrs

		# DO NOT USE [{}] * len(dynamic_data)
		# It would create list of copies of the same dict object
		error_list = {"dynamic": [{} for _ in range(len(dynamic_data))]}

		for attr_name, attr_setup in custom_attrs_spec.items():
		for index, observation in enumerate(dynamic_data):
		# If custom attribute is required then can not be null or empty
		if attr_setup.get("required", False):
		custom_attrs = observation.get("attrs", [])

		if attr_name not in custom_attrs:
		error_list["dynamic"][index][attr_name] = [
		_("This field is required.")
		]
		else:
		value = custom_attrs.get(attr_name, None)

		if value is None:
		error_list["dynamic"][index][attr_name] = [
		_("This field may not be null.")
		]

		# Extra validation for field type string
		if isinstance(value, str):
		value = value.strip()

		if not value:
		error_list["dynamic"][index][attr_name] = [
		_("This field may not be empty.")
		]

		if any(error_list["dynamic"]):
		raise serializers.ValidationError(error_list)
		self._custom_dynamic_attrs_validation(dynamic_data, custom_attrs_spec)

		def _check_classification(self):
		user = self.context["request"].user