flux & audio reply

.gitignore

@@ -0,0 +1,6 @@
+# directories
+/images
+/audio_replies
+
+# env
+.env

FLUX.py

@@ -0,0 +1,146 @@
+import requests
+import time
+import io
+import os
+import re
+import json
+from PIL import Image, UnidentifiedImageError
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# Load the ImgBB API key from the environment variables.
+IMGBB_API_KEY = os.getenv("IMGBB_API_KEY")
+
+def upload_to_imgbb(image_path, file_name):
+    """
+    Uploads the image located at image_path to ImgBB.
+    Returns:
+        str: URL of the uploaded image on ImgBB or None if failed.
+    """
+    try:
+        with open(image_path, 'rb') as f:
+            image_data = f.read()
+        response = requests.post(
+            "https://api.imgbb.com/1/upload",
+            params={"key": IMGBB_API_KEY},
+            files={"image": (file_name, image_data)}
+        )
+        response.raise_for_status()
+        result = response.json()
+        if result.get("data") and "url" in result["data"]:
+            return result["data"]["url"]
+        else:
+            print("Failed to upload image to ImgBB.")
+            return None
+    except requests.RequestException as e:
+        print(f"Error uploading image to ImgBB: {e}")
+        return None
+    except Exception as e:
+        print(f"Unexpected error uploading image to ImgBB: {e}")
+        return None
+
+def generate_image(prompt, request_id, current_request_id, image_dir, attempt=0):
+    """
+    Generate an image using the Pollinations API.
+    
+    Parameters:
+        prompt (str): The prompt for image generation.
+        width (int): Desired image width.
+        height (int): Desired image height.
+        request_id (int): The request id for the current operation.
+        current_request_id (int): The current active request id.
+        image_dir (str): Directory where image will be saved.
+        attempt (int): Current attempt count (zero-indexed).
+    
+    Returns:
+        tuple: (PIL.Image object, image_path (str), returned_prompt (str), image_url (str))
+        or None if image fetch fails or request id mismatches.
+    """
+    model = "flux"
+    width = 1920
+    height = 1080
+    enhance_param = "true"
+    url = f"https://image.pollinations.ai/prompt/{prompt}?nologo=true&safe=false&private=true&model={model}&enhance={enhance_param}&width={width}&height={height}"
+    print(f"Attempt {attempt + 1}: Fetching image with URL: {url}")
+    
+    try:
+        response = requests.get(url, timeout=45)
+    except Exception as e:
+        print(f"Error fetching image: {e}")
+        return None
+
+    if response.status_code != 200:
+        print(f"Failed to fetch image. Status code: {response.status_code}")
+        return None
+
+    if request_id != current_request_id:
+        print("Request ID mismatch. Operation cancelled.")
+        return None
+
+    print("Image fetched successfully.")
+    image_data = response.content
+
+    try:
+        image = Image.open(io.BytesIO(image_data))
+        actual_width, actual_height = image.size
+        print(f"Actual image dimensions: {actual_width}x{actual_height}")
+        
+        # Extract metadata from EXIF if available
+        exif_data = image.info.get('exif', b'')
+        returned_prompt = prompt
+        if exif_data:
+            json_match = re.search(b'{"prompt":.*}', exif_data)
+            if json_match:
+                json_str = json_match.group(0).decode('utf-8')
+                try:
+                    metadata_dict = json.loads(json_str)
+                    returned_prompt = metadata_dict.get('prompt', prompt)
+                except json.JSONDecodeError as e:
+                    print(f"Failed to parse JSON in metadata: {e}")
+            else:
+                print("No JSON data found in EXIF")
+        
+        if (actual_width, actual_height) != (width, height):
+            print(f"Warning: Received image dimensions ({actual_width}x{actual_height}) do not match requested dimensions ({width}x{height})")
+    except UnidentifiedImageError:
+        print("Error: Received data is not a valid image.")
+        raise
+
+    timestamp = int(time.time())
+    image_filename = f"background_{timestamp}.png"
+    image_path = os.path.join(image_dir, image_filename)
+    
+    # Ensure the image directory exists
+    os.makedirs(image_dir, exist_ok=True)
+    
+    try:
+        image.save(image_path, 'PNG')
+        print(f"Image saved to {image_path}")
+        # Upload image to ImgBB
+        image_url = upload_to_imgbb(image_path, image_filename)
+        if image_url:
+            print(f"Image uploaded to ImgBB: {image_url}")
+        else:
+            print("Failed to upload image to ImgBB.")
+    except Exception as e:
+        print(f"Error saving image: {e}")
+        return None
+
+    return image, image_path, returned_prompt, image_url
+
+if __name__ == "__main__":
+    # Example usage
+    prompt = "Beach party, anime style, vibrant colors"
+    request_id = 1
+    current_request_id = 1
+    image_dir = "./images"
+    
+    image, image_path, returned_prompt, image_url = generate_image(prompt, request_id, current_request_id, image_dir)
+    
+    if image:
+        print(f"Image generated and saved at {image_path}")
+        print(f"Returned prompt: {returned_prompt}")
+        print(f"Image URL: {image_url}")
+    else:
+        print("Failed to generate image.")

