feat: [five-c] Create rules-based engine #409

dje/admin.py

@@ -96,6 +96,7 @@
 from dje.views import manage_tab_permissions_view
 from dje.views import object_compare_view
 from dje.views import object_copy_view
+from policy.rules import RULE_REGISTRY
 
 EXTERNAL_SOURCE_LOOKUP = "external_references__external_source_id"
 
@@ -1046,12 +1047,11 @@ def render(self, name, value, attrs=None, renderer=None):
 
 
 class DataspaceConfigurationForm(forms.ModelForm):
-    """
-    Configure Dataspace settings.
+    """Configure Dataspace integration settings, with sensitive values hidden in the UI."""
 
-    This form includes fields for various API keys, with sensitive values
-    hidden in the UI using the HiddenValueWidget.
-    """
+    class Meta:
+        model = DataspaceConfiguration
+        exclude = ["policy_rules_config"]
 
     hidden_value_fields = [
         "scancodeio_api_key",
@@ -1078,6 +1078,73 @@ def clean(self):
                 del self.cleaned_data[field_name]
 
 
+class PolicyRulesConfigurationForm(forms.ModelForm):
+    """Form for configuring policy rule overrides stored in policy_rules_config."""
+
+    class Meta:
+        model = DataspaceConfiguration
+        fields = []
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.add_policy_rule_config_fields()
+
+    def add_policy_rule_config_fields(self):
+        """Inject per-rule form fields with initial values from the saved policy_rules_config."""
+        config = getattr(self.instance, "policy_rules_config", {}) or {}
+        for rule_type, handler in RULE_REGISTRY.items():
+            rule_config = config.get(rule_type, {})
+            self.fields[f"rule_{rule_type}_disabled"] = forms.BooleanField(
+                label="Disable this rule",
+                required=False,
+                initial=not rule_config.get("is_active", True),
+            )
+            self.fields[f"rule_{rule_type}_threshold"] = forms.IntegerField(
+                label="Threshold",
+                required=False,
+                min_value=0,
+                initial=rule_config.get("threshold"),
+                widget=forms.NumberInput(
+                    attrs={"placeholder": f"Default: {handler.default_threshold}"}
+                ),
+                help_text="Minimum violations to trigger the rule. Leave blank to use the default.",
+            )
+            for param_name, param_desc in handler.parameters_schema.items():
+                self.fields[f"rule_{rule_type}_param_{param_name}"] = forms.FloatField(
+                    label=param_name.replace("_", " ").title(),
+                    required=False,
+                    initial=(rule_config.get("parameters") or {}).get(param_name),
+                    help_text=param_desc,
+                )
+
+    def build_policy_rules_config(self):
+        """Serialize the per-rule form fields back into the policy_rules_config dict."""
+        policy_rules_config = {}
+        for rule_type, handler in RULE_REGISTRY.items():
+            rule_config = {}
+            if self.cleaned_data.get(f"rule_{rule_type}_disabled"):
+                rule_config["is_active"] = False
+            threshold = self.cleaned_data.get(f"rule_{rule_type}_threshold")
+            if threshold is not None:
+                rule_config["threshold"] = threshold
+            parameters = {}
+            for param_name in handler.parameters_schema:
+                param_value = self.cleaned_data.get(f"rule_{rule_type}_param_{param_name}")
+                if param_value is not None:
+                    parameters[param_name] = param_value
+            if parameters:
+                rule_config["parameters"] = parameters
+            if rule_config:
+                policy_rules_config[rule_type] = rule_config
+        return policy_rules_config
+
+    def save(self, commit=True):
+        self.instance.policy_rules_config = self.build_policy_rules_config()
+        if commit:
+            self.instance.save(update_fields=["policy_rules_config"])
+        return self.instance
+
+
 class DataspaceConfigurationInline(DataspacedFKMixin, admin.StackedInline):
     model = DataspaceConfiguration
     form = DataspaceConfigurationForm
@@ -1142,12 +1209,13 @@ class DataspaceConfigurationInline(DataspacedFKMixin, admin.StackedInline):
     ]
     # Do not include the Dataspace related FKs on addition as the Dataspace does not exist yet
     fieldsets = [("", {"fields": ("homepage_layout",)})] + add_fieldsets
+    inline_classes = ("grp-collapse grp-open",)
     can_delete = False
 
     def get_fieldsets(self, request, obj=None):
         if not obj:
             return self.add_fieldsets
-        return super().get_fieldsets(request, obj)
+        return [("", {"fields": ("homepage_layout",)})] + self.add_fieldsets
 
     def get_readonly_fields(self, request, obj=None):
         """Only a user from the current Dataspace can edit Dataspace related FKs."""
@@ -1160,6 +1228,38 @@ def get_readonly_fields(self, request, obj=None):
         return readonly_fields
 
 
+class PolicyRulesConfigurationInline(DataspacedFKMixin, admin.StackedInline):
+    model = DataspaceConfiguration
+    form = PolicyRulesConfigurationForm
+    verbose_name_plural = _("Policy Rules Configuration")
+    verbose_name = _("Policy Rules Configuration")
+    classes = ("grp-collapse grp-open",)
+    inline_classes = ("grp-collapse grp-open",)
+    can_delete = False
+
+    def get_fieldsets(self, request, obj=None):
+        if not obj:
+            return []
+        rule_fieldsets = []
+        for rule_type, handler in RULE_REGISTRY.items():
+            fields = [f"rule_{rule_type}_disabled", f"rule_{rule_type}_threshold"]
+            for param_name in handler.parameters_schema:
+                fields.append(f"rule_{rule_type}_param_{param_name}")
+            rule_fieldsets.append((
+                handler.label,
+                {
+                    "fields": fields,
+                    "description": handler.description,
+                    "classes": ("grp-collapse grp-open",),
+                },
+            ))
+        return rule_fieldsets
+
+    def get_formset(self, request, obj=None, **kwargs):
+        kwargs["fields"] = []
+        return super().get_formset(request, obj, **kwargs)
+
+
 @admin.register(Dataspace, site=dejacode_site)
 class DataspaceAdmin(
     ReferenceOnlyPermissions,
@@ -1239,7 +1339,7 @@ class DataspaceAdmin(
         ),
     )
     search_fields = ("name",)
