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'\b(stop|should\s+stop)\s+using\b', re.I),
    re.compile(r'\bconsult\s+(your|a)\s+(doctor|physician|nurse)\b', re.I),
    re.compile(r'\b(blood\s+pressure|heart\s+rate|pulse|oxygen)\b', re.I),
    re.compile(r'(\$|CAD|USD)\s?\d+', re.I),  # No price mentions
]

# Universal hard-escalate: ANY equipment category - fire / smoke / sparks /
# burning / injury / trapped is always an immediate escalation. Word
# boundaries prevent "unhurt" matching "hurt" and "fireman" matching "fire".
UNIVERSAL_ESCALATION_RE = re.compile(
    r'\b(fire|smoke|burning|spark|injur(y|ed)|hurt|bleeding|trapped)\b',
    re.I,
)

# Category-specific safety symptoms - only fire if the category is flagged
# safety_critical=True on fusion.repair.product.category (stairlifts,
# porch lifts, power wheelchairs). "won.?t" handles both "won't" and "wont".
SAFETY_SYMPTOMS_RE = re.compile(
    r"\b(stuck|motor|brake\s*fail|won.?t\s*stop|overshoot)\b",
    re.I,
)


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)
        # Universal: fire / smoke / spark / burning / injury / trapped escalate
        # regardless of equipment category. Electrical fire on a hospital bed
        # is exactly as urgent as on a stairlift.
        if UNIVERSAL_ESCALATION_RE.search(text):
            return True
        # Category-specific: 'stuck', 'motor', 'brake fail', etc. only escalate
        # on safety-critical categories (stairlifts, porch lifts, power chairs).
        if category and category.safety_critical and SAFETY_SYMPTOMS_RE.search(text):
            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 _normalise(self, text):
        """Strip punctuation + lowercase so 'wont move' matches 'won't move'
        and vice versa.

        IMPORTANT: apostrophes are REMOVED (not replaced with space), so
        "won't" -> "wont" matches user input "wont" (without apostrophe).
        Other punctuation collapses to a single space.
        """
        s = (text or "").lower()
        # Remove ALL apostrophe variants (straight + curly) so contraction
        # forms collide with apostrophe-less forms.
        for apos in ("'", "\u2019", "\u2018", "\u02bc"):
            s = s.replace(apos, "")
        # Everything else non-alphanumeric -> single space.
        return re.sub(r"[^a-z0-9 ]+", " ", s)

    @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 = self._normalise(' '.join(symptoms))
            rules = Rule.search([
                ('category_id', '=', category.id),
                ('active', '=', True),
            ], order='sequence')
            for r in rules:
                kws = [
                    self._normalise(k)
                    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])