Skip to content

Commit

Permalink
update call for api entreprise v3
Browse files Browse the repository at this point in the history
  • Loading branch information
@SebastienReuiller
SebastienReuiller committed Dec 17, 2024
1 parent 7887eb6 commit 7d77737
Showing 1 changed file with 111 additions and 100 deletions.
211 changes: 111 additions & 100 deletions lemarche/utils/apis/api_entreprise.py
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
# https://github.com/betagouv/itou/blob/master/itou/utils/apis/api_entreprise.py

import logging
from dataclasses import dataclass
from datetime import date, datetime

import requests
Expand All @@ -13,21 +14,40 @@

logger = logging.getLogger(__name__)


API_ENTREPRISE_REASON = "Mise à jour données Marché de la plateforme de l'Inclusion"
TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S" # "2016-12-31T00:00:00+01:00" # timezone not managed
DATE_FORMAT = "%Y-%m-%d"


@dataclass
class Entreprise:
forme_juridique: str
forme_juridique_code: str


@dataclass
class Etablissement:
naf: str
is_closed: bool
is_head_office: bool
employees: str
employees_date_reference: str
date_constitution: date


@dataclass
class Exercice:
chiffre_affaires: str
date_fin_exercice: date


def entreprise_get_or_error(siren, reason="Inscription au marché de l'inclusion"):
"""
Obtain company data from entreprises.api.gouv.fr
documentation: https://doc.entreprise.api.gouv.fr/?json#entreprise-v2
Obtain company data from entreprise.api.gouv.fr
documentation: https://entreprise.api.gouv.fr/developpeurs/openapi#tag/Informations-generales/paths/~1v3~1insee~1sirene~1unites_legales~1%7Bsiren%7D/get # noqa
Format info:
- "date_mise_a_jour": 1449183600
- "date_derniere_mise_a_jour": 1449183600
"""
data = None
entreprise = None
error = None

query_string = urlencode(
Expand All @@ -37,43 +57,44 @@ def entreprise_get_or_error(siren, reason="Inscription au marché de l'inclusion
"object": reason,
}
)

url = f"{settings.API_ENTREPRISE_BASE_URL}/entreprises/{siren}?{query_string}"
url = f"{settings.API_ENTREPRISE_BASE_URL}/insee/sirene/unites_legales/{siren}?{query_string}"
headers = {"Authorization": f"Bearer {settings.API_ENTREPRISE_TOKEN}"}

try:
r = requests.get(url, headers=headers)
r.raise_for_status()
data = r.json()
response = r.json()
except requests.exceptions.HTTPError as e:
status_code = e.response.status_code

if status_code == 422:
error = f"SIREN « {siren} » non reconnu."
elif status_code == 404:
error = f"SIREN « {siren} » 404 ?"
else:
# logger.error("Error while fetching `%s`: %s", url, e)
error = "Problème de connexion à la base Sirene. Essayez ultérieurement."
match status_code:
case 422:
error = f"SIREN « {siren} » non reconnu."
case 404:
error = f"SIREN « {siren} » 404 ?"
case _:
logger.error("Error while fetching `%s`: %s", url, e)
error = "Problème de connexion à la base Sirene. Essayez ultérieurement."
return None, error
except requests.ReadTimeout as e: # noqa
# logger.error("Error while fetching `%s`: %s", url, e)
logger.error("Error while fetching `%s`: %s", url, e)
error = "The read operation timed out"
return None, error

if data and data.get("errors"):
error = data["errors"][0]
if response and response.get("errors"):
error = response["errors"][0]
return None, error

if not data.get("entreprise") or not data["entreprise"].get("forme_juridique"):
# logger.error("Invalid format of response from API Entreprise")
data = response["data"]
if not data.get("forme_juridique"):
logger.error(f"Invalid format of response from API Entreprise - Entreprise (siren={siren}): {response}")
error = "Le format de la réponse API Entreprise est non valide."
return None, error

entreprise = {
"forme_juridique": data["entreprise"]["forme_juridique"],
"forme_juridique_code": data["entreprise"]["forme_juridique_code"],
}
entreprise = Entreprise(
forme_juridique=data["forme_juridique"]["libelle"],
forme_juridique_code=data["forme_juridique"]["code"],
)
return entreprise, None


Expand All @@ -85,12 +106,11 @@ def siae_update_entreprise(siae):
return 0, error

update_data = dict()

if entreprise:
if entreprise["forme_juridique"]:
update_data["api_entreprise_forme_juridique"] = entreprise["forme_juridique"]
if entreprise["forme_juridique_code"]:
update_data["api_entreprise_forme_juridique_code"] = entreprise["forme_juridique_code"]
if entreprise.forme_juridique:
update_data["api_entreprise_forme_juridique"] = entreprise.forme_juridique
if entreprise.forme_juridique_code:
update_data["api_entreprise_forme_juridique_code"] = entreprise.forme_juridique_code

