govoplan-core/src/govoplan_core/privacy/retention.py

"""Compatibility facade for privacy retention policy.

Policy-owned behavior is provided by ``govoplan-policy`` through the
``policy.privacyRetention`` capability. This module keeps legacy imports stable
without importing policy implementation code from core.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Mapping

from govoplan_core.core.policy import CAPABILITY_POLICY_PRIVACY_RETENTION, PrivacyRetentionService
from govoplan_core.core.runtime import get_registry
from govoplan_core.privacy.schemas import (
    RETENTION_DAY_KEYS as RETENTION_DAY_KEYS,
    RETENTION_POLICY_FIELD_KEYS,
    default_allow_lower_level_limits as default_allow_lower_level_limits,
    normalize_allow_lower_level_limits,
)


PRIVACY_POLICY_SETTINGS_KEY = "privacy_retention_policy"
_DEFAULT_POLICY = {
    "store_raw_campaign_json": True,
    "raw_campaign_json_retention_days": None,
    "generated_eml_retention_days": None,
    "stored_report_detail_retention_days": None,
    "mock_mailbox_retention_days": None,
    "audit_detail_retention_days": None,
    "audit_detail_level": "full",
}

_EXPORTS = (
    "PRIVACY_POLICY_SETTINGS_KEY",
    "RETENTION_DAY_KEYS",
    "RETENTION_POLICY_FIELD_KEYS",
    "PrivacyRetentionPolicy",
    "PrivacyRetentionPolicyPatch",
    "PrivacyPolicyError",
    "default_allow_lower_level_limits",
    "privacy_policy_from_settings",
    "privacy_policy_from_session",
    "set_privacy_policy",
    "effective_privacy_policy",
    "parent_privacy_policy",
    "effective_privacy_policy_sources",
    "parent_privacy_policy_sources",
    "get_privacy_policy_for_scope",
    "set_privacy_policy_for_scope",
    "sanitize_audit_details_for_policy",
    "apply_retention_policy",
)


class PrivacyPolicyError(RuntimeError):
    pass


class _UnavailablePolicyModel:
    def __init__(self, *args: Any, **kwargs: Any) -> None:
        del args, kwargs
        _raise_unavailable()

    @classmethod
    def model_validate(cls, value: Any) -> Any:
        del value
        _raise_unavailable()


PrivacyRetentionPolicy = _UnavailablePolicyModel
PrivacyRetentionPolicyPatch = _UnavailablePolicyModel


@dataclass(frozen=True, slots=True)
class _SystemPrivacyPolicy:
    payload: Mapping[str, Any]

    def model_dump(self, *args: Any, **kwargs: Any) -> dict[str, Any]:
        del args, kwargs
        return dict(self.payload)


def _raise_unavailable() -> None:
    raise PrivacyPolicyError("Privacy retention policy requires the active govoplan-policy module.")


def _runtime_service() -> PrivacyRetentionService | None:
    registry = get_registry()
    if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(CAPABILITY_POLICY_PRIVACY_RETENTION):
        return None
    capability = registry.require_capability(CAPABILITY_POLICY_PRIVACY_RETENTION)
    if not isinstance(capability, PrivacyRetentionService):
        raise PrivacyPolicyError("Privacy retention policy capability is invalid")
    return capability


def _service() -> PrivacyRetentionService:
    service = _runtime_service()
    if service is None:
        _raise_unavailable()
    return service


def _settings_payload(item: object) -> dict[str, Any]:
    settings = getattr(item, "settings", None)
    return dict(settings or {}) if isinstance(settings, Mapping) else {}


def _policy_data(settings_payload: Mapping[str, Any] | None) -> dict[str, Any]:
    raw = settings_payload.get(PRIVACY_POLICY_SETTINGS_KEY) if isinstance(settings_payload, Mapping) else None
    if not isinstance(raw, Mapping):
        raw = {}
    payload = dict(_DEFAULT_POLICY)
    for key in RETENTION_POLICY_FIELD_KEYS:
        if key in raw:
            payload[key] = raw[key]
    payload["allow_lower_level_limits"] = normalize_allow_lower_level_limits(raw.get("allow_lower_level_limits"), fill_defaults=True)
    return payload


def privacy_policy_from_settings(item: object) -> Any:
    service = _runtime_service()
    if service is not None:
        return service.privacy_policy_from_settings(item)
    return _SystemPrivacyPolicy(_policy_data(_settings_payload(item)))


def privacy_policy_from_session(*args: Any, **kwargs: Any) -> Any:
    return _service().privacy_policy_from_session(*args, **kwargs)


def set_privacy_policy(item: object, policy: Any) -> Any:
    service = _runtime_service()
    if service is not None:
        return service.set_privacy_policy(item, policy)
    value = policy.model_dump(mode="json") if hasattr(policy, "model_dump") else dict(policy or {})
    next_settings = _settings_payload(item)
    next_settings[PRIVACY_POLICY_SETTINGS_KEY] = _policy_data({PRIVACY_POLICY_SETTINGS_KEY: value})
    setattr(item, "settings", next_settings)
    return _SystemPrivacyPolicy(next_settings[PRIVACY_POLICY_SETTINGS_KEY])


def effective_privacy_policy(*args: Any, **kwargs: Any) -> Any:
    return _service().effective_privacy_policy(*args, **kwargs)


def parent_privacy_policy(*args: Any, **kwargs: Any) -> Any:
    return _service().parent_privacy_policy(*args, **kwargs)


def effective_privacy_policy_sources(*args: Any, **kwargs: Any) -> Any:
    return _service().effective_privacy_policy_sources(*args, **kwargs)


def parent_privacy_policy_sources(*args: Any, **kwargs: Any) -> Any:
    return _service().parent_privacy_policy_sources(*args, **kwargs)


def get_privacy_policy_for_scope(*args: Any, **kwargs: Any) -> Any:
    return _service().get_privacy_policy_for_scope(*args, **kwargs)


def set_privacy_policy_for_scope(*args: Any, **kwargs: Any) -> Any:
    return _service().set_privacy_policy_for_scope(*args, **kwargs)


def sanitize_audit_details_for_policy(*args: Any, **kwargs: Any) -> dict[str, Any]:
    service = _runtime_service()
    if service is not None:
        return dict(service.sanitize_audit_details_for_policy(*args, **kwargs))
    details = args[1] if len(args) > 1 else kwargs.get("details")
    return dict(details or {})


def apply_retention_policy(*args: Any, **kwargs: Any) -> Any:
    return _service().apply_retention_policy(*args, **kwargs)


__all__ = list(_EXPORTS)