feat(campaign): add privacy-safe aggregate reports

src/govoplan_campaign/backend/report_privacy_policy.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import Any, Mapping
+
+from sqlalchemy.orm import Session
+
+from govoplan_core.tenancy.scope import Tenant
+
+
+DEFAULT_SMALL_CELL_THRESHOLD = 5
+MIN_SMALL_CELL_THRESHOLD = 2
+MAX_SMALL_CELL_THRESHOLD = 100
+SMALL_CELL_THRESHOLD_ENV = "GOVOPLAN_CAMPAIGN_REPORT_SMALL_CELL_THRESHOLD"
+CAMPAIGN_REPORT_POLICY_SETTINGS_KEY = "campaign_report_privacy_policy"
+SMALL_CELL_THRESHOLD_SETTINGS_KEY = "small_cell_threshold"
+
+
+class CampaignReportPrivacyPolicyError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True, slots=True)
+class CampaignReportPrivacyPolicy:
+    small_cell_threshold: int
+    source: str
+    deployment_small_cell_threshold: int
+    tenant_small_cell_threshold: int | None = None
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "small_cell_threshold": self.small_cell_threshold,
+            "source": self.source,
+            "deployment_small_cell_threshold": self.deployment_small_cell_threshold,
+            "tenant_small_cell_threshold": self.tenant_small_cell_threshold,
+            "deployment_setting": SMALL_CELL_THRESHOLD_ENV,
+            "tenant_setting": (
+                f"tenant.settings.{CAMPAIGN_REPORT_POLICY_SETTINGS_KEY}."
+                f"{SMALL_CELL_THRESHOLD_SETTINGS_KEY}"
+            ),
+        }
+
+
+def effective_campaign_report_privacy_policy(
+    session: Session,
+    *,
+    tenant_id: str,
+    environ: Mapping[str, str] | None = None,
+) -> CampaignReportPrivacyPolicy:
+    env = os.environ if environ is None else environ
+    deployment_raw = env.get(SMALL_CELL_THRESHOLD_ENV)
+    deployment_value = _configured_threshold(
+        deployment_raw,
+        source=SMALL_CELL_THRESHOLD_ENV,
+        default=DEFAULT_SMALL_CELL_THRESHOLD,
+    )
+    tenant = session.get(Tenant, tenant_id)
+    tenant_raw = _tenant_threshold_value(tenant.settings if tenant is not None else None)
+    if tenant_raw is None:
+        return CampaignReportPrivacyPolicy(
+            small_cell_threshold=deployment_value,
+            source="deployment" if deployment_raw not in (None, "") else "deployment_default",
+            deployment_small_cell_threshold=deployment_value,
+        )
+
+    tenant_value = _configured_threshold(
+        tenant_raw,
+        source=(
+            f"tenant.settings.{CAMPAIGN_REPORT_POLICY_SETTINGS_KEY}."
+            f"{SMALL_CELL_THRESHOLD_SETTINGS_KEY}"
+        ),
+    )
+    effective_value = max(deployment_value, tenant_value)
+    return CampaignReportPrivacyPolicy(
+        small_cell_threshold=effective_value,
+        source="tenant" if tenant_value >= deployment_value else "deployment_floor",
+        deployment_small_cell_threshold=deployment_value,
+        tenant_small_cell_threshold=tenant_value,
+    )
+
+
+def _tenant_threshold_value(settings: Mapping[str, Any] | None) -> object | None:
+    if not isinstance(settings, Mapping):
+        return None
+    policy = settings.get(CAMPAIGN_REPORT_POLICY_SETTINGS_KEY)
+    if not isinstance(policy, Mapping):
+        return None
+    return policy.get(SMALL_CELL_THRESHOLD_SETTINGS_KEY)
+
+
+def _configured_threshold(value: object, *, source: str, default: int | None = None) -> int:
+    if value is None or (isinstance(value, str) and not value.strip()):
+        if default is not None:
+            return default
+        raise CampaignReportPrivacyPolicyError(f"{source} must be configured as an integer")
+    if isinstance(value, bool):
+        raise CampaignReportPrivacyPolicyError(f"{source} must be an integer, not a boolean")
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError) as exc:
+        raise CampaignReportPrivacyPolicyError(f"{source} must be an integer") from exc
+    if str(parsed) != str(value).strip() and not isinstance(value, int):
+        raise CampaignReportPrivacyPolicyError(f"{source} must be an integer")
+    if parsed < MIN_SMALL_CELL_THRESHOLD or parsed > MAX_SMALL_CELL_THRESHOLD:
+        raise CampaignReportPrivacyPolicyError(
+            f"{source} must be between {MIN_SMALL_CELL_THRESHOLD} and {MAX_SMALL_CELL_THRESHOLD}"
+        )
+    return parsed