Odoo-Modules/fusion_helpdesk_central/utils.py

# -*- coding: utf-8 -*-
# Copyright 2026 Nexa Systems Inc.
# License OPL-1
"""Pure helpers for the owner-approval engagement flow.

No Odoo environment, no `request`, no network — just data in, data out.
Everything here is unit-testable in isolation, which is the same pattern
fusion_helpdesk/utils.py already follows. The OpenAI client lives in
`openai_client.py` so this file stays IO-free and trivially testable.
"""
import json
import logging
import urllib.error
import urllib.request
from markupsafe import Markup, escape

_logger = logging.getLogger(__name__)


# Frozen as a Python constant rather than an ICP key — the prompt is
# load-bearing on output quality and we want every change reviewed in git.
# Use {placeholders} to template; see build_summary_prompt for the fill.
SUMMARY_PROMPT = """You are summarising a customer support ticket for a busy executive
who needs to decide whether to approve the work.

Output rules:
- 4-6 short bullet points, plain text (no markdown).
- First bullet: the ask, in one sentence.
- Second bullet: the business impact if approved.
- Third bullet: the business impact if NOT approved (or "none material").
- Optional bullets: cost / effort signals if any are mentioned.
- Final bullet: open questions the approver should think about.
- Do not invent facts. If the thread doesn't say, write "not stated".
- No greetings, no sign-offs, no preamble.

Ticket title: {name}
Original report:
{description_plain}

Replies so far:
{messages_plain}
"""

# Bound the prompt sent to OpenAI so a runaway thread doesn't blow cost or
# context. 8000 chars * ~0.25 tokens/char ~= 2000 tokens, well under any model
# cap and cheap on gpt-4o-mini.
OPENAI_PROMPT_MAX_CHARS = 8000


def build_summary_prompt(ticket_name, description_plain, messages):
    """Render SUMMARY_PROMPT with ticket + chatter content.

    `messages` is a list of dicts with at minimum {author, date, body_plain}.
    We render one line per message so the model can attribute who said what.
    Empty inputs collapse to "(none)" so the prompt never has bare blanks
    that the model could interpret as a missing section.
    """
    name = (ticket_name or '').strip() or '(untitled)'
    desc = (description_plain or '').strip() or '(no description)'
    if messages:
        lines = []
        for m in messages:
            author = (m.get('author') or 'unknown').strip()
            date = (m.get('date') or '').strip()
            body = (m.get('body_plain') or '').strip() or '(empty)'
            lines.append('%s (%s): %s' % (author, date, body))
        msgs = '\n---\n'.join(lines)
    else:
        msgs = '(no replies yet)'
    return SUMMARY_PROMPT.format(
        name=name, description_plain=desc, messages_plain=msgs,
    )


def truncate_for_openai(prompt, max_chars=OPENAI_PROMPT_MAX_CHARS):
    """Cap prompt length so a 50-message thread can't run up the bill.

    Truncates from the END (keeps the title + description + earliest messages),
    appending a marker so the model knows context was cut. Earliest replies
    are usually the most relevant for an approval decision; later replies tend
    to be operational chatter.
    """
    if len(prompt) <= max_chars:
        return prompt
    marker = '\n\n[...truncated for length...]'
    keep = max_chars - len(marker)
    if keep <= 0:
        return marker[:max_chars]
    return prompt[:keep] + marker


def format_engagement_chatter(decision, owner_name, comment=None):
    """Build the chatter message posted on the ticket when the owner decides.

    Returns Markup so message_post renders it as HTML (not escaped). Decision
    must be 'approved' or 'rejected' — anything else raises ValueError so a
    typo upstream fails loud instead of posting a half-rendered message.
    """
    if decision == 'approved':
        prefix, label = '✓', 'Approved'
    elif decision == 'rejected':
        prefix, label = '✗', 'Rejected'
    else:
        raise ValueError(
            'format_engagement_chatter: decision must be "approved" or '
            '"rejected", got %r' % (decision,)
        )
    name = escape((owner_name or '').strip() or 'the owner')
    text = (comment or '').strip()
    body = '<p><b>%s %s by %s</b></p>' % (prefix, label, name)
    if text:
        body += '<blockquote style="margin:6px 0 0 0; padding:6px 12px; ' \
                'border-left:3px solid #d8dadd; color:#444;"><i>%s</i></blockquote>' \
                % escape(text).replace('\n', '<br/>')
    return Markup(body)


def call_openai_chat(api_key, model, prompt, timeout=15):
    """Single non-streaming call to OpenAI's chat-completions endpoint.

    Returns the response text on success, or '' on any failure (network,
    timeout, non-2xx, malformed JSON, empty choices). Never raises — the
    caller treats '' as "AI unavailable" and shows the manual-summary
    fallback in the wizard.

    Uses urllib.request from the stdlib so we don't add a pip dependency
    just for one HTTP call. JSON shape per OpenAI's chat-completions spec
    as of 2026: messages, model, response choices[0].message.content.
    """
    if not api_key or not prompt:
        return ''
    payload = json.dumps({
        'model': model or 'gpt-4o-mini',
        'messages': [{'role': 'user', 'content': prompt}],
        'temperature': 0.2,    # we want deterministic-ish summaries
    }).encode('utf-8')
    req = urllib.request.Request(
        'https://api.openai.com/v1/chat/completions',
        data=payload,
        headers={
            'Authorization': 'Bearer %s' % api_key,
            'Content-Type': 'application/json',
        },
        method='POST',
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            data = json.loads(resp.read().decode('utf-8'))
    except urllib.error.HTTPError as e:
        _logger.warning(
            'fusion_helpdesk_central: OpenAI returned HTTP %s', e.code,
        )
        return ''
    except (urllib.error.URLError, TimeoutError, OSError) as e:
        _logger.warning(
            'fusion_helpdesk_central: OpenAI network error: %s', e,
        )
        return ''
    except (ValueError, json.JSONDecodeError) as e:
        _logger.warning(
            'fusion_helpdesk_central: OpenAI returned malformed JSON: %s', e,
        )
        return ''
    choices = (data or {}).get('choices') or []
    if not choices:
        return ''
    content = (choices[0].get('message') or {}).get('content') or ''
    return content.strip()