Public Access
202 lines
7.9 KiB
Python
202 lines
7.9 KiB
Python
import os
|
|
import json
|
|
import time
|
|
import requests
|
|
from datetime import datetime
|
|
import tiktoken
|
|
from modules.logger import log_event
|
|
|
|
# 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_FILE = os.path.join(BASE_DIR, "tarifs_openrouter.json")
|
|
|
|
VIP_ADMIN_USERS = ["xavier", "céline", "celine", "anaëlle", "anaelle", "renan", "audric"]
|
|
|
|
def obtenir_tarifs_dynamiques() -> dict:
|
|
maintenant = time.time()
|
|
|
|
if os.path.exists(TARIFS_FILE):
|
|
if maintenant - os.path.getmtime(TARIFS_FILE) < 7200:
|
|
try:
|
|
with open(TARIFS_FILE, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
pass
|
|
|
|
tarifs = {}
|
|
try:
|
|
resp = requests.get("https://openrouter.ai/api/v1/models", timeout=10)
|
|
if resp.status_code == 200:
|
|
data = resp.json().get("data", [])
|
|
for m in data:
|
|
pricing = m.get("pricing", {})
|
|
p_prompt = float(pricing.get("prompt", 0.0))
|
|
p_comp = float(pricing.get("completion", 0.0))
|
|
tarifs[m["id"]] = max(p_prompt, p_comp)
|
|
|
|
with open(TARIFS_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(tarifs, f, indent=4)
|
|
log_event("INFO", "BILLING", "Tarifs OpenRouter mis à jour dynamiquement.")
|
|
return tarifs
|
|
except Exception as e:
|
|
log_event("ERROR", "BILLING", f"Échec récupération des tarifs : {e}")
|
|
|
|
if os.path.exists(TARIFS_FILE):
|
|
try:
|
|
with open(TARIFS_FILE, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
pass
|
|
return {}
|
|
|
|
def charger_soldes() -> dict:
|
|
if os.path.exists(BALANCES_FILE):
|
|
try:
|
|
with open(BALANCES_FILE, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
pass
|
|
soldes = {"cagnotte_commune_vip": 35.00}
|
|
sauvegarder_soldes(soldes)
|
|
return soldes
|
|
|
|
def sauvegarder_soldes(soldes: dict):
|
|
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"):
|
|
entree = {
|
|
"timestamp": time.time(),
|
|
"date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
"utilisateur": utilisateur,
|
|
"rang": rang,
|
|
"type_operation": type_op,
|
|
"modele": modele,
|
|
"tokens": tokens,
|
|
"cout_brut_api": cout_brut,
|
|
"prix_facture_client": prix_facture
|
|
}
|
|
audits = []
|
|
if os.path.exists(AUDIT_FILE):
|
|
try:
|
|
with open(AUDIT_FILE, "r", encoding="utf-8") as f:
|
|
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:
|
|
try:
|
|
enc = tiktoken.get_encoding(modele)
|
|
return len(enc.encode(texte))
|
|
except Exception as e:
|
|
log_event("ERROR", "BILLING", f"Erreur comptage tokens : {e}")
|
|
return 0
|
|
|
|
def estimer_cout_facture(prompt: str, modele_id: str, rang_utilisateur: str = "standard", multi_ia: bool = False) -> tuple[float, float, int]:
|
|
tokens = compter_tokens(prompt)
|
|
tarifs = obtenir_tarifs_dynamiques()
|
|
cout_unitaire = tarifs.get(modele_id, 0.0)
|
|
cout_brut = tokens * cout_unitaire
|
|
|
|
if multi_ia and cout_brut > 0:
|
|
cout_brut *= 2
|
|
|
|
if cout_brut == 0.0:
|
|
return 0.0, 0.0, tokens
|
|
|
|
rang = rang_utilisateur.lower()
|
|
if rang in ["super admin", "admin", "vip"] or rang in VIP_ADMIN_USERS:
|
|
prix_facture = cout_brut
|
|
elif rang == "famille":
|
|
prix_facture = cout_brut / 0.925
|
|
else:
|
|
prix_facture = cout_brut / 0.85
|
|
|
|
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):
|
|
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()
|
|
|
|
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"
|
|
else:
|
|
solde_disponible = soldes.get(utilisateur_cle, 0.0)
|
|
cle_compte = utilisateur_cle
|
|
|
|
if solde_disponible < prix_facture:
|
|
log_event("WARNING", "BILLING", f"Rejet : solde insuffisant pour {utilisateur_nom} ({solde_disponible} $ < {prix_facture} $)")
|
|
raise Exception(f"SOLDE_INSUFFISANT|{prix_facture}|{solde_disponible}")
|
|
|
|
soldes[cle_compte] = round(solde_disponible - prix_facture, 6)
|
|
sauvegarder_soldes(soldes)
|
|
|
|
enregistrer_transaction_audit(utilisateur_nom, rang_utilisateur, modele_id, tokens, cout_brut, prix_facture)
|
|
log_event("INFO", "BILLING", f"Requête validée [{rang_utilisateur}] - Débit : {prix_facture} $ - Solde restant : {soldes[cle_compte]} $")
|
|
return True, prix_facture
|
|
|
|
def obtenir_solde(utilisateur_nom: str) -> float:
|
|
# --- SYNCHRONISATION DIRECTE POUR L'ADMINISTRATEUR PRINCIPAL ---
|
|
if utilisateur_nom.lower() in ["xavier"]:
|
|
api_key = os.getenv("OPENROUTER_API_KEY")
|
|
if api_key:
|
|
try:
|
|
resp = requests.get("https://openrouter.ai/api/v1/auth/key", headers={"Authorization": f"Bearer {api_key}"}, timeout=3)
|
|
if resp.status_code == 200:
|
|
data = resp.json().get("data", {})
|
|
|
|
credit_total = data.get('usage', 0.0)
|
|
limite_restante = data.get('limit_remaining')
|
|
balance = data.get('balance', 0.0)
|
|
|
|
vrai_solde = float(limite_restante) if limite_restante is not None else float(balance)
|
|
|
|
if vrai_solde > 0 or limite_restante is not None:
|
|
soldes = charger_soldes()
|
|
soldes["cagnotte_commune_vip"] = vrai_solde
|
|
soldes["xavier"] = vrai_solde
|
|
sauvegarder_soldes(soldes)
|
|
return vrai_solde
|
|
except Exception as e:
|
|
log_event("WARNING", "BILLING", f"Échec de synchronisation OpenRouter : {e}")
|
|
|
|
# --- COMPORTEMENT STANDARD POUR LES AUTRES UTILISATEURS ---
|
|
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:
|
|
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)
|
|
except Exception:
|
|
return []
|
|
|
|
dossier = []
|
|
for item in tous_les_audits:
|
|
if item.get("utilisateur", "").lower() == utilisateur_nom.lower():
|
|
if est_admin:
|
|
dossier.append(item)
|
|
else:
|
|
dossier.append({
|
|
"date": item.get("date"),
|
|
"type_operation": item.get("type_operation"),
|
|
"modele": item.get("modele"),
|
|
"tokens": item.get("tokens"),
|
|
"montant_facture_ttc": item.get("prix_facture_client")
|
|
})
|
|
return dossier |