Add bitmask fields for ChannelMetadata categories

contentcuration/contentcuration/tests/helpers.py

@@ -1,7 +1,9 @@
 from importlib import import_module
+from urllib.parse import urlencode
 
 import mock
 from celery import states
+from django.urls import reverse
 from django_celery_results.models import TaskResult
 from search.models import ContentNodeFullTextSearch
 
@@ -53,3 +55,19 @@ def __new__(cls, *args, **kwargs):
             return mock.Mock(spec_set=cls)
 
     return MockClass()
+
+
+def reverse_with_query(
+    viewname, urlconf=None, args=None, kwargs=None, current_app=None, query=None
+):
+    """
+    This helper wraps the Django `reverse` function to support the `query` argument.
+    This argument is supported natively since Django 5.2, so when Django is updated
+    above this version, this helper can be removed.
+    """
+    url = reverse(
+        viewname, urlconf=urlconf, args=args, kwargs=kwargs, current_app=current_app
+    )
+    if query:
+        return f"{url}?{urlencode(query)}"
+    return url

contentcuration/contentcuration/tests/viewsets/test_community_library_submission.py

@@ -1,6 +1,5 @@
 import datetime
 from unittest import mock
-from urllib.parse import urlencode
 
 import pytz
 from django.urls import reverse
@@ -12,25 +11,10 @@
 from contentcuration.models import CommunityLibrarySubmission
 from contentcuration.tests import testdata
 from contentcuration.tests.base import StudioAPITestCase
+from contentcuration.tests.helpers import reverse_with_query
 from contentcuration.viewsets.sync.constants import ADDED_TO_COMMUNITY_LIBRARY
 
 
-def reverse_with_query(
-    viewname, urlconf=None, args=None, kwargs=None, current_app=None, query=None
-):
-    """
-    This helper wraps the Django `reverse` function to support the `query` argument.
-    This argument is supported natively since Django 5.2, so when Django is updated
-    above this version, this helper can be removed.
-    """
-    url = reverse(
-        viewname, urlconf=urlconf, args=args, kwargs=kwargs, current_app=current_app
-    )
-    if query:
-        return f"{url}?{urlencode(query)}"
-    return url
-
-
 class CRUDTestCase(StudioAPITestCase):
     @property
     def new_submission_metadata(self):

contentcuration/kolibri_public/migrations/0008_channelmetadata_categories_bitmask_0.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.2.24 on 2025-08-08 17:32
+from django.db import migrations
+from django.db import models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("kolibri_public", "0007_new_channel_metadata"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="channelmetadata",
+            name="categories_bitmask_0",
+            field=models.BigIntegerField(blank=True, default=0, null=True),
+        ),
+    ]

contentcuration/kolibri_public/models.py

@@ -1,9 +1,11 @@
 from django.db import models
-from django.db.models import F
 from kolibri_content import base_models
 from kolibri_content.fields import JSONField
-from kolibri_public.search import bitmask_fieldnames
-from kolibri_public.search import metadata_bitmasks
+from kolibri_public.search import channelmetadata_bitmask_fieldnames
+from kolibri_public.search import channelmetadata_metadata_bitmasks
+from kolibri_public.search import contentnode_bitmask_fieldnames
+from kolibri_public.search import contentnode_metadata_bitmasks
+from kolibri_public.search import has_all_labels
 from mptt.managers import TreeManager
 from mptt.querysets import TreeQuerySet
 
@@ -17,31 +19,7 @@ class ContentTag(base_models.ContentTag):
 
 class ContentNodeQueryset(TreeQuerySet):
     def has_all_labels(self, field_name, labels):
