config/clients/openai_cli.py

import base64
import os
from io import BytesIO
from typing import Dict, List, Optional

from openai import APIConnectionError, AuthenticationError, OpenAI, RateLimitError
from PIL import Image


# Configuration
API_KEY_ENV_VAR = "OPENAI_API_KEY"
DEFAULT_TEXT_MODEL = "gpt-4o-mini"
DEFAULT_VISION_MODEL = "gpt-4o"


def _get_api_key(provided_key: Optional[str] = None) -> Optional[str]:
    """
    Returns the provided key if valid, otherwise falls back to the environment variable.
    """
    if provided_key and provided_key.strip():
        return provided_key.strip()
    return os.getenv(API_KEY_ENV_VAR)


def call_api(messages: List[Dict], model_name: Optional[str] = None, image: Optional[Image.Image] = None, api_key: Optional[str] = None, **kwargs):
    """
    Calls the OpenAI API (GPT models).
    """
    # 1. Resolve API Key
    final_api_key = _get_api_key(api_key)

    if not final_api_key:
        return "Error: Authentication required. Please provide an OpenAI API Key."

    # 2. Initialize Client
    try:
        client = OpenAI(api_key=final_api_key)
    except Exception as e:
        return f"Error: Failed to initialize OpenAI client. {e}"

    # 3. Prepare Messages & Payload
    final_messages = []

    # Handle Vision (Multimodal)
    if image:
        print("Making a VISION call to OpenAI.")
        buffered = BytesIO()
        image.save(buffered, format="PNG")
        image_base64 = base64.b64encode(buffered.getvalue()).decode("utf-8")

        # Extract user text from messages to combine with image
        user_text = "Analyze this image."
        for msg in messages:
            if msg["role"] == "user":
                user_text = msg["content"]
                break

        # Format strictly for OpenAI Vision
        content = [{"type": "text", "text": user_text}, {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image_base64}"}}]

        # Add system prompt if exists
        for msg in messages:
            if msg["role"] == "system":
                final_messages.append(msg)

        final_messages.append({"role": "user", "content": content})
        final_model = model_name or DEFAULT_VISION_MODEL

    # Handle Text-Only
    else:
        print("Making a TEXT call to OpenAI.")
        final_messages = messages
        final_model = model_name or DEFAULT_TEXT_MODEL

    # 4. Call API
    try:
        print(f"Calling OpenAI API with model: {final_model}")
        completion = client.chat.completions.create(model=final_model, messages=final_messages, **kwargs)
        return completion.choices[0].message.content
    except AuthenticationError as e:
        raise ValueError(f"Authentication Failed: {e.body.get('message', str(e)) if e.body else str(e)}")
    except RateLimitError:
        raise RuntimeError("Rate Limit Exceeded (429). Please try again later.")
    except APIConnectionError:
        raise ConnectionError("Failed to connect to API. Check your internet.")
    except Exception as e:
        raise RuntimeError(f"API Error: {str(e)}")