141 lines
5.1 KiB
Python
141 lines
5.1 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 HartOMat 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 HartOMat 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, tenant_config: dict | None = None) -> dict:
|
|
"""
|
|
Validate thumbnail orientation using Azure GPT-4o Vision.
|
|
Updates the order_item AI validation fields in DB.
|
|
|
|
If tenant_config is provided and tenant_config["ai_enabled"] is True,
|
|
the tenant's own Azure credentials are used instead of global settings.
|
|
"""
|
|
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, tenant_config)
|
|
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,
|
|
tenant_config: dict | None = None,
|
|
) -> dict:
|
|
"""Call Azure OpenAI GPT-4o with a base64-encoded thumbnail.
|
|
|
|
Credential resolution order:
|
|
1. tenant_config (if provided and ai_enabled=True)
|
|
2. Global settings (azure_openai_* env vars)
|
|
"""
|
|
import json
|
|
|
|
# Resolve credentials from tenant config or global settings
|
|
if tenant_config and tenant_config.get("ai_enabled"):
|
|
api_key = tenant_config.get("ai_api_key") or settings.azure_openai_api_key
|
|
endpoint = tenant_config.get("ai_endpoint") or settings.azure_openai_endpoint
|
|
deployment = tenant_config.get("ai_deployment") or settings.azure_openai_deployment
|
|
api_version = tenant_config.get("ai_api_version") or settings.azure_openai_api_version
|
|
max_tokens = int(tenant_config.get("ai_max_tokens", 500))
|
|
temperature = float(tenant_config.get("ai_temperature", 0.1))
|
|
prompt = tenant_config.get("ai_validation_prompt") or VALIDATION_PROMPT
|
|
else:
|
|
api_key = settings.azure_openai_api_key
|
|
endpoint = settings.azure_openai_endpoint
|
|
deployment = settings.azure_openai_deployment
|
|
api_version = settings.azure_openai_api_version
|
|
max_tokens = 500
|
|
temperature = 0.1
|
|
prompt = VALIDATION_PROMPT
|
|
|
|
if not api_key or not 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=api_key,
|
|
azure_endpoint=endpoint,
|
|
api_version=api_version,
|
|
)
|
|
|
|
with open(thumbnail_path, "rb") as f:
|
|
image_b64 = base64.b64encode(f.read()).decode("utf-8")
|
|
|
|
response = client.chat.completions.create(
|
|
model=deployment,
|
|
messages=[
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "text", "text": prompt},
|
|
{
|
|
"type": "image_url",
|
|
"image_url": {"url": f"data:image/png;base64,{image_b64}"},
|
|
},
|
|
],
|
|
}
|
|
],
|
|
max_completion_tokens=max_tokens,
|
|
temperature=temperature,
|
|
)
|
|
|
|
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
|