update_data["api_entreprise_entreprise_last_sync_date"] = timezone.now()
Siae.objects.filter(id=siae.id).update(**update_data)
Expand All @@ -101,17 +121,15 @@ def siae_update_entreprise(siae):

def etablissement_get_or_error(siret, reason="Inscription au marché de l'inclusion"):
"""
Obtain company data from entreprises.api.gouv.fr
documentation: https://doc.entreprise.api.gouv.fr/?json#etablissements-v2
Obtain company data from entreprise.api.gouv.fr
documentation: https://entreprise.api.gouv.fr/developpeurs/openapi#tag/Informations-generales/paths/~1v3~1insee~1sirene~1etablissements~1%7Bsiret%7D/get # noqa
Format info:
- "date_mise_a_jour": 1449183600
- "date_reference": "2014"
- "date_creation_etablissement": 1108594800
- "date_creation": 1108594800
- "date_fermeture": 1315173600
"""
data = None
etablissement = None
error = None

query_string = urlencode(
Expand All @@ -121,55 +139,51 @@ def etablissement_get_or_error(siret, reason="Inscription au marché de l'inclus
"object": reason,
}
)

url = f"{settings.API_ENTREPRISE_BASE_URL}/etablissements/{siret}?{query_string}"
url = f"{settings.API_ENTREPRISE_BASE_URL}/insee/sirene/etablissements/{siret}?{query_string}"
headers = {"Authorization": f"Bearer {settings.API_ENTREPRISE_TOKEN}"}

try:
r = requests.get(url, headers=headers)
r.raise_for_status()
data = r.json()
response = r.json()
except requests.exceptions.HTTPError as e:
status_code = e.response.status_code

if status_code == 422:
error = f"SIRET « {siret} » non reconnu."
elif status_code == 404:
error = f"SIRET « {siret} » 404 ?"
else:
# logger.error("Error while fetching `%s`: %s", url, e)
error = "Problème de connexion à la base Sirene. Essayez ultérieurement."
match status_code:
case 422:
error = f"SIRET « {siret} » non reconnu." # TODO: check v3 error code
case 404:
error = f"SIRET « {siret} » 404 ?"
case _:
# logger.error("Error while fetching `%s`: %s", url, e)
error = f"Problème de connexion à la base Sirene. Essayez ultérieurement. ({status_code})"
return None, error
except requests.ReadTimeout as e: # noqa
# logger.error("Error while fetching `%s`: %s", url, e)
logger.error("Error while fetching `%s`: %s", url, e)
error = "The read operation timed out"
return None, error

if data and data.get("errors"):
error = data["errors"][0]
if response and response.get("errors"):
error = response["errors"][0]
return None, error

if not data.get("etablissement") or not data["etablissement"].get("adresse"):
# logger.error("Invalid format of response from API Entreprise")
data = response["data"]
if (
not data.get("activite_principale")
or not data.get("etat_administratif")
or not data.get("tranche_effectif_salarie")
or not data.get("date_creation")
):
logger.error(f"Invalid format of response from API Entreprise - Etablissement (siret={siret}): {response}")
error = "Le format de la réponse API Entreprise est non valide."
return None, error

# address = data["etablissement"]["adresse"]
etablissement = {
# name=address["l1"],
# # FIXME To check (l4 => line_1)
# address_line_1=address["l4"],
# address_line_2=address["l3"],
# post_code=address["code_postal"],
# city=address["localite"],
# department=department_from_postcode(address["code_postal"]),
"naf": data["etablissement"]["naf"],
"is_closed": data["etablissement"]["etat_administratif"]["value"] == "F",
"is_head_office": data["etablissement"].get("siege_social", False),
"employees": data["etablissement"]["tranche_effectif_salarie_etablissement"]["intitule"],
"employees_date_reference": data["etablissement"]["tranche_effectif_salarie_etablissement"]["date_reference"],
"date_constitution": date.fromtimestamp(data["etablissement"]["date_creation_etablissement"]),
}
etablissement = Etablissement(
naf=data["activite_principale"]["code"],
is_closed=data["etat_administratif"] == "F",
is_head_office=data.get("siege_social", False),
employees=data["tranche_effectif_salarie"]["intitule"],
employees_date_reference=data["tranche_effectif_salarie"]["date_reference"],
date_constitution=date.fromtimestamp(data["date_creation"]),
)
return etablissement, None


Expand All @@ -180,40 +194,36 @@ def siae_update_etablissement(siae):
return 0, error

update_data = dict()