-    inlines = [DataspaceConfigurationInline]
+    inlines = [DataspaceConfigurationInline, PolicyRulesConfigurationInline]
     form = DataspaceAdminForm
     change_form_template = "admin/dje/dataspace/change_form.html"
     change_list_template = "admin/change_list_extended.html"

dje/migrations/0016_dataspaceconfiguration_policy_rules_config.py

@@ -0,0 +1,18 @@
+# Generated by Django 6.0.6 on 2026-07-15 08:25
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('dje', '0015_alter_dataspaceconfiguration_purldb_api_key_and_more'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='dataspaceconfiguration',
+            name='policy_rules_config',
+            field=models.JSONField(blank=True, default=dict, help_text='Override default policy rule settings for this dataspace.'),
+        ),
+    ]

dje/models.py

@@ -614,6 +614,12 @@ class DataspaceConfiguration(DataspaceForeignKeyValidationMixin, models.Model):
         ),
     )
 
+    policy_rules_config = models.JSONField(
+        blank=True,
+        default=dict,
+        help_text=_("Override default policy rule settings for this dataspace."),
+    )
+
     def __str__(self):
         return f"{self.dataspace}"
 

policy/engine.py

@@ -0,0 +1,81 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# DejaCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: AGPL-3.0-only
+# See https://github.com/aboutcode-org/dejacode for support or download.
+# See https://aboutcode.org for more information about AboutCode FOSS projects.
+#
+
+from django.utils import timezone
+
+from policy.rules import RULE_REGISTRY
+from product_portfolio.models import ProductPolicyViolation
+
+
+def get_effective_config(rule_type, dataspace):
+    """
+    Resolve threshold, parameters, and is_active for a rule type in a given dataspace.
+
+    Reads the dataspace-level override from DataspaceConfiguration.policy_rules_config,
+    falling back to the code defaults defined on the rule handler.
+    """
+    handler = RULE_REGISTRY[rule_type]
+    try:
+        rule_config = dataspace.configuration.policy_rules_config.get(rule_type, {})
+    except AttributeError:
+        rule_config = {}
+
+    return {
+        "is_active": rule_config.get("is_active", True),
+        "threshold": rule_config.get("threshold", handler.default_threshold),
+        "parameters": rule_config.get("parameters", {}),
+    }
+
+
+def evaluate_rule(rule_type, product, threshold, parameters):
+    """Evaluate a single rule against a product and record the violation if triggered."""
+    handler = RULE_REGISTRY[rule_type]
+    violation_count = handler.count_violations(product, threshold, parameters)
+
+    lookup = {"rule_type": rule_type, "product": product, "resolved": False}
+
+    if violation_count > 0:
+        violation, created = ProductPolicyViolation.objects.get_or_create(
+            **lookup,
+            defaults={"dataspace": product.dataspace, "violation_count": violation_count},
+        )
+        if not created:
+            violation.violation_count = violation_count
+            violation.save()
+        return violation
+
+    else:
+        ProductPolicyViolation.objects.filter(**lookup).update(
+            resolved=True,
+            resolved_date=timezone.now(),
+        )
+        return
+
+
+def evaluate_rules(product):
+    """
+    Evaluate all rules in RULE_REGISTRY for the given product.
+
+    Returns the list of active ProductPolicyViolation instances.
+    """
+    violations = []
+    for rule_type in RULE_REGISTRY:
+        config = get_effective_config(rule_type, product.dataspace)
+        if not config["is_active"]:
+            # Explicitly resolve open violations so disabling a rule clears its history
+            # rather than leaving stale unresolved records.
+            ProductPolicyViolation.objects.filter(
+                rule_type=rule_type, product=product, resolved=False,
+            ).update(resolved=True, resolved_date=timezone.now())
+            continue
+
+        violation = evaluate_rule(rule_type, product, config["threshold"], config["parameters"])
+        if violation:
+            violations.append(violation)
+
+    return violations

policy/models.py

@@ -244,3 +244,32 @@ def save(self, *args, **kwargs):
         if self.from_policy.content_type == self.to_policy.content_type:
             raise AssertionError
         super().save(*args, **kwargs)
+
+
+class AbstractPolicyViolation(models.Model):
+    """Shared fields for all concrete policy violation models. No DB table."""
+
+    violation_count = models.PositiveIntegerField(
+        default=0,
+        help_text=_("Number of objects currently violating the rule."),
+    )
+    detected_date = models.DateTimeField(
+        auto_now_add=True,
+        help_text=_("The date and time when this violation was first detected."),
+    )
+    last_checked = models.DateTimeField(
+        auto_now=True,
+        help_text=_("The date and time of the last evaluation."),
+    )
+    resolved = models.BooleanField(
+        default=False,
+        help_text=_("Indicates whether this violation has been resolved."),
+    )
+    resolved_date = models.DateTimeField(
+        null=True,
+        blank=True,
+        help_text=_("The date and time when this violation was resolved."),
+    )
+
+    class Meta:
+        abstract = True