from flask import Flask, Response, request, stream_with_context
from google import genai
from google.genai import types
import os
from PIL import Image
import io
import json
# --- Configuration ---
GOOGLE_API_KEY = os.environ.get("GEMINI_API_KEY")
if not GOOGLE_API_KEY:
raise ValueError("La variable d'environnement GEMINI_API_KEY n'est pas définie.")
app = Flask(__name__)
# --- Initialisation client Gemini ---
def initialize_gemini_client():
try:
return genai.GenerativeModel(
model_name="gemini-1.5-flash-latest", # Modèle par défaut
api_key=GOOGLE_API_KEY,
generation_config=types.GenerationConfig(
temperature=0.7,
),
)
except Exception as e:
print(f"Erreur lors de l'initialisation du client GenAI : {e}")
return None
client = initialize_gemini_client()
# --- Code HTML/CSS/JS pour le Frontend ---
HTML_PAGE = """
Gemini Image Solver
Résoudre une image avec Gemini
Prêt à recevoir une image.
Réponse de Gemini:
"""
# --- Fonctions d'utilitaires pour le traitement d'images ---
def process_image(image_bytes):
"""Traite une image pour l'envoyer à Gemini"""
try:
img = Image.open(io.BytesIO(image_bytes))
# Assurez-vous que le format est supporté par Gemini
if img.format not in ['PNG', 'JPEG', 'WEBP', 'HEIC', 'HEIF']:
print(f"Format d'image original {img.format} non optimal, conversion en PNG.")
output_format = "PNG"
else:
output_format = img.format
buffered = io.BytesIO()
img.save(buffered, format=output_format)
return buffered.getvalue(), output_format.lower()
except Exception as e:
print(f"Erreur lors du traitement de l'image: {e}")
raise
# --- Routes Flask ---
@app.route('/')
def index():
return HTML_PAGE
@app.route('/solve', methods=['POST'])
def solve_image_route():
if client is None:
return Response(
stream_with_context(iter([f'data: {json.dumps({"error": "Le client Gemini n\'est pas initialisé."})}\n\n'])),
mimetype='text/event-stream'
)
if 'image' not in request.files:
return Response(
stream_with_context(iter([f'data: {json.dumps({"error": "Aucun fichier image fourni."})}\n\n'])),
mimetype='text/event-stream'
)
file = request.files['image']
if file.filename == '':
return Response(
stream_with_context(iter([f'data: {json.dumps({"error": "Aucun fichier sélectionné."})}\n\n'])),
mimetype='text/event-stream'
)
try:
image_data = file.read()
# Préparer l'image pour Gemini
processed_image, output_format = process_image(image_data)
# Le prompt pour Gemini
prompt_parts = [
types.Part.from_data(data=processed_image, mime_type=f'image/{output_format}'),
types.Part.from_text("Résous ceci. Explique clairement ta démarche en français. Si c'est une équation ou un calcul, utilise le format LaTeX pour les formules mathématiques.")
]
def generate_stream():
current_mode = 'starting'
try:
# Utilisation de generate_content avec stream=True
response_stream = client.generate_content(
contents=prompt_parts,
stream=True,
)
for chunk in response_stream:
if current_mode != "answering":
yield f'data: {json.dumps({"mode": "answering"})}\n\n'
current_mode = "answering"
if chunk.parts:
for part in chunk.parts:
if hasattr(part, 'text') and part.text:
yield f'data: {json.dumps({"content": part.text})}\n\n'
elif hasattr(chunk, 'text') and chunk.text:
yield f'data: {json.dumps({"content": chunk.text})}\n\n'
except types.generation_types.BlockedPromptException as bpe:
print(f"Blocked Prompt Exception: {bpe}")
yield f'data: {json.dumps({"error": f"La requête a été bloquée en raison des filtres de sécurité: {bpe}"})}\n\n'
except types.generation_types.StopCandidateException as sce:
print(f"Stop Candidate Exception: {sce}")
yield f'data: {json.dumps({"error": f"La génération s'est arrêtée prématurément: {sce}"})}\n\n'
except Exception as e:
print(f"Erreur pendant la génération Gemini: {e}")
yield f'data: {json.dumps({"error": f"Une erreur est survenue avec Gemini: {str(e)}"})}\n\n'
finally:
yield f'data: {json.dumps({"mode": "finished"})}\n\n'
return Response(
stream_with_context(generate_stream()),
mimetype='text/event-stream',
headers={
'Cache-Control': 'no-cache',
'X-Accel-Buffering': 'no', # Important pour Nginx
'Connection': 'keep-alive'
}
)
except Exception as e:
print(f"Erreur générale dans /solve: {e}")
return Response(
stream_with_context(iter([f'data: {json.dumps({"error": f"Une erreur inattendue est survenue sur le serveur: {str(e)}"})}\n\n'])),
mimetype='text/event-stream'
)
if __name__ == '__main__':
# Vérification finale avant de lancer
if not GOOGLE_API_KEY:
print("ERREUR CRITIQUE: GEMINI_API_KEY n'est pas défini. L'application ne peut pas démarrer correctement.")
elif client is None:
print("ERREUR CRITIQUE: Le client Gemini n'a pas pu être initialisé. Vérifiez votre clé API et la connectivité.")
else:
print("Serveur démarré. Accédez à http://localhost:5000 pour utiliser l'application.")
app.run(debug=True, host='0.0.0.0', port=5000)