VoiceReply.py

@@ -0,0 +1,67 @@
+import requests
+import os
+import time
+import urllib.parse
+
+def generate_voice_reply(prompt, model="openai-audio", voice="coral", audio_dir="."):
+    """
+    Generate an audio file (MP3) using the Pollinations text-to-speech API.
+    
+    The prompt is appended with a system message explaining that this is Eve,
+    a WhatsApp bot that can generate images from Flux and help users generate images.
+    To generate an image, the user should send a message like: /imagine {prompt}
+
+    Parameters:
+        prompt (str): The text prompt for audio generation.
+        model (str): Model identifier (default: openai-audio).
+        voice (str): Voice parameter (default: coral).
+        audio_dir (str): Directory where the audio file will be saved.
+
+    Returns:
+        tuple: (audio_file_path (str), audio_data (bytes)) or None if failed.
+    """
+    # Append system prompt to the user's prompt
+    system_prompt = ("Your name is Eve, a WhatsApp bot that can generate images from Flux and help users generate images. "
+                     "To generate an image, send a message like: /imagine prompt. Rules: Don't say that you can't send whatsapp messages or generate images. Be friendly and helpful to users. help them with prompts. You can engage in general conversation. ")
+    full_prompt = f"{prompt}\n\n{system_prompt}"
+    encoded_prompt = urllib.parse.quote(full_prompt)
+    url = f"http://text.pollinations.ai/{encoded_prompt}?model={model}&voice={voice}"
+    print(f"Fetching audio with URL: {url}")
+    
+    try:
+        response = requests.get(url, timeout=30)
+    except Exception as e:
+        print(f"Error fetching audio: {e}")
+        return None
+
+    if response.status_code != 200:
+        print(f"Failed to fetch audio. Status code: {response.status_code}")
+        return None
+
+    audio_data = response.content
+    timestamp = int(time.time())
+    file_name = f"voice_reply_{timestamp}.mp3"
+    audio_file_path = os.path.join(audio_dir, file_name)
+    
+    try:
+        with open(audio_file_path, "wb") as f:
+            f.write(audio_data)
+        print(f"Audio saved to {audio_file_path}")
+    except Exception as e:
+        print(f"Error saving audio file: {e}")
+        return None
+
+    return audio_file_path, audio_data
+
+
+if __name__ == "__main__":
+    # Example usage
+    prompt = "Hi. how are you."
+    audio_dir = "./audio_replies"
+    os.makedirs(audio_dir, exist_ok=True)
+    
+    audio_file_path, audio_data = generate_voice_reply(prompt, audio_dir=audio_dir)
+    if audio_file_path:
+        print(f"Generated audio file: {audio_file_path}")
+    else:
+        print("Failed to generate audio file.")

app.py

@@ -1,10 +1,12 @@
 import os
 import threading
 import requests
-from flask import Flask, request, jsonify
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.responses import PlainTextResponse, JSONResponse
+from FLUX import generate_image
+from VoiceReply import generate_voice_reply
 
 from llm import generate_llm
-from sd import generate_sd
 
 GREEN_API_URL = os.getenv("GREEN_API_URL")
 GREEN_API_MEDIA_URL = os.getenv("GREEN_API_MEDIA_URL", "https://api.green-api.com")
@@ -12,16 +14,15 @@ GREEN_API_TOKEN = os.getenv("GREEN_API_TOKEN")
 GREEN_API_ID_INSTANCE = os.getenv("GREEN_API_ID_INSTANCE")
 WEBHOOK_AUTH_TOKEN = os.getenv("WEBHOOK_AUTH_TOKEN")
 PORT = 7860
+image_dir = "/tmp/images"
+audio_dir = "/tmp/audio"
 
 if not all([GREEN_API_URL, GREEN_API_TOKEN, GREEN_API_ID_INSTANCE, WEBHOOK_AUTH_TOKEN]):
     raise ValueError("Environment variables are not set properly")
 
-app = Flask(__name__)
+app = FastAPI()
 
 def send_message(message_id, to_number, message, retries=3):
-    """
-    Send a text message using Green API with retry logic.
-    """
     if to_number.endswith('@g.us'):
         chat_id = to_number
     else:
@@ -45,9 +46,6 @@ def send_message(message_id, to_number, message, retries=3):
             return {"error": str(e)}
 
 def send_image(message_id, to_number, image_path, retries=3):
-    """
-    Send an image using Green API with retry logic.
-    """
     if to_number.endswith('@g.us'):
         chat_id = to_number
     else:
