voici mon backend je veux que tu ajoute les gros modeles stp en faisant une requête sur groq

app.py

from gevent import monkey
monkey.patch_all()
import os
import re
import json
import logging
from typing import Any, Dict, List, Optional, Tuple, Generator
import time
import requests
from bs4 import BeautifulSoup
from concurrent.futures import ThreadPoolExecutor, as_completed

from flask import Flask, request, Response, jsonify, stream_with_context
from flask_cors import CORS
from groq import Groq
from e2b_code_interpreter import Sandbox
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import httpx

Configuration

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(name)
app = Flask(name)
CORS(app)

Environment variables

GROQ_API_KEY = os.environ.get("GROQ_API_KEY")
E2B_API_KEY = os.environ.get("E2B_API_KEY")

Constants

PISTON_API_URL = "https://emkc.org/api/v2/piston/execute"
WHOOGLE_INSTANCE_URL = "https://chatapp-1-u8t2.onrender.com"
VALID_MODELS_LOW_TOKEN = ["qwen-2.5-32b", "llama-3.1-8b-instant", "gemma2-9b-it", "mixtral-8x7b-32768"]
DEFAULT_CHAT_MODEL_LOW_TOKEN = "qwen-2.5-32b"
SUMMARIZATION_MODEL = "qwen-2.5-32b"
CODE_PATTERN = re.compile(r"python\s*([\s\S]*?)")
CODE_REQUIRED_MARKER = "[CODE_REQUIRED]"
CANVAS_START_MARKER = "[CANVAS_START]"
CANVAS_END_MARKER = "[CANVAS_END]"
MAX_CONTEXT_CHARS_PER_PAGE = 1000
MAX_TOTAL_CONTEXT_CHARS = 5000
MAX_WEB_LINKS = 10 # Increased to 10 sources
CHAT_HISTORY_MESSAGE_LIMIT = 10
WHOOGLE_CACHE: Dict[str, str] = {}

ALL_AVAILABLE_MODELS = [
"distil-whisper-large-v3-en",
"gemma2-9b-it",
"llama-3.3-70b-versatile",
"llama-3.1-8b-instant",
"llama-guard-3-8b",
"llama3-70b-8192",
"llama3-8b-8192",
"mixtral-8x7b-32768",
"whisper-large-v3",
"whisper-large-v3-turbo",
"qwen-2.5-32b",
"deepseek-r1-distill-qwen-32b",
"deepseek-r1-distill-llama-70b-specdec",
"deepseek-r1-distill-llama-70b",
"llama-3.3-70b-specdec",
"llama-3.2-1b-preview",
"llama-3.2-3b-preview",
"llama-3.2-11b-vision-preview",
"llama-3.2-90b-vision-preview"
]

Initialize clients - Check for API Keys

groq_client = Groq(api_key=GROQ_API_KEY) if GROQ_API_KEY else None
if groq_client is None:
logger.error("GROQ_API_KEY is not set. Groq functionalities will be disabled.")
sandbox = Sandbox(api_key=E2B_API_KEY, timeout=600) if E2B_API_KEY else None
if sandbox is None:
logger.error("E2B_API_KEY is not set. Code execution in sandbox will be disabled.")

http_session = requests.Session()
http_session.headers.update({"User-Agent": "PlinkoChatBot/1.0"})

def operate_requests(prompt) -> Generator[Dict[str, str], None, Tuple[str, List[str]]]:
"""
Fonction regroupant toutes les opérations :
1. Recherche sur Whoogle.
2. Limitation à MAX_WEB_LINKS liens pertinents.
3. Extraction et résumé du contenu de chaque page pertinente (en parallèle).
4. Retourne le contexte obtenu et la liste des sources pertinentes.
"""
global GROQ_API_KEY
groq_client_local = Groq(api_key=GROQ_API_KEY) if GROQ_API_KEY else None # Use local client to avoid global issues
if groq_client_local is None:
yield {"type": "web_search_status", "content": "error"}
yield "Error: GROQ_API_KEY is not set.", []
return

