185 lines
5.4 KiB
Python
185 lines
5.4 KiB
Python
import json
|
|
import time
|
|
import urllib.request
|
|
import urllib.parse
|
|
import logging
|
|
import os
|
|
import requests
|
|
import base64
|
|
import re
|
|
from PIL import Image
|
|
import io
|
|
|
|
|
|
|
|
|
|
def call_API_urlibrequest(
|
|
data={},
|
|
verbose=False,
|
|
url="",
|
|
headers=[],
|
|
method="post",
|
|
base=2, # number of seconds to wait
|
|
max_tries=3,
|
|
):
|
|
|
|
if verbose:
|
|
logging.info("input_data:%s", data)
|
|
|
|
# Allow multiple attempts to call the API incase of downtime.
|
|
# Return provided response to user after 3 failed attempts.
|
|
wait_seconds = [base**i for i in range(max_tries)]
|
|
|
|
for num_tries in range(max_tries):
|
|
try:
|
|
|
|
if method == "get":
|
|
|
|
# Encode the parameters and append them to the URL
|
|
query_string = urllib.parse.urlencode(data)
|
|
|
|
url_with_params = f"{url}?{query_string}"
|
|
request = urllib.request.Request(url_with_params, method="GET")
|
|
for ele in headers:
|
|
|
|
request.add_header(ele[0], ele[1])
|
|
|
|
elif method == "post":
|
|
# Convert the dictionary to a JSON formatted string and encode it to bytes
|
|
data_to_send = json.dumps(data).encode("utf-8")
|
|
|
|
request = urllib.request.Request(url, data=data_to_send, method="POST")
|
|
for ele in headers:
|
|
|
|
request.add_header(ele[0], ele[1])
|
|
else:
|
|
return {"error_message": "method_not_allowed"}
|
|
|
|
# Send the request and capture the response
|
|
|
|
with urllib.request.urlopen(request, timeout=300) as response:
|
|
# Read and decode the response
|
|
|
|
response_json = json.loads(response.read().decode("utf-8"))
|
|
logging.info("response_json:%s", response_json)
|
|
|
|
logging.info("response.status_code:%s", response.getcode())
|
|
return response_json
|
|
|
|
except Exception as e:
|
|
|
|
logging.error("error message:%s", e)
|
|
response_json = {"error": e}
|
|
|
|
logging.info("num_tries:%s", num_tries)
|
|
logging.info(
|
|
"Waiting %s seconds before automatically trying again.",
|
|
str(wait_seconds[num_tries]),
|
|
)
|
|
time.sleep(wait_seconds[num_tries])
|
|
|
|
logging.info(
|
|
"Tried %s times to make API call to get a valid response object", max_tries
|
|
)
|
|
logging.info("Returning provided response")
|
|
return response_json
|
|
|
|
|
|
def parse_mllm_alt_text_response(mllm_response):
|
|
"""
|
|
Parse an MLLM response string and extract key attributes into a JSON object.
|
|
|
|
from mllm response like:
|
|
```json\n{\n\"Original alt-text assessment\"... etc
|
|
to a structured dictionary.
|
|
|
|
Args:
|
|
mllm_response (str): The raw MLLM response text containing JSON data
|
|
|
|
Returns:
|
|
dict: A dictionary containing the extracted attributes, or None if parsing fails
|
|
"""
|
|
try:
|
|
# Handle NaN or None values
|
|
if mllm_response is None or mllm_response == "":
|
|
return {
|
|
"original_alt_text_assessment": None,
|
|
"assessment": None,
|
|
"evaluation_result": None,
|
|
"new_alt_text": None
|
|
}
|
|
|
|
# Extract JSON content between ```json and ``` markers
|
|
json_match = re.search(r'```json\s*(.*?)\s*```', mllm_response, re.DOTALL)
|
|
|
|
if not json_match:
|
|
# Try to find JSON without markdown code blocks
|
|
json_match = re.search(r'\{.*\}', mllm_response, re.DOTALL)
|
|
|
|
if not json_match:
|
|
return {
|
|
"original_alt_text_assessment": None,
|
|
"assessment": None,
|
|
"evaluation_result": None,
|
|
"new_alt_text": None
|
|
}
|
|
|
|
json_str = json_match.group(1) if '```json' in mllm_response else json_match.group(0)
|
|
|
|
# Parse the JSON string
|
|
parsed_data = json.loads(json_str)
|
|
|
|
# Create a structured output with the key attributes
|
|
result = {
|
|
"original_alt_text_assessment": parsed_data.get("Original alt-text assessment", ""),
|
|
"assessment": parsed_data.get("Assessment", ""),
|
|
"evaluation_result": parsed_data.get("EvaluationResult", ""),
|
|
"new_alt_text": parsed_data.get("New alt-text", "")
|
|
}
|
|
|
|
return result
|
|
|
|
except json.JSONDecodeError as e:
|
|
print(f"JSON parsing error: {e}")
|
|
return {
|
|
"original_alt_text_assessment": None,
|
|
"assessment": None,
|
|
"evaluation_result": None,
|
|
"new_alt_text": None
|
|
}
|
|
except Exception as e:
|
|
print(f"Error parsing MLLM response: {e}")
|
|
return {
|
|
"original_alt_text_assessment": None,
|
|
"assessment": None,
|
|
"evaluation_result": None,
|
|
"new_alt_text": None
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def encode_image_from_url(image_url):
|
|
response = requests.get(image_url)
|
|
|
|
# Open image and convert to RGB
|
|
image = Image.open(io.BytesIO(response.content))
|
|
|
|
# Convert to RGB (handles RGBA, grayscale, etc.)
|
|
if image.mode != 'RGB':
|
|
image = image.convert('RGB')
|
|
|
|
# Save to bytes buffer
|
|
buffer = io.BytesIO()
|
|
image.save(buffer, format='PNG') # or 'JPEG'
|
|
buffer.seek(0)
|
|
|
|
# Encode to base64
|
|
return base64.b64encode(buffer.getvalue()).decode("utf-8")
|
|
|
|
|
|
|