(anct) fetch and display organization names of communes

ANCT-specific extraction of organization names for communes, front
end changes to match.
This commit is contained in:
Laurent Bossavit
2024-12-05 15:44:30 +01:00
committed by Laurent Bossavit
parent d495ef3e19
commit 20cc173e93
10 changed files with 93 additions and 22 deletions

View File

@@ -3,6 +3,7 @@
import logging
import requests
from requests.adapters import HTTPAdapter, Retry
from core.plugins.base import BaseOrganizationPlugin
@@ -33,20 +34,17 @@ class NameFromSiretOrganizationPlugin(BaseOrganizationPlugin):
logger.warning("Empty list 'liste_enseignes' in %s", organization_data)
return None
def _get_organization_name_from_siret(self, siret):
"""Return the organization name from the SIRET."""
try:
response = requests.get(self._api_url.format(siret=siret), timeout=10)
response.raise_for_status()
data = response.json()
except requests.RequestException as exc:
logger.exception("%s: Unable to fetch organization name from SIRET", exc)
return None
def get_organization_name_from_results(self, data, siret):
"""Return the organization name from the results of a SIRET search."""
for result in data["results"]:
nature = "nature_juridique"
commune = nature in result and result[nature] == "7210"
for organization in result["matching_etablissements"]:
if organization.get("siret") == siret:
return self._extract_name_from_organization_data(organization)
if commune:
return organization["libelle_commune"].title()
return self._extract_name_from_organization_data(organization)
logger.warning("No organization name found for SIRET %s", siret)
return None
@@ -63,10 +61,21 @@ class NameFromSiretOrganizationPlugin(BaseOrganizationPlugin):
# In the nominal case, there is only one registration ID because
# the organization as been created from it.
name = self._get_organization_name_from_siret(
organization.registration_id_list[0]
)
if not name:
try:
# Retry logic as the API may be rate limited
s = requests.Session()
retries = Retry(total=5, backoff_factor=0.1, status_forcelist=[429])
s.mount("https://", HTTPAdapter(max_retries=retries))
siret = organization.registration_id_list[0]
response = s.get(self._api_url.format(siret=siret), timeout=10)
response.raise_for_status()
data = response.json()
name = self.get_organization_name_from_results(data, siret)
if not name:
return
except requests.RequestException as exc:
logger.exception("%s: Unable to fetch organization name from SIRET", exc)
return
organization.name = name