Privacy Engineering Architecture

import asyncio
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum

class DataProcessingPurpose(Enum):
    ANALYTICS = "analytics"
    MARKETING = "marketing"
    PERSONALIZATION = "personalization"
    SECURITY = "security"
    OPERATIONS = "operations"

class ConsentStatus(Enum):
    GRANTED = "granted"
    WITHDRAWN = "withdrawn"
    EXPIRED = "expired"
    PENDING = "pending"

@dataclass
class DataSubject:
    subject_id: str
    consent_grants: Dict[DataProcessingPurpose, ConsentStatus]
    consent_timestamps: Dict[DataProcessingPurpose, datetime]
    retention_preferences: Dict[str, int]  # days
    data_categories: List[str]

@dataclass
class PrivacyPolicy:
    purpose: DataProcessingPurpose
    max_retention_days: int
    requires_explicit_consent: bool
    lawful_basis: str
    data_minimization_rules: List[str]

class PrivacyComplianceEngine:
    def __init__(self):
        self.data_subjects = {}
        self.privacy_policies = {}
        self.processing_logs = []
        self.retention_scheduler = RetentionScheduler()

    async def validate_processing_request(
        self,
        subject_id: str,
        purpose: DataProcessingPurpose,
        data_fields: List[str]
    ) -> Dict:
        """Validate if data processing is allowed under privacy policies"""

        subject = self.data_subjects.get(subject_id)
        if not subject:
            return {
                'allowed': False,
                'reason': 'Subject not found',
                'compliance_status': 'violation'
            }

        policy = self.privacy_policies.get(purpose)
        if not policy:
            return {
                'allowed': False,
                'reason': 'No privacy policy for purpose',
                'compliance_status': 'violation'
            }

        # Check consent requirements
        consent_status = subject.consent_grants.get(purpose)
        if policy.requires_explicit_consent:
            if consent_status != ConsentStatus.GRANTED:
                return {
                    'allowed': False,
                    'reason': f'Missing consent for {purpose.value}',
                    'compliance_status': 'consent_violation',
                    'required_action': 'obtain_consent'
                }

            # Check consent expiry (GDPR Article 7.3)
            consent_time = subject.consent_timestamps.get(purpose)
            if consent_time and (datetime.utcnow() - consent_time).days > 365:
                return {
                    'allowed': False,
                    'reason': 'Consent expired - refresh required',
                    'compliance_status': 'consent_expired',
                    'required_action': 'refresh_consent'
                }

        # Apply data minimization (GDPR Article 5.1.c)
        allowed_fields = self.apply_data_minimization(
            data_fields, purpose, policy.data_minimization_rules
        )

        if len(allowed_fields) < len(data_fields):
            filtered_fields = set(data_fields) - set(allowed_fields)
            return {
                'allowed': True,
                'filtered_fields': list(filtered_fields),
                'allowed_fields': allowed_fields,
                'compliance_status': 'compliant_with_minimization',
                'reason': 'Data minimization applied'
            }

        return {
            'allowed': True,
            'compliance_status': 'compliant',
            'retention_deadline': self.calculate_retention_deadline(subject_id, purpose)
        }

    def apply_data_minimization(
        self,
        requested_fields: List[str],
        purpose: DataProcessingPurpose,
        minimization_rules: List[str]
    ) -> List[str]:
        """Apply data minimization principles to field selection"""

        # Purpose limitation - only allow relevant fields
        purpose_field_map = {
            DataProcessingPurpose.ANALYTICS: [
                'user_id', 'session_id', 'event_type', 'timestamp',
                'page_url', 'user_agent', 'country'
            ],
            DataProcessingPurpose.MARKETING: [
                'email', 'preferences', 'segments', 'campaign_history'
            ],
            DataProcessingPurpose.PERSONALIZATION: [
                'user_id', 'preferences', 'behavior_history', 'demographics'
            ],
            DataProcessingPurpose.SECURITY: [
                'ip_address', 'device_fingerprint', 'login_history',
                'security_events', 'risk_score'
            ]
        }

        allowed_base_fields = purpose_field_map.get(purpose, [])

        # Apply custom minimization rules
        filtered_fields = []
        for field in requested_fields:
            if field in allowed_base_fields:
                # Apply additional rule-based filtering
                if self.passes_minimization_rules(field, minimization_rules):
                    filtered_fields.append(field)

        return filtered_fields

    def passes_minimization_rules(self, field: str, rules: List[str]) -> bool:
        """Check if field passes data minimization rules"""
        for rule in rules:
            if rule.startswith('exclude:') and field in rule.replace('exclude:', ''):
                return False
            elif rule.startswith('require_consent:') and field in rule.replace('require_consent:', ''):
                # Additional consent check would be performed here
                pass
        return True

    async def process_data_subject_request(
        self,
        subject_id: str,
        request_type: str
    ) -> Dict:
        """Handle GDPR Articles 15-21 data subject rights"""

        subject = self.data_subjects.get(subject_id)
        if not subject:
            return {'status': 'error', 'message': 'Subject not found'}

        if request_type == 'access':  # Article 15
            return await self.handle_access_request(subject_id)
        elif request_type == 'rectification':  # Article 16
            return await self.handle_rectification_request(subject_id)
        elif request_type == 'erasure':  # Article 17 (Right to be forgotten)
            return await self.handle_erasure_request(subject_id)
        elif request_type == 'portability':  # Article 20
            return await self.handle_portability_request(subject_id)
        elif request_type == 'object':  # Article 21
            return await self.handle_objection_request(subject_id)

        return {'status': 'error', 'message': 'Unknown request type'}

    async def handle_erasure_request(self, subject_id: str) -> Dict:
        """Right to be forgotten implementation"""

        # Check if erasure is legally required
        erasure_blockers = []

        # Legal obligation check (GDPR Article 17.3.b)
        if self.has_legal_retention_obligation(subject_id):
            erasure_blockers.append('legal_retention_requirement')

        # Freedom of expression check (GDPR Article 17.3.a)
        if self.involves_freedom_of_expression(subject_id):
            erasure_blockers.append('freedom_of_expression')

        if erasure_blockers:
            return {
                'status': 'partially_fulfilled',
                'blockers': erasure_blockers,
                'action_taken': 'Data processing restricted where legally permissible'
            }

        # Perform cascading deletion across all systems
        deletion_tasks = [
            self.delete_from_primary_database(subject_id),
            self.delete_from_analytics_systems(subject_id),
            self.delete_from_backup_systems(subject_id),
            self.delete_from_cdn_logs(subject_id),
            self.notify_data_processors(subject_id, 'erasure')
        ]

        results = await asyncio.gather(*deletion_tasks, return_exceptions=True)

        # Log compliance action
        self.log_compliance_action(subject_id, 'erasure', results)

        return {
            'status': 'completed',
            'deletion_results': results,
            'compliance_certificate': self.generate_erasure_certificate(subject_id)
        }