text
def whoogle_search(query):
    """
    Effectue une recherche via Whoogle et retourne une liste de liens et titres.
    """
    base_url = WHOOGLE_INSTANCE_URL + "/search"
    params = {"q": query}
    headers = {"User-Agent": "Mozilla/5.0"}

    try:
        response = requests.get(base_url, params=params, headers=headers)
        response.raise_for_status()
    except requests.exceptions.RequestException as e:
        logger.error(f"Erreur lors de la requête Whoogle: {e}")
        return []

    soup = BeautifulSoup(response.text, "html.parser")
    results = []
    for result in soup.find_all("div", class_="result"): # Adjust class if needed based on Whoogle instance
        link_tag = result.find("a", href=True)
        title_tag = result.find("h3") # Assuming title is in h3, adjust if needed
        if link_tag and title_tag and link_tag["href"].startswith("http"):
            results.append({"link": link_tag["href"], "title": title_tag.text})
    return results

def is_relevant_title(title: str, query: str) -> bool:
    """
    Détermine si le titre d'une page est pertinent par rapport à la requête.
    """
    query_keywords = query.lower().split()
    title_lower = title.lower()
    relevant_keywords_count = 0
    for keyword in query_keywords:
        if keyword in title_lower:
            relevant_keywords_count += 1
    # Heuristic: Title is relevant if at least 1 keyword is present (can be adjusted)
    return relevant_keywords_count >= 1

@retry(stop=stop_after_attempt(5),
       wait=wait_exponential(multiplier=1, min=4, max=60),
       retry=retry_if_exception_type(httpx.HTTPError))
def summarize_with_groq(query, text_content):
    """
    Utilise l'API Groq pour extraire les informations importantes du texte.
    Retry avec tenacity en cas d'erreur 429.
    """
    if groq_client_local is None: # Use local client
        raise Exception("GROQ_API_KEY is not set, cannot summarize.")
    try:
        response = groq_client_local.chat.completions.create( # Use local client
            model=SUMMARIZATION_MODEL,
            messages=[
                {
                    "role": "system",
                    "content": (
                        "Vous êtes un assistant expert en extraction d'informations. "
                        "Votre tâche est de lire le texte fourni et d'extraire les informations les plus importantes "
                        "qui répondent à la question de recherche suivante : '{}'. "
                        "Créez une note concise résumant les points clés et les informations pertinentes. "
                        "Soyez précis et factuel."
                    ).format(query)
                },
                {
                    "role": "user",
                    "content": text_content[:MAX_CONTEXT_CHARS_PER_PAGE]
                }
            ],
            temperature=0.2
        )
        return response.choices[0].message.content.strip()
    except httpx.HTTPError as e:
        if e.response.status_code == 429:
            logger.warning(f"Erreur 429 (Too Many Requests) lors de l'appel à l'API Groq (summarize_with_groq). Reessai en cours... Details: {e}")
            raise e
        else:
            logger.error(f"Erreur HTTP lors de l'appel à l'API Groq (summarize_with_groq) (non-429). Details: {e}")
            raise
    except Exception as e:
        logger.error(f"Erreur inattendue lors de l'appel à l'API Groq (summarize_with_groq). Détails: {e}")
        return "Erreur lors de la synthèse avec Groq."

def fetch_and_summarize(link, query):
    """
    Récupère le contenu d'une page et utilise Groq pour en extraire une note résumée.
    """
    try:
        r = requests.get(link, timeout=20)
        r.raise_for_status()
        soup_page = BeautifulSoup(r.text, 'html.parser')
        text_content = soup_page.get_text(separator='\n', strip=True)
        print(f"{link} : OK - Extraction du contenu...")

        summary_note = summarize_with_groq(query, text_content)
        print(f"{link} : OK - Note Groq ajoutée.")
        return f"Source: {link}\nNote:\n{summary_note}\n\n"

    except requests.exceptions.RequestException as e:
        print(f"{link} : Échec de la récupération de la page.")
        return f"Source: {link}\nNote:\nErreur de récupération de la page.\n\n"
    except Exception as e:
        print(f"{link} : Erreur lors du traitement de la page : {e}")
        return f"Source: {link}\nNote:\nErreur lors du traitement de la page: {e}\n\n"