-        bitmasks = metadata_bitmasks[field_name]
-        bits = {}
-        for label in labels:
-            if label in bitmasks:
-                bitmask_fieldname = bitmasks[label]["bitmask_field_name"]
-                if bitmask_fieldname not in bits:
-                    bits[bitmask_fieldname] = 0
-                bits[bitmask_fieldname] += bitmasks[label]["bits"]
-
-        filters = {}
-        annotations = {}
-        for bitmask_fieldname, bits in bits.items():
-            annotation_fieldname = "{}_{}".format(bitmask_fieldname, "masked")
-            # To get the correct result, i.e. an AND that all the labels are present,
-            # we need to check that the aggregated value is euqal to the bits.
-            # If we wanted an OR (which would check for any being present),
-            # we would have to use GREATER THAN 0 here.
-            filters[annotation_fieldname] = bits
-            # This ensures that the annotated value is the result of the AND operation
-            # so if all the values are present, the result will be the same as the bits
-            # but if any are missing, it will not be equal to the bits, but will only be
-            # 0 if none of the bits are present.
-            annotations[annotation_fieldname] = F(bitmask_fieldname).bitand(bits)
-
-        return self.annotate(**annotations).filter(**filters)
+        return has_all_labels(self, contentnode_metadata_bitmasks, field_name, labels)
 
 
 class ContentNodeManager(
@@ -79,7 +57,7 @@ class ContentNode(base_models.ContentNode):
     objects = ContentNodeManager()
 
 
-for field_name in bitmask_fieldnames:
+for field_name in contentnode_bitmask_fieldnames:
     field = models.BigIntegerField(default=0, null=True, blank=True)
     field.contribute_to_class(ContentNode, field_name)
 
@@ -96,6 +74,17 @@ class AssessmentMetaData(base_models.AssessmentMetaData):
     pass
 
 
+class ChannelMetadataQueryset(models.QuerySet):
+    def has_all_labels(self, field_name, labels):
+        return has_all_labels(
+            self, channelmetadata_metadata_bitmasks, field_name, labels
+        )
+
+
+class ChannelMetadataManager(models.Manager.from_queryset(ChannelMetadataQueryset)):
+    pass
+
+
 class ChannelMetadata(base_models.ChannelMetadata):
     # Note: The `categories` field should contain a _list_, NOT a _dict_.
 
@@ -110,6 +99,13 @@ class ChannelMetadata(base_models.ChannelMetadata):
     categories = models.JSONField(null=True, blank=True)
     countries = models.ManyToManyField(Country, related_name="public_channels")
 
+    objects = ChannelMetadataManager()
+
+
+for field_name in channelmetadata_bitmask_fieldnames:
+    field = models.BigIntegerField(default=0, null=True, blank=True)
+    field.contribute_to_class(ChannelMetadata, field_name)
+
 
 class MPTTTreeIDManager(models.Model):
     """

contentcuration/kolibri_public/search.py

@@ -26,37 +26,54 @@
 from le_utils.constants.labels.subjects import SUBJECTSLIST
 
 
-metadata_lookup = {
+contentnode_metadata_lookup = {
     "learning_activities": LEARNINGACTIVITIESLIST,
     "categories": SUBJECTSLIST,
     "grade_levels": LEVELSLIST,
     "accessibility_labels": ACCESSIBILITYCATEGORIESLIST,
     "learner_needs": NEEDSLIST,
 }
+contentnode_metadata_bitmasks = {}
+contentnode_bitmask_fieldnames = {}
 
-
-metadata_bitmasks = {}
-
-bitmask_fieldnames = {}
-
-
-for key, labels in metadata_lookup.items():
-    bitmask_lookup = {}
-    i = 0
-    while labels[i : i + 64]:
-        bitmask_field_name = "{}_bitmask_{}".format(key, i)
-        bitmask_fieldnames[bitmask_field_name] = []
-        for j, label in enumerate(labels):
-            info = {
-                "bitmask_field_name": bitmask_field_name,
-                "field_name": key,
-                "bits": 2 ** j,
-                "label": label,
-            }
-            bitmask_lookup[label] = info
-            bitmask_fieldnames[bitmask_field_name].append(info)
-        i += 64
-    metadata_bitmasks[key] = bitmask_lookup
+channelmetadata_metadata_lookup = {
+    "categories": SUBJECTSLIST,
+}
+channelmetadata_metadata_bitmasks = {}
+channelmetadata_bitmask_fieldnames = {}
+
+
+def _populate_bitmask_data(metadata_lookup, metadata_bitmasks, bitmask_fieldnames):
+
+    for key, labels in metadata_lookup.items():
+        bitmask_lookup = {}
+        i = 0
+        while (chunk := labels[i : i + 64]) :
+            bitmask_field_name = "{}_bitmask_{}".format(key, i)
+            bitmask_fieldnames[bitmask_field_name] = []
+            for j, label in enumerate(chunk):
+                info = {
+                    "bitmask_field_name": bitmask_field_name,
+                    "field_name": key,
+                    "bits": 2 ** j,
+                    "label": label,
+                }
+                bitmask_lookup[label] = info
+                bitmask_fieldnames[bitmask_field_name].append(info)
+            i += 64
+        metadata_bitmasks[key] = bitmask_lookup
+
+
+_populate_bitmask_data(
+    contentnode_metadata_lookup,
+    contentnode_metadata_bitmasks,
+    contentnode_bitmask_fieldnames,
+)
+_populate_bitmask_data(
+    channelmetadata_metadata_lookup,
+    channelmetadata_metadata_bitmasks,
+    channelmetadata_bitmask_fieldnames,
+)
 
 
 def _get_available_languages(base_queryset):
@@ -87,7 +104,7 @@ def _get_available_channels(base_queryset):
 # Remove the SQLite Bitwise OR definition as not needed.
 
 
-def get_available_metadata_labels(base_queryset):
+def get_contentnode_available_metadata_labels(base_queryset):
     # Updated to use the kolibri_public ChannelMetadata model
     from kolibri_public.models import ChannelMetadata
 
@@ -101,12 +118,12 @@ def get_available_metadata_labels(base_queryset):
     if cache_key not in cache:
         base_queryset = base_queryset.order_by()
         aggregates = {}
-        for field in bitmask_fieldnames:
+        for field in contentnode_bitmask_fieldnames:
             field_agg = field + "_agg"
             aggregates[field_agg] = BitOr(field)
         output = {}
         agg = base_queryset.aggregate(**aggregates)
-        for field, values in bitmask_fieldnames.items():
+        for field, values in contentnode_bitmask_fieldnames.items():
             bit_value = agg[field + "_agg"]
             for value in values:
                 if value["field_name"] not in output:
@@ -123,10 +140,12 @@ def get_all_contentnode_label_metadata():
     # Updated to use the kolibri_public ContentNode model
     from kolibri_public.models import ContentNode
 
-    return get_available_metadata_labels(ContentNode.objects.filter(available=True))
+    return get_contentnode_available_metadata_labels(
+        ContentNode.objects.filter(available=True)
+    )
 
 
-def annotate_label_bitmasks(queryset):
+def annotate_label_bitmasks(queryset, bitmask_fieldnames):
     update_statements = {}
     for bitmask_fieldname, label_info in bitmask_fieldnames.items():
         update_statements[bitmask_fieldname] = sum(
@@ -142,3 +161,39 @@ def annotate_label_bitmasks(queryset):
             for info in label_info
         )
     queryset.update(**update_statements)
+
+
+def annotate_contentnode_label_bitmasks(queryset):
+    return annotate_label_bitmasks(queryset, contentnode_bitmask_fieldnames)
+
+
+def annotate_channelmetadata_label_bitmasks(queryset):
+    return annotate_label_bitmasks(queryset, channelmetadata_bitmask_fieldnames)
+
+
+def has_all_labels(queryset, metadata_bitmasks, field_name, labels):
+    bitmasks = metadata_bitmasks[field_name]
+    bits = {}
+    for label in labels:
+        if label in bitmasks:
+            bitmask_fieldname = bitmasks[label]["bitmask_field_name"]
+            if bitmask_fieldname not in bits:
+                bits[bitmask_fieldname] = 0
+            bits[bitmask_fieldname] += bitmasks[label]["bits"]
+
+    filters = {}
+    annotations = {}
+    for bitmask_fieldname, bits in bits.items():
+        annotation_fieldname = "{}_{}".format(bitmask_fieldname, "masked")
+        # To get the correct result, i.e. an AND that all the labels are present,
+        # we need to check that the aggregated value is euqal to the bits.
+        # If we wanted an OR (which would check for any being present),
+        # we would have to use GREATER THAN 0 here.
+        filters[annotation_fieldname] = bits
+        # This ensures that the annotated value is the result of the AND operation
+        # so if all the values are present, the result will be the same as the bits
+        # but if any are missing, it will not be equal to the bits, but will only be
+        # 0 if none of the bits are present.
+        annotations[annotation_fieldname] = F(bitmask_fieldname).bitand(bits)
+
+    return queryset.annotate(**annotations).filter(**filters)