Files
2026-03-31 13:10:46 +02:00

116 lines
3.5 KiB
Python

"""Routes API — Conseils IA."""
from fastapi import APIRouter, Depends, Request, HTTPException
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import AsyncSession
from backend.database.db import get_db
from backend.database.models import AIDecision
router = APIRouter()
class AdviceRequest(BaseModel):
session_id: int | None = None
turn: int = 0
tavern_tier: int = 1
gold: int = 3
hero_id: str = ""
hero_hp: int = 40
board_minions: list = []
tavern_minions: list = []
hand_minions: list = []
freeze: bool = False
can_upgrade: bool = True
upgrade_cost: int = 5
available_spells: list = []
current_placement: int = 5
player_count: int = 8
phase: str = "recruit"
def serialize_advice(advice) -> dict:
"""Sérialise un objet FullAdvice en dict JSON-serializable."""
return {
"main_decision": {
"action": advice.main_decision.action,
"target": advice.main_decision.target,
"priority": advice.main_decision.priority,
"confidence": advice.main_decision.confidence,
"reasoning": advice.main_decision.reasoning,
"synergies": advice.main_decision.synergies_highlighted,
"warnings": advice.main_decision.warnings,
},
"secondary_decisions": [
{
"action": d.action,
"target": d.target,
"priority": d.priority,
"confidence": d.confidence,
"reasoning": d.reasoning,
}
for d in advice.secondary_decisions
],
"board_analysis": advice.board_analysis,
"strategy_long_term": advice.strategy_long_term,
"threat_assessment": advice.threat_assessment,
"processing_ms": advice.processing_ms,
"model_used": advice.model_used,
"confidence_overall": advice.confidence_overall,
}
@router.post("/")
async def get_advice(
req: AdviceRequest,
request: Request,
db: AsyncSession = Depends(get_db),
):
"""Génère un conseil IA pour l'état de jeu fourni."""
ai = request.app.state.ai_service
if not ai:
raise HTTPException(503, "Service IA non initialisé")
state = req.model_dump()
advice = await ai.get_advice(state)
data = serialize_advice(advice)
# Persister si session active
if req.session_id:
dec = AIDecision(
session_id=req.session_id,
turn=req.turn,
phase=req.phase,
game_state=state,
recommendation=data,
reasoning=advice.main_decision.reasoning,
confidence=advice.confidence_overall,
model_used=advice.model_used,
processing_ms=advice.processing_ms,
)
db.add(dec)
await db.flush()
data["decision_id"] = dec.id
return data
@router.get("/from-screen")
async def advice_from_screen(request: Request, db: AsyncSession = Depends(get_db)):
"""Génère un conseil depuis la capture d'écran en cours."""
vis = request.app.state.vision_service
ai = request.app.state.ai_service
if not vis:
raise HTTPException(503, "Service vision non disponible")
# Obtenir l'état actuel (ou déclencher une capture)
state = vis.get_current_state()
if not state:
state = await vis.capture_now()
advice = await ai.get_advice(state)
return {
"advice": serialize_advice(advice),
"screenshot": vis.get_screenshot_b64(),
"extracted_state": state,
}