def fetch_pages_and_summarize(links: List[str], query: str) -> str:
    """
    Utilise un pool de threads pour traiter plusieurs pages en parallèle et retourne le contexte concaténé.
    """
    results = []
    with ThreadPoolExecutor(max_workers=5) as executor: # Reduced max_workers for concurrency control
        future_to_link = {executor.submit(fetch_and_summarize, link, query): link for link in links}
        for future in as_completed(future_to_link):
            result = future.result()
            results.append(result)
    return "".join(results)

# --- Déroulement principal de la fonction operate_requests ---
yield {"type": "web_search_status", "content": "start"}

# 1. Recherche sur Whoogle
search_results = whoogle_search(prompt)
if not search_results:
    print("Aucun résultat de recherche trouvé.")
    yield {"type": "web_search_status", "content": "end"}
    yield "", []
    return

# 2. Filtrer les liens pertinents basés sur le titre
relevant_links_with_titles = []
for result in search_results:
    if is_relevant_title(result["title"], prompt):
        relevant_links_with_titles.append(result)
    else:
        print(f"Titre non pertinent ignoré: {result['title']}")

# 3. Limiter à MAX_WEB_LINKS liens pertinents
limited_relevant_links = [item["link"] for item in relevant_links_with_titles[:MAX_WEB_LINKS]]
print(f"{len(limited_relevant_links)} lien(s) pertinent(s) utilisé(s) pour la recherche.")

if not limited_relevant_links:
    print("Aucun lien pertinent trouvé après analyse des titres.")
    yield {"type": "web_search_status", "content": "end"}
    yield "", []
    return


# 4. Extraction et résumé en parallèle des liens pertinents
context = fetch_pages_and_summarize(limited_relevant_links, prompt)
yield {"type": "web_search_status", "content": "end"}
yield context, limited_relevant_links
return

def extract_code_blocks(text: str) -> Tuple[Optional[str], Optional[str]]:
"""Extrait les blocs de code Python et gère le marqueur [CODE_REQUIRED]."""
if CODE_REQUIRED_MARKER in text:
text_no_marker = text.replace(CODE_REQUIRED_MARKER, "", 1).strip()
code_match = CODE_PATTERN.search(text_no_marker)
if code_match:
return code_match.group(1).strip(), None
else:
code_match = CODE_PATTERN.search(text)
if code_match:
return None, code_match.group(1).strip()
return None, None

def run_code_in_sandbox(code: str) -> str:
"""Exécute le code Python dans le sandbox e2b et retourne le résultat."""
if sandbox is None:
return "Error: E2B_API_KEY is not set. Code execution is disabled."
start_time = time.time()
try:
logger.info("Début exécution code dans sandbox e2b.")
execution = None
method_name = next((method for method in ["run_code_sync", "execute_sync", "execute_code", "execute", "run_code", "run"] if hasattr(sandbox, method)), None)
if method_name:
logger.info(f"Utilisation de la méthode sandbox.{method_name}")
execution = getattr(sandbox, method_name)(code)
else:
error_msg = "Aucune méthode d'exécution de code disponible dans le sandbox."
logger.error(error_msg)
return error_msg
logs = execution.logs
stdout_logs = logs.stdout if hasattr(logs, 'stdout') else []
stderr_logs = logs.stderr if hasattr(logs, 'stderr') else []
stdout = "\n".join(stdout_logs)
stderr = "\n".join(stderr_logs)
if stderr:
return f"Stdout:\n{stdout}\n\nStderr:\n{stderr}"
return stdout
except Exception as e:
duration = time.time() - start_time
error_msg = f"Erreur lors de l'exécution du code dans e2b après {duration:.2f} secondes: {str(e)}"
logger.exception(error_msg)
return f"Erreur exécution code: {str(e)}"
finally:
if sandbox and hasattr(sandbox, 'close'): # Ensure sandbox cleanup
try:
sandbox.close() # Explicitly close sandbox resources
logger.info("Sandbox resources closed.")
except Exception as e:
logger.warning(f"Erreur lors de la fermeture du sandbox: {e}")

