116 lines
4.3 KiB
Python
116 lines
4.3 KiB
Python
import os
|
|
import json
|
|
import logging
|
|
import asyncio
|
|
from typing import Dict, Any, Optional
|
|
from google import genai
|
|
from google.genai import types
|
|
from sqlalchemy import select
|
|
from app.db.session import SessionLocal
|
|
from app.models import SystemParameter
|
|
|
|
logger = logging.getLogger("AI-Service")
|
|
|
|
class AIService:
|
|
"""
|
|
AI Service v1.2.4 - Production Ready
|
|
- Robot 2 (Technical Enrichment) & Robot 3 (OCR)
|
|
- Fix: JSON response cleaning and array-to-dict transformation.
|
|
"""
|
|
api_key = os.getenv("GEMINI_API_KEY")
|
|
client = genai.Client(api_key=api_key) if api_key else None
|
|
PRIMARY_MODEL = "gemini-2.0-flash"
|
|
|
|
@classmethod
|
|
async def get_config_delay(cls) -> float:
|
|
"""Lekéri az adminisztrálható késleltetést az adatbázisból."""
|
|
try:
|
|
async with SessionLocal() as db:
|
|
stmt = select(SystemParameter).where(SystemParameter.key == "AI_REQUEST_DELAY")
|
|
res = await db.execute(stmt)
|
|
param = res.scalar_one_or_none()
|
|
return float(param.value) if param else 1.0
|
|
except Exception:
|
|
return 1.0
|
|
|
|
@classmethod
|
|
async def get_clean_vehicle_data(cls, make: str, raw_model: str, v_type: str) -> Optional[Dict[str, Any]]:
|
|
"""Robot 2: Gépjármű technikai adatok dúsítása."""
|
|
if not cls.client:
|
|
return None
|
|
|
|
await asyncio.sleep(await cls.get_config_delay())
|
|
|
|
prompt = f"""
|
|
Jármű: {make} {raw_model} ({v_type}).
|
|
Adj technikai adatokat JSON formátumban.
|
|
FONTOS: A 'technical_code' mező NEM lehet üres. Ha nem tudod a gyári kódot, adj 'N/A' értéket!
|
|
|
|
Várt struktúra:
|
|
{{
|
|
"marketing_name": "tiszta marketing név",
|
|
"technical_code": "gyári kód vagy N/A",
|
|
"ccm": egész szám,
|
|
"kw": egész szám,
|
|
"maintenance": {{
|
|
"oil_type": "viszkozitás",
|
|
"oil_qty": tizedes tört literben,
|
|
"spark_plug": "gyertya típus",
|
|
"coolant": "hűtőfolyadék"
|
|
}}
|
|
}}
|
|
"""
|
|
config = types.GenerateContentConfig(
|
|
system_instruction="Profi gépjárműtechnikus vagy. Kizárólag tiszta JSON-t válaszolsz.",
|
|
response_mime_type="application/json",
|
|
temperature=0.1
|
|
)
|
|
|
|
try:
|
|
response = cls.client.models.generate_content(model=cls.PRIMARY_MODEL, contents=prompt, config=config)
|
|
res_json = json.loads(response.text)
|
|
|
|
if isinstance(res_json, list) and len(res_json) > 0:
|
|
res_json = res_json[0]
|
|
|
|
return res_json if isinstance(res_json, dict) else None
|
|
except Exception as e:
|
|
logger.error(f"❌ AI hiba ({make} {raw_model}): {e}")
|
|
return None
|
|
|
|
@classmethod
|
|
async def analyze_document_image(cls, image_data: bytes, doc_type: str) -> Optional[Dict[str, Any]]:
|
|
"""Robot 3: Multimodális OCR elemzés (Képbeolvasás)."""
|
|
if not cls.client:
|
|
return None
|
|
|
|
await asyncio.sleep(await cls.get_config_delay())
|
|
|
|
prompts = {
|
|
"identity": "Személyes okmány adatok.",
|
|
"vehicle_reg": "Rendszám, alvázszám, technikai adatok.",
|
|
"invoice": "Számla adatok, összegek, dátumok.",
|
|
"odometer": "Csak a kilométeróra állása számként."
|
|
}
|
|
|
|
config = types.GenerateContentConfig(
|
|
system_instruction="Profi OCR dokumentum-elemző vagy. Csak tiszta JSON-t válaszolsz.",
|
|
response_mime_type="application/json"
|
|
)
|
|
|
|
try:
|
|
response = cls.client.models.generate_content(
|
|
model=cls.PRIMARY_MODEL,
|
|
contents=[
|
|
f"Elemezd ezt a képet ({doc_type}): {prompts.get(doc_type, '')}",
|
|
types.Part.from_bytes(data=image_data, mime_type="image/jpeg")
|
|
],
|
|
config=config
|
|
)
|
|
res_json = json.loads(response.text)
|
|
if isinstance(res_json, list) and len(res_json) > 0:
|
|
res_json = res_json[0]
|
|
return res_json if isinstance(res_json, dict) else None
|
|
except Exception as e:
|
|
logger.error(f"❌ AI OCR hiba ({doc_type}): {e}")
|
|
return None |