[Demandeur d’emploi] Reprise de stock des critères certifiés

clevercloud/cron.json

@@ -17,6 +17,7 @@
   "0 */6 * * * $ROOT/clevercloud/run_management_command.sh sync_s3_files",
 
   "1 0 * * * $ROOT/clevercloud/run_management_command.sh update_prescriber_organization_with_api_entreprise --verbosity 2",
+  "15 0 * * * $ROOT/clevercloud/run_management_command.sh certify_selected_administrative_criteria --wet-run",
   "30 0 * * * $ROOT/clevercloud/run_management_command.sh collect_analytics_data --save",
   "45 0 * * * $ROOT/clevercloud/run_management_command.sh import_advisor_information shared_bucket/imports-gps/export_gps.xlsx --wet-run",
   "30 1 * * * $ROOT/clevercloud/run_management_command.sh new_users_to_brevo --wet-run",

config/settings/dev.py

@@ -85,4 +85,5 @@
 # Don't use json formatter in dev
 del LOGGING["handlers"]["console"]["formatter"]  # noqa: F405
 
-API_PARTICULIER_BASE_URL = "https://staging.particulier.api.gouv.fr/api/"
+API_PARTICULIER_BASE_URL = os.getenv("API_PARTICULIER_BASE_URL", "https://staging.particulier.api.gouv.fr/api/")
+API_PARTICULIER_TOKEN = os.getenv("API_PARTICULIER_TOKEN")

itou/users/management/commands/certify_selected_administrative_criteria.py

