Alwrity - WIP - main_config

lib/gpt_providers/image_to_text_gen/gemini_image_details.py

@@ -0,0 +1,79 @@
+"""
+"""
+import os
+import logging
+from pathlib import Path
+
+import google.generativeai as genai
+logging.basicConfig(level=logging.INFO, format='%(asctime)s-%(levelname)s-%(module)s-%(lineno)d-%(message)s')
+from dotenv import load_dotenv
+load_dotenv(Path('../../.env'))
+
+from tenacity import (
+    retry,
+    stop_after_attempt,
+    wait_random_exponential,
+) # for exponential backoff
+
+
+@retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(6))
+def gemini_get_img_info(prompt, img_path):
+    """ Get image details from arxiv papers. """
+    logging.info(f"Get image details from Gemini Pro.")
+    try:
+        genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
+    except Exception as e:
+        logging.error(f"Could not load gemini API key: {e}")
+        raise e
+
+    # Set up the model
+    generation_config = {
+        "temperature": 0.9,
+        "top_p": 1,
+        "top_k": 1,
+        "max_output_tokens": 1096,
+    }
+
+    safety_settings = [{
+        "category": "HARM_CATEGORY_HARASSMENT",
+        "threshold": "BLOCK_NONE"
+    },
+    {
+        "category": "HARM_CATEGORY_HATE_SPEECH",
+        "threshold": "BLOCK_NONE"
+    },
+    {
+        "category": "HARM_CATEGORY_SEXUALLY_EXPLICIT",
+        "threshold": "BLOCK_NONE"
+    },
+    {
+        "category": "HARM_CATEGORY_DANGEROUS_CONTENT",
+        "threshold": "BLOCK_NONE"
+    },]
+
+    try:
+        model = genai.GenerativeModel(model_name="gemini-pro-vision",
+            generation_config=generation_config,
+            safety_settings=safety_settings)
+    except Exception as e:
+        logging.error(f"Could not create GenerativeModel: {e}")
+        raise e
+
+    # Validate that an image is present
+    if not (img := Path(img_path)).exists():
+        raise FileNotFoundError(f"Could not find image: {img}")
+
+    image_parts = [{
+        "mime_type": "image/png",
+        "data": Path(img_path).read_bytes()
+    },]
+
+    prompt_parts = [f"{prompt}", image_parts[0],]
+
+    try:
+        response = model.generate_content(prompt_parts)
+        return response.text
+    except Exception as e:
+        logging.error(f"Gemini is blocking this request: {response.prompt_feedback.block_reason}")
+        logging.error(f"Gemini Vision, Failed to give image Details: {e}\n{response.prompt_feedback}")
+        raise e

lib/gpt_providers/image_to_text_gen/openai_vision_img_details.py

@@ -0,0 +1,94 @@
+import requests
+import re
+import base64
+import os
+import sys
+from tenacity import (
+    retry,
+    stop_after_attempt,
+    wait_random_exponential,
+)  # for exponential backoff
+ 
+@retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(6))
+def analyze_and_extract_details_from_image(image_path):
+    """
+    Analyzes an image using OpenAI's Vision API to extract Alt Text, Description, Title, and Caption.
+    
+    This function encodes an image to a base64 string and sends a request to the OpenAI API.
+    It interprets the contents of the image, returning a textual description.
+
+    Args:
+        image_path (str): Path to the image file.
+
+    Returns:
+        dict: A dictionary with extracted details including Alt Text, Description, Title, and Caption.
+        None: If an error occurs during processing.
+
+    Raises:
+        SystemExit: If a critical error occurs that prevents the function from executing successfully.
+    """
+    try:
+        logger.info("Starting image analysis using OpenAI's Vision API.")
+
+        def encode_image(path):
+            """ Encodes an image to a base64 string. """
+            with open(path, "rb") as image_file:
+                return base64.b64encode(image_file.read()).decode('utf-8')
+
+        base64_image = encode_image(image_path)
+        logger.info("Image encoded to base64 successfully.")
+
+        headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {os.environ.get('OPENAI_API_KEY')}"
+        }
+
+        payload = {
+            "model": "gpt-4-vision-preview",
+            "messages": [
+                {
+                    "role": "user",
+                    "content": [
+                        {
+                            "type": "text",
+                            "text": "Analyze the given image and suggest the following: Alternative text(Alt Text), description, title, caption."
+                        },
+                        {
+                            "type": "image_url",
+                            "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}
+                        }
+                    ]
+                }
+            ],
+            "max_tokens": 300
+        }
+
+        response = requests.post("https://api.openai.com/v1/chat/completions", headers=headers, json=payload)
+        response.raise_for_status()
+
+        assistant_message = response.json()['choices'][0]['message']['content']
+        logger.info("Received response from OpenAI API.")
+
+        # Extracting details using regular expressions
+        alt_text_match = re.search(r'Alt Text: "(.*?)"', assistant_message)
+        description_match = re.search(r'Description: (.*?)\n\n', assistant_message)
+        title_match = re.search(r'Title: "(.*?)"', assistant_message)
+        caption_match = re.search(r'Caption: "(.*?)"', assistant_message)
+
+        image_details = {
+            'alt_text': alt_text_match.group(1) if alt_text_match else "N/A",
+            'description': description_match.group(1) if description_match else "N/A",
+            'title': title_match.group(1) if title_match else "N/A",
+            'caption': caption_match.group(1) if caption_match else "N/A"
+        }
+
+        logger.info("Image analysis completed successfully.")
+        return image_details
+
+    except requests.RequestException as e:
+        logger.error(f"GPT-Vision API communication failure. Error: {e}")
+        sys.exit(f"Exiting due to GPT-Vision API communication failure: {e}")
+
+    except Exception as e:
+        logger.error(f"Unexpected error occurred during image analysis: {e}")
+        sys.exit(f"Exiting due to an unexpected error: {e}")