Odoo-Modules/fusion_repairs/models/repair_ai_service.py

# -*- coding: utf-8 -*-
# Copyright 2024-2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)

"""Repair AI Service - single guardrailed entry point for client-portal AI.

Per the design spec (Appendix A), this AbstractModel:
1) Builds a strict system prompt forbidding medical advice, diagnoses,
   stop-using recommendations, etc.
2) Calls fusion.api.service.call_openai() if available (try/fallback per
   fusion-api-integration rule - never installs as a hard dep)
3) JSON-schema validates the response and runs a forbidden-phrase regex
4) Always falls back to deterministic fusion.repair.self.check.rule
   records on any failure - intake must never be blocked by AI

System prompt + JSON schema live in ir.config_parameter so the office can
refine them without code changes.
"""

import hashlib
import json
import logging
import re

from odoo import api, fields, models

_logger = logging.getLogger(__name__)


# ----- Safety filters -----
FORBIDDEN_PATTERNS = [
    re.compile(r'\b(diagnos(e|is|ed|ing))\b', re.I),
    re.compile(r'\byou have\b', re.I),
    re.compile(r'\bmedical condition\b', re.I),
    re.compile(r'\bstop using\b', re.I),
    re.compile(r'\bconsult\s+(your|a)\s+(doctor|physician|nurse)\b', re.I),
    re.compile(r'(\$|CAD|USD)\s?\d+', re.I),  # No price mentions
]

# Categories where motor/safety symptoms always escalate without asking AI.
SAFETY_CATEGORY_CODES = ('stairlift', 'porch_lift')
SAFETY_SYMPTOMS = (
    'smoke', 'burning', 'spark', 'fire', 'stuck', 'trapped',
    'motor', 'brake fail', "won't stop", 'overshoot',
)


DEFAULT_SYSTEM_PROMPT = (
    "You are a triage assistant for Fusion Repairs, a Canadian medical "
    "equipment service company. Your ONLY job is to suggest 1-3 safe, "
    "reversible self-check steps a client can try on their medical equipment "
    "before scheduling a technician visit.\n\n"
    "ABSOLUTE RULES:\n"
    "1. NEVER provide medical advice, diagnoses, or health recommendations.\n"
    "2. NEVER claim a definitive cause for the problem.\n"
    "3. NEVER recommend stopping use of medical equipment.\n"
    "4. NEVER use phrases like 'you have', 'I diagnose', 'you should stop', "
    "'medical condition', 'consult your doctor'.\n"
    "5. ONLY suggest steps that are: safe, reversible, require no tools, "
    "take under 2 minutes, and pose zero risk to the client or equipment.\n"
    "6. If symptoms involve smoke, sparks, burning smell, motors on "
    "stairlifts/porch lifts, OR if you are uncertain -> return "
    "escalate_immediately: true.\n"
    "7. Maximum 3 steps. Each step <= 1 sentence. Grade-6 reading level. "
    "No technical jargon.\n"
    "8. NEVER reference part numbers, prices, or other clients.\n"
    "9. If client reports injury, equipment fire, or person trapped -> "
    "escalate_immediately: true with escalation_reason: 'emergency'.\n"
    "10. You MUST output valid JSON matching the provided schema. No prose, "
    "no markdown, no commentary."
)