@@ -67,49 +65,75 @@ def send_image(message_id, to_number, image_path, retries=3):
                 continue
             return {"error": str(e)}
 
-def response_text(message_id, chat_id, prompt):
+def send_audio(message_id, to_number, audio_path, retries=3):
     """
-    Generate a response using the LLM and send it to the user.
+    Send an audio file using the Green API similar to send_image.
     """
+    if to_number.endswith('@g.us'):
+        chat_id = to_number
+    else:
+        chat_id = to_number
+
+    url = f"{GREEN_API_MEDIA_URL}/waInstance{GREEN_API_ID_INSTANCE}/sendFileByUpload/{GREEN_API_TOKEN}"
+    payload = {'chatId': chat_id, 'caption': 'Here is your voice reply!', 'quotedMessageId': message_id}
+    files = [('file', ('audio.mp3', open(audio_path, 'rb'), 'audio/mpeg'))]
+
+    for attempt in range(retries):
+        try:
+            response = requests.post(url, data=payload, files=files)
+            response.raise_for_status()
+            return response.json()
+        except requests.RequestException as e:
+            if attempt < retries - 1:
+                continue
+            return {"error": str(e)}
+
+def response_text(message_id, chat_id, prompt):
     try:
         msg = generate_llm(prompt)
         send_message(message_id, chat_id, msg)
     except Exception as e:
         send_message(message_id, chat_id, "There was an error processing your request.")
 
+def response_audio(message_id, chat_id, prompt):
+    try:
+        audio_file_path, audio_data = generate_voice_reply(prompt, model="openai-audio", voice="coral", audio_dir=audio_dir)
+        if audio_file_path:
+            send_audio(message_id, chat_id, audio_file_path)
+            os.remove(audio_file_path)  # Clean up the file after sending
+        else:
+            response_text(message_id, chat_id, prompt=prompt)
+    except Exception as e:
+        send_message(message_id, chat_id, "There was an error generating the audio. Please try again later.")
+
 def handle_image_generation(message_id, chat_id, prompt):
-    """
-    Generate an image from the provided prompt and send it to the user.
-    """
     try:
-        image_data, image_path = generate_sd(prompt)
-        if image_data:
+        image, image_path, returned_prompt, image_url = generate_image(prompt, message_id, message_id, image_dir)
+        if image:
             send_image(message_id, chat_id, image_path)
+            send_message(message_id, chat_id, f"Image generated successfully! You can view it here: {image_url}. \nPrompt: > _{returned_prompt}_")
         else:
             send_message(message_id, chat_id, "Failed to generate image. Please try again later.")
     except Exception as e:
         send_message(message_id, chat_id, "There was an error generating the image. Please try again later.")
 
-@app.route('/', methods=['GET'])
+@app.get("/", response_class=PlainTextResponse)
 def index():
-    """
-    Basic endpoint to check if the script is running.
-    """
     return "Server is running!"
 
-@app.route('/whatsapp', methods=['POST'])
-def whatsapp_webhook():
-    """
-    Handle incoming WhatsApp messages.
-    """
-    data = request.get_json()
+@app.post("/whatsapp")
+async def whatsapp_webhook(request: Request):
     auth_header = request.headers.get('Authorization', '').strip()
-
     if auth_header != f"Bearer {WEBHOOK_AUTH_TOKEN}":
-        return jsonify({"error": "Unauthorized"}), 403
+        raise HTTPException(status_code=403, detail="Unauthorized")
+
+    try:
+        data = await request.json()
+    except Exception:
+        return JSONResponse(content={"error": "Invalid JSON"}, status_code=400)
 
     if data.get('typeWebhook') != 'incomingMessageReceived':
-        return jsonify(success=True)
+        return {"success": True}
 
     try:
         chat_id = data['senderData']['chatId']
@@ -121,10 +145,9 @@ def whatsapp_webhook():
         elif 'extendedTextMessageData' in message_data:
             body = message_data['extendedTextMessageData']['text'].strip()
         else:
-            return jsonify(success=True)
-
+            return {"success": True}
     except KeyError as e:
-        return jsonify({"error": f"Missing key in data: {e}"}), 200
+        return JSONResponse(content={"error": f"Missing key in data: {e}"}, status_code=200)
 
     if body.lower().startswith('/imagine'):
         prompt = body.replace('/imagine', '').strip()
@@ -136,7 +159,8 @@ def whatsapp_webhook():
     else:
         threading.Thread(target=response_text, args=(message_id, chat_id, body)).start()
 
-    return jsonify(success=True)
+    return {"success": True}
 
 if __name__ == '__main__':
-    app.run(debug=True, port=PORT, host="0.0.0.0")
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=PORT, debug=True)

requirements.txt

@@ -2,4 +2,5 @@ fastapi
 uvicorn[standard]
 openai
 flask
-pillow
+pillow
+requests