MAJ Globale : Chemins absolus (NAS), Model Updater auto, Timeout 90s OpenRouter, Pop-up et traçabilité financière
Build and Push Docker Image / build-and-push (push) Successful in 1m48s

This commit is contained in:
xavier committed 2026-08-22 03:08:57 +02:00
1 parent b4ed2947d4
commit 632c23069a
9 files changed
+66 -89

No files matched your search

+5 -27
View File
@@ -5,8 +5,11 @@ from datetime import datetime
import tiktoken
from modules.logger import log_event
BALANCES_FILE = "aethas_balances.json"
AUDIT_FILE = "aethas_financial_audit.json"
# FORCAGE DES CHEMINS ABSOLUS
BASE_DIR = "/DATA/AppData/MULTI-IA-AETHAS38"
os.makedirs(BASE_DIR, exist_ok=True)
BALANCES_FILE = os.path.join(BASE_DIR, "aethas_balances.json")
AUDIT_FILE = os.path.join(BASE_DIR, "aethas_financial_audit.json")
TARIFS_OPENROUTER = {
"meta-llama/llama-3.1-8b-instruct:free": 0.0,
@@ -21,25 +24,21 @@ TARIFS_OPENROUTER = {
VIP_ADMIN_USERS = ["xavier", "céline", "celine", "anaëlle", "anaelle", "renan", "audric"]
def charger_soldes() -> dict:
"""Charge les soldes réels depuis le stockage persistant."""
if os.path.exists(BALANCES_FILE):
try:
with open(BALANCES_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
# Initialisation de base : cagnotte commune de 35.00 $
soldes = {"cagnotte_commune_vip": 35.00}
sauvegarder_soldes(soldes)
return soldes
def sauvegarder_soldes(soldes: dict):
"""Sauvegarde les soldes sur disque."""
with open(BALANCES_FILE, "w", encoding="utf-8") as f:
json.dump(soldes, f, indent=4)
def enregistrer_transaction_audit(utilisateur: str, rang: str, modele: str, tokens: int, cout_brut: float, prix_facture: float, type_op: str = "DEBIT_REQUETE"):
"""Journalise chaque opération financière de manière inaltérable pour les litiges."""
entree = {
"timestamp": time.time(),
"date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
@@ -51,7 +50,6 @@ def enregistrer_transaction_audit(utilisateur: str, rang: str, modele: str, toke
"cout_brut_api": cout_brut,
"prix_facture_client": prix_facture
}
audits = []
if os.path.exists(AUDIT_FILE):
try:
@@ -59,13 +57,11 @@ def enregistrer_transaction_audit(utilisateur: str, rang: str, modele: str, toke
audits = json.load(f)
except Exception:
audits = []
audits.append(entree)
with open(AUDIT_FILE, "w", encoding="utf-8") as f:
json.dump(audits, f, indent=4)
def compter_tokens(texte: str, modele: str = "cl100k_base") -> int:
"""Compte le nombre de tokens d'un texte."""
try:
enc = tiktoken.get_encoding(modele)
return len(enc.encode(texte))
@@ -74,12 +70,6 @@ def compter_tokens(texte: str, modele: str = "cl100k_base") -> int:
return 0
def estimer_cout_facture(prompt: str, modele_id: str, rang_utilisateur: str = "standard", multi_ia: bool = False) -> tuple[float, float, int]:
"""
Calcule le coût brut API et le prix facturé selon le rang :
- VIP / Admin : marge 0 % (coût réel déduit de la cagnotte commune)
- Famille : marge 7,5 % (coût / 0.925)
- Standard : marge 15 % (coût / 0.85)
"""
tokens = compter_tokens(prompt)
cout_unitaire = TARIFS_OPENROUTER.get(modele_id, 0.0)
cout_brut = tokens * cout_unitaire
@@ -101,16 +91,13 @@ def estimer_cout_facture(prompt: str, modele_id: str, rang_utilisateur: str = "s
return round(cout_brut, 6), round(prix_facture, 6), tokens
def verifier_et_valider_budget(prompt: str, modele_id: str, utilisateur_nom: str = "xavier", rang_utilisateur: str = "super admin", multi_ia: bool = False):
"""Vérifie le solde avant exécution et applique le débit approprié."""
cout_brut, prix_facture, tokens = estimer_cout_facture(prompt, modele_id, rang_utilisateur, multi_ia)
if prix_facture == 0.0:
return True, 0.0
soldes = charger_soldes()
utilisateur_cle = utilisateur_nom.lower()
# Distinction cagnotte commune VIP/Admin vs compte individuel
if utilisateur_cle in VIP_ADMIN_USERS or rang_utilisateur.lower() in ["super admin", "admin", "vip"]:
solde_disponible = soldes.get("cagnotte_commune_vip", 0.0)
cle_compte = "cagnotte_commune_vip"
@@ -122,7 +109,6 @@ def verifier_et_valider_budget(prompt: str, modele_id: str, utilisateur_nom: str
log_event("WARNING", "BILLING", f"Rejet : solde insuffisant pour {utilisateur_nom} ({solde_disponible} $ < {prix_facture} $)")
raise Exception(f"SOLDE_INSUFFISANT|{prix_facture}|{solde_disponible}")
# Débit strict
soldes[cle_compte] = round(solde_disponible - prix_facture, 6)
sauvegarder_soldes(soldes)
@@ -131,21 +117,14 @@ def verifier_et_valider_budget(prompt: str, modele_id: str, utilisateur_nom: str
return True, prix_facture
def obtenir_solde(utilisateur_nom: str) -> float:
"""Retourne le solde actuel de l'utilisateur ou de la cagnotte commune."""
soldes = charger_soldes()
if utilisateur_nom.lower() in VIP_ADMIN_USERS:
return soldes.get("cagnotte_commune_vip", 0.0)
return soldes.get(utilisateur_nom.lower(), 0.0)
def generer_dossier_litige(utilisateur_nom: str, est_admin: bool = False) -> list:
"""
Génère les éléments d'audit pour un litige :
- Pour l'utilisateur : date, modèle, tokens, prix facturé (marge masquée).
- Pour l'Admin : audit complet avec coût brut API.
"""
if not os.path.exists(AUDIT_FILE):
return []
try:
with open(AUDIT_FILE, "r", encoding="utf-8") as f:
tous_les_audits = json.load(f)
@@ -158,7 +137,6 @@ def generer_dossier_litige(utilisateur_nom: str, est_admin: bool = False) -> lis
if est_admin:
dossier.append(item)
else:
# Filtrage des marges pour le client
dossier.append({
"date": item.get("date"),
"type_operation": item.get("type_operation"),