238 lines
8.7 KiB
Python
238 lines
8.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Copyright 2026 Nexa Systems Inc.
|
|
# License OPL-1 (Odoo Proprietary License v1.0)
|
|
|
|
import base64
|
|
import logging
|
|
import os
|
|
import tempfile
|
|
|
|
from odoo import http
|
|
from odoo.http import request
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
class FusionNotesController(http.Controller):
|
|
|
|
# ------------------------------------------------------------------
|
|
# Settings endpoint (for frontend to fetch user preferences)
|
|
# ------------------------------------------------------------------
|
|
|
|
@http.route('/fusion_notes/get_settings', type='jsonrpc', auth='user')
|
|
def get_settings(self, **kwargs):
|
|
"""Return Fusion Notes settings for the frontend."""
|
|
ICP = request.env['ir.config_parameter'].sudo()
|
|
return {
|
|
'review_mode': ICP.get_param(
|
|
'fusion_notes.default_review_mode', 'True') == 'True',
|
|
'ai_format': ICP.get_param(
|
|
'fusion_notes.default_ai_format', 'False') == 'True',
|
|
'max_seconds': int(
|
|
ICP.get_param('fusion_notes.max_recording_seconds', '300')),
|
|
'has_api_key': bool(
|
|
ICP.get_param('fusion_notes.openai_api_key', '')),
|
|
}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Whisper transcription
|
|
# ------------------------------------------------------------------
|
|
|
|
@http.route('/fusion_notes/transcribe', type='jsonrpc', auth='user')
|
|
def transcribe(self, audio_base64, mime_type='audio/webm', **kwargs):
|
|
"""Transcribe audio using OpenAI Whisper API.
|
|
|
|
Args:
|
|
audio_base64: Base64-encoded audio data.
|
|
mime_type: MIME type of the audio (default audio/webm).
|
|
|
|
Returns:
|
|
dict with 'text' on success or 'error' on failure.
|
|
"""
|
|
ICP = request.env['ir.config_parameter'].sudo()
|
|
api_key = ICP.get_param('fusion_notes.openai_api_key', '')
|
|
|
|
if not api_key:
|
|
return {
|
|
'error': 'OpenAI API key not configured. '
|
|
'Go to Settings > Fusion Notes.',
|
|
}
|
|
|
|
try:
|
|
import requests as req
|
|
except ImportError:
|
|
return {'error': 'Python requests library not available.'}
|
|
|
|
# Decode audio from base64
|
|
try:
|
|
audio_data = base64.b64decode(audio_base64)
|
|
except Exception as e:
|
|
_logger.error('Failed to decode audio: %s', e)
|
|
return {'error': 'Invalid audio data.'}
|
|
|
|
if len(audio_data) < 1000:
|
|
return {'error': 'Audio too short. Please record a longer message.'}
|
|
|
|
# Determine file extension from MIME type
|
|
ext_map = {
|
|
'audio/webm': '.webm',
|
|
'audio/ogg': '.ogg',
|
|
'audio/wav': '.wav',
|
|
'audio/mp4': '.m4a',
|
|
'audio/mpeg': '.mp3',
|
|
}
|
|
suffix = ext_map.get(mime_type, '.webm')
|
|
|
|
# Write to temp file and send to Whisper
|
|
tmp_path = None
|
|
try:
|
|
with tempfile.NamedTemporaryFile(
|
|
suffix=suffix, delete=False
|
|
) as tmp:
|
|
tmp.write(audio_data)
|
|
tmp_path = tmp.name
|
|
|
|
with open(tmp_path, 'rb') as audio_file:
|
|
# Use translations endpoint to always output English
|
|
# (auto-translates any spoken language to English)
|
|
response = req.post(
|
|
'https://api.openai.com/v1/audio/translations',
|
|
headers={'Authorization': f'Bearer {api_key}'},
|
|
files={
|
|
'file': (
|
|
f'recording{suffix}', audio_file, mime_type,
|
|
),
|
|
},
|
|
data={'model': 'whisper-1'},
|
|
timeout=60,
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
text = result.get('text', '').strip()
|
|
|
|
if not text:
|
|
return {'error': 'No speech detected in the recording.'}
|
|
|
|
return {'text': text}
|
|
|
|
except req.exceptions.Timeout:
|
|
return {
|
|
'error': 'Transcription timed out. '
|
|
'Please try a shorter recording.',
|
|
}
|
|
except req.exceptions.HTTPError as e:
|
|
body = e.response.text if e.response else ''
|
|
_logger.error('Whisper API error: %s - %s', e, body)
|
|
return {'error': f'Transcription failed: {e}'}
|
|
except Exception as e:
|
|
_logger.error('Transcription error: %s', e)
|
|
return {'error': f'Transcription failed: {e}'}
|
|
finally:
|
|
if tmp_path and os.path.exists(tmp_path):
|
|
try:
|
|
os.unlink(tmp_path)
|
|
except OSError:
|
|
pass
|
|
|
|
# ------------------------------------------------------------------
|
|
# GPT formatting
|
|
# ------------------------------------------------------------------
|
|
|
|
@http.route('/fusion_notes/format', type='jsonrpc', auth='user')
|
|
def format_note(self, text, **kwargs):
|
|
"""Format transcribed text into a professional note using GPT.
|
|
|
|
Args:
|
|
text: Raw transcription text.
|
|
|
|
Returns:
|
|
dict with 'text' on success or 'error' on failure.
|
|
"""
|
|
ICP = request.env['ir.config_parameter'].sudo()
|
|
api_key = ICP.get_param('fusion_notes.openai_api_key', '')
|
|
|
|
if not api_key:
|
|
return {'error': 'OpenAI API key not configured.'}
|
|
|
|
model = ICP.get_param('fusion_notes.ai_model', 'gpt-4o-mini')
|
|
|
|
try:
|
|
import requests as req
|
|
except ImportError:
|
|
return {'error': 'Python requests library not available.'}
|
|
|
|
system_prompt = (
|
|
"You are a professional note formatter. Rewrite the following "
|
|
"voice transcription as a clean, professional log entry.\n"
|
|
"Rules:\n"
|
|
"- ALWAYS output in English regardless of input language\n"
|
|
"- If the input is in another language, translate it to English\n"
|
|
"- Fix grammar and punctuation\n"
|
|
"- Remove filler words (um, uh, like, you know, so, basically)\n"
|
|
"- Keep ALL facts, names, dates, and details exactly as stated\n"
|
|
"- Organize into clear, concise sentences\n"
|
|
"- Use professional tone appropriate for clinical/business notes\n"
|
|
"- Do NOT add information that was not in the original\n"
|
|
"- Do NOT add headers, bullet points, or extra formatting "
|
|
"unless the content clearly warrants it\n"
|
|
"- Return ONLY the formatted text, no explanations or preamble"
|
|
)
|
|
|
|
try:
|
|
response = req.post(
|
|
'https://api.openai.com/v1/chat/completions',
|
|
headers={
|
|
'Authorization': f'Bearer {api_key}',
|
|
'Content-Type': 'application/json',
|
|
},
|
|
json={
|
|
'model': model,
|
|
'messages': [
|
|
{'role': 'system', 'content': system_prompt},
|
|
{'role': 'user', 'content': text},
|
|
],
|
|
'max_tokens': 1000,
|
|
'temperature': 0.3,
|
|
},
|
|
timeout=30,
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
formatted = result['choices'][0]['message']['content'].strip()
|
|
return {'text': formatted}
|
|
|
|
except Exception as e:
|
|
_logger.error('GPT formatting error: %s', e)
|
|
return {'error': f'Formatting failed: {e}'}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Post note to chatter
|
|
# ------------------------------------------------------------------
|
|
|
|
@http.route('/fusion_notes/post_note', type='jsonrpc', auth='user')
|
|
def post_note(self, thread_model, thread_id, body, **kwargs):
|
|
"""Post a voice note to the specified record's chatter.
|
|
|
|
Args:
|
|
thread_model: The Odoo model name (e.g. 'sale.order').
|
|
thread_id: The record ID.
|
|
body: The note text to post.
|
|
|
|
Returns:
|
|
dict with 'success' or 'error'.
|
|
"""
|
|
try:
|
|
record = request.env[thread_model].browse(int(thread_id))
|
|
if not record.exists():
|
|
return {'error': 'Record not found.'}
|
|
|
|
record.message_post(
|
|
body=body,
|
|
message_type='comment',
|
|
subtype_xmlid='mail.mt_note',
|
|
)
|
|
return {'success': True}
|
|
except Exception as e:
|
|
_logger.error('Failed to post voice note: %s', e)
|
|
return {'error': f'Failed to post note: {e}'}
|