225 lines
No EOL
13 KiB
Python
225 lines
No EOL
13 KiB
Python
import httpx
|
|
import logging
|
|
import time
|
|
from typing import Optional, Dict, Any, List, Tuple
|
|
import asyncio
|
|
|
|
from core.config import settings
|
|
from schemas.france_travail import FranceTravailSearchResponse, OffreDetail
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class FranceTravailOfferService:
|
|
def __init__(self):
|
|
self.client = httpx.AsyncClient(base_url=settings.FRANCE_TRAVAIL_API_BASE_URL)
|
|
self.auth_client = httpx.AsyncClient(base_url=settings.FRANCE_TRAVAIL_TOKEN_URL.split('?')[0])
|
|
self.token_info = {"token": None, "expires_at": 0}
|
|
self.geo_api_client = httpx.AsyncClient() # Client for geo.api.gouv.fr
|
|
|
|
async def _get_access_token(self):
|
|
if self.token_info["token"] and self.token_info["expires_at"] > time.time() + 60: # Refresh 1 min before expiry
|
|
logger.info("Utilisation du token France Travail depuis le cache.")
|
|
return self.token_info["token"]
|
|
|
|
logger.info("Obtention d'un nouveau token France Travail...")
|
|
try:
|
|
token_url_with_realm = settings.FRANCE_TRAVAIL_TOKEN_URL
|
|
response = await self.auth_client.post(
|
|
token_url_with_realm,
|
|
data={
|
|
"grant_type": "client_credentials",
|
|
"client_id": settings.FRANCE_TRAVAIL_CLIENT_ID,
|
|
"client_secret": settings.FRANCE_TRAVAIL_CLIENT_SECRET,
|
|
"scope": "o2dsoffre api_offresdemploiv2"
|
|
}
|
|
)
|
|
response.raise_for_status()
|
|
token_data = response.json()
|
|
self.token_info["token"] = token_data["access_token"]
|
|
self.token_info["expires_at"] = time.time() + token_data["expires_in"]
|
|
logger.info("Nouveau token France Travail obtenu et mis en cache.")
|
|
return self.token_info["token"]
|
|
except httpx.HTTPStatusError as e:
|
|
logger.error(f"Échec de l'obtention du token France Travail: {e.response.status_code} - {e.response.text}")
|
|
raise RuntimeError(f"Échec de l'obtention du token France Travail: {e.response.text}")
|
|
except httpx.RequestError as e:
|
|
logger.error(f"Erreur réseau lors de l'obtention du token France Travail: {e}")
|
|
raise RuntimeError(f"Erreur réseau lors de l'obtention du token France Travail: {e}")
|
|
|
|
async def get_insee_and_postal_code_for_commune(self, commune_name: str) -> Optional[Tuple[str, str, float, float]]:
|
|
logger.info(f"Début de la recherche Geo API pour: '{commune_name}'")
|
|
try:
|
|
# First, try exact match by name
|
|
geo_url_by_name = f"https://geo.api.gouv.fr/communes?nom={commune_name}&fields=codesPostaux,code,nom,centre&format=json&limit=5"
|
|
logger.info(f"Recherche par nom via API Geo.gouv.fr: {geo_url_by_name}")
|
|
response = await self.geo_api_client.get(geo_url_by_name)
|
|
response.raise_for_status()
|
|
communes = response.json()
|
|
|
|
for commune in communes:
|
|
# Prioritize exact name match
|
|
if commune['nom'].lower() == commune_name.lower():
|
|
insee_code = commune['code']
|
|
postal_code = commune['codesPostaux'][0] if commune['codesPostaux'] else None
|
|
latitude = commune['centre']['coordinates'][1]
|
|
longitude = commune['centre']['coordinates'][0]
|
|
logger.info(f"Correspondance exacte par nom trouvée: INSEE='{insee_code}', CP='{postal_code}', Lat='{latitude}', Long='{longitude}' pour '{commune_name}'.")
|
|
return insee_code, postal_code, latitude, longitude
|
|
|
|
# If no exact name match, try the first result if available
|
|
if communes:
|
|
commune = communes[0]
|
|
insee_code = commune['code']
|
|
postal_code = commune['codesPostaux'][0] if commune['codesPostaux'] else None
|
|
latitude = commune['centre']['coordinates'][1]
|
|
longitude = commune['centre']['coordinates'][0]
|
|
logger.info(f"Aucune correspondance exacte par nom, première commune trouvée: INSEE='{insee_code}', CP='{postal_code}', Lat='{latitude}', Long='{longitude}' pour '{commune_name}'.")
|
|
return insee_code, postal_code, latitude, longitude
|
|
|
|
# If not found by name, try by postal code if commune_name looks like a postal code
|
|
if commune_name.isdigit() and len(commune_name) == 5:
|
|
geo_url_by_cp = f"https://geo.api.gouv.fr/communes?codePostal={commune_name}&fields=codesPostaux,code,nom,centre&format=json&limit=1"
|
|
logger.info(f"Recherche par code postal via API Geo.gouv.fr: {geo_url_by_cp}")
|
|
response = await self.geo_api_client.get(geo_url_by_cp)
|
|
response.raise_for_status()
|
|
communes_by_cp = response.json()
|
|
if communes_by_cp:
|
|
commune = communes_by_cp[0]
|
|
insee_code = commune['code']
|
|
postal_code = commune['codesPostaux'][0] if commune['codesPostaux'] else None
|
|
latitude = commune['centre']['coordinates'][1]
|
|
longitude = commune['centre']['coordinates'][0]
|
|
logger.info(f"Correspondance par code postal trouvée: INSEE='{insee_code}', CP='{postal_code}', Lat='{latitude}', Long='{longitude}' pour code postal '{commune_name}'.")
|
|
return insee_code, postal_code, latitude, longitude
|
|
|
|
logger.warning(f"Aucune correspondance trouvée pour la commune/code postal: '{commune_name}' dans l'API Geo.gouv.fr.")
|
|
return None
|
|
except httpx.HTTPStatusError as e:
|
|
logger.error(f"Erreur HTTP lors de l'appel à l'API Geo.gouv.fr pour '{commune_name}': {e.response.status_code} - {e.response.text}")
|
|
return None
|
|
except httpx.RequestError as e:
|
|
logger.error(f"Erreur réseau lors de l'appel à l'API Geo.gouv.fr pour '{commune_name}': {e}")
|
|
return None
|
|
|
|
async def search_offers(
|
|
self,
|
|
motsCles: Optional[str] = None,
|
|
commune: Optional[str] = None, # Reste le nom de la commune
|
|
codePostal: Optional[str] = None,
|
|
latitude: Optional[float] = None,
|
|
longitude: Optional[float] = None,
|
|
distance: Optional[int] = None,
|
|
# codeDepartement: Optional[str] = None, # Ce paramètre sera maintenant dérivé en interne
|
|
range_start: int = 0,
|
|
range_end: int = 14,
|
|
typeContrat: Optional[List[str]] = None,
|
|
experience: Optional[List[str]] = None
|
|
) -> FranceTravailSearchResponse:
|
|
token = await self._get_access_token()
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
params: Dict[str, Any] = {
|
|
"range": f"{range_start}-{range_end}"
|
|
}
|
|
|
|
if motsCles:
|
|
params["motsCles"] = motsCles
|
|
if typeContrat:
|
|
params["typeContrat"] = ','.join(typeContrat)
|
|
if experience:
|
|
params["experience"] = ','.join(experience)
|
|
|
|
# Logique de localisation améliorée
|
|
# insee_code = None # Non utilisé directement comme paramètre pour l'API France Travail
|
|
# postal_code_for_api = None # Non utilisé directement comme paramètre pour l'API France Travail
|
|
# latitude_for_api = None # Non utilisé directement comme paramètre pour l'API France Travail
|
|
# longitude_for_api = None # Non utilisé directement comme paramètre pour l'API France Travail
|
|
|
|
# Le codeDepartement sera déterminé ici si une commune est fournie
|
|
derived_departement_code = None
|
|
|
|
if commune:
|
|
logger.info(f"Traitement de la commune spécifiée: '{commune}' pour dériver le département.")
|
|
geo_data = await self.get_insee_and_postal_code_for_commune(commune)
|
|
if geo_data:
|
|
insee_code, postal_code_from_geo, latitude_from_geo, longitude_from_geo = geo_data
|
|
|
|
# Dériver le code départemental du code postal ou INSEE
|
|
if postal_code_from_geo and len(postal_code_from_geo) >= 2:
|
|
derived_departement_code = postal_code_from_geo[:2]
|
|
# Cas spécifiques pour la Corse
|
|
if postal_code_from_geo.startswith('2A'):
|
|
derived_departement_code = '2A'
|
|
elif postal_code_from_geo.startswith('2B'):
|
|
derived_departement_code = '2B'
|
|
|
|
logger.info(f"Département dérivé de '{commune}': {derived_departement_code}")
|
|
|
|
# Si un département est dérivé, l'utiliser prioritairement
|
|
if derived_departement_code:
|
|
params["departement"] = derived_departement_code
|
|
logger.info(f"Paramètre 'departement' utilisé (dérivé de la commune): {derived_departement_code}")
|
|
elif latitude_from_geo and longitude_from_geo and distance is not None:
|
|
# Fallback sur latitude/longitude si la dérivation du département échoue
|
|
params["latitude"] = latitude_from_geo
|
|
params["longitude"] = longitude_from_geo
|
|
params["distance"] = distance
|
|
logger.info(f"Paramètres de localisation utilisés (dérivés de la commune): Latitude/Longitude et Distance: Lat={latitude_from_geo}, Long={longitude_from_geo}, Dist={distance}")
|
|
else:
|
|
logger.warning(f"Impossible de dériver le département ou d'obtenir des coordonnées valides pour la commune '{commune}'. Recherche sans localisation précise.")
|
|
else:
|
|
logger.warning(f"Impossible d'obtenir les données géographiques pour la commune '{commune}'. Recherche sans localisation précise.")
|
|
elif codePostal:
|
|
params["codePostal"] = codePostal
|
|
if distance is not None:
|
|
params["distance"] = distance
|
|
logger.info(f"Paramètres de localisation utilisés: Code Postal={codePostal}, Distance={distance}")
|
|
elif latitude is not None and longitude is not None:
|
|
params["latitude"] = latitude
|
|
params["longitude"] = longitude
|
|
if distance is not None:
|
|
params["distance"] = distance
|
|
logger.info(f"Paramètres de localisation utilisés: Latitude/Longitude: Lat={latitude}, Long={longitude}, Dist={distance}")
|
|
else:
|
|
logger.warning("Aucun paramètre de localisation valide (commune, code postal, lat/long) n'a été spécifié. La recherche sera nationale.")
|
|
|
|
|
|
logger.info(f"Appel à l'API France Travail pour search_offers avec paramètres FINAUX: {params}")
|
|
try:
|
|
response = await self.client.get("/v2/offres/search", headers=headers, params=params)
|
|
response.raise_for_status()
|
|
logger.info(f"Réponse brute de l'API France Travail (search_offers): {response.json()}")
|
|
return FranceTravailSearchResponse(**response.json())
|
|
except httpx.HTTPStatusError as e:
|
|
logger.error(f"Échec de la recherche d'offres France Travail: {e.response.status_code} - {e.response.text}")
|
|
if e.response.status_code == 400 and "incorrect value" in e.response.text.lower() and "commune" in e.response.text.lower():
|
|
raise RuntimeError(f"L'API France Travail a renvoyé une erreur 400: La valeur du paramètre 'commune' est incorrecte. Veuillez vérifier le code INSEE ou envisager une recherche par département ou latitude/longitude.")
|
|
else:
|
|
raise RuntimeError(f"Échec de la recherche d'offres France Travail: {e.response.text}")
|
|
except httpx.RequestError as e:
|
|
logger.error(f"Erreur réseau lors de la recherche d'offres France Travail: {e}")
|
|
raise RuntimeError(f"Erreur réseau lors de la recherche d'offres France Travail: {e}")
|
|
|
|
async def get_offer_details(self, offer_id: str) -> OffreDetail:
|
|
token = await self._get_access_token()
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
logger.info(f"Appel à l'API France Travail pour get_offer_details avec id: {offer_id}")
|
|
try:
|
|
response = await self.client.get(f"/v2/offres/{offer_id}", headers=headers)
|
|
response.raise_for_status()
|
|
logger.info(f"Réponse brute de l'API France Travail (get_offer_details): {response.json()}")
|
|
return OffreDetail(**response.json())
|
|
except httpx.HTTPStatusError as e:
|
|
logger.error(f"Échec de la récupération des détails de l'offre {offer_id}: {e.response.status_code} - {e.response.text}")
|
|
raise RuntimeError(f"Échec de la récupération des détails de l'offre {offer_id}: {e.response.text}")
|
|
except httpx.RequestError as e:
|
|
logger.error(f"Erreur réseau lors de la récupération des détails de l'offre {offer_id}: {e}")
|
|
raise RuntimeError(f"Erreur réseau lors de la récupération des détails de l'offre {offer_id}: {e}")
|
|
|
|
france_travail_offer_service = FranceTravailOfferService() |