62 lines
No EOL
2.6 KiB
Python
62 lines
No EOL
2.6 KiB
Python
import httpx
|
|
import logging
|
|
from typing import List, Dict, Any
|
|
|
|
from core.config import settings
|
|
from services.france_travail_auth_service import france_travail_auth_service
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class RomeoService:
|
|
def __init__(self):
|
|
# CORRIGÉ ICI: Utilise 'FRANCE_TRAVAIL_ROMEO_API_URL' comme suggéré par l'erreur
|
|
self.base_url = settings.FRANCE_TRAVAIL_ROMEO_API_URL
|
|
self._http_client = httpx.AsyncClient()
|
|
logger.info(f"RomeoService initialized with base_url: {self.base_url}")
|
|
|
|
async def _call_api(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Appel générique à une endpoint de l'API Romeo.
|
|
Récupère le jeton d'accès via le service d'authentification.
|
|
"""
|
|
access_token = await france_travail_auth_service.get_access_token()
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
url = f"{self.base_url}{endpoint}"
|
|
|
|
logger.info(f"Appel API Romeo: {url} avec données: {data.keys()}")
|
|
|
|
try:
|
|
response = await self._http_client.post(url, json=data, headers=headers)
|
|
response.raise_for_status() # Lève une exception pour les codes d'état HTTP 4xx/5xx
|
|
logger.info(f"Réponse API Romeo reçue (status: {response.status_code}).")
|
|
return response.json()
|
|
except httpx.HTTPStatusError as e:
|
|
logger.error(f"Erreur HTTP lors de l'appel à l'API Romeo: {e.response.status_code} - {e.response.text}")
|
|
raise RuntimeError(f"Erreur lors de l'appel à l'API Romeo: {e.response.text}")
|
|
except Exception as e:
|
|
logger.error(f"Erreur inattendue lors de l'appel à l'API Romeo: {e}")
|
|
raise RuntimeError(f"Erreur inattendue lors de l'appel à l'API Romeo: {e}")
|
|
|
|
async def predict_metiers(self, text: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
Prédit les métiers ROME à partir d'un texte donné.
|
|
"""
|
|
endpoint = "/predire/metiers"
|
|
data = {"texte": text}
|
|
response_data = await self._call_api(endpoint, data)
|
|
return response_data.get("predictions", [])
|
|
|
|
async def predict_competences(self, text: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
Prédit les compétences ROME à partir d'un texte donné.
|
|
"""
|
|
endpoint = "/predire/competences"
|
|
data = {"texte": text}
|
|
response_data = await self._call_api(endpoint, data)
|
|
return response_data.get("predictions", [])
|
|
|
|
# Instanciation unique du service Romeo
|
|
romeo_service = RomeoService() |