def extract_numeric_result(exec_result: str) -> Optional[str]:
"""Extrait la dernière ligne numérique du résultat d'exécution de code."""
if "Stdout:" in exec_result:
stdout_section = exec_result.split("Stdout:\n", 1)[1].split("\n\nStderr:")[0].strip()
else:
stdout_section = exec_result.strip()
lines = [l.strip() for l in stdout_section.splitlines() if l.strip()]
if not lines:
return None
last_line = lines[-1]
if re.search(r'\d', last_line):
return last_line
return None

@app.route('/models', methods=['GET'])
def get_models() -> Response:
"""Retourne la liste des modèles valides."""
return jsonify({"models": ALL_AVAILABLE_MODELS})

@app.route('/chat', methods=['POST'])
def chat_endpoint() -> Response:
"""
Point d'entrée principal pour le chat.
Gère le streaming SSE, l'exécution de code et les canvas.
Intègre la recherche web si activée (configurable via requête).
"""

text
@stream_with_context
@retry(stop=stop_after_attempt(5),
       wait=wait_exponential(multiplier=1, min=4, max=60),
       retry=retry_if_exception_type(httpx.HTTPError))
def stream_response():
    if groq_client is None:
        yield format_sse_event({"type": "error", "content": "GROQ_API_KEY non configurée côté serveur."})
        return

    data = request.get_json()
    if not data:
        logger.warning("Requête chat invalide: corps de requête manquant.")
        yield format_sse_event({"type": "error", "content": "Requête chat invalide : corps de requête manquant."})
        return

    enable_web_search = data.get('enable_web_search', True)
    messages = data.get('messages', [])
    model: str = data.get('model', DEFAULT_CHAT_MODEL_LOW_TOKEN).strip()
    execute_code: bool = data.get('execute_code', True)

    if not messages or not isinstance(messages, list) or not model or model not in VALID_MODELS_LOW_TOKEN:
        logger.warning("Requête chat invalide: messages=%s, model=%s", messages, model)
        yield format_sse_event({"type": "error", "content": "Requête chat invalide : messages ou modèle incorrects."})
        return


    system_web_context = ""
    last_user_message = messages[-1]['content'] if messages else ""
    web_sources: List[str] = []

    if enable_web_search and last_user_message:
        web_search_generator = operate_requests(last_user_message)
        try:
            while True:
                try:
                    event = next(web_search_generator)
                    if isinstance(event, dict) and event.get('type') == 'web_search_status':
                        yield format_sse_event(event)
                    elif isinstance(event, tuple) and len(event) == 2:
                        system_web_context, web_sources = event
                        break
                    elif isinstance(event, tuple) and event == ("", []):
                        system_web_context = ""
                        web_sources = []
                        break
                except StopIteration:
                    system_web_context = ""
                    web_sources = []
                    break # Correctly placed break statement
        except Exception as e:
            logger.error(f"Web search error: {str(e)}")
            system_web_context = ""
            web_sources = []


    # Prompt système factorisé et simplifié
    base_prompt = "🎯 **Rôle : Expert en calcul et canvas.**\n" \
                  "**Mission :** Répondre, calculer, canvas interactifs. Style expert. Concise.\n\n" \
                  "**MODÈLE IA UTILISÉ :** {model}\n\n"
    web_context_prompt_part = f"**CONTEXTE WEB :**\n{system_web_context}\n\n" if system_web_context else ""
    instructions_prompt_part = "**CALCUL NUMÉRIQUE :**\n" \
                               "1. Phrase intro calcul Python.\n" \
                               "2. `[CODE_REQUIRED]`.\n" \
                               "3. Bloc PYTHON minimal, lisible, `print(resultat_numerique)`.\n" \
                               "**[APRES EXECUTION CODE]** Continue, explique résultat code en DETAILS. Fais avancer conversation.\n\n" \
                               "**CANVAS CODE :**\n" \
                               "1. Canvas code interactif.\n" \
                               "2. `[CANVAS_START]`...`[CANVAS_END]`.\n" \
                               "**CANVAS TEXTE :**\n" \
                               "1. Canvas texte interactif (markdown).\n" \
                               "2. `[CANVAS_START]`...`[CANVAS_END]`.\n" \
                               "**CHAT CANVAS :** MODIFIER canvas. NOUVEAU canvas complet `[CANVAS_START]`...`[CANVAS_END]`.\n" \
                               "**PAS CALCUL NI CANVAS :** Réponse directe, concise. Contexte conversation."


    system_prompt_content = base_prompt.format(model=model) + web_context_prompt_part + instructions_prompt_part


    system_prompt = {"role": "system", "content": system_prompt_content}
    messages_with_system_prompt = [system_prompt] + messages[-CHAT_HISTORY_MESSAGE_LIMIT:]

    full_reply_text = ""
    code_to_execute = None
    code_to_display = None
    is_code_used = False
    canvas_content = None
    code_executed = False
    prompt_tokens = 0
    completion_tokens = 0
    total_tokens = 0
    initial_response_completed = False

    yield format_sse_event({"type": "analysis_status", "content": "start"})

    try:
        response_stream = groq_client.chat.completions.create(
            model=model,
            messages=messages_with_system_prompt,
            temperature=0.0,
            stream=True,
        )
        for chunk in response_stream:
            if initial_response_completed:
                break
            delta_content = chunk.choices[0].delta.content or ""
            delta_content_no_marker = delta_content.replace(CODE_REQUIRED_MARKER, "")
            full_reply_text += delta_content_no_marker
            if CANVAS_START_MARKER in full_reply_text and CANVAS_END_MARKER in full_reply_text:
                start_index = full_reply_text.find(CANVAS_START_MARKER) + len(CANVAS_START_MARKER)
                end_index = full_reply_text.find(CANVAS_END_MARKER)
                canvas_content_extracted = full_reply_text[start_index:end_index].strip()
                canvas_content = canvas_content_extracted
                delta_content_for_chat = delta_content_no_marker.replace(CANVAS_START_MARKER, "").replace(CANVAS_END_MARKER, "")
                yield format_sse_event({"type": "text", "content": delta_content_for_chat})
                yield format_sse_event({"type": "canvas_start", "content": canvas_content})
                continue
            else:
                yield format_sse_event({"type": "text", "content": delta_content_no_marker})
            code_to_execute, code_to_display = extract_code_blocks(full_reply_text)
            if code_to_execute:
                initial_response_completed = True

        if execute_code and code_to_execute:
            is_code_used = True
            logger.info("Envoi événement CODE_BLOCK")
            yield format_sse_event({"type": "code_block", "content": code_to_execute})
            yield format_sse_event({"type": "code_execution_start"})
            execution_result = run_code_in_sandbox(code_to_execute)
            numeric_result = extract_numeric_result(execution_result)
            if numeric_result:
                code_executed = True
                logger.info("Envoi événement EXECUTED_CODE_RESULT")
                yield format_sse_event({"type": "executed_code_result", "content": numeric_result})
                conclusion_prompt_content = (
                    f"**[APRES CODE]** Resultat: **{numeric_result}**. Continue conversation."
                )
                conclusion_prompt = [{"role": "user", "content": conclusion_prompt_content}]
                conclusion_stream = groq_client.chat.completions.create(
                    model=model,
                    messages=conclusion_prompt,
                    temperature=0.0,
                    stream=True,
                )
                for chunk in conclusion_stream:
                    delta_content = chunk.choices[0].delta.content or ""
                    yield format_sse_event({"type": "text", "content": delta_content})
                yield format_sse_event({"type": "final"})
            else:
                code_executed = True
                logger.info("Envoi événement EXECUTED_CODE_ERROR")
                yield format_sse_event({"type": "executed_code_error", "content": execution_result})
                yield format_sse_event({"type": "final", "result": "Fin de la réponse (erreur ou pas de résultat numérique)."})

        if web_sources:
            yield format_sse_event({"type": "web_sources", "content": web_sources})

        yield format_sse_event({"type": "sources_start"})
        for source_url in web_sources:
            yield format_sse_event({"type": "source_item", "content": source_url})
        yield format_sse_event({"type": "sources_end"})


        prompt_tokens = 0 # Token count is not implemented in this version
        completion_tokens = 0 # Token count is not implemented in this version
        total_tokens = prompt_tokens + completion_tokens # Token count is not implemented in this version
        if not code_executed:
            logger.info("Envoi événement FINAL (sans code exécuté)")
            yield format_sse_event({"type": "final", "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "total_tokens": total_tokens})
        else:
            yield format_sse_event({"type": "final", "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "total_tokens": total_tokens})
    except httpx.HTTPError as e:
        if e.response.status_code == 429:
            error_message = f"Erreur 429 (Too Many Requests) lors de l'appel à l'API Groq (stream_response). Réessai automatique par Tenacity. Détails: {e}"
            logger.warning(error_message)
            yield format_sse_event({"type": "error", "content": "Erreur temporaire de l'IA. Réessayez svp."})
            raise e
        else:
            error_message = f"Erreur HTTP lors de l'appel à l'API Groq (stream_response) (non-429). Détails: {e}"
            logger.exception(error_message)
            yield format_sse_event({"type": "error", "content": error_message})
    except Exception as e:
        error_message = f"Erreur serveur IA: {str(e)}"
        logger.exception(error_message)
        yield format_sse_event({"type": "error", "content": error_message})
    finally:
        yield format_sse_event({"type": "analysis_status", "content": "end"})
        yield format_sse_event({"type": "done", "is_code_execution": is_code_used})

return Response(stream_with_context(stream_response()), mimetype='text/event-stream')

@app.route('/canvas_modify', methods=['POST'])
def canvas_modify_endpoint() -> Response:
if groq_client is None:
return jsonify({"error": "GROQ_API_KEY non configurée côté serveur."}), 500

text
data = request.get_json()
if not data:
    logger.warning("Requête canvas_modify sans données.")
    return jsonify({"error": "Requête canvas_modify invalide : corps de requête manquant."}), 400
user_message = data.get('message')
canvas_content = data.get('canvas_content')
model = data.get('model', DEFAULT_CHAT_MODEL_LOW_TOKEN).strip()
if not user_message or canvas_content is None:
    logger.warning("Requête canvas_modify incomplète: message=%s, canvas_content=%s", user_message, canvas_content)
    return jsonify({"error": "Requête canvas_modify invalide : message ou contenu canvas manquant."}), 400
canvas_chat_messages = [
    {"role": "system", "content": ("Chat canvas. Modifier canvas. "
                                     "NOUVEAU canvas complet `[CANVAS_START]`...`[CANVAS_END]`.")},
    {"role": "assistant", "content": f"[CANVAS_START]\\n{canvas_content}\\n[CANVAS_END] Canvas actuel. Modifier?"},
    {"role": "user", "content": user_message}
]

@retry(stop=stop_after_attempt(5),
       wait=wait_exponential(multiplier=1, min=4, max=60),
       retry=retry_if_exception_type(httpx.HTTPError))
def canvas_modify_groq_call(canvas_chat_messages, model):
    """Fonction retryable pour l'appel à Groq dans canvas_modify_endpoint"""
    return groq_client.chat.completions.create(
        model=model,
        messages=canvas_chat_messages,
        temperature=0.0,
        stream=False,
    )

try:
    response_stream = canvas_modify_groq_call(canvas_chat_messages, model)
    ai_response_text = response_stream.choices[0].message.content or ""
    if CANVAS_START_MARKER in ai_response_text and CANVAS_END_MARKER in ai_response_text:
        start_index = ai_response_text.find(CANVAS_START_MARKER) + len(CANVAS_START_MARKER)
        end_index = ai_response_text.find(CANVAS_END_MARKER)
        updated_canvas_content = ai_response_text[start_index:end_index].strip()
        response_message = "Canvas mis à jour."
        logger.info("Canvas modifié avec succès.")
        return jsonify({"updated_canvas_content": updated_canvas_content, "response_message": response_message})
    else:
        logger.error("Réponse IA canvas_modify invalide: marqueurs CANVAS_START/END manquants.")
        return jsonify({"error": "Réponse IA invalide pour la modification du canvas", "full_response": ai_response_text}), 500
except httpx.HTTPError as e:
    if e.response.status_code == 429:
        error_message = f"Erreur 429 (Too Many Requests) lors de l'appel à l'API Groq (canvas_modify_endpoint). Après plusieurs essais, opération échouée. Details: {e}"
    else:
        error_message = f"Erreur HTTP lors de l'appel à l'API Groq (canvas_modify_endpoint) (non-429). Détails: {e}"
    logger.exception(error_message)
    return jsonify({"error": "Erreur lors de la modification du canvas après plusieurs tentatives.", "details": error_message}), 500
except Exception as e:
    error_message = f"Erreur serveur (canvas_modify): {str(e)}"
    logger.exception(error_message)
    return jsonify({"error": error_message}), 500

@app.route('/run', methods=['POST'])
def run_code_endpoint() -> Response:
"""
Point d'accès pour exécuter du code via l'API Piston (pour canvas code).
Langages supportés par Piston.
"""
data = request.get_json()
if not data:
logger.warning("Requête run sans données.")
return jsonify({"error": "Requête run invalide : corps de requête manquant."}), 400
language = data.get("language")
code = data.get("code")
stdin = data.get("stdin", "")
if not language or not code:
logger.warning("Requête run incomplète: language=%s, code=%s", language, code)
return jsonify({"error": "Requête run invalide : langage et code requis."}), 400
payload = {
"language": language,
"version": "*",
"files": [{"content": code}],
"stdin": stdin
}
try:
piston_response = http_session.post(PISTON_API_URL, json=payload, timeout=30)
piston_response.raise_for_status()
logger.info("Code exécuté avec Piston avec succès.")
return jsonify(piston_response.json())
except requests.RequestException as e:
error_message = f"Erreur lors de l'appel à l'API Piston: {str(e)}"
logger.exception(error_message)
piston_text = piston_response.text if 'piston_response' in locals() and piston_response is not None else ""
return jsonify({"error": "Erreur d'exécution du code via l'API Piston.", "piston_response": piston_text, "details": error_message}), 500
except Exception as e:
error_message = f"Erreur inattendue lors de l'exécution du code: {str(e)}"
logger.exception(error_message)
return jsonify({"error": "Erreur inattendue lors de l'exécution du code.", "details": error_message}), 500

@app.route('/run', methods=['OPTIONS'])
def run_code_options() -> Response:
"""Gère les requêtes OPTIONS pour l'endpoint /run (CORS preflight)."""
return jsonify({'message': 'OPTIONS requests allowed'}), 200

def format_sse_event(data: Dict[str, Any]) -> str:
"""Formate les données en event SSE (Server-Sent Events)."""
return f"data: {json.dumps(data)}\n\n"
if name == 'main':
app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 5000)))

Je ne veux pas que les modeles sois chargé en local

je veux que tu fasse une requête sur groq pour les modeles en plus de ceci je veux tout les modeles si quand tu recois les modeles tu les tries en fonction de vision resonning et voix