class RetentionScheduler:
    """Automated data retention policy enforcement"""

    def __init__(self):
        self.retention_policies = {}
        self.scheduled_deletions = []

    async def schedule_retention_cleanup(self):
        """Daily job to enforce retention policies"""

        current_time = datetime.utcnow()

        for policy_name, policy in self.retention_policies.items():
            # Find data past retention period
            expired_data = await self.find_expired_data(policy)

            for data_record in expired_data:
                # Grace period check
                if self.in_grace_period(data_record, policy):
                    continue

                # Legal hold check
                if self.has_legal_hold(data_record):
                    continue

                # Schedule for deletion
                await self.schedule_deletion(data_record, policy)

    async def find_expired_data(self, policy: Dict) -> List:
        """Find data that has exceeded retention period"""
        # Implementation would query various data stores
        # using the retention policy criteria
        pass

    def generate_retention_report(self) -> Dict:
        """Generate compliance report for retention policies"""
        return {
            'total_policies': len(self.retention_policies),
            'scheduled_deletions': len(self.scheduled_deletions),
            'compliance_rate': self.calculate_compliance_rate(),
            'storage_saved_gb': self.calculate_storage_savings(),
        }

import React, { useState, useEffect } from 'react';
import { BarChart, Bar, LineChart, Line, XAxis, YAxis, CartesianGrid,
         Tooltip, ResponsiveContainer, PieChart, Pie, Cell } from 'recharts';

interface ComplianceMetrics {
  framework: string;
  overallScore: number;
  requirements: {
    category: string;
    score: number;
    status: 'compliant' | 'warning' | 'violation';
    findings: string[];
  }[];
}

interface PrivacyIncident {
  id: string;
  type: 'consent_violation' | 'data_breach' | 'retention_violation' | 'access_violation';
  severity: 'low' | 'medium' | 'high' | 'critical';
  timestamp: string;
  description: string;
  affectedSubjects: number;
  status: 'open' | 'investigating' | 'resolved';
}

export const PrivacyComplianceDashboard: React.FC = () => {
  const [complianceMetrics, setComplianceMetrics] = useState<ComplianceMetrics[]>([]);
  const [incidents, setIncidents] = useState<PrivacyIncident[]>([]);
  const [timeRange, setTimeRange] = useState<'24h' | '7d' | '30d' | '90d'>('30d');

  const fetchComplianceData = async () => {
    // Fetch compliance data from APIs
    const [complianceRes, incidentsRes] = await Promise.all([
      fetch(`/api/privacy/compliance?range=${timeRange}`),
      fetch(`/api/privacy/incidents?range=${timeRange}`)
    ]);

    setComplianceMetrics(await complianceRes.json());
    setIncidents(await incidentsRes.json());
  };

  const overallComplianceScore = complianceMetrics.reduce(
    (avg, metric) => avg + metric.overallScore, 0
  ) / (complianceMetrics.length || 1);

  return (
    <div className="p-6 space-y-6">
      <div className="grid grid-cols-1 md:grid-cols-3 gap-6">
        <div className="bg-white rounded-lg p-6 shadow-lg">
          <h3 className="text-sm font-medium text-gray-500">Overall Compliance</h3>
          <p className="text-2xl font-bold text-green-600">
            {overallComplianceScore.toFixed(1)}%
          </p>
        </div>

        <div className="bg-white rounded-lg p-6 shadow-lg">
          <h3 className="text-sm font-medium text-gray-500">Open Incidents</h3>
          <p className="text-2xl font-bold text-red-600">
            {incidents.filter(i => i.status !== 'resolved').length}
          </p>
        </div>

        <div className="bg-white rounded-lg p-6 shadow-lg">
          <h3 className="text-sm font-medium text-gray-500">Critical Issues</h3>
          <p className="text-2xl font-bold text-orange-600">
            {incidents.filter(i => i.severity === 'critical').length}
          </p>
        </div>
      </div>

      <div className="bg-white rounded-lg p-6 shadow-lg">
        <h3 className="text-lg font-semibold mb-4">Compliance Framework Scores</h3>
        <ResponsiveContainer width="100%" height={300}>
          <BarChart data={complianceMetrics}>
            <CartesianGrid strokeDasharray="3 3" />
            <XAxis dataKey="framework" />
            <YAxis />
            <Tooltip />
            <Bar dataKey="overallScore" fill="#8b5cf6" />
          </BarChart>
        </ResponsiveContainer>
      </div>
    </div>
  );
};