160 lines
6.1 KiB
Python
160 lines
6.1 KiB
Python
from fastapi import APIRouter, Request
|
|
from fastapi.responses import JSONResponse
|
|
import logging
|
|
from pydantic import BaseModel
|
|
import json
|
|
from typing import Dict, List
|
|
from datetime import datetime, timezone
|
|
|
|
from dependences.utils import (
|
|
disclaim_bool_string,
|
|
prepare_output_folder,
|
|
create_folder,
|
|
db_persistence_insert,
|
|
)
|
|
from dependences.image_extractor import ImageExtractor
|
|
from dependences.mllm_management import MLLMManager
|
|
|
|
invalid_json_input_msg = "Invalid JSON format"
|
|
unexpected_error_msg = "Unexpected Error: could not end the process"
|
|
|
|
|
|
class WCAGAltTextValuation(BaseModel):
|
|
page_url: str = "https://www.bbc.com"
|
|
context_levels: int = 5
|
|
pixel_distance_threshold: int = 200
|
|
number_of_images: int = 10
|
|
save_images: bool = True
|
|
save_elaboration: bool = True
|
|
specific_images_urls: List[str] = []
|
|
|
|
|
|
class WCAGAltTextValuationRoutes:
|
|
|
|
def __init__(self, connection_db, mllm_settings):
|
|
self.connection_db = connection_db
|
|
self.mllm_settings = mllm_settings
|
|
self.router = APIRouter()
|
|
|
|
self.router.add_api_route(
|
|
"/wgag_alttext_validation",
|
|
self.wgag_alttext_validation,
|
|
methods=["POST"],
|
|
tags=["Wcag Alt Text Validation"],
|
|
description="WCAG validator alt_text validation",
|
|
name="wgag alttext validation",
|
|
dependencies=[],
|
|
)
|
|
|
|
logging.info("wcag alttext routes correctly initialized.")
|
|
|
|
async def wgag_alttext_validation(
|
|
self, request: Request, data: WCAGAltTextValuation
|
|
) -> JSONResponse:
|
|
"""Return the alt text validation assessment based on WCAG guidelines"""
|
|
try:
|
|
json_content = json.loads(data.model_dump_json())
|
|
mllm_model_id = self.mllm_settings["mllm_model_id"]
|
|
|
|
# prepare output folders if needed---
|
|
images_output_dir = ""
|
|
if (
|
|
disclaim_bool_string(json_content["save_elaboration"]) == True
|
|
or disclaim_bool_string(json_content["save_images"]) == True
|
|
): # if something to save
|
|
url_path = (
|
|
json_content["page_url"]
|
|
.replace(":", "")
|
|
.replace("//", "_")
|
|
.replace("/", "_")
|
|
)
|
|
now = datetime.now(timezone.utc)
|
|
now_str = now.strftime("%Y_%m_%d-%H_%M_%S")
|
|
folder_str = mllm_model_id.replace(":", "-") + "_" + now_str
|
|
output_dir = prepare_output_folder(url_path, folder_str)
|
|
|
|
if disclaim_bool_string(json_content["save_images"]) == True:
|
|
images_output_dir = create_folder(
|
|
output_dir, directory_separator="/", next_path="images"
|
|
)
|
|
print("save images path:", images_output_dir)
|
|
|
|
# ---------------------
|
|
|
|
# Create extractor
|
|
image_extractor = ImageExtractor(
|
|
json_content["page_url"],
|
|
context_levels=json_content["context_levels"],
|
|
pixel_distance_threshold=json_content["pixel_distance_threshold"],
|
|
number_of_images=json_content["number_of_images"],
|
|
save_images=json_content["save_images"],
|
|
save_images_path=images_output_dir,
|
|
)
|
|
# Extract images
|
|
logging.info(f"Extracting images from: {json_content['page_url']}")
|
|
images = await image_extractor.extract_images(
|
|
specific_images_urls=json_content["specific_images_urls"]
|
|
)
|
|
# MLLM settings
|
|
mllm_end_point = self.mllm_settings["mllm_end_point"]
|
|
mllm_api_key = self.mllm_settings["mllm_api_key"]
|
|
|
|
logging.info("mllm_end_point:%s", mllm_end_point)
|
|
logging.info("mllm_model_id:%s", mllm_model_id)
|
|
|
|
# Create MLLM manager
|
|
mllm_manager = MLLMManager(mllm_end_point, mllm_api_key, mllm_model_id)
|
|
logging.info("mllm_manager.end_point:%s", mllm_manager.end_point)
|
|
mllm_responses = mllm_manager.make_alt_text_evaluation(
|
|
images,
|
|
openai_model=self.mllm_settings["openai_model"],
|
|
)
|
|
|
|
mllm_responses_object = {"mllm_model_id": mllm_model_id, "mllm_responses": mllm_responses}
|
|
|
|
returned_object = {"images": images, "mllm_validations": mllm_responses_object}
|
|
|
|
try:
|
|
# Persist to local db
|
|
# Convert JSON data to string
|
|
json_in_str = json.dumps(images, ensure_ascii=False)
|
|
json_out_str = json.dumps(mllm_responses_object, ensure_ascii=False)
|
|
db_persistence_insert(
|
|
connection_db=self.connection_db,
|
|
insert_type="wcag_alttext_validation",
|
|
json_in_str=json_in_str,
|
|
json_out_str=json_out_str,
|
|
table="wcag_validator_results",
|
|
)
|
|
except Exception as e:
|
|
logging.error("error persisting to local db: %s", e)
|
|
|
|
# save extracted images info
|
|
if (
|
|
disclaim_bool_string(json_content["save_elaboration"]) == True
|
|
): # Optionally save to JSON
|
|
|
|
await image_extractor.save_elaboration(
|
|
images, output_dir=output_dir + "/extracted_images.json"
|
|
)
|
|
|
|
# save mllm responses
|
|
with open(
|
|
output_dir + "/mllm_responses.json", "w", encoding="utf-8"
|
|
) as f:
|
|
json.dump(mllm_responses, f, indent=2, ensure_ascii=False)
|
|
|
|
return JSONResponse(content=returned_object, status_code=200)
|
|
|
|
except json.JSONDecodeError:
|
|
logging.error(invalid_json_input_msg)
|
|
return JSONResponse(
|
|
content={"error": invalid_json_input_msg}, status_code=400
|
|
)
|
|
|
|
except Exception as e:
|
|
logging.error(unexpected_error_msg + " %s", e)
|
|
return JSONResponse(
|
|
content={"error": unexpected_error_msg}, status_code=500
|
|
)
|