@@ -0,0 +1,142 @@
+import datetime
+import logging
+from math import ceil
+
+from django.db.models import Exists, OuterRef, Q
+from django.utils.timezone import make_aware
+
+from itou.eligibility.models.geiq import GEIQAdministrativeCriteria, GEIQSelectedAdministrativeCriteria
+from itou.eligibility.models.iae import AdministrativeCriteria, SelectedAdministrativeCriteria
+from itou.users.models import User
+from itou.utils.apis import api_particulier
+from itou.utils.command import BaseCommand
+from itou.utils.iterators import chunks
+
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """
+    Certify selected administrative criteria from eligiility diagnosis made during the last 6 months
+    by calling the Particulier API.
+    """
+
+    def add_arguments(self, parser):
+        parser.add_argument("--limit", dest="limit", action="store", type=int)
+        parser.add_argument("--verbose", dest="verbose", action="store_true")
+        parser.add_argument("--wet-run", dest="wet_run", action="store_true")
+
+    def call_api_and_store_result(
+        self,
+        SelectedAdministrativeCriteriaModel,
+        AdministrativeCriteriaModel,
+        limit=None,
+        verbose=False,
+        wet_run=False,
+    ):
+        if not verbose:
+            # Don't log HTTP requests detail.
+            logging.getLogger("httpx").setLevel(logging.WARNING)
+            logging.getLogger("httpcore").setLevel(logging.WARNING)
+
+        total_criteria_with_certification = 0
+        found_beneficiaries = set()
+        found_not_beneficiaries = set()
+        not_found_users = set()  # 404
+        server_errors = 0  # 429, 503, 504
+
+        criteria = AdministrativeCriteriaModel.objects.certifiable()
+        period = (make_aware(datetime.datetime(2024, 1, 1)), make_aware(datetime.datetime(2024, 10, 1)))
+        criteria_pks = (
+            SelectedAdministrativeCriteriaModel.objects.filter(
+                administrative_criteria__in=criteria,
+                eligibility_diagnosis__created_at__range=period,
+                eligibility_diagnosis__job_seeker__jobseeker_profile__birth_country__isnull=False,
+                eligibility_diagnosis__job_seeker__jobseeker_profile__birthdate__isnull=False,
+                eligibility_diagnosis__job_seeker__first_name__isnull=False,
+                eligibility_diagnosis__job_seeker__last_name__isnull=False,
+                eligibility_diagnosis__job_seeker__title__isnull=False,
+            )
+            .exclude(Q(certified__isnull=False) | Q(data_returned_by_api__error__contains="not_found"))  # exclude 404
+            .order_by("pk")
+            .values_list("pk", flat=True)
+        )
+        if limit:
+            criteria_pks = criteria_pks[:limit]
+
+        total_criteria = len(criteria_pks)
+        if total_criteria == 0:
+            logger.info("No criteria to certify. Stop now and enjoy your day! ")
+            return
+
+        users_count = User.objects.filter(
+            Exists(
+                SelectedAdministrativeCriteria.objects.filter(
+                    eligibility_diagnosis__job_seeker_id=OuterRef("pk"),
+                    id__in=criteria_pks,
+                )
+            )
+        ).count()
+        logger.info(
+            f"Candidats à certifier pour le modèle {SelectedAdministrativeCriteriaModel.__name__}: {users_count}"
+        )
+
+        chunks_total = ceil(total_criteria / 1000)
+        chunks_count = 0
+        for criteria_pk_subgroup in chunks(criteria_pks, 1000):
+            criteria = SelectedAdministrativeCriteriaModel.objects.filter(pk__in=criteria_pk_subgroup).select_related(
+                "administrative_criteria",
+                "eligibility_diagnosis__job_seeker",
+                "eligibility_diagnosis__job_seeker__jobseeker_profile",
+                "eligibility_diagnosis__job_seeker__jobseeker_profile__birth_place",
+                "eligibility_diagnosis__job_seeker__jobseeker_profile__birth_country",
+            )
+
+            with api_particulier.client() as client:
+                for criterion in criteria:
+                    criterion.certify(client, save=False)
+                    data_returned_by_api = criterion.data_returned_by_api
+                    if data_returned_by_api is None:
+                        continue
+
+                    if data_returned_by_api.get("status"):
+                        total_criteria_with_certification += 1
+                        if criterion.certified:
+                            found_beneficiaries.add(criterion.eligibility_diagnosis.job_seeker.pk)
+                        else:
+                            found_not_beneficiaries.add(criterion.eligibility_diagnosis.job_seeker.pk)
+
+                    if data_returned_by_api.get("error"):
+                        if data_returned_by_api["error"] == "not_found":
+                            not_found_users.add(criterion.eligibility_diagnosis.job_seeker.pk)
+                        else:
+                            server_errors += 1
+
+            if wet_run:
+                SelectedAdministrativeCriteriaModel.objects.bulk_update(
+                    criteria,
+                    fields=[
+                        "data_returned_by_api",
+                        "certified",
+                        "certification_period",
+                        "certified_at",
+                    ],
+                )
+
+            chunks_count += 1
+            logger.info(f"########### {chunks_count/chunks_total*100:.2f}%")
+
+        logger.info(f"Total criteria to be certified: {total_criteria}")
+        logger.info(f"Total criteria with certification: {total_criteria_with_certification}")
+        logger.info(f"Not beneficiaries: {len(found_not_beneficiaries)}")
+        logger.info(f"Beneficiaries: {len(found_beneficiaries)}")
+        logger.info(f"Not found: {len(not_found_users)}")
+        logger.info(f"Server errors: {server_errors}")
+        users_found = total_criteria_with_certification / total_criteria * 100
+        logger.info(f"That's {users_found:.2f}% users found.")
+
+    def handle(self, limit, verbose, wet_run, *args, **kwargs):
+        options = {"wet_run": wet_run, "limit": limit, "verbose": verbose}
+        self.call_api_and_store_result(GEIQSelectedAdministrativeCriteria, GEIQAdministrativeCriteria, **options)
+        self.call_api_and_store_result(SelectedAdministrativeCriteria, AdministrativeCriteria, **options)

itou/utils/apis/api_particulier.py

@@ -11,6 +11,13 @@
 logger = logging.getLogger("APIParticulierClient")
 
 
