init

app.py

@@ -0,0 +1,142 @@
+import os
+import threading
+import requests
+from flask import Flask, request, jsonify
+
+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")
+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 = int(os.getenv("PORT"))
+
+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__)
+
+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:
+        chat_id = to_number
+
+    url = f"{GREEN_API_URL}/waInstance{GREEN_API_ID_INSTANCE}/sendMessage/{GREEN_API_TOKEN}"
+    payload = {
+        "chatId": chat_id,
+        "message": message,
+        "quotedMessageId": message_id,
+    }
+
+    for attempt in range(retries):
+        try:
+            response = requests.post(url, json=payload)
+            response.raise_for_status()
+            return response.json()
+        except requests.RequestException as e:
+            if attempt < retries - 1:
+                continue
+            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:
+        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 you go!', 'quotedMessageId': message_id}
+    files = [('file', ('image.jpg', open(image_path, 'rb'), 'image/jpeg'))]
+
+    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):
+    """
+    Generate a response using the LLM and send it to the user.
+    """
+    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 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:
+            send_image(message_id, chat_id, image_path)
+        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'])
+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()
+    auth_header = request.headers.get('Authorization', '').strip()
+
+    if auth_header != f"Bearer {WEBHOOK_AUTH_TOKEN}":
+        return jsonify({"error": "Unauthorized"}), 403
+
+    if data.get('typeWebhook') != 'incomingMessageReceived':
+        return jsonify(success=True)
+
+    try:
+        chat_id = data['senderData']['chatId']
+        message_id = data['idMessage']
+        message_data = data.get('messageData', {})
+
+        if 'textMessageData' in message_data:
+            body = message_data['textMessageData']['textMessage'].strip()
+        elif 'extendedTextMessageData' in message_data:
+            body = message_data['extendedTextMessageData']['text'].strip()
+        else:
+            return jsonify(success=True)
+
+    except KeyError as e:
+        return jsonify({"error": f"Missing key in data: {e}"}), 200
+
+    if body.lower().startswith('/imagine'):
+        prompt = body.replace('/imagine', '').strip()
+        if not prompt:
+            send_message(message_id, chat_id, "Please provide a prompt after /imagine.")
+        else:
+            send_message(message_id, chat_id, "Generating...")
+            threading.Thread(target=handle_image_generation, args=(message_id, chat_id, prompt)).start()
+    else:
+        threading.Thread(target=response_text, args=(message_id, chat_id, body)).start()
+
+    return jsonify(success=True)
+
+if __name__ == '__main__':
+    app.run(debug=True, port=PORT, host="0.0.0.0")

config.yaml

@@ -0,0 +1,18 @@
+config:
+  llm:
+    model: koboldcpp/HF_SPACE_Tiefighter-13B
+    system_prompt: |
+      Your name is {char}. You are an AI capable of creating images. users can access you through the telegram bot https://t.me/eve_uni_bot and the whatsapp group https://chat.whatsapp.com/ETDeb3AOywhCI59gbau8wU. in whatsapp just join the whatsapp group and send the command to group chat then the bot will generate the image and send it to you through the group chat so don't send command to the whatsapp bot account. so if users want to generate images privatly tell them to use the telegram bot instead of whatsapp group. because in telegram users are using the telegram bot its private. users can only create images when /imagine command is provided. example: /imagine a white cat,realistic,bight colors. if someone wrongly use the command help them with instruction on how to use generate image using /imagine command. if user struggle with prompt help them with the prompt. if user asks questions or start a conversation chat with them. if the command is mispelled tell the user about it.
+    char: Eve
+  SD:
+    steps: 30
+    negative_prompt: nsfw, nude, low quality, lowres, low details, oversaturated, undersaturated, underexposed,blurry, grainy, morbid, ugly, asymmetrical, mutated malformed, mutilated, poorly lit, bad shadow, draft, cropped, out of frame, cut off, jpeg artifacts, out of focus, glitch, duplicate,bad body parts, bad anatomy, bad hands, bad face, bad eyes, bad mouth, bad ears, bad legs, ugly face, ugly eyes, watermark, text, error, missing fingers
+    width: 1024
+    height: 1024
+    sampler_name: DPM++ 2M
+    cfg_scale: 7.0
+  app:
+    whatsapp_bot_enabled: true
+    telegram_bot_enabled: true
+    version: "0.3 Beta"
+    debug: true

