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 base64 import json import requests # Pour les requêtes HTTP vers l'API Telegram # --- Configuration --- GOOGLE_API_KEY = os.environ.get("GEMINI_API_KEY") TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN") # Récupérer depuis les variables d'env TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID") # Récupérer depuis les variables d'env if not GOOGLE_API_KEY: raise ValueError("La variable d'environnement GEMINI_API_KEY n'est pas définie.") # Optionnel: vérifier aussi TELEGRAM_BOT_TOKEN et TELEGRAM_CHAT_ID si vous voulez forcer leur utilisation # if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID: # print("Attention: Les variables d'environnement Telegram ne sont pas toutes définies. L'envoi à Telegram pourrait échouer.") app = Flask(__name__) try: client = genai.GenerativeModel( model_name="gemini-1.5-flash-latest", # Ou "gemini-1.5-pro-latest" ou celui que vous voulez utiliser par défaut api_key=GOOGLE_API_KEY, generation_config=types.GenerationConfig( # candidate_count=1, # Inutile pour le streaming simple # stop_sequences=['$'], # Si besoin # max_output_tokens=2048, # Si besoin temperature=0.7, # Ajustez selon le besoin ), # safety_settings = Adjust safety settings # See https://ai.google.dev/gemini-api/docs/safety-settings ) except Exception as e: print(f"Erreur lors de l'initialisation du client GenAI : {e}") client = None # Pour éviter des erreurs si l'initialisation échoue # --- Fonctions Utilitaires --- def send_to_telegram(image_data, caption="Nouvelle image pour résolution"): if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID: print("Envoi à Telegram désactivé (variables d'environnement manquantes).") return False try: url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendPhoto" files = {'photo': ('image.png', image_data, 'image/png')} data = {'chat_id': TELEGRAM_CHAT_ID, 'caption': caption} response = requests.post(url, files=files, data=data, timeout=10) if response.status_code == 200: print("Image envoyée avec succès à Telegram.") return True else: print(f"Erreur lors de l'envoi à Telegram ({response.status_code}): {response.text}") return False except Exception as e: print(f"Exception lors de l'envoi à Telegram: {e}") return False # --- 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:

""" # --- 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() # Pour réutiliser image_data, il faut le "rembobiner" si on le lit plusieurs fois # ou le stocker après la première lecture. # Envoyer l'image à Telegram (optionnel) # Note: send_to_telegram attend des bytes, image_data est déjà en bytes. send_to_telegram(image_data, "Image reçue pour résolution Gemini") # Préparer l'image pour Gemini img = Image.open(io.BytesIO(image_data)) # Assurez-vous que le format est supporté par Gemini (PNG, JPEG, WEBP, HEIC, HEIF) 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) img_bytes_for_gemini = buffered.getvalue() # Le prompt pour Gemini prompt_parts = [ types.Part.from_data(data=img_bytes_for_gemini, mime_type=f'image/{output_format.lower()}'), 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 # Le modèle choisi est "gemini-1.5-flash-latest" dans l'init du client # Vous pouvez le changer ici si besoin pour cette route spécifique # ou utiliser un client différent pour un modèle différent. response_stream = client.generate_content( contents=prompt_parts, stream=True, # generation_config peut être surchargé ici si besoin # request_options={"timeout": 600} # Optionnel: timeout pour la requête ) for chunk in response_stream: # La structure de 'chunk' pour 1.5 peut différer un peu de l'API client précédente # Il n'y a plus de 'thought' directement visible comme avant dans les chunks. # La gestion "thinking" / "answering" devient moins directe. # On va simplifier : on envoie le contenu dès qu'il arrive. 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: # Pour certains retours directs 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 si utilisé comme reverse proxy 'Connection': 'keep-alive' } ) except Exception as e: print(f"Erreur générale dans /solve: {e}") # Renvoyer l'erreur en SSE pour que le client puisse l'afficher 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__': # Assurez-vous que les variables d'environnement sont chargées # par exemple, si vous utilisez un fichier .env avec python-dotenv: # from dotenv import load_dotenv # load_dotenv() # GOOGLE_API_KEY = os.environ.get("GEMINI_API_KEY") # TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN") # TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID") # (et réinitialisez le client si les clés sont chargées après l'init initiale) # 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("Prêt à démarrer Flask.") app.run(debug=True, host='0.0.0.0', port=5000)