Mise en place gestion financiére
Build and Push Docker Image / build-and-push (push) Successful in 35s

This commit is contained in:
xavier committed 2026-08-29 08:30:53 +02:00
1 parent 5d4ac62b87
commit 7ef450da91
3 files changed
+70 -9

No files matched your search

+65 -4
View File
@@ -13,8 +13,50 @@ 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 est conservé pour la compatibilité avec l'ancien code,
# mais la logique stricte des rôles est maintenant dans la fonction d'estimation.
VIP_ADMIN_USERS = ["xavier", "céline", "celine", "anaëlle", "anaelle", "renan", "audric"]
# -----------------------------------------------------------------------------
# NOUVEAU MODULE : CONVERSION DE DEVISES (API FRANKFURTER)
# -----------------------------------------------------------------------------
TAUX_DEVISE_FILE = os.path.join(BASE_DIR, "taux_devise.json")
def obtenir_taux_usd_eur() -> float:
"""
Récupère le taux de change USD -> EUR via l'API Frankfurter.
Met en cache le résultat pendant 12 heures pour ne pas ralentir le serveur.
"""
maintenant = time.time()
# Vérification du cache (12 heures = 43200 secondes)
if os.path.exists(TAUX_DEVISE_FILE):
if maintenant - os.path.getmtime(TAUX_DEVISE_FILE) < 43200:
try:
with open(TAUX_DEVISE_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
return float(data.get("taux", 0.92)) # 0.92 par défaut (sécurité)
except Exception:
pass
try:
# Appel à l'API publique (gratuite et sans clé) basée sur la BCE
resp = requests.get("https://api.frankfurter.app/latest?from=USD&to=EUR", timeout=5)
if resp.status_code == 200:
taux = resp.json().get("rates", {}).get("EUR", 0.92)
# Sauvegarde en cache
with open(TAUX_DEVISE_FILE, "w", encoding="utf-8") as f:
json.dump({"taux": taux, "timestamp": maintenant}, f)
log_event("INFO", "BILLING", f"Taux USD->EUR mis à jour : {taux}")
return taux
except Exception as e:
log_event("WARNING", "BILLING", f"Échec récupération du taux de change : {e}")
return 0.92 # Taux de repli fixe en cas de panne réseau
# -----------------------------------------------------------------------------
# GESTION DES TARIFS OPENROUTER
# -----------------------------------------------------------------------------
def obtenir_tarifs_dynamiques() -> dict:
maintenant = time.time()
@@ -68,6 +110,11 @@ def sauvegarder_soldes(soldes: dict):
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"):
# Conversion du coût final en Euros pour l'export comptable URSSAF
taux_actuel = obtenir_taux_usd_eur()
cout_brut_eur = cout_brut * taux_actuel
prix_facture_eur = prix_facture * taux_actuel
entree = {
"timestamp": time.time(),
"date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
@@ -77,7 +124,10 @@ def enregistrer_transaction_audit(utilisateur: str, rang: str, modele: str, toke
"modele": modele,
"tokens": tokens,
"cout_brut_api": cout_brut,
"prix_facture_client": prix_facture
"prix_facture_client": prix_facture,
# Données Euro pour le Livre de Recettes (Pandas)
"cout_brut_api_eur": cout_brut_eur,
"prix_facture_client_eur": prix_facture_eur
}
audits = []
if os.path.exists(AUDIT_FILE):
@@ -98,6 +148,9 @@ def compter_tokens(texte: str, modele: str = "cl100k_base") -> int:
log_event("ERROR", "BILLING", f"Erreur comptage tokens : {e}")
return 0
# -----------------------------------------------------------------------------
# MOTEUR DE CALCUL DES MARGES COMMERCIALES (Règles strictes)
# -----------------------------------------------------------------------------
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()
@@ -111,10 +164,16 @@ def estimer_cout_facture(prompt: str, modele_id: str, rang_utilisateur: str = "s
return 0.0, 0.0, tokens
rang = rang_utilisateur.lower()
if rang in ["super admin", "admin", "vip"] or rang in VIP_ADMIN_USERS:
# 1. PAS DE MARGE (Super Admin, Admin, Famille)
if rang in ["super admin", "admin", "famille"]:
prix_facture = cout_brut
elif rang == "famille":
# 2. MARGE RÉDUITE 7.5% (Utilisateurs VIP choisis par Admin)
elif rang == "vip":
prix_facture = cout_brut / 0.925
# 3. MARGE STANDARD 15% (Utilisateurs par défaut)
else:
prix_facture = cout_brut / 0.85
@@ -128,9 +187,11 @@ def verifier_et_valider_budget(prompt: str, modele_id: str, utilisateur_nom: str
soldes = charger_soldes()
utilisateur_cle = utilisateur_nom.lower()
if utilisateur_cle in VIP_ADMIN_USERS or rang_utilisateur.lower() in ["super admin", "admin", "vip"]:
# Les rôles protégés tapent dans la cagnotte globale (carte bancaire de l'admin)
if utilisateur_cle in VIP_ADMIN_USERS or rang_utilisateur.lower() in ["super admin", "admin", "famille"]:
solde_disponible = soldes.get("cagnotte_commune_vip", 0.0)
cle_compte = "cagnotte_commune_vip"
# Les utilisateurs payants (VIP / Standard) utilisent leur propre recharge
else:
solde_disponible = soldes.get(utilisateur_cle, 0.0)
cle_compte = utilisateur_cle
+3 -3
View File
@@ -37,9 +37,9 @@ def generer_export_comptable():
"Référence (N° Facture / Chat ID)": "EXEMPLE-001",
"Client (Utilisateur)": "Exemple_Client",
"Nature de la prestation": "Services techniques IA (Prestation de services)",
"Coût API Brut HT (€)": 0.05,
"Marge Commerciale (€)": 0.45,
"Montant encaissé HT (€)": 0.50
"Coût API Brut HT (€)": float(transaction.get("cout_brut_api_eur", transaction.get("cout_brut_api", 0.0))),
"Marge Commerciale (€)": float(transaction.get("prix_facture_client_eur", 0.0)) - float(transaction.get("cout_brut_api_eur", 0.0)),
"Montant encaissé HT (€)": float(transaction.get("prix_facture_client_eur", transaction.get("prix_facture_client", 0.0)))
})
else:
for transaction in donnees_brutes: