158 lines
4.7 KiB
Python
158 lines
4.7 KiB
Python
from fastapi import APIRouter, HTTPException, Query, Path
|
|
import httpx
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
from schemas import EntrepriseSearch, EntrepriseSearchResponse
|
|
from utils.enterprise import (
|
|
calculer_tva_intracommunautaire,
|
|
mapper_resultat_api,
|
|
rechercher_entreprise_api,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/entreprises", tags=["Entreprises"])
|
|
|
|
|
|
@router.get("/search", response_model=EntrepriseSearchResponse)
|
|
async def rechercher_entreprise(
|
|
q: str = Query(..., min_length=2, description="Nom d'entreprise, SIREN ou SIRET"),
|
|
per_page: int = Query(5, ge=1, le=25, description="Nombre de résultats (max 25)"),
|
|
):
|
|
try:
|
|
logger.info(f" Recherche entreprise: '{q}'")
|
|
|
|
api_response = await rechercher_entreprise_api(q, per_page)
|
|
|
|
resultats_api = api_response.get("results", [])
|
|
|
|
if not resultats_api:
|
|
logger.info(f"Aucun résultat pour: {q}")
|
|
return EntrepriseSearchResponse(total_results=0, results=[], query=q)
|
|
|
|
entreprises = []
|
|
for data in resultats_api:
|
|
entreprise = mapper_resultat_api(data)
|
|
if entreprise:
|
|
entreprises.append(entreprise)
|
|
|
|
logger.info(f" {len(entreprises)} résultat(s) trouvé(s)")
|
|
|
|
return EntrepriseSearchResponse(
|
|
total_results=len(entreprises), results=entreprises, query=q
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
|
|
except Exception as e:
|
|
logger.error(f"Erreur recherche entreprise: {e}", exc_info=True)
|
|
raise HTTPException(
|
|
status_code=500, detail=f"Erreur lors de la recherche: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.get("/siren/{siren}", response_model=EntrepriseSearch)
|
|
async def lire_entreprise_par_siren(
|
|
siren: str = Path(
|
|
...,
|
|
min_length=9,
|
|
max_length=9,
|
|
pattern=r"^\d{9}$",
|
|
description="Numéro SIREN (9 chiffres)",
|
|
),
|
|
):
|
|
try:
|
|
logger.info(f"Lecture entreprise SIREN: {siren}")
|
|
|
|
api_response = await rechercher_entreprise_api(siren, per_page=1)
|
|
|
|
resultats = api_response.get("results", [])
|
|
|
|
if not resultats:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Aucune entreprise trouvée pour le SIREN {siren}",
|
|
)
|
|
|
|
entreprise_data = resultats[0]
|
|
|
|
if entreprise_data.get("siren") != siren:
|
|
raise HTTPException(status_code=404, detail=f"SIREN {siren} introuvable")
|
|
|
|
entreprise = mapper_resultat_api(entreprise_data)
|
|
|
|
if not entreprise:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail="Erreur lors du traitement des données entreprise",
|
|
)
|
|
|
|
if not entreprise.is_active:
|
|
logger.warning(f" Entreprise CESSÉE: {siren}")
|
|
|
|
return entreprise
|
|
|
|
except HTTPException:
|
|
raise
|
|
|
|
except Exception as e:
|
|
logger.error(f"Erreur lecture SIREN {siren}: {e}", exc_info=True)
|
|
raise HTTPException(
|
|
status_code=500, detail=f"Erreur lors de la récupération: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.get("/tva/{siren}")
|
|
async def calculer_tva(
|
|
siren: str = Path(
|
|
...,
|
|
min_length=9,
|
|
max_length=9,
|
|
pattern=r"^\d{9}$",
|
|
description="Numéro SIREN (9 chiffres)",
|
|
),
|
|
):
|
|
tva_number = calculer_tva_intracommunautaire(siren)
|
|
|
|
if not tva_number:
|
|
raise HTTPException(status_code=400, detail=f"SIREN invalide: {siren}")
|
|
|
|
return {
|
|
"siren": siren,
|
|
"vat_number": tva_number,
|
|
"format": "FR + Clé (2 chiffres) + SIREN (9 chiffres)",
|
|
}
|
|
|
|
|
|
@router.get("/health")
|
|
async def health_check_api_sirene():
|
|
try:
|
|
async with httpx.AsyncClient(timeout=5.0) as client:
|
|
response = await client.get(
|
|
"https://recherche-entreprises.api.gouv.fr/search",
|
|
params={"q": "test", "per_page": 1},
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return {
|
|
"status": "healthy",
|
|
"api_sirene": "disponible",
|
|
"response_time_ms": response.elapsed.total_seconds() * 1000,
|
|
"timestamp": datetime.now().isoformat(),
|
|
}
|
|
else:
|
|
return {
|
|
"status": "degraded",
|
|
"api_sirene": f"statut {response.status_code}",
|
|
"timestamp": datetime.now().isoformat(),
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Health check failed: {e}")
|
|
return {
|
|
"status": "unhealthy",
|
|
"api_sirene": "indisponible",
|
|
"error": str(e),
|
|
"timestamp": datetime.now().isoformat(),
|
|
}
|