if etablissement:
# update_data"nature"] = siae_constants.NATURE_HEAD_OFFICE if etablissement["is_head_office"] else siae_constants.NATURE_ANTENNA # noqa
# update_data"is_active"] = False if not etablissement["is_closed"] else True
if etablissement["employees"]:
if etablissement.employees:
update_data["api_entreprise_employees"] = (
etablissement["employees"]
if (etablissement["employees"] != "Unités non employeuses")
etablissement.employees
if (etablissement.employees != "Unités non employeuses")
else "Non renseigné"
)
if etablissement["employees_date_reference"]:
update_data["api_entreprise_employees_year_reference"] = etablissement["employees_date_reference"]
if etablissement["date_constitution"]:
update_data["api_entreprise_date_constitution"] = etablissement["date_constitution"]
if etablissement.employees_date_reference:
update_data["api_entreprise_employees_year_reference"] = etablissement.employees_date_reference
if etablissement.date_constitution:
update_data["api_entreprise_date_constitution"] = etablissement.date_constitution

update_data["api_entreprise_etablissement_last_sync_date"] = timezone.now()
Siae.objects.filter(id=siae.id).update(**update_data)

return 1, etablissement

return 0, f"SIAE {siae.id} without SIRET"


def exercice_get_or_error(siret, reason="Inscription au marché de l'inclusion"):
"""
Obtain company data from entreprises.api.gouv.fr
documentation: https://entreprise.api.gouv.fr/catalogue/#a-exercices
documentation: https://entreprise.api.gouv.fr/developpeurs/openapi#tag/Informations-financieres/paths/~1v3~1dgfip~1etablissements~1%7Bsiret%7D~1chiffres_affaires/get # noqa
Format info:
- "date_fin_exercice": "2016-12-31T00:00:00+01:00"
- "date_fin_exercice": "2024-12-17"
Often returns errors: 404, 422, 502
"""
data = None
exercice = None
error = None

query_string = urlencode(
Expand All @@ -223,57 +233,58 @@ def exercice_get_or_error(siret, reason="Inscription au marché de l'inclusion")
"object": reason,
}
)

url = f"{settings.API_ENTREPRISE_BASE_URL}/exercices/{siret}?{query_string}"
url = f"{settings.API_ENTREPRISE_BASE_URL}/dgfip/etablissements/{siret}/chiffres_affaires?{query_string}"
headers = {"Authorization": f"Bearer {settings.API_ENTREPRISE_TOKEN}"}

try:
r = requests.get(url, headers=headers)
r.raise_for_status()
data = r.json()
response = r.json()
except requests.exceptions.HTTPError as e:
status_code = e.response.status_code

if status_code == 422:
error = f"SIRET {siret} non reconnu."
else:
# logger.error("Error while fetching `%s`: %s", url, e)
error = "Problème de connexion à la base Sirene. Essayez ultérieurement."
logger.error("Error while fetching `%s`: %s", url, e)
error = f"Problème de connexion à la base Sirene. Essayez ultérieurement. ({status_code})"
return None, error
except requests.ReadTimeout as e: # noqa
# logger.error("Error while fetching `%s`: %s", url, e)
logger.error("Error while fetching `%s`: %s", url, e)
error = "The read operation timed out"
return None, error

if data and data.get("errors"):
error = data["errors"][0]
if response and response.get("errors"):
error = response["errors"][0]
return None, error

if not data.get("exercices") or not len(data["exercices"]):
# logger.error("Invalid format of response from API Entreprise")
if not response.get("data") or not len(response["data"]):
logger.error(f"Invalid format of response from API Entreprise - Exercice (siret={siret}): {response}")
error = "Le format de la réponse API Entreprise est non valide."
return None, error

exercice = data["exercices"][0]

exercice = Exercice(
chiffre_affaires=response["data"][0]["data"]["chiffre_affaires"],
date_fin_exercice=response["data"][0]["data"]["date_fin_exercice"],
)
return exercice, None


def siae_update_exercice(siae):
if siae.siret:
exercice, error = exercice_get_or_error(siae.siret, reason=API_ENTREPRISE_REASON) # noqa
exercice, error = exercice_get_or_error(siae.siret, reason=API_ENTREPRISE_REASON)
if error:
return 0, error

update_data = dict()

if exercice:
update_data = dict()
if exercice["ca"]:
update_data["api_entreprise_ca"] = exercice["ca"]
if exercice["date_fin_exercice"]:
if exercice.chiffre_affaires:
update_data["api_entreprise_ca"] = exercice.chiffre_affaires
if exercice.date_fin_exercice:
update_data["api_entreprise_ca_date_fin_exercice"] = datetime.strptime(
exercice["date_fin_exercice"][:-6], TIMESTAMP_FORMAT
exercice.date_fin_exercice, DATE_FORMAT
).date()

update_data["api_entreprise_exercice_last_sync_date"] = timezone.now()
Expand Down

0 comments on commit 7d77737

Please sign in to comment.