133 lines
6.0 KiB
Python
133 lines
6.0 KiB
Python
from ..repositories.ai_model_repository import AiModelRepository
|
|
from ..models import AiModel
|
|
from typing import Optional, List
|
|
import requests
|
|
from django.conf import settings
|
|
from django.core.cache import cache
|
|
from django.core.files.uploadedfile import UploadedFile
|
|
|
|
class ConnectionError(Exception):
|
|
"""Custom Exception สำหรับการเชื่อมต่อล้มเหลว"""
|
|
pass
|
|
|
|
class AiModelService:
|
|
def __init__(self, repository: AiModelRepository):
|
|
self.repo = repository
|
|
|
|
def get_all_models(self) -> List[AiModel]:
|
|
return self.repo.get_all()
|
|
|
|
def create_model(self, validated_data: dict) -> AiModel:
|
|
# Business Logic: ตรวจสอบความซ้ำซ้อน, การตั้งค่าเริ่มต้น
|
|
new_model = AiModel(**validated_data)
|
|
return self.repo.save(new_model)
|
|
|
|
def test_connection(self, pk: int) -> dict:
|
|
"""Logic สำหรับเรียก HTTP Ping ไปยัง AI Service ภายนอก"""
|
|
model = self.repo.get_by_id(pk)
|
|
if not model:
|
|
raise ValueError("Model not found")
|
|
|
|
# ใช้ Root URL ของ Service สำหรับ Ping
|
|
test_url = model.base_url.rstrip('/') + '/'
|
|
|
|
try:
|
|
# ใช้ requests เพื่อเรียก HTTP (ใช้ httpx ถ้าต้องการ Async)
|
|
response = requests.get(test_url, timeout=5)
|
|
response.raise_for_status() # Raise error for 4xx or 5xx
|
|
|
|
return {
|
|
"status": "success",
|
|
"message": f"Successfully pinged {test_url}",
|
|
"response_status": response.status_code,
|
|
"response_data": response.json()
|
|
}
|
|
except requests.exceptions.RequestException as e:
|
|
# แปลง HTTP Error เป็น Custom Exception ของ Business Logic
|
|
raise ConnectionError(f"Connection failed to {test_url}: {e}")
|
|
|
|
def set_status(self, pk: int, new_status: str) -> Optional[AiModel]:
|
|
# Business Logic: การเปลี่ยนสถานะต้องผ่าน Service
|
|
model = self.repo.get_by_id(pk)
|
|
if model:
|
|
# Logic: มีเงื่อนไขว่าต้องผ่าน TESTING ก่อนไป ACTIVE
|
|
if model.status == 'TESTING' and new_status == 'ACTIVE':
|
|
# Logic: Trigger Hot Swap ใน FastAPI Service ก่อนเปลี่ยนสถานะ (ถ้าต้องการ)
|
|
pass
|
|
|
|
model.status = new_status
|
|
return self.repo.save(model)
|
|
return None
|
|
|
|
# -----------------------------------------------
|
|
# Logic สำหรับ Proxy AI Inference (พร้อม Caching)
|
|
# -----------------------------------------------
|
|
def run_inference(self, pk: int, file_data: UploadedFile, user_id: int) -> dict:
|
|
model = self.repo.get_by_id(pk)
|
|
if not model:
|
|
raise ValueError(f"Model ID {pk} not found.")
|
|
|
|
if model.status != 'ACTIVE':
|
|
raise PermissionError(f"Model '{model.name}' is not currently ACTIVE.")
|
|
|
|
# --- Redis Caching Logic ---
|
|
# 1. สร้าง Cache Key จาก Model ID และ Hash/Metadata ของไฟล์
|
|
# การใช้ชื่อไฟล์และขนาดไฟล์เป็น hash อย่างง่ายก็เพียงพอ
|
|
# ใน Production ควรใช้ Hash (e.g., hashlib.sha256) ของเนื้อหาไฟล์ทั้งหมด
|
|
file_hash_key = f"{file_data.name}_{file_data.size}_{file_data.content_type}"
|
|
cache_key = f'inference_result:{pk}:{file_hash_key}'
|
|
|
|
# 2. ตรวจสอบ Cache
|
|
cached_result = cache.get(cache_key)
|
|
if cached_result:
|
|
print(f"Cache HIT for Model {pk} with file {file_data.name}")
|
|
# คืนผลลัพธ์จาก Redis ทันที
|
|
return cached_result
|
|
|
|
print(f"Cache MISS for Model {pk} - Proxying request...")
|
|
|
|
# --- Proxy Execution Logic ---
|
|
|
|
# 1. สร้าง Full Inference URL
|
|
full_url = model.full_inference_url()
|
|
|
|
# 2. จัดการ Headers (แนบ Internal Auth Key ถ้าจำเป็น)
|
|
headers = {}
|
|
if model.auth_required:
|
|
# ใช้ชื่อตัวแปรสภาพแวดล้อมที่คาดว่าจะถูกกำหนดไว้
|
|
internal_key = getattr(settings, 'AI_INTERNAL_AUTH_KEY', 'default_secret')
|
|
headers['X-Internal-Auth'] = internal_key
|
|
|
|
# 3. จัดการ Payload (ไฟล์)
|
|
# requests.post จะจัดการ Content-Type: multipart/form-data ให้อัตโนมัติ
|
|
files = {
|
|
'file': (file_data.name, file_data.file, file_data.content_type)
|
|
}
|
|
|
|
# 4. ส่ง Request ไปยัง AI Service ภายนอก
|
|
try:
|
|
response = requests.post(
|
|
full_url,
|
|
files=files,
|
|
headers=headers,
|
|
timeout=600 # 10 นาที
|
|
)
|
|
response.raise_for_status()
|
|
|
|
# 5. ประมวลผลผลลัพธ์
|
|
result = response.json()
|
|
|
|
# 6. บันทึกผลลัพธ์ลง Redis Cache
|
|
# ตั้ง TTL (Time To Live) 1 ชั่วโมง
|
|
cache.set(cache_key, result, timeout=3600)
|
|
|
|
# 7. บันทึก Audit Log (Logic ในอนาคต)
|
|
# self.repo.log_inference_request(pk, user_id, full_url, response.status_code)
|
|
|
|
return result
|
|
|
|
except requests.exceptions.HTTPError as e:
|
|
detail = f"AI Service returned error {response.status_code}: {response.text}"
|
|
raise ConnectionError(detail)
|
|
except requests.exceptions.RequestException as e:
|
|
raise ConnectionError(f"Network error or timeout connecting to {full_url}: {e}") |