Public Access
142 lines
6.3 KiB
Python
142 lines
6.3 KiB
Python
import os
|
|
import requests
|
|
import time
|
|
import concurrent.futures
|
|
import base64
|
|
from modules.logger import log_event
|
|
|
|
def generer_image_cloudflare(prompt: str) -> bytes:
|
|
account_id = os.getenv("CLOUDFLARE_ACCOUNT_ID")
|
|
api_token = os.getenv("CLOUDFLARE_API_TOKEN")
|
|
if not account_id or not api_token: return None
|
|
|
|
url = f"https://api.cloudflare.com/client/v4/accounts/{account_id}/ai/run/@cf/stabilityai/stable-diffusion-xl-base-1.0"
|
|
headers = {"Authorization": f"Bearer {api_token}"}
|
|
|
|
try:
|
|
resp = requests.post(url, headers=headers, json={"prompt": prompt}, timeout=120)
|
|
if resp.status_code == 200:
|
|
if "application/json" in resp.headers.get("Content-Type", ""):
|
|
data = resp.json()
|
|
if data.get("success") and "result" in data:
|
|
return base64.b64decode(data["result"])
|
|
else:
|
|
return resp.content
|
|
except Exception as e:
|
|
log_event("WARNING", "MEDIA", f"Échec Cloudflare : {e}")
|
|
return None
|
|
|
|
def generer_image_openrouter(prompt: str) -> tuple:
|
|
api_key = os.getenv("OPENROUTER_API_KEY")
|
|
if not api_key: return None, None
|
|
|
|
url = "https://openrouter.ai/api/v1/images"
|
|
headers = {
|
|
"Authorization": f"Bearer {api_key}",
|
|
"HTTP-Referer": "https://aethas38.duckdns.org",
|
|
"X-Title": "Aethas38"
|
|
}
|
|
modele = "bytedance-seed/seedream-4.5"
|
|
payload = {"prompt": prompt, "model": modele}
|
|
|
|
try:
|
|
resp = requests.post(url, headers=headers, json=payload, timeout=120)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
if "data" in data and len(data["data"]) > 0 and "b64_json" in data["data"][0]:
|
|
img_bytes = base64.b64decode(data["data"][0]["b64_json"])
|
|
return img_bytes, modele
|
|
except Exception as e:
|
|
log_event("WARNING", "MEDIA", f"Échec OpenRouter Image : {e}")
|
|
return None, None
|
|
|
|
def generer_image_huggingface(prompt: str) -> tuple:
|
|
api_key = os.getenv("HUGGINGFACE_API_KEY")
|
|
if not api_key: return None, "Clé API manquante"
|
|
|
|
url = "https://router.huggingface.co/hf-inference/models/stabilityai/stable-diffusion-xl-base-1.0"
|
|
headers = {"Authorization": f"Bearer {api_key}"}
|
|
|
|
try:
|
|
resp = requests.post(url, headers=headers, json={"inputs": prompt}, timeout=120)
|
|
if resp.status_code == 200:
|
|
return resp.content, "Succès"
|
|
else:
|
|
return None, f"Erreur HTTP {resp.status_code} : {resp.text}"
|
|
except Exception as e:
|
|
log_event("WARNING", "MEDIA", f"Échec HF : {e}")
|
|
return None, f"Exception système : {str(e)}"
|
|
|
|
def generer_les_images(prompt: str, fichiers: list, premium: bool) -> tuple:
|
|
trace = []
|
|
if fichiers:
|
|
trace.append("🟢 Mode Image-to-Image")
|
|
return f"*(BETA)* Le mode Image-to-Image est désactivé temporairement. \n\nFichiers : {', '.join(fichiers)}", "#ff9800", "Système", "En dev", trace
|
|
|
|
trace.append(f"🟢 Lancement Text-to-Image (Premium: {premium})")
|
|
|
|
# Exécution en parallèle (OpenRouter n'est lancé QUE si premium est True)
|
|
with concurrent.futures.ThreadPoolExecutor() as executor:
|
|
future_or = executor.submit(generer_image_openrouter, prompt) if premium else None
|
|
future_cf = executor.submit(generer_image_cloudflare, prompt)
|
|
future_hf = executor.submit(generer_image_huggingface, prompt)
|
|
|
|
img_or, nom_modele_or = future_or.result() if future_or else (None, None)
|
|
img_cf = future_cf.result()
|
|
img_hf, msg_hf = future_hf.result()
|
|
|
|
dossier_media = "/DATA/AppData/MULTI-IA-AETHAS38/RAG/Media"
|
|
os.makedirs(dossier_media, exist_ok=True)
|
|
|
|
reponse_md = f"**Prompt analysé :** {prompt}\n\n"
|
|
fournisseurs_actifs = []
|
|
|
|
# Traitement OpenRouter
|
|
if future_or is None:
|
|
trace.append("⚪ OR : Ignoré (Mode Gratuit)")
|
|
elif img_or:
|
|
nom_or = f"img_or_{int(time.time())}.png"
|
|
with open(os.path.join(dossier_media, nom_or), 'wb') as f: f.write(img_or)
|
|
reponse_md += f"### 🌐 OpenRouter ({nom_modele_or})\n\n"
|
|
reponse_md += f"**[📥 Télécharger l'image OR](/Media/{nom_or})**\n\n---\n\n"
|
|
fournisseurs_actifs.append("OpenRouter")
|
|
trace.append("🟢 OR : Succès")
|
|
else:
|
|
trace.append("🔴 OR : Échec")
|
|
|
|
# Traitement Cloudflare
|
|
if img_cf:
|
|
nom_cf = f"img_cf_{int(time.time())}.png"
|
|
with open(os.path.join(dossier_media, nom_cf), 'wb') as f: f.write(img_cf)
|
|
reponse_md += f"### ☁️ Cloudflare Workers AI\n\n"
|
|
reponse_md += f"**[📥 Télécharger l'image CF](/Media/{nom_cf})**\n\n---\n\n"
|
|
fournisseurs_actifs.append("Cloudflare")
|
|
trace.append("🟢 CF : Succès")
|
|
else:
|
|
trace.append("🔴 CF : Échec")
|
|
|
|
# Traitement HuggingFace
|
|
if img_hf:
|
|
nom_hf = f"img_hf_{int(time.time())}.png"
|
|
with open(os.path.join(dossier_media, nom_hf), 'wb') as f: f.write(img_hf)
|
|
reponse_md += f"### 🤗 HuggingFace (SDXL)\n\n"
|
|
reponse_md += f"**[📥 Télécharger l'image HF](/Media/{nom_hf})**\n\n"
|
|
fournisseurs_actifs.append("HuggingFace")
|
|
trace.append("🟢 HF : Succès")
|
|
else:
|
|
erreur_courte = str(msg_hf)[:50] + "..." if len(str(msg_hf)) > 50 else str(msg_hf)
|
|
trace.append(f"🔴 HF : Échec ({erreur_courte})")
|
|
reponse_md += f"### 🤗 HuggingFace (SDXL)\n*⚠️ Échec de la génération : {msg_hf}*\n\n"
|
|
|
|
if not img_or and not img_cf and not img_hf:
|
|
return "*Échec critique : Aucune API n'a pu générer d'image.*", "#ff4444", "Système", "Erreur", trace
|
|
|
|
return reponse_md, "#00e5ff", " + ".join(fournisseurs_actifs) if fournisseurs_actifs else "Aucun", "Moteur Média", trace
|
|
|
|
def router_media(prompt: str, type_media: str, premium: bool, fichiers: list) -> tuple:
|
|
trace = [f"🟢 Routage Média : {type_media.upper()}"]
|
|
if type_media == "image":
|
|
return generer_les_images(prompt, fichiers, premium)
|
|
elif type_media == "video":
|
|
return "*La génération vidéo nécessite une API payante.* \n\nPrompt : " + prompt, "#ff4444", "Système", "En attente API", trace
|
|
return "Type de média non reconnu.", "#333", "Erreur", "Aucun", trace |