class FusionRepairAIService(models.AbstractModel):
    _name = 'fusion.repair.ai.service'
    _description = 'Repair AI Service - guardrailed self-check engine'

    # ------------------------------------------------------------------
    # PUBLIC API
    # ------------------------------------------------------------------
    @api.model
    def suggest_self_check(self, product_category_id=None, symptoms=None, urgency=None):
        """Return a list of safe self-check steps for the client portal.

        Returns a dict with shape:
          {
            'escalate_immediately': bool,
            'escalation_reason': str | None,
            'confidence': 'high' | 'medium' | 'low',
            'steps': [{'instruction': str, 'expected_result': str,
                       'safety_note': str | None}, ...],
            'source': 'ai' | 'fallback' | 'escalated',
            'disclaimer': str,
          }
        """
        symptoms = [s for s in (symptoms or []) if s]
        category = (
            self.env['fusion.repair.product.category'].sudo().browse(product_category_id)
            if product_category_id else False
        )

        # Pre-check: hard-escalate for safety-critical category + symptom combos
        # without consulting AI. This is BEFORE any AI call so even if AI is
        # down we still escalate the right way.
        if self._should_hard_escalate(category, symptoms, urgency):
            return self._escalated_response('safety')

        # Try the AI, fall back to deterministic rules on any failure.
        ai_result = self._try_ai(category, symptoms)
        if ai_result:
            ai_result['source'] = 'ai'
            ai_result['disclaimer'] = self._disclaimer()
            return ai_result

        return self._deterministic_fallback(category, symptoms)

    # ------------------------------------------------------------------
    # HARD ESCALATION
    # ------------------------------------------------------------------
    @api.model
    def _should_hard_escalate(self, category, symptoms, urgency):
        if urgency == 'safety':
            return True
        text = ' '.join(symptoms).lower()
        if category and category.code in SAFETY_CATEGORY_CODES:
            if any(kw in text for kw in SAFETY_SYMPTOMS):
                return True
        # Anyone reporting fire / injury / trapped person, regardless of category.
        if any(kw in text for kw in ('fire', 'injury', 'hurt', 'bleeding', 'trapped')):
            return True
        return False

    @api.model
    def _escalated_response(self, reason):
        return {
            'escalate_immediately': True,
            'escalation_reason': reason,
            'confidence': 'high',
            'steps': [],
            'source': 'escalated',
            'disclaimer': self._disclaimer(),
        }

    # ------------------------------------------------------------------
    # AI CALL (try/fallback)
    # ------------------------------------------------------------------
    @api.model
    def _try_ai(self, category, symptoms):
        try:
            ApiService = self.env.get('fusion.api.service')
            if not ApiService:
                return None
            messages = [
                {'role': 'system', 'content': self._system_prompt()},
                {'role': 'user', 'content': self._user_prompt(category, symptoms)},
            ]
            cache_key = self._cache_key(category, symptoms)
            cached = self._cache_get(cache_key)
            if cached:
                return cached

            raw = ApiService.call_openai(
                consumer='fusion_repairs',
                feature='client_self_triage',
                messages=messages,
                max_tokens=400,
            )
            if not raw:
                return None
            parsed = self._safe_parse(raw)
            if not parsed:
                self._log_incident('parse_failed', raw)
                return None
            self._cache_set(cache_key, parsed)
            return parsed
        except Exception as e:
            _logger.info('AI self-check skipped: %s', e)
            return None

    @api.model
    def _system_prompt(self):
        ICP = self.env['ir.config_parameter'].sudo()
        return ICP.get_param(
            'fusion_repairs.ai_self_check_system_prompt',
            DEFAULT_SYSTEM_PROMPT,
        )

    @api.model
    def _user_prompt(self, category, symptoms):
        cat_name = category.name if category else 'medical equipment'
        return (
            f"Equipment category: {cat_name}\n"
            f"Reported symptoms: {'; '.join(symptoms) if symptoms else '(none provided)'}\n"
            "Output the JSON object only."
        )

    # ------------------------------------------------------------------
    # SAFE PARSE + VALIDATE
    # ------------------------------------------------------------------
    @api.model
    def _safe_parse(self, raw):
        """Parse the AI response, validate against the JSON schema, and run
        forbidden-phrase regex filters. Returns None on any failure - caller
        falls back to deterministic rules."""
        if not raw:
            return None
        text = raw.strip()
        # Strip code-fence wrapping if AI added it.
        if text.startswith('```'):
            text = re.sub(r'^```[a-zA-Z]*\n?', '', text)
            text = re.sub(r'\n?```$', '', text)
        try:
            data = json.loads(text)
        except (ValueError, TypeError):
            return None
        # Schema check (minimal - we don't pull in jsonschema as a dep)
        if not isinstance(data, dict):
            return None
        if not isinstance(data.get('escalate_immediately'), bool):
            return None
        confidence = data.get('confidence')
        if confidence not in ('high', 'medium', 'low'):
            return None
        steps = data.get('steps')
        if not isinstance(steps, list) or len(steps) > 3:
            return None
        # Coherence: not-escalated must have at least one step.
        if not data['escalate_immediately'] and not steps:
            return None
        # Per-step validation + forbidden-phrase scan.
        cleaned_steps = []
        for step in steps:
            if not isinstance(step, dict):
                return None
            instr = step.get('instruction')
            expected = step.get('expected_result')
            if not isinstance(instr, str) or not instr.strip():
                return None
            if not isinstance(expected, str) or not expected.strip():
                return None
            if len(instr) > 200 or len(expected) > 200:
                return None
            if self._contains_forbidden(instr) or self._contains_forbidden(expected):
                return None
            note = step.get('safety_note')
            if note is not None and (not isinstance(note, str) or len(note) > 200):
                return None
            if note and self._contains_forbidden(note):
                return None
            cleaned_steps.append({
                'instruction': instr.strip(),
                'expected_result': expected.strip(),
                'safety_note': (note or '').strip() or None,
            })
        return {
            'escalate_immediately': data['escalate_immediately'],
            'escalation_reason': data.get('escalation_reason') or None,
            'confidence': confidence,
            'steps': cleaned_steps,
        }

    @api.model
    def _contains_forbidden(self, text):
        if not text:
            return False
        return any(p.search(text) for p in FORBIDDEN_PATTERNS)

    # ------------------------------------------------------------------
    # DETERMINISTIC FALLBACK
    # ------------------------------------------------------------------
    @api.model
    def _deterministic_fallback(self, category, symptoms):
        """Look up fusion.repair.self.check.rule records for the category
        and return the matching steps. Used when AI is unavailable or
        returns invalid / unsafe content."""
        Rule = self.env['fusion.repair.self.check.rule'].sudo()
        steps = []
        if category:
            haystack = ' '.join(symptoms).lower()
            rules = Rule.search([
                ('category_id', '=', category.id),
                ('active', '=', True),
            ], order='sequence')
            for r in rules:
                kws = [k.strip().lower() for k in (r.symptom_keywords or '').split(',') if k.strip()]
                if not kws or any(kw and kw in haystack for kw in kws):
                    steps.append({
                        'instruction': r.instruction or '',
                        'expected_result': r.expected_result or '',
                        'safety_note': r.safety_note or None,
                    })
                if len(steps) >= 3:
                    break
        return {
            'escalate_immediately': len(steps) == 0,
            'escalation_reason': None if steps else 'no_match',
            'confidence': 'medium' if steps else 'low',
            'steps': steps,
            'source': 'fallback',
            'disclaimer': self._disclaimer(),
        }

    # ------------------------------------------------------------------
    # CACHE (in-memory per worker, 24h)
    # ------------------------------------------------------------------
    _CACHE = {}
    _CACHE_TTL = 24 * 3600

    @api.model
    def _cache_key(self, category, symptoms):
        symptom_hash = hashlib.sha256(
            ('|'.join(sorted(s.lower() for s in symptoms))).encode()
        ).hexdigest()[:16]
        return f"{category.code if category else 'none'}:{symptom_hash}"

    @api.model
    def _cache_get(self, key):
        import time
        entry = self._CACHE.get(key)
        if not entry:
            return None
        ts, value = entry
        if time.time() - ts > self._CACHE_TTL:
            self._CACHE.pop(key, None)
            return None
        return value

    @api.model
    def _cache_set(self, key, value):
        import time
        # Bound cache size to ~512 entries.
        if len(self._CACHE) > 512:
            self._CACHE.clear()
        self._CACHE[key] = (time.time(), value)

    # ------------------------------------------------------------------
    # MISC
    # ------------------------------------------------------------------
    @api.model
    def _disclaimer(self):
        return ("This is not medical advice. If you're unsure, schedule a "
                "technician visit. In an emergency, call 9-1-1.")

    @api.model
    def _log_incident(self, kind, raw):
        _logger.warning('AI self-check incident (%s): %s', kind, (raw or '')[:300])