openbot/app.py

import base64
import json
import os
import smtplib
import tempfile
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from pathlib import Path

from dotenv import load_dotenv
from flask import Flask, jsonify, request, send_from_directory
from groq import Groq

load_dotenv()

ROOT = Path(__file__).resolve().parent
STATIC_DIR = ROOT / "static"
CONTACTS_FILE = ROOT / "contacts.json"

GROQ_API_KEY = os.getenv("GROQ_API_KEY", "")
GMAIL_EMAIL = os.getenv("GMAIL_EMAIL", "")
GMAIL_PASSWORD = os.getenv("GMAIL_PASSWORD", "")

CHAT_MODEL = "llama-3.3-70b-versatile"
STT_MODEL = "whisper-large-v3-turbo"
TTS_MODEL = "canopylabs/orpheus-v1-english"
TTS_VOICE = "troy"

app = Flask(__name__, static_folder=str(STATIC_DIR), static_url_path="")
app.secret_key = os.getenv("APP_SECRET", "dev-secret-change-me")

client = Groq(api_key=GROQ_API_KEY)
histories = {}

SYSTEM_PROMPT = """Tu es l'assistant personnel vocal de Laurent.
Tu peux discuter normalement en francais et aider a envoyer des emails via Gmail.

Quand l'utilisateur veut envoyer un email, collecte les informations manquantes :
destinataire, sujet, contenu.

Quand tu as toutes les informations pour envoyer un email, reponds uniquement avec ce JSON :
{"action": "envoyer_email", "destinataire": "...", "sujet": "...", "corps": "..."}

Pour tout le reste, reponds en francais naturel, avec des phrases courtes.
En mode voiture, sois bref : une ou deux phrases maximum."""


def charger_contacts():
    try:
        return json.loads(CONTACTS_FILE.read_text(encoding="utf-8"))
    except FileNotFoundError:
        return {"laurent": "barontil@hotmail.com"}


def construire_system_prompt():
    contacts = charger_contacts()
    lignes = [f"- {nom.title()} : {email}" for nom, email in contacts.items()]
    return SYSTEM_PROMPT + "\n\nContacts connus :\n" + "\n".join(lignes)


def resoudre_contact(destinataire):
    contacts = charger_contacts()
    valeur = destinataire.strip()
    return contacts.get(valeur.lower(), valeur)


def extraire_action_email(reponse):
    decodeur = json.JSONDecoder()
    for index, caractere in enumerate(reponse):
        if caractere != "{":
            continue
        try:
            data, _ = decodeur.raw_decode(reponse[index:])
        except json.JSONDecodeError:
            continue
        if isinstance(data, dict) and data.get("action") == "envoyer_email":
            return data
    return None


def envoyer_email(destinataire, sujet, corps):
    destinataire = resoudre_contact(destinataire)
    msg = MIMEMultipart()
    msg["From"] = GMAIL_EMAIL
    msg["To"] = destinataire
    msg["Subject"] = sujet
    msg.attach(MIMEText(corps, "plain", "utf-8"))

    with smtplib.SMTP("smtp.gmail.com", 587) as serveur:
        serveur.starttls()
        serveur.login(GMAIL_EMAIL, GMAIL_PASSWORD)
        serveur.sendmail(GMAIL_EMAIL, destinataire, msg.as_string())

    return destinataire


def demander_au_llm(session_id, texte):
    historique = histories.setdefault(session_id, [])
    historique.append({"role": "user", "content": texte})
    del historique[:-20]

    reponse = client.chat.completions.create(
        model=CHAT_MODEL,
        messages=[{"role": "system", "content": construire_system_prompt()}] + historique,
        temperature=0.45,
        max_tokens=900,
    )
    contenu = reponse.choices[0].message.content
    historique.append({"role": "assistant", "content": contenu})
    return contenu


def transcrire_audio(fichier):
    suffix = Path(fichier.filename or "voice.webm").suffix or ".webm"
    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
        fichier.save(tmp)
        chemin = tmp.name

    try:
        with open(chemin, "rb") as audio:
            transcription = client.audio.transcriptions.create(
                model=STT_MODEL,
                file=audio,
                language="fr",
                response_format="text",
            )
        return transcription if isinstance(transcription, str) else transcription.text
    finally:
        try:
            os.remove(chemin)
        except OSError:
            pass


def generer_audio(texte):
    texte = texte.strip()
    if len(texte) > 900:
        texte = texte[:897] + "..."

    reponse_audio = client.audio.speech.create(
        model=TTS_MODEL,
        voice=TTS_VOICE,
        input=texte,
        response_format="wav",
        speed=1.0,
    )

    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
        chemin = tmp.name

    try:
        reponse_audio.write_to_file(chemin)
        audio = Path(chemin).read_bytes()
        return "data:audio/wav;base64," + base64.b64encode(audio).decode("ascii")
    finally:
        try:
            os.remove(chemin)
        except OSError:
            pass


def traiter_message(session_id, texte, avec_audio=False):
    reponse = demander_au_llm(session_id, texte)
    action = extraire_action_email(reponse)
    email_sent = False

    if action:
        destinataire = envoyer_email(action["destinataire"], action["sujet"], action["corps"])
        reponse = f"Email envoye a {destinataire}."
        email_sent = True

    audio_url = None
    if avec_audio:
        audio_url = generer_audio(reponse)

    return {
        "reply": reponse,
        "audioUrl": audio_url,
        "emailSent": email_sent,
    }


@app.get("/")
def index():
    return send_from_directory(STATIC_DIR, "index.html")


@app.get("/health")
def health():
    return jsonify({"ok": True})


@app.post("/api/chat")
def chat():
    data = request.get_json(force=True)
    texte = (data.get("text") or "").strip()
    session_id = data.get("sessionId") or "default"
    avec_audio = bool(data.get("withAudio"))

    if not texte:
        return jsonify({"error": "Message vide"}), 400

    try:
        return jsonify(traiter_message(session_id, texte, avec_audio))
    except Exception as e:
        print(f"Erreur /api/chat : {e}", flush=True)
        return jsonify({"error": str(e)}), 500


@app.post("/api/voice")
def voice():
    fichier = request.files.get("audio")
    session_id = request.form.get("sessionId") or "default"

    if not fichier:
        return jsonify({"error": "Audio manquant"}), 400

    try:
        texte = transcrire_audio(fichier).strip()
        resultat = traiter_message(session_id, texte, avec_audio=True)
        resultat["transcript"] = texte
        return jsonify(resultat)
    except Exception as e:
        print(f"Erreur /api/voice : {e}", flush=True)
        return jsonify({"error": str(e)}), 500


@app.post("/api/reset")
def reset():
    data = request.get_json(silent=True) or {}
    session_id = data.get("sessionId") or "default"
    histories.pop(session_id, None)
    return jsonify({"ok": True})


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=int(os.getenv("PORT", "8080")), debug=True)