Public Access
77 lines
3.2 KiB
Python
77 lines
3.2 KiB
Python
import os
|
|
import json
|
|
import time
|
|
import requests
|
|
from modules.logger import log_event
|
|
|
|
CACHE_MODELS_FILE = "openrouter_models_cache.json"
|
|
CACHE_TTL_SECONDES = 86400 # Rafraîchissement automatique toutes les 24 heures
|
|
|
|
def synchroniser_catalogue_openrouter(forcer: bool = False) -> dict:
|
|
"""
|
|
Interroge l'API OpenRouter pour récupérer en direct tous les modèles,
|
|
leurs spécialités et leurs tarifs exacts (Prompt & Completion).
|
|
Met en cache le résultat localement.
|
|
"""
|
|
# Vérification du cache local
|
|
if not forcer and os.path.exists(CACHE_MODELS_FILE):
|
|
try:
|
|
with open(CACHE_MODELS_FILE, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
if time.time() - data.get("derniere_synchro", 0) < CACHE_TTL_SECONDES:
|
|
return data.get("modeles", {})
|
|
except Exception as e:
|
|
log_event("WARNING", "OPENROUTER_SYNC", f"Erreur lecture cache modèles : {e}")
|
|
|
|
url = "https://openrouter.ai/api/v1/models"
|
|
headers = {
|
|
"Authorization": f"Bearer {os.getenv('OPENROUTER_API_KEY', '')}",
|
|
"HTTP-Referer": "https://aethas38.duckdns.org",
|
|
"X-Title": "Aethas38 Multi-IA"
|
|
}
|
|
|
|
try:
|
|
reponse = requests.get(url, headers=headers, timeout=10)
|
|
reponse.raise_for_status()
|
|
catalogue_brut = reponse.json().get("data", [])
|
|
|
|
catalogue_traite = {}
|
|
for item in catalogue_brut:
|
|
modele_id = item.get("id")
|
|
pricing = item.get("pricing", {})
|
|
|
|
# Conversion des prix en float (prix par token unitaire)
|
|
prix_prompt = float(pricing.get("prompt", 0.0) or 0.0)
|
|
prix_completion = float(pricing.get("completion", 0.0) or 0.0)
|
|
|
|
catalogue_traite[modele_id] = {
|
|
"nom": item.get("name", modele_id),
|
|
"contexte_max": item.get("context_length", 4096),
|
|
"est_gratuit": prix_prompt == 0.0 and prix_completion == 0.0,
|
|
"prix_token_prompt": prix_prompt,
|
|
"prix_token_completion": prix_completion
|
|
}
|
|
|
|
# Sauvegarde en cache
|
|
with open(CACHE_MODELS_FILE, "w", encoding="utf-8") as f:
|
|
json.dump({
|
|
"derniere_synchro": time.time(),
|
|
"total_modeles": len(catalogue_traite),
|
|
"modeles": catalogue_traite
|
|
}, f, indent=4)
|
|
|
|
log_event("INFO", "OPENROUTER_SYNC", f"Catalogue OpenRouter synchronisé : {len(catalogue_traite)} modèles chargés.")
|
|
return catalogue_traite
|
|
|
|
except Exception as e:
|
|
log_event("ERROR", "OPENROUTER_SYNC", f"Échec de synchronisation OpenRouter : {e}")
|
|
# En cas d'échec réseau, on tente de renvoyer le vieux cache existant
|
|
if os.path.exists(CACHE_MODELS_FILE):
|
|
with open(CACHE_MODELS_FILE, "r", encoding="utf-8") as f:
|
|
return json.load(f).get("modeles", {})
|
|
return {}
|
|
|
|
def obtenir_modeles_gratuits() -> list:
|
|
"""Renvoie uniquement la liste des slugs des modèles 100 % gratuits."""
|
|
catalogue = synchroniser_catalogue_openrouter()
|
|
return [slug for slug, info in catalogue.items() if info.get("est_gratuit", False)] |