PyPaC_engine.py

# PyPaC_engine.py (WIP/draft) embeds privacy compliance into Python applications
# Not ready for production

import datetime
from typing import Any, Dict, List, Optional, Callable

class PrivacyViolation(Exception):
    pass

class AccessLogEntry:
    def __init__(self, field: str, actor: str, purpose: str, processor: str):
        self.timestamp = datetime.datetime.now()
        self.field = field
        self.actor = actor
        self.purpose = purpose
        self.processor = processor

    def __repr__(self):
        return f"[{self.timestamp}] Field='{self.field}' Actor='{self.actor}' Purpose='{self.purpose}' Processor='{self.processor}'"

class PrivacyField:
    def __init__(
            self,
            value: Any,
            purpose: Optional[str] = None,
            retention_days: Optional[int] = None,
            consent_required: bool = False,
            access_roles: Optional[List[str]] = None,
            redact_after_expiry: bool = False,
            allowed_processors: Optional[List[str]] = None,
            mask_fn: Optional[Callable[[Any], str]] = None,
    ):
        self.value = value
        self.purpose = purpose
        self.retention_days = retention_days
        self.consent_required = consent_required
        self.access_roles = access_roles or []
        self.created_at = datetime.datetime.now()
        self.redact_after_expiry = redact_after_expiry
        self.allowed_processors = allowed_processors or []
        self.mask_fn = mask_fn

    def is_expired(self) -> bool:
        if self.retention_days is None:
            return False
        return (datetime.datetime.now() - self.created_at).days > self.retention_days

    def get_value(self, role: Optional[str], consent_given: bool, purpose: Optional[str], processor: Optional[str]) -> Any:
        if self.allowed_processors and (processor not in self.allowed_processors):
            raise PrivacyViolation(f"Processor '{processor}' not allowed for this field")

        if self.consent_required and not consent_given:
            raise PrivacyViolation("Consent required")

        if self.purpose and self.purpose != purpose:
            raise PrivacyViolation(f"Purpose mismatch: expected {self.purpose}, got {purpose}")

        if role and self.access_roles and role not in self.access_roles:
            raise PrivacyViolation(f"Access denied for role: {role}")

        if self.is_expired():
            if self.redact_after_expiry:
                return "[REDACTED]"
            raise PrivacyViolation("Data expired")

        if role and self.access_roles and self.mask_fn and role not in self.access_roles:
            return self.mask_fn(self.value)

        return self.value

class PrivacyModel:
    def __init__(self, user_id: str, data: Dict[str, PrivacyField]):
        self.user_id = user_id
        self.data = data
        self.access_log: List[AccessLogEntry] = []
        self.user_requests: List[str] = []

    def get(self, key: str, role: Optional[str] = None, consent_given: bool = False,
            purpose: Optional[str] = None, processor: Optional[str] = None, actor: Optional[str] = "unknown"):
        field = self.data.get(key)
        if not field:
            raise KeyError(f"Field '{key}' not found")

        value = field.get_value(role, consent_given, purpose, processor)
        self.access_log.append(AccessLogEntry(field=key, actor=actor, purpose=purpose or "unspecified", processor=processor or "unspecified"))
        return value

    def export_user_data(self) -> Dict[str, Any]:
        return {
            k: (v.value if not v.is_expired() else "[EXPIRED]")
            for k, v in self.data.items()
        }

    def delete_expired_fields(self):
        self.data = {
            k: v for k, v in self.data.items()
            if not v.is_expired() or v.redact_after_expiry
        }

    def redact_all(self):
        for field in self.data.values():
            field.value = "[REDACTED]"

    def purge_user_data(self):
        self.data.clear()

    def record_request(self, request_type: str):
        timestamp = datetime.datetime.now().isoformat()
        self.user_requests.append(f"{timestamp} - {request_type}")

    def get_access_log(self) -> List[AccessLogEntry]:
        return self.access_log

    def get_request_log(self) -> List[str]:
        return self.user_requests