llm.py

@@ -0,0 +1,35 @@
+from openai import OpenAI
+from dotenv import load_dotenv
+import os
+from utils import read_config
+
+load_dotenv()
+BASE_URL = os.getenv("LLM_BASE_URL")
+
+# Initialize the OpenAI API with your API key
+client = OpenAI(
+    api_key="koboldcpp",
+    base_url=BASE_URL
+)
+
+def pre_process():
+    # Read the config each time pre_process is called
+    config = read_config()
+    system_prompt = config['llm']['system_prompt']
+    char = config['llm']['char']
+
+    system_prompt = system_prompt.replace("{char}", char)
+    return system_prompt
+
+# Function to handle messages and stream response from OpenAI
+def generate_llm(prompt):
+    system_prompt = pre_process()
+    response = client.chat.completions.create(
+        model="koboldcpp/HF_SPACE_Tiefighter-13B",
+        messages=[
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": prompt}
+        ]
+    )
+    msg = response.choices[0].message.content
+    return msg

requirements.txt

@@ -0,0 +1,4 @@
+fastapi
+uvicorn[standard]
+openai
+flask

sd.py

@@ -0,0 +1,76 @@
+import requests
+import base64
+import os
+from datetime import datetime
+import io
+import yaml
+
+IMAGE_GENERATION_URL = os.getenv("SD_API_URL")
+CACHE_DIR = '/tmp/cache'  # Use /tmp directory which is writable
+
+# Ensure the cache directory exists
+os.makedirs(CACHE_DIR, exist_ok=True)
+
+def load_config():
+    with open('config.yaml', 'r') as f:
+        return yaml.safe_load(f)
+
+def generate_sd(prompt):
+    """
+    Generate an image using the specified prompt and save it to the cache directory.
+
+    Returns:
+        image_data (bytes): Raw image data as bytes.
+        image_path (str): Absolute path to the saved image file using forward slashes.
+    """
+    config = load_config()
+    config = config['config']
+    # Define payload for the POST request
+    payload = {
+        "prompt": prompt,
+        "steps": config['SD']['steps'],
+        "width": config['SD']['width'],
+        "height": config['SD']['height'],
+        "sampler_name": config['SD']['sampler_name'],
+        "seed": -1,
+        "cfg_scale": config['SD']['cfg_scale'],
+        "negative_prompt": config['SD']['negative_prompt'],
+    }
+
+    try:
+        # Send POST request to the image generation endpoint
+        response = requests.post(IMAGE_GENERATION_URL, json=payload)
+        response.raise_for_status()  # Raise an exception for HTTP errors
+        r = response.json()
+
+        # Decode the image and save it to the cache folder
+        if 'images' in r and r['images']:
+            image_data = base64.b64decode(r['images'][0])
+            timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
+            image_filename = f"image_{timestamp}.jpg"
+            prompt_filename = f"image_{timestamp}.yaml"
+            image_path = os.path.abspath(os.path.join(CACHE_DIR, image_filename))
+            prompt_path = os.path.abspath(os.path.join(CACHE_DIR, prompt_filename))
+
+            # Save the prompt to a YAML file
+            with open(prompt_path, 'w') as f:
+                yaml.dump(payload, f)
+
+            # Save the image as a JPEG file
+            with open(image_path, 'wb') as f:
+                image = io.BytesIO(image_data)
+                f.write(image.getbuffer())
+
+            # Convert the path to use forward slashes
+            image_path = image_path.replace('\\', '/')
+
+            return image_data, image_path
+        else:
+            print("No images found in the response.")
+            return None, None
+    except requests.RequestException as e:
+        print(f"HTTP request error: {e}")
+        return None, None
+    except Exception as e:
+        print(f"Error generating image: {e}")
+        return None, None

utils.py

@@ -0,0 +1,7 @@
+def read_config():
+    import yaml
+
+    with open("config.yaml", "r") as yamlfile:
+        config = yaml.load(yamlfile, Loader=yaml.FullLoader)
+
+    return config['config']