+class ShouldRetryException(httpx.HTTPStatusError):
+    """
+    This exception can be used to ask Tenacity to retry
+    while attaching a response and a request to it.
+    """
+
+
 def client():
     return httpx.Client(
         base_url=settings.API_PARTICULIER_BASE_URL,
@@ -57,21 +64,40 @@ def _build_params_from(job_seeker):
 @tenacity.retry(
     wait=tenacity.wait_fixed(2),
     stop=tenacity.stop_after_attempt(4),
-    retry=tenacity.retry_if_exception_type(httpx.RequestError),
+    retry=tenacity.retry_if_exception_type(ShouldRetryException),
 )
 def _request(client, endpoint, job_seeker):
     params = _build_params_from(job_seeker=job_seeker)
     response = client.get(endpoint, params=params)
-    if response.status_code == 504:
-        reason = response.json().get("reason")
-        logger.error(f"{response.url=} {reason=}")
-        raise httpx.RequestError(message=reason)
+    error_message = None
+    # Bad Request or Unauthorized
+    # Same as 503 except we don't retry
+    if response.status_code in [400, 401]:
+        error_message = "Bad Request" if response.status_code == 400 else "Unauthorized"
+        logger.error(error_message, extra={"response": response.json()})
+        raise httpx.HTTPStatusError(message=error_message, request=response.request, response=response)
+    # Too Many Requests
+    elif response.status_code == 429:
+        errors = response.json().get("errors")
+        if errors:
+            error_message = errors[0]
+            logger.error(error_message)
+        raise ShouldRetryException(message=error_message, request=response.request, response=response)
+    # Service unavailable
     elif response.status_code == 503:
-        errors = response.json()["errors"]
-        reason = errors[0].get("title")
-        for error in errors:
-            logger.error(f"{response.url=} {error['title']}")
-        raise httpx.RequestError(message=reason)
+        error_message = response.json().get("error")
+        if error_message:
+            error_message = response.json().get("reason")
+        else:
+            errors = response.json().get("errors")
+            error_message = errors[0].get("title")
+        logger.error(error_message)
+        raise ShouldRetryException(message=error_message, request=response.request, response=response)
+    #  Server error
+    elif response.status_code == 504:
+        error_message = response.json().get("reason")
+        logger.error(error_message)
+        raise ShouldRetryException(message=error_message, request=response.request, response=response)
     else:
         response.raise_for_status()
     return response.json()
@@ -82,23 +108,22 @@ def revenu_solidarite_active(client, job_seeker):
         "start_at": None,
         "end_at": None,
         "is_certified": None,
-        "raw_response": "",
+        "raw_response": None,
     }
     try:
-        data = _request(client, "/v2/revenu-solidarite-active", job_seeker)
+        response_data = _request(client, "/v2/revenu-solidarite-active", job_seeker)
     except httpx.HTTPStatusError as exc:  # not 5XX.
-        logger.info(f"Beneficiary not found. {job_seeker.public_id=}")
         data["raw_response"] = exc.response.json()
-    except tenacity.RetryError as retry_err:  # 503 or 504
+    except tenacity.RetryError as retry_err:  # 429, 503 or 504
         exc = retry_err.last_attempt._exception
-        data["raw_response"] = str(exc)
+        data["raw_response"] = exc.response.json()
     except KeyError as exc:
-        data["raw_response"] = str(exc)
+        logger.info(str(exc))  # FIXME: should be removed
     else:
         data = {
-            "start_at": _parse_date(data["dateDebut"]),
-            "end_at": _parse_date(data["dateFin"]),
-            "is_certified": data["status"] == "beneficiaire",
-            "raw_response": data,
+            "start_at": _parse_date(response_data["dateDebut"]),
+            "end_at": _parse_date(response_data["dateFin"]),
+            "is_certified": response_data["status"] == "beneficiaire",
+            "raw_response": response_data,
         }
     return data