"""Génération de messages par IA via l'abonnement ChatGPT (provider `chatgpt` de LiteLLM). L'accès se fait par le flow « Sign in with ChatGPT » (device code) : aucune clé API, c'est l'abonnement ChatGPT de l'utilisateur qui est utilisé. Le token est stocké LOCALEMENT dans le dossier `.chatgpt/` du projet pour rester portable (utilisable en tournée sur un portable). Usage strictement personnel : le backend `chatgpt.com/backend-api/codex` n'est pas une API publique documentée et peut changer ; on s'appuie sur LiteLLM qui en absorbe les détails. """ import os import threading from pathlib import Path DOSSIER = Path(__file__).parent TOKEN_DIR = DOSSIER / ".chatgpt" # Doit être défini AVANT tout import de litellm pour que le token soit écrit dans le projet. os.environ.setdefault("CHATGPT_TOKEN_DIR", str(TOKEN_DIR)) MODELE_DEFAUT = "chatgpt/gpt-5.4" # État du login device-code en cours, partagé entre la route /api/ia/login et le thread de polling. _verrou_login = threading.Lock() _etat_login = { "en_cours": False, "user_code": None, "verification_url": None, "erreur": None, } class IANonConnecte(Exception): """Aucune session ChatGPT valide : il faut se connecter d'abord.""" class IAErreur(Exception): """Échec d'un appel à l'IA (réseau, rate limit, token expiré...).""" def _authenticator(): from litellm.llms.chatgpt.authenticator import Authenticator return Authenticator() def est_connecte(): """Vrai si un token ChatGPT exploitable (access ou refresh) est stocké localement.""" try: data = _authenticator()._read_auth_file() return bool(data and (data.get("access_token") or data.get("refresh_token"))) except Exception: return False def lancer_login(): """Démarre le flow device-code et renvoie le code + l'URL à afficher dans l'UI. Le polling (attente que l'utilisateur valide dans le navigateur) tourne en arrière-plan ; l'UI suit l'avancement via `statut_login()`. """ if est_connecte(): return {"deja_connecte": True} from litellm.llms.chatgpt.common_utils import CHATGPT_DEVICE_VERIFY_URL with _verrou_login: if _etat_login["en_cours"]: return { "user_code": _etat_login["user_code"], "verification_url": _etat_login["verification_url"], } auth = _authenticator() device = auth._request_device_code() auth._record_device_code_request() _etat_login.update({ "en_cours": True, "user_code": device["user_code"], "verification_url": CHATGPT_DEVICE_VERIFY_URL, "erreur": None, }) def _attendre_validation(): try: code = auth._poll_for_authorization_code(device) tokens = auth._exchange_code_for_tokens(code) auth._write_auth_file(auth._build_auth_record(tokens)) except Exception as e: _etat_login["erreur"] = str(e) finally: _etat_login["en_cours"] = False threading.Thread(target=_attendre_validation, daemon=True).start() return { "user_code": _etat_login["user_code"], "verification_url": _etat_login["verification_url"], } def statut_login(): """État courant pour l'UI : connecté ? login en cours ? code à afficher ? erreur ?""" en_cours = _etat_login["en_cours"] return { "connecte": est_connecte(), "en_cours": en_cours, "user_code": _etat_login["user_code"] if en_cours else None, "verification_url": _etat_login["verification_url"] if en_cours else None, "erreur": _etat_login["erreur"], } def _message_modele(prospect, modele): """Substitution {nom}/{ville}/{type} — réplique de messagePour() côté frontend.""" ville = (prospect.get("Ville") or "").strip() return (modele or "") \ .replace("{nom}", prospect.get("Nom du prospect") or "") \ .replace("{ville}", f" à {ville}" if ville else "") \ .replace("{type}", (prospect.get("Type") or "").lower()) def _infos_prospect(prospect): champs = [ ("Nom", prospect.get("Nom du prospect")), ("Ville", prospect.get("Ville")), ("Type de lieu", prospect.get("Type")), ("Infos du lieu", prospect.get("Infos du lieu")), ] return "\n".join(f"- {cle} : {val}" for cle, val in champs if (val or "").strip()) def _infos_groupe(groupe): """Bloc décrivant le groupe (depuis les réglages), injecté dans le prompt système. Ne liste que les champs renseignés ; renvoie "" si rien n'est fourni. """ if not groupe: return "" champs = [ ("Nom du groupe", groupe.get("nom")), ("Style", groupe.get("style")), ("Description", groupe.get("description")), ("Lien", groupe.get("lien")), ] lignes = [f"- {cle} : {(val or '').strip()}" for cle, val in champs if (val or "").strip()] if not lignes: return "" return "Tu représentes le groupe suivant :\n" + "\n".join(lignes) def _message_erreur(exc): bas = str(exc).lower() if "rate" in bas or "429" in bas or "limit" in bas or "quota" in bas: return "Limite de l'abonnement ChatGPT atteinte (fenêtre de 5 h). Réessayez plus tard." if "401" in bas or "403" in bas or "expired" in bas or "token" in bas or "unauthor" in bas: return "Session ChatGPT expirée ou invalide. Reconnectez-vous dans les Paramètres." return f"Échec de la génération : {exc}" def generer_message(prospect, modele, mode="generer", nom_modele=None, groupe=None): """Génère un message de prise de contact pour un prospect. mode == "generer" : l'IA rédige un message sur mesure (modèle = guide de ton/style). mode == "peaufiner" : on substitue d'abord le modèle, puis l'IA le reformule sans en changer le sens. `groupe` : dict optionnel (nom, style, description, lien) issu des réglages ; ses champs renseignés sont injectés dans le prompt système pour personnaliser la rédaction. """ nom_modele = nom_modele or MODELE_DEFAUT if not est_connecte(): raise IANonConnecte( "Connectez-vous à ChatGPT dans les Paramètres avant de générer un message." ) infos = _infos_prospect(prospect) or "- (aucune information détaillée disponible)" bloc_groupe = _infos_groupe(groupe) if mode == "peaufiner": brouillon = _message_modele(prospect, modele) parties = [ "Tu rédiges des messages de prise de contact en français pour proposer " "l'organisation de concerts à des établissements." ] if bloc_groupe: parties.append(bloc_groupe) parties.append( "On te donne un brouillon : reformule-le pour le rendre plus naturel, chaleureux " "et engageant, sans inventer d'information ni changer le sens. Réponds UNIQUEMENT " "par le message final, sans commentaire." ) systeme = "\n\n".join(parties) utilisateur = f"Brouillon à améliorer :\n{brouillon}\n\nInfos sur l'établissement :\n{infos}" else: parties = [ "Tu rédiges des messages de prise de contact en français pour proposer " "l'organisation de concerts à des établissements (bars, restaurants, salles...)." ] if bloc_groupe: parties.append(bloc_groupe) parties.append( "Le message doit être court, personnalisé, chaleureux et se terminer par une " "question ouvrant l'échange. Réponds UNIQUEMENT par le message final, sans " "commentaire ni objet d'e-mail." ) systeme = "\n\n".join(parties) utilisateur = ( f"Rédige un message de prise de contact pour cet établissement :\n{infos}\n\n" f"Inspire-toi de ce modèle pour le ton et le style :\n{modele}" ) # Streaming OBLIGATOIRE : le backend Codex envoie le texte par deltas et renvoie un # `output` vide dans l'événement final, ce qui casse la voie non-streaming de LiteLLM # (« Unknown items in responses API response: [] »). On accumule donc les morceaux. try: import litellm flux = litellm.completion( model=nom_modele, messages=[ {"role": "system", "content": systeme}, {"role": "user", "content": utilisateur}, ], stream=True, ) morceaux = [] for chunk in flux: if chunk.choices: delta = chunk.choices[0].delta.content if delta: morceaux.append(delta) except Exception as e: raise IAErreur(_message_erreur(e)) return "".join(morceaux).strip()