ab/ay
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

mise en place de romeo

Browse source
This commit is contained in:
quasar 2025-06-29 21:23:43 +02:00
parent a4d1bae9f0
commit 6b53a419c9
2 changed files with 91 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

22
backend/core/config.py
View file

@ -7,29 +7,27 @@ class Settings(BaseSettings):
# Par défaut, un dossier 'uploads' dans le répertoire 'backend'
UPLOADS_DIR: str = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "uploads")
# Secret key pour les JWT (à générer une valeur forte en production)
SECRET_KEY: str = os.getenv("SECRET_KEY") # Assurez-vous que c'est le même que celui utilisé dans security.py si vous l'avez hardcodé là-bas
SECRET_KEY: str = os.getenv("SECRET_KEY")
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
MISTRAL_API_KEY: Optional[str] = None
GEMINI_API_KEY: Optional[str] = None
LLM_PROVIDER: str = "gemini" # Votre choix par défaut
# --- AJOUTEZ CES DEUX LIGNES ---
GEMINI_MODEL_NAME: Optional[str] = "gemini-1.5-flash" # Ou le nom de modèle Gemini que vous utilisez
MISTRAL_MODEL_NAME: Optional[str] = "mistral-tiny" # Ou le nom de modèle Mistral par défaut si vous l'utilisez
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
# --- Nouvelles variables pour l'API France Travail ---
# --- Variables pour les APIs France Travail ---
FRANCE_TRAVAIL_CLIENT_ID: str
FRANCE_TRAVAIL_CLIENT_SECRET: str
FRANCE_TRAVAIL_TOKEN_URL: str = "https://francetravail.io/connexion/oauth2/access_token?realm=%2Fpartenaire"
FRANCE_TRAVAIL_API_BASE_URL: str = "https://api.francetravail.io/partenaire/offresdemploi"
FRANCE_TRAVAIL_API_SCOPE: str = "o2dsoffre api_offresdemploiv2" # Les scopes requis par l'API
FRANCE_TRAVAIL_ROMEO_API_URL: str = "https://api.francetravail.io/partenaire/romeo/v2"
# Si vous avez un scope API par défaut pour les offres d'emploi, vous pouvez le spécifier ici, par exemple :
# FRANCE_TRAVAIL_OFFER_API_SCOPE: str = "api_offresdemploiv1"
settings = Settings()
print(f"DEBUG: FRANCE_TRAVAIL_CLIENT_ID chargé: {settings.FRANCE_TRAVAIL_CLIENT_ID}")
print(f"DEBUG: FRANCE_TRAVAIL_CLIENT_SECRET chargé: {settings.FRANCE_TRAVAIL_CLIENT_SECRET}")
# Créer le dossier d'uploads s'il n'existe pas
os.makedirs(settings.UPLOADS_DIR, exist_ok=True)
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
# --- AJOUT CRUCIAL : Instanciation de l'objet settings ---
settings = Settings()

81
backend/services/romeo_service.py Normal file
View file

@ -0,0 +1,81 @@
import httpx
import logging
from typing import List, Dict, Any, Optional
from core.config import settings
from services.oauth_service import OAuthService # Assurez-vous que ce service existe
logger = logging.getLogger(__name__)
class RomeoService:
def __init__(self):
self.base_url = settings.FRANCE_TRAVAIL_ROMEO_API_URL # URL de base de l'API Romeo
self.scope = "api_romeov2" # Scope spécifique pour Romeo
self.oauth_service = OAuthService(settings.FRANCE_TRAVAIL_OAUTH_URL, settings.FRANCE_TRAVAIL_CLIENT_ID, settings.FRANCE_TRAVAIL_CLIENT_SECRET)
self.client = httpx.AsyncClient()
logger.info(f"RomeoService initialized with base_url: {self.base_url}")
async def _get_access_token(self) -> str:
"""Récupère le token d'accès spécifique à l'API Romeo."""
try:
token_response = await self.oauth_service.get_access_token(self.scope)
return token_response.access_token
except Exception as e:
logger.error(f"Erreur lors de la récupération du token d'accès pour Romeo: {e}")
raise RuntimeError(f"Impossible de récupérer le token d'accès pour Romeo: {e}")
async def _call_api(self, endpoint: str, text: str) -> Optional[Dict[str, Any]]:
"""
Appelle un endpoint de l'API Romeo avec le texte donné.
Gère l'authentification et les erreurs de base.
"""
token = await self._get_access_token()
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json; charset=utf-8"
}
# Les APIs Romeo attendent le texte dans un champ 'texte' de l'objet JSON
data = {"texte": text}
url = f"{self.base_url}{endpoint}"
logger.info(f"Calling Romeo API: {url} with text length {len(text)}")
try:
response = await self.client.post(url, headers=headers, json=data, timeout=30.0)
response.raise_for_status() # Lève une exception pour les codes d'erreur HTTP (4xx ou 5xx)
return response.json()
except httpx.HTTPStatusError as e:
logger.error(f"Erreur HTTP lors de l'appel à Romeo {endpoint}: {e.response.status_code} - {e.response.text}")
raise RuntimeError(f"Erreur de l'API Romeo: {e.response.text}")
except httpx.RequestError as e:
logger.error(f"Erreur réseau ou de requête lors de l'appel à Romeo {endpoint}: {e}")
raise RuntimeError(f"Erreur de communication avec l'API Romeo: {e}")
except Exception as e:
logger.error(f"Une erreur inattendue est survenue lors de l'appel à Romeo {endpoint}: {e}")
raise RuntimeError(f"Erreur inattendue avec 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é.
Retourne une liste de dictionnaires avec les détails des prédictions métiers.
"""
if not text:
return [] # Retourne une liste vide si le texte est vide
response_data = await self._call_api("/predictionMetiers", text)
# Romeo renvoie souvent une liste directe de prédictions si successful
return response_data if response_data is not None else []
async def predict_competences(self, text: str) -> List[Dict[str, Any]]:
"""
Prédit les compétences ROME à partir d'un texte donné.
Retourne une liste de dictionnaires avec les détails des prédictions de compétences.
"""
if not text:
return [] # Retourne une liste vide si le texte est vide
response_data = await self._call_api("/predictionCompetences", text)
# Romeo renvoie souvent une liste directe de prédictions si successful
return response_data if response_data is not None else []
# Instanciation unique du service Romeo
romeo_service = RomeoService()