589 lines
19 KiB
Python
589 lines
19 KiB
Python
import base64
|
|
import json
|
|
import os
|
|
import re
|
|
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
|
|
from memory import (
|
|
add_memory,
|
|
add_repository,
|
|
add_task,
|
|
init_db,
|
|
search_memory,
|
|
upsert_client,
|
|
upsert_project,
|
|
)
|
|
|
|
load_dotenv()
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
STATIC_DIR = ROOT / "static"
|
|
CONTACTS_FILE = Path(os.getenv("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"
|
|
TTS_PROVIDER = os.getenv("TTS_PROVIDER", "browser").lower()
|
|
|
|
app = Flask(__name__, static_folder=str(STATIC_DIR), static_url_path="")
|
|
app.secret_key = os.getenv("APP_SECRET", "dev-secret-change-me")
|
|
init_db()
|
|
|
|
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": "..."}
|
|
|
|
Quand l'utilisateur veut ajouter, memoriser ou modifier un contact email, collecte le nom et l'adresse email.
|
|
Quand tu as les deux informations, reponds uniquement avec ce JSON :
|
|
{"action": "ajouter_contact", "nom": "...", "email": "..."}
|
|
|
|
Tu disposes aussi d'une memoire professionnelle persistante pour les clients, projets, taches, notes et depots Git.
|
|
Quand l'utilisateur te demande de noter, memoriser, ajouter ou associer une information professionnelle, reponds uniquement avec un JSON.
|
|
Actions disponibles :
|
|
{"action": "ajouter_client", "nom": "...", "notes": "...", "statut": "actif"}
|
|
{"action": "ajouter_projet", "nom": "...", "client": "...", "statut": "...", "resume": "...", "prochaine_action": "..."}
|
|
{"action": "ajouter_tache", "titre": "...", "client": "...", "projet": "...", "details": "...", "echeance": "..."}
|
|
{"action": "ajouter_repo", "nom": "...", "projet": "...", "chemin": "...", "url": "...", "branche": "..."}
|
|
{"action": "ajouter_note", "contenu": "...", "sujet": "...", "client": "...", "projet": "...", "type": "note"}
|
|
{"action": "chercher_memoire", "requete": "..."}
|
|
|
|
Si une phrase contient plusieurs choses a enregistrer, utilise :
|
|
{"actions": [{...}, {...}]}
|
|
|
|
Quand l'utilisateur demande ce que tu sais d'un client, d'un projet, d'une tache ou d'un depot, utilise chercher_memoire.
|
|
|
|
Pour tout le reste, reponds en francais naturel, avec des phrases courtes.
|
|
N'utilise pas d'anglais sauf si l'utilisateur le demande explicitement.
|
|
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 sauvegarder_contacts(contacts):
|
|
CONTACTS_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
CONTACTS_FILE.write_text(
|
|
json.dumps(contacts, ensure_ascii=False, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
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(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") in ACTIONS or isinstance(data.get("actions"), list)):
|
|
return data
|
|
return None
|
|
|
|
|
|
ACTIONS = {
|
|
"envoyer_email",
|
|
"ajouter_contact",
|
|
"ajouter_client",
|
|
"ajouter_projet",
|
|
"ajouter_tache",
|
|
"ajouter_repo",
|
|
"ajouter_note",
|
|
"chercher_memoire",
|
|
}
|
|
|
|
|
|
def ajouter_contact(nom, email):
|
|
contacts = charger_contacts()
|
|
nom_normalise = " ".join(nom.strip().lower().split())
|
|
email_normalise = email.strip()
|
|
|
|
if not nom_normalise or "@" not in email_normalise:
|
|
raise ValueError("Contact incomplet")
|
|
|
|
contacts[nom_normalise] = email_normalise
|
|
sauvegarder_contacts(contacts)
|
|
return nom_normalise, email_normalise
|
|
|
|
|
|
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 detecter_action_directe(texte):
|
|
texte_nettoye = texte.strip()
|
|
texte_min = texte_nettoye.lower()
|
|
|
|
clients_a_ajouter = extraire_liste_clients(texte_nettoye)
|
|
if clients_a_ajouter:
|
|
return {
|
|
"actions": [
|
|
{
|
|
"action": "ajouter_client",
|
|
"nom": nom,
|
|
"notes": "Ajoute depuis une liste dictee par Laurent.",
|
|
"statut": "actif",
|
|
}
|
|
for nom in clients_a_ajouter
|
|
]
|
|
}
|
|
|
|
recherche_match = re.search(
|
|
r"(?:qu(?:'| )?est[- ]?ce que tu sais sur|que sais[- ]?tu sur|cherche dans la memoire|memoire sur)\s+(.+)",
|
|
texte_nettoye,
|
|
flags=re.IGNORECASE,
|
|
)
|
|
if recherche_match:
|
|
return {
|
|
"action": "chercher_memoire",
|
|
"requete": recherche_match.group(1).strip(" ?.,;:"),
|
|
}
|
|
|
|
if re.search(r"\b(liste|affiche|montre)\b.*\bclients?\b", texte_min):
|
|
return {"action": "chercher_memoire", "requete": ""}
|
|
|
|
if re.search(r"\b(liste|affiche|montre)\b.*\b(projets?|taches?|repos?|depots?)\b", texte_min):
|
|
return {"action": "chercher_memoire", "requete": ""}
|
|
|
|
repo_match = re.search(
|
|
r"(?:associe|ajoute|note)\s+(?:le\s+)?repo\s+(\S+)(?:\s+au\s+projet\s+(.+))?",
|
|
texte_nettoye,
|
|
flags=re.IGNORECASE,
|
|
)
|
|
if repo_match:
|
|
chemin_ou_url = repo_match.group(1).strip(" .,;:")
|
|
projet = (repo_match.group(2) or "").strip(" .,;:")
|
|
return {
|
|
"action": "ajouter_repo",
|
|
"chemin": chemin_ou_url if not chemin_ou_url.startswith(("http://", "https://", "git@")) else "",
|
|
"url": chemin_ou_url if chemin_ou_url.startswith(("http://", "https://", "git@")) else "",
|
|
"projet": projet,
|
|
}
|
|
|
|
client_match = re.search(
|
|
r"\b([A-Z0-9][A-Z0-9&.\- ]{2,})\s+est\s+(?:un|une)\s+nouveau\s+client\b",
|
|
texte_nettoye,
|
|
flags=re.IGNORECASE,
|
|
)
|
|
if not client_match:
|
|
return None
|
|
|
|
nom_client = client_match.group(1).strip(" .,;:")
|
|
actions = [
|
|
{
|
|
"action": "ajouter_client",
|
|
"nom": nom_client,
|
|
"notes": texte_nettoye,
|
|
"statut": "actif",
|
|
}
|
|
]
|
|
|
|
rappel_match = re.search(
|
|
r"(?:je\s+dois\s+|il\s+faut\s+|note\s+(?:quelque\s+part\s+)?que\s+je\s+dois\s+)(rappeler\s+.+)",
|
|
texte_nettoye,
|
|
flags=re.IGNORECASE,
|
|
)
|
|
if rappel_match:
|
|
actions.append(
|
|
{
|
|
"action": "ajouter_tache",
|
|
"titre": rappel_match.group(1).strip(" .,;:"),
|
|
"client": nom_client,
|
|
"details": texte_nettoye,
|
|
}
|
|
)
|
|
|
|
return {"actions": actions}
|
|
|
|
|
|
def extraire_liste_clients(texte):
|
|
if not re.search(r"\bclients?\b", texte, flags=re.IGNORECASE):
|
|
return []
|
|
if not re.search(r"\b(voici|liste|note|notes|noter|memorise|ajoute|ajoutes)\b", texte, flags=re.IGNORECASE):
|
|
return []
|
|
|
|
lignes = [ligne.strip(" -\t\r.,;") for ligne in texte.splitlines()]
|
|
candidats = []
|
|
apres_entete = False
|
|
|
|
for ligne in lignes:
|
|
if not ligne:
|
|
continue
|
|
ligne_min = ligne.lower()
|
|
if "client" in ligne_min:
|
|
apres_entete = True
|
|
apres_deux_points = ligne.split(":", 1)[1].strip() if ":" in ligne else ""
|
|
if apres_deux_points:
|
|
candidats.extend(separer_noms_clients(apres_deux_points))
|
|
continue
|
|
if not apres_entete:
|
|
continue
|
|
if re.search(r"\b(note|notes|noter|quelque part|merci|stp|s'il te plait)\b", ligne_min):
|
|
continue
|
|
candidats.extend(separer_noms_clients(ligne))
|
|
|
|
noms = []
|
|
deja_vus = set()
|
|
for candidat in candidats:
|
|
nom = nettoyer_nom_client(candidat)
|
|
if not nom:
|
|
continue
|
|
cle = nom.lower()
|
|
if cle not in deja_vus:
|
|
noms.append(nom)
|
|
deja_vus.add(cle)
|
|
return noms
|
|
|
|
|
|
def separer_noms_clients(texte):
|
|
morceaux = re.split(r"[,;/]|\s+\bet\b\s+", texte)
|
|
return [morceau.strip() for morceau in morceaux if morceau.strip()]
|
|
|
|
|
|
def nettoyer_nom_client(nom):
|
|
nom = nom.strip(" -\t\r.,;:")
|
|
if not nom or len(nom) < 2:
|
|
return ""
|
|
if re.search(r"\b(liste|clients?|voici|note|notes|noter|quelque|part)\b", nom, flags=re.IGNORECASE):
|
|
return ""
|
|
if len(nom.split()) > 4:
|
|
return ""
|
|
return nom
|
|
|
|
|
|
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 executer_action(action):
|
|
nom_action = action.get("action")
|
|
|
|
if nom_action == "envoyer_email":
|
|
destinataire = envoyer_email(action["destinataire"], action["sujet"], action["corps"])
|
|
return f"Email envoye a {destinataire}.", {"emailSent": True}
|
|
|
|
if nom_action == "ajouter_contact":
|
|
nom, email = ajouter_contact(action["nom"], action["email"])
|
|
return f"Contact ajoute : {nom.title()}.", {"contactSaved": True}
|
|
|
|
if nom_action == "ajouter_client":
|
|
client_mem = upsert_client(
|
|
action.get("nom", ""),
|
|
notes=action.get("notes", ""),
|
|
status=action.get("statut", "actif"),
|
|
)
|
|
return f"C'est note : {client_mem['name']} est dans tes clients.", {"memorySaved": True}
|
|
|
|
if nom_action == "ajouter_projet":
|
|
projet = upsert_project(
|
|
action.get("nom", ""),
|
|
client_name=action.get("client", ""),
|
|
status=action.get("statut", "actif"),
|
|
summary=action.get("resume", ""),
|
|
next_action=action.get("prochaine_action", ""),
|
|
)
|
|
return f"C'est note pour le projet {projet['name']}.", {"memorySaved": True}
|
|
|
|
if nom_action == "ajouter_tache":
|
|
tache = add_task(
|
|
action.get("titre", ""),
|
|
client_name=action.get("client", ""),
|
|
project_name=action.get("projet", ""),
|
|
details=action.get("details", ""),
|
|
due_at=action.get("echeance", ""),
|
|
)
|
|
return f"Tache ajoutee : {tache['title']}.", {"memorySaved": True}
|
|
|
|
if nom_action == "ajouter_repo":
|
|
repo = add_repository(
|
|
action.get("nom", ""),
|
|
project_name=action.get("projet", ""),
|
|
local_path=action.get("chemin", ""),
|
|
remote_url=action.get("url", ""),
|
|
main_branch=action.get("branche", ""),
|
|
)
|
|
return f"Depot associe : {repo['name']}.", {"memorySaved": True}
|
|
|
|
if nom_action == "ajouter_note":
|
|
note = add_memory(
|
|
action.get("contenu", ""),
|
|
kind=action.get("type", "note"),
|
|
subject=action.get("sujet", ""),
|
|
client_name=action.get("client", ""),
|
|
project_name=action.get("projet", ""),
|
|
)
|
|
sujet = note["subject"] or "memoire professionnelle"
|
|
return f"C'est note dans {sujet}.", {"memorySaved": True}
|
|
|
|
if nom_action == "chercher_memoire":
|
|
resultat = search_memory(action.get("requete", ""))
|
|
return formater_memoire(resultat), {"memoryRead": True}
|
|
|
|
return "", {}
|
|
|
|
|
|
def executer_actions(action_data):
|
|
actions = action_data.get("actions")
|
|
if not isinstance(actions, list):
|
|
actions = [action_data]
|
|
|
|
reponses = []
|
|
flags = {
|
|
"emailSent": False,
|
|
"contactSaved": False,
|
|
"memorySaved": False,
|
|
"memoryRead": False,
|
|
}
|
|
actions_executees = []
|
|
for action in actions:
|
|
if not isinstance(action, dict):
|
|
continue
|
|
reponse, action_flags = executer_action(action)
|
|
actions_executees.append(action)
|
|
if reponse:
|
|
reponses.append(reponse)
|
|
for cle in flags:
|
|
flags[cle] = flags[cle] or bool(action_flags.get(cle))
|
|
|
|
if actions_executees and all(action.get("action") == "ajouter_client" for action in actions_executees):
|
|
noms = [action.get("nom", "").strip() for action in actions_executees if action.get("nom", "").strip()]
|
|
if len(noms) > 1:
|
|
return f"C'est note : {len(noms)} clients ajoutes ({', '.join(noms)}).", flags
|
|
|
|
return " ".join(reponses), flags
|
|
|
|
|
|
def formater_memoire(resultat):
|
|
lignes = []
|
|
for client_mem in resultat["clients"][:5]:
|
|
detail = f"Client : {client_mem['name']}"
|
|
if client_mem.get("notes"):
|
|
detail += f" - {client_mem['notes']}"
|
|
lignes.append(detail)
|
|
|
|
for projet in resultat["projects"][:5]:
|
|
detail = f"Projet : {projet['name']}"
|
|
if projet.get("client_name"):
|
|
detail += f" ({projet['client_name']})"
|
|
if projet.get("status"):
|
|
detail += f" - statut : {projet['status']}"
|
|
if projet.get("next_action"):
|
|
detail += f" - prochaine action : {projet['next_action']}"
|
|
lignes.append(detail)
|
|
|
|
for tache in resultat["tasks"][:5]:
|
|
contexte = tache.get("client_name") or tache.get("project_name") or "general"
|
|
lignes.append(f"Tache : {tache['title']} ({contexte})")
|
|
|
|
for repo in resultat["repositories"][:5]:
|
|
chemin = repo.get("local_path") or repo.get("remote_url") or "chemin non renseigne"
|
|
lignes.append(f"Depot : {repo['name']} - {chemin}")
|
|
|
|
for note in resultat["memories"][:5]:
|
|
sujet = note.get("subject") or note.get("client_name") or note.get("project_name") or "note"
|
|
lignes.append(f"Note : {sujet} - {note['content']}")
|
|
|
|
if not lignes:
|
|
return "Je n'ai rien trouve dans la memoire professionnelle."
|
|
return "Voici ce que j'ai trouve :\n" + "\n".join(f"- {ligne}" for ligne in lignes[:10])
|
|
|
|
|
|
def traiter_message(session_id, texte, avec_audio=False):
|
|
action_directe = detecter_action_directe(texte)
|
|
reponse = json.dumps(action_directe, ensure_ascii=False) if action_directe else demander_au_llm(session_id, texte)
|
|
action = extraire_action(reponse)
|
|
flags = {
|
|
"emailSent": False,
|
|
"contactSaved": False,
|
|
"memorySaved": False,
|
|
"memoryRead": False,
|
|
}
|
|
|
|
if action:
|
|
reponse, flags = executer_actions(action)
|
|
|
|
audio_url = None
|
|
if avec_audio and TTS_PROVIDER == "groq":
|
|
audio_url = generer_audio(reponse)
|
|
|
|
return {
|
|
"reply": reponse,
|
|
"audioUrl": audio_url,
|
|
"emailSent": flags["emailSent"],
|
|
"contactSaved": flags["contactSaved"],
|
|
"memorySaved": flags["memorySaved"],
|
|
"memoryRead": flags["memoryRead"],
|
|
}
|
|
|
|
|
|
@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"
|
|
avec_audio = request.form.get("withAudio", "false").lower() == "true"
|
|
|
|
if not fichier:
|
|
return jsonify({"error": "Audio manquant"}), 400
|
|
|
|
try:
|
|
texte = transcrire_audio(fichier).strip()
|
|
resultat = traiter_message(session_id, texte, avec_audio=avec_audio)
|
|
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=os.getenv("FLASK_DEBUG", "0") == "1",
|
|
)
|