# -*- coding: utf-8 -*- """ Stockage des projets : SQLite pour les états, disque pour les fichiers. Le principe qui gouverne ce module : le worker peut être tué à tout instant par l'hébergeur. Tout ce qui est enregistré l'est donc de façon atomique (écriture dans un fichier temporaire puis renommage), et l'état d'un projet suffit toujours à reprendre le travail au tour suivant. projets// entrees/ brief.md, charte.pdf, logo.png, photos… conversation.json l'état complet du dialogue avec le modèle site/ ce que le modèle écrit livraison.zip """ import json import os import re import sqlite3 import time import unicodedata import config # --- États d'un projet ------------------------------------------------------ BROUILLON = "brouillon" # saisi, pas encore lancé EN_ATTENTE = "en_attente" # dans la file, le worker va le prendre EN_COURS = "en_cours" # génération commencée TERMINE = "termine" # le modèle a appelé terminer() ECHEC = "echec" # abandon après erreurs répétées INTERROMPU = "interrompu" # arrêté par l'opérateur ETATS_ACTIFS = (EN_ATTENTE, EN_COURS) def _connexion(): cx = sqlite3.connect(config.BASE, timeout=30) cx.row_factory = sqlite3.Row cx.execute("PRAGMA journal_mode=WAL") cx.execute("PRAGMA busy_timeout=30000") return cx def initialiser(): """Crée la base si besoin. Idempotent : appelé à chaque démarrage.""" os.makedirs(config.DOSSIER_PROJETS, exist_ok=True) with _connexion() as cx: cx.executescript( """ CREATE TABLE IF NOT EXISTS projets ( id TEXT PRIMARY KEY, nom TEXT NOT NULL, etat TEXT NOT NULL, cree_le REAL NOT NULL, maj_le REAL NOT NULL, tours INTEGER NOT NULL DEFAULT 0, resume TEXT, erreur TEXT, echecs INTEGER NOT NULL DEFAULT 0, jetons_entree INTEGER NOT NULL DEFAULT 0, jetons_sortie INTEGER NOT NULL DEFAULT 0, jetons_cache_lecture INTEGER NOT NULL DEFAULT 0, jetons_cache_ecriture INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS journal ( id INTEGER PRIMARY KEY AUTOINCREMENT, projet_id TEXT NOT NULL, horodatage REAL NOT NULL, evenement TEXT NOT NULL, detail TEXT ); CREATE INDEX IF NOT EXISTS idx_journal_projet ON journal (projet_id, id); CREATE INDEX IF NOT EXISTS idx_projets_etat ON projets (etat, maj_le); """ ) def identifiant(nom): """Fabrique un identifiant de dossier lisible à partir du nom du client.""" base = unicodedata.normalize("NFKD", nom or "projet") base = base.encode("ascii", "ignore").decode("ascii").lower() base = re.sub(r"[^a-z0-9]+", "-", base).strip("-") or "projet" base = base[:40] prefixe = time.strftime("%Y%m%d-%H%M") return "%s-%s" % (prefixe, base) # --- Chemins ---------------------------------------------------------------- def dossier(projet_id): return os.path.join(config.DOSSIER_PROJETS, projet_id) def dossier_entrees(projet_id): return os.path.join(dossier(projet_id), "entrees") def dossier_site(projet_id): return os.path.join(dossier(projet_id), "site") def chemin_conversation(projet_id): return os.path.join(dossier(projet_id), "conversation.json") def chemin_archive(projet_id): return os.path.join(dossier(projet_id), "livraison.zip") # --- Écriture atomique ------------------------------------------------------ def _ecrire_atomique(chemin, texte): """Écrit un fichier sans jamais laisser de version tronquée derrière soi.""" temporaire = "%s.tmp.%d" % (chemin, os.getpid()) with open(temporaire, "w", encoding="utf-8") as f: f.write(texte) f.flush() os.fsync(f.fileno()) os.replace(temporaire, chemin) def lire_conversation(projet_id): chemin = chemin_conversation(projet_id) if not os.path.exists(chemin): return None with open(chemin, encoding="utf-8") as f: return json.load(f) def ecrire_conversation(projet_id, conversation): _ecrire_atomique( chemin_conversation(projet_id), json.dumps(conversation, ensure_ascii=False, indent=1), ) # --- Projets ---------------------------------------------------------------- def creer(nom): projet_id = identifiant(nom) os.makedirs(dossier_entrees(projet_id), exist_ok=True) os.makedirs(dossier_site(projet_id), exist_ok=True) maintenant = time.time() with _connexion() as cx: cx.execute( "INSERT INTO projets (id, nom, etat, cree_le, maj_le) VALUES (?,?,?,?,?)", (projet_id, nom, BROUILLON, maintenant, maintenant), ) journaliser(projet_id, "cree", nom) return projet_id def lire(projet_id): with _connexion() as cx: ligne = cx.execute("SELECT * FROM projets WHERE id=?", (projet_id,)).fetchone() return dict(ligne) if ligne else None def lister(limite=50): with _connexion() as cx: lignes = cx.execute( "SELECT * FROM projets ORDER BY cree_le DESC LIMIT ?", (limite,) ).fetchall() return [dict(l) for l in lignes] def changer_etat(projet_id, etat, erreur=None, resume=None): with _connexion() as cx: cx.execute( "UPDATE projets SET etat=?, maj_le=?, " "erreur=COALESCE(?, erreur), resume=COALESCE(?, resume) WHERE id=?", (etat, time.time(), erreur, resume, projet_id), ) journaliser(projet_id, "etat", etat if not erreur else "%s : %s" % (etat, erreur)) def compter_tour(projet_id, usage=None): """Enregistre un tour d'API et cumule la consommation de jetons.""" usage = usage or {} with _connexion() as cx: cx.execute( "UPDATE projets SET tours = tours + 1, maj_le = ?, " "jetons_entree = jetons_entree + ?, " "jetons_sortie = jetons_sortie + ?, " "jetons_cache_lecture = jetons_cache_lecture + ?, " "jetons_cache_ecriture = jetons_cache_ecriture + ? " "WHERE id = ?", ( time.time(), int(usage.get("input_tokens") or 0), int(usage.get("output_tokens") or 0), int(usage.get("cache_read_input_tokens") or 0), int(usage.get("cache_creation_input_tokens") or 0), projet_id, ), ) def compter_echec(projet_id, remise_a_zero=False): """Suit les échecs consécutifs, pour abandonner une génération qui boucle.""" with _connexion() as cx: if remise_a_zero: cx.execute("UPDATE projets SET echecs=0 WHERE id=?", (projet_id,)) return 0 cx.execute( "UPDATE projets SET echecs = echecs + 1, maj_le=? WHERE id=?", (time.time(), projet_id), ) ligne = cx.execute("SELECT echecs FROM projets WHERE id=?", (projet_id,)).fetchone() return ligne["echecs"] if ligne else 0 def prochain_en_attente(): """Le projet le plus ancien à traiter. En_cours d'abord : on finit ce qu'on a commencé avant d'en démarrer un autre.""" with _connexion() as cx: ligne = cx.execute( "SELECT * FROM projets WHERE etat IN (?,?) " "ORDER BY CASE etat WHEN ? THEN 0 ELSE 1 END, maj_le ASC LIMIT 1", (EN_COURS, EN_ATTENTE, EN_COURS), ).fetchone() return dict(ligne) if ligne else None # --- Journal ---------------------------------------------------------------- def journaliser(projet_id, evenement, detail=None): with _connexion() as cx: cx.execute( "INSERT INTO journal (projet_id, horodatage, evenement, detail) VALUES (?,?,?,?)", (projet_id, time.time(), evenement, detail), ) def journal(projet_id, depuis=0, limite=200): with _connexion() as cx: lignes = cx.execute( "SELECT * FROM journal WHERE projet_id=? AND id>? ORDER BY id ASC LIMIT ?", (projet_id, depuis, limite), ).fetchall() return [dict(l) for l in lignes] # --- Coût ------------------------------------------------------------------- # Tarifs Opus 5, en dollars par million de jetons. À ajuster si la grille bouge. TARIFS = { "entree": 5.0, "sortie": 25.0, "cache_lecture": 0.50, # un dixième de l'entrée "cache_ecriture": 6.25, # 1,25 fois l'entrée } def cout_dollars(projet): """Coût cumulé d'un projet, en dollars.""" return round( projet.get("jetons_entree", 0) / 1e6 * TARIFS["entree"] + projet.get("jetons_sortie", 0) / 1e6 * TARIFS["sortie"] + projet.get("jetons_cache_lecture", 0) / 1e6 * TARIFS["cache_lecture"] + projet.get("jetons_cache_ecriture", 0) / 1e6 * TARIFS["cache_ecriture"], 4, )