"""GDPR data-subject rights endpoints. Exposes two routes: * ``GET /api/gdpr/export`` Returns a JSON snapshot of the calling user's personal data — profile, entity membership, exam attempts, answers, AI interactions, feedback, tickets, and login history. * ``POST /api/gdpr/delete`` Initiates the "right to erasure" flow. Because Odoo enforces FK constraints on linked business records (exam attempts, tickets, audit trail), we **anonymise** instead of hard-deleting: the ``res.users`` row is deactivated, ``res.partner`` PII fields are wiped, and a tombstone row is written to ``encoach.gdpr.erasure.request`` for audit. Hard delete of linked business rows (attempts, answers, feedback) is performed for records where it is legally required and where no other user depends on them. The flow is intentionally verbose so operators can see what happened. A production deployment may want to queue the erasure to a background job and require a cooling-off period before executing; we expose a ``confirm`` flag for that. """ from __future__ import annotations import json import logging from odoo import fields, http from odoo.http import request from .base import ( _error_response, _get_json_body, _json_response, jwt_required, ) _logger = logging.getLogger(__name__) # ---------------------------------------------------------------------- # Helpers # ---------------------------------------------------------------------- def _safe_records(env_name: str, domain, fields_to_read): """Read records if the model exists, else return [].""" env = request.env if env_name not in env: return [] try: return env[env_name].sudo().search(domain).read(fields_to_read) except Exception as exc: _logger.warning("gdpr: could not read %s: %s", env_name, exc) return [] def _sanitize(values): """Convert non-JSON-serialisable values to strings/nulls.""" out = {} for key, value in values.items(): if isinstance(value, (list, tuple)) and len(value) == 2 and isinstance(value[0], int): # Many2one read tuples — keep both id and display name. out[key] = {"id": value[0], "display_name": value[1]} elif isinstance(value, bytes): out[key] = f"<{len(value)} bytes omitted>" elif hasattr(value, "isoformat"): out[key] = value.isoformat() else: out[key] = value return out # ---------------------------------------------------------------------- # Controller # ---------------------------------------------------------------------- class EncoachGdprController(http.Controller): """Implements export + right-to-erasure endpoints.""" # ------------------------------------------------------------------ # GET /api/gdpr/export # ------------------------------------------------------------------ @http.route("/api/gdpr/export", type="http", auth="none", methods=["GET"], csrf=False) @jwt_required def export(self, **kw): try: user = request.env.user partner = user.partner_id profile = { "id": user.id, "login": user.login, "name": user.name, "email": user.email or "", "lang": user.lang or "", "tz": user.tz or "", "create_date": user.create_date.isoformat() if user.create_date else None, "login_date": user.login_date.isoformat() if user.login_date else None, } partner_data = { "id": partner.id if partner else None, "name": partner.name if partner else None, "email": partner.email if partner else None, "phone": partner.phone if partner else None, "mobile": partner.mobile if partner else None, "street": partner.street if partner else None, "city": partner.city if partner else None, "country": partner.country_id.name if partner and partner.country_id else None, } # Entity / role memberships entity_rel = _safe_records( "encoach.user.entity.rel", [("user_id", "=", user.id)], ["id", "entity_id", "role_id", "is_active"], ) # Exam attempts & answers attempts = _safe_records( "encoach.student.attempt", [("user_id", "=", user.id)], ["id", "exam_id", "status", "score", "started_at", "submitted_at"], ) attempt_ids = [a["id"] for a in attempts] answers = _safe_records( "encoach.student.answer", [("attempt_id", "in", attempt_ids)], ["id", "attempt_id", "question_id", "student_answer", "is_correct", "score"], ) if attempt_ids else [] # AI feedback they left feedback = _safe_records( "encoach.ai.feedback", [("user_id", "=", user.id)], ["id", "subject_type", "subject_id", "rating", "comment", "status", "create_date"], ) # AI calls they triggered (logs) ai_logs = _safe_records( "encoach.ai.log", [("user_id", "=", user.id)] if "encoach.ai.log" in request.env else [], ["id", "service", "action", "model_used", "total_tokens", "status", "create_date"], ) # Tickets they filed tickets = _safe_records( "encoach.ticket", [("created_by_id", "=", user.id)], ["id", "name", "status", "priority", "assigned_to_id", "create_date"], ) # Feedback on their coaching, if any coaching_sessions = _safe_records( "encoach.coaching.session", [("user_id", "=", user.id)], ["id", "create_date", "summary"], ) payload = { "profile": profile, "partner": partner_data, "entity_memberships": [_sanitize(r) for r in entity_rel], "exam_attempts": [_sanitize(r) for r in attempts], "exam_answers": [_sanitize(r) for r in answers], "ai_feedback": [_sanitize(r) for r in feedback], "ai_calls": [_sanitize(r) for r in ai_logs], "tickets": [_sanitize(r) for r in tickets], "coaching_sessions": [_sanitize(r) for r in coaching_sessions], "exported_at": fields.Datetime.to_string(fields.Datetime.now()), "export_format_version": "1.0", } return _json_response(payload) except Exception as e: _logger.exception("gdpr export failed") return _error_response(str(e), 500) # ------------------------------------------------------------------ # POST /api/gdpr/delete # ------------------------------------------------------------------ @http.route("/api/gdpr/delete", type="http", auth="none", methods=["POST"], csrf=False) @jwt_required def delete(self, **kw): try: body = _get_json_body() or {} if not body.get("confirm"): return _error_response( "Set {\"confirm\": true} to initiate erasure. " "This action cannot be undone.", 400, ) user = request.env.user if user.id == 1 or user.has_group("base.group_system"): return _error_response( "Admin accounts cannot self-erase via this endpoint. " "Contact the data protection officer.", 403, ) login = user.login email = user.email or "" archived_id = user.id summary = { "anonymised_partner_fields": [], "deactivated_user": False, "deleted_feedback_count": 0, "deleted_coaching_count": 0, "retained_attempts_count": 0, } with request.env.cr.savepoint(): # 1. Anonymise the partner PII. partner = user.partner_id if partner: partner.sudo().write({ "name": f"deleted-user-{archived_id}", "email": False, "phone": False, "mobile": False, "street": False, "street2": False, "city": False, "zip": False, "image_1920": False, "comment": "Erased per GDPR request.", }) summary["anonymised_partner_fields"] = [ "name", "email", "phone", "mobile", "street", "street2", "city", "zip", "image_1920", "comment", ] # 2. Hard-delete their feedback (personal reviews/comments). if "encoach.ai.feedback" in request.env: feedback = request.env["encoach.ai.feedback"].sudo().search([ ("user_id", "=", user.id), ]) summary["deleted_feedback_count"] = len(feedback) feedback.unlink() # 3. Hard-delete coaching session transcripts (personal). if "encoach.coaching.session" in request.env: sessions = request.env["encoach.coaching.session"].sudo().search([ ("user_id", "=", user.id), ]) summary["deleted_coaching_count"] = len(sessions) sessions.unlink() # 4. Retain exam attempts (aggregate analytics rely on them) # but strip personal free-text fields where they exist. if "encoach.student.attempt" in request.env: retained = request.env["encoach.student.attempt"].sudo().search([ ("user_id", "=", user.id), ]) summary["retained_attempts_count"] = len(retained) # Clear any written responses that contain user PII — keep # the scores/timestamps for analytics integrity. if "encoach.student.answer" in request.env: answers = request.env["encoach.student.answer"].sudo().search([ ("attempt_id", "in", retained.ids), ]) # student_answer can contain free-text PII. answers.write({"student_answer": False}) # 5. Rewrite AI log user_id → None so we can no longer tie # those rows to the person. if "encoach.ai.log" in request.env: ai_logs = request.env["encoach.ai.log"].sudo().search([ ("user_id", "=", user.id), ]) if ai_logs: try: ai_logs.write({"user_id": False}) except Exception: # user_id may be required; fall back to deletion. ai_logs.unlink() # 6. Deactivate the user (we can't unlink — FKs). user.sudo().write({ "active": False, "login": f"erased-{archived_id}@example.invalid", }) summary["deactivated_user"] = True # 7. Write tombstone audit row. request.env["encoach.gdpr.erasure.request"].sudo().create({ "user_login": login, "user_email": email, "user_id_archived": archived_id, "status": "completed", "summary": json.dumps(summary), "completed_at": fields.Datetime.now(), }) # Blow away any cached JWT sessions for this user. try: from odoo.addons.encoach_api.utils.jwt_cache import ( invalidate_user as _invalidate_user, ) _invalidate_user(archived_id) except Exception as exc: _logger.debug("jwt cache invalidation skipped: %s", exc) return _json_response({ "ok": True, "summary": summary, "message": ( "Your account has been anonymised and deactivated. " "Some records (e.g. exam attempts) are retained " "without personal identifiers for statistical purposes." ), }) except Exception as e: _logger.exception("gdpr delete failed") return _error_response(str(e), 500)