111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
"""
|
|
Azure OpenAI GPT-4o Vision validator for thumbnail orientation.
|
|
"""
|
|
import base64
|
|
import logging
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
VALIDATION_PROMPT = """You are a quality control expert for Schaeffler bearing product catalog images.
|
|
|
|
Analyze this thumbnail of a bearing/mechanical component and evaluate:
|
|
1. Is the component orientation correct for a standard product catalog? (typically isometric view, 30° elevation, 45° rotation)
|
|
2. Are the key features visible? (rolling elements, rings, cage if present)
|
|
3. Does it match standard Schaeffler catalog angle conventions?
|
|
|
|
Respond in JSON with exactly these fields:
|
|
{
|
|
"passed": true/false,
|
|
"confidence": 0.0-1.0,
|
|
"feedback": "Brief explanation",
|
|
"suggested_rotation": "Description of recommended adjustment if needed"
|
|
}"""
|
|
|
|
|
|
def validate_thumbnail(order_item_id: str) -> dict:
|
|
"""
|
|
Validate thumbnail orientation using Azure GPT-4o Vision.
|
|
Updates the order_item AI validation fields in DB.
|
|
"""
|
|
from app.config import settings
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
from app.models.order_item import OrderItem, AIValidationStatus
|
|
|
|
engine = create_engine(settings.database_url_sync)
|
|
with Session(engine) as session:
|
|
item = session.get(OrderItem, uuid.UUID(order_item_id))
|
|
if not item:
|
|
logger.error(f"OrderItem not found: {order_item_id}")
|
|
return {}
|
|
|
|
item.ai_validation_status = AIValidationStatus.pending
|
|
session.commit()
|
|
|
|
try:
|
|
result = _call_azure_vision(item.thumbnail_path, settings)
|
|
item.ai_validation_status = AIValidationStatus.completed
|
|
item.ai_validation_result = result
|
|
except Exception as exc:
|
|
logger.error(f"AI validation failed for {order_item_id}: {exc}")
|
|
item.ai_validation_status = AIValidationStatus.failed
|
|
item.ai_validation_result = {"error": str(exc)}
|
|
result = {}
|
|
|
|
session.commit()
|
|
return result
|
|
|
|
|
|
def _call_azure_vision(thumbnail_path: str | None, settings) -> dict:
|
|
"""Call Azure OpenAI GPT-4o with a base64-encoded thumbnail."""
|
|
import json
|
|
|
|
if not settings.azure_openai_api_key or not settings.azure_openai_endpoint:
|
|
raise ValueError("Azure OpenAI credentials not configured")
|
|
|
|
if not thumbnail_path or not Path(thumbnail_path).exists():
|
|
raise FileNotFoundError(f"Thumbnail not found: {thumbnail_path}")
|
|
|
|
try:
|
|
from openai import AzureOpenAI
|
|
|
|
client = AzureOpenAI(
|
|
api_key=settings.azure_openai_api_key,
|
|
azure_endpoint=settings.azure_openai_endpoint,
|
|
api_version=settings.azure_openai_api_version,
|
|
)
|
|
|
|
with open(thumbnail_path, "rb") as f:
|
|
image_b64 = base64.b64encode(f.read()).decode("utf-8")
|
|
|
|
response = client.chat.completions.create(
|
|
model=settings.azure_openai_deployment,
|
|
messages=[
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "text", "text": VALIDATION_PROMPT},
|
|
{
|
|
"type": "image_url",
|
|
"image_url": {"url": f"data:image/png;base64,{image_b64}"},
|
|
},
|
|
],
|
|
}
|
|
],
|
|
max_tokens=500,
|
|
temperature=0.1,
|
|
)
|
|
|
|
content = response.choices[0].message.content or ""
|
|
# Extract JSON from response
|
|
start = content.find("{")
|
|
end = content.rfind("}") + 1
|
|
if start >= 0 and end > start:
|
|
return json.loads(content[start:end])
|
|
return {"passed": False, "confidence": 0.0, "feedback": content, "suggested_rotation": ""}
|
|
|
|
except Exception as exc:
|
|
raise RuntimeError(f"Azure OpenAI call failed: {exc}") from exc
|