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: """Récupère les tarifs via API toutes les 2 heures maximum pour éviter le hardcoding.""" 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)) # On sécurise l'estimation en prenant la valeur la plus haute entre tokens in et out 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}") # Repli sur le cache si l'API est injoignable 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: 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 def calculer_prix_recharge_ttc(montant_credits_usd: float, methode: str = "paypal") -> float: """ Calcule le coût final pour l'utilisateur qui souhaite acheter une recharge de crédits. Le coût final répercute les frais bancaires pour que l'entreprise soit à l'équilibre. Frais estimés (Variables selon devises, basés sur des profils Sandbox/Standard) : - PayPal : 2.9% + 0.35$ fixes par transaction - Revolut Business : 1.0% + 0.20$ fixes par transaction en ligne Formule d'absorption des frais : Montant_Final = (Credits + Frais_Fixes) / (1 - Pourcentage_Frais) """ if methode.lower() == "paypal": frais_fixes = 0.35 frais_variable = 0.029 elif methode.lower() == "revolut": frais_fixes = 0.20 frais_variable = 0.010 else: return montant_credits_usd # Aucune méthode bancaire spécifiée montant_a_payer = (montant_credits_usd + frais_fixes) / (1 - frais_variable) return round(montant_a_payer, 2)