Check closed with API Entreprises

erp/management/commands/api_entreprises_clean_closed.py

@@ -0,0 +1,77 @@
+from datetime import timedelta
+
+from django.core.management.base import BaseCommand
+from django.db.models import Q
+from django.utils import timezone
+
+from erp.models import Activite, Erp
+from erp.provider.entreprise import check_closed
+
+IGNORED_ACTIVITIES = ["Administration publique", "Mairie", "Gendarmerie", "Bureau de poste"]
+
+
+class Command(BaseCommand):
+    help = "Check for closed ERPs from API Entreprises"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--start_pk",
+            type=int,
+            required=False,
+            default=0,
+            help="Resume the check since this given ERP PK in our DB.",
+        )
+        parser.add_argument(
+            "--write",
+            default=False,
+            action="store_true",
+            help="Actually edit the database",
+        )
+
+        parser.add_argument(
+            "--nb_days",
+            type=int,
+            required=False,
+            default=60,
+            help="Check the ERPs which have not been checked in the last nb_days.",
+        )
+
+    def _flag_erp_as_closed(self, existing_erp):
+        print(f"Flag permanently closed ERP: {existing_erp} - {existing_erp.get_absolute_uri()}")
+        if not self.write:
+            print("Dry run mode, no DB action, use --write to apply this deletion")
+            return
+
+        existing_erp.permanently_closed = True
+        existing_erp.save()
+
+    def handle(self, *args, **options):
+        self.write = options["write"]
+        self.start_pk = options.get("start_pk")
+
+        ignored_activities = Activite.objects.filter(nom__in=IGNORED_ACTIVITIES)
+        if ignored_activities.count() != len(IGNORED_ACTIVITIES):
+            print("Please check the IGNORED_ACTIVITIES list, at least one activity has not been found. Exit...")
+            return
+
+        limit_date = timezone.now() - timedelta(days=options["nb_days"])
+        qs = Erp.objects.published().filter(Q(check_closed_at=None) | Q(check_closed_at__lte=limit_date))
+        qs = qs.exclude(activite__in=ignored_activities)
+        if self.start_pk:
+            qs = qs.filter(pk__gte=self.start_pk)
+        qs = qs.order_by("pk")
+
+        for erp in qs.iterator():
+            print(f"Checking ERP with PK {erp.pk}")
+            query = f"{erp.numero} {erp.voie}" if erp.numero else erp.lieu_dit
+            query = f"{erp.nom}, {query} {erp.code_postal} {erp.commune}"
+
+            query_address = f"{erp.numero} {erp.voie}" if erp.numero else erp.lieu_dit
+            query_address = f"{query_address} {erp.code_postal} {erp.commune}"
+            if check_closed(erp.nom, query_address, erp.commune_ext.code_insee):
+                self._flag_erp_as_closed(erp)
+                continue
+
+            if self.write:
+                erp.check_closed_at = timezone.now()
+                erp.save(update_fields=("check_closed_at",))

erp/provider/entreprise.py

@@ -121,3 +121,38 @@ def search(terms, code_insee, activities):
         return []
     except requests.exceptions.RequestException as err:
         raise RuntimeError(f"entreprise api error: {err}")
+
+
+def check_closed(name, address, code_insee):
+    term = f"{name}, {address}"
+    payload = {
+        "per_page": MAX_PER_PAGE,
+        "page": 1,
+        "q": term,
+        "code_insee": code_insee,
+        "categorie_entreprise": "PME,ETI",
+    }
+    res = requests.get(f"{BASE_URL_ENTERPRISE_API}", payload, timeout=5)
+    try:
+        if not (len(results := (res.json().get("results") or [])) == 1):
+            return False
+    except (requests.exceptions.JSONDecodeError, requests.exceptions.ReadTimeout):
+        return False
+
+    closed = results[0].get("siege", {}).get("date_fermeture") is not None
+    if not closed:
+        return False
+
+    # Do the same request without the erp name parameter nor the code_insee to obtain other ERPs at the same address.
+    # If we have more than one result, we can't consider the ERP as closed.
+    payload["q"] = address
+    del payload["code_insee"]
+
+    res = requests.get(f"{BASE_URL_ENTERPRISE_API}", payload, timeout=5)
+    try:
+        if len(res.json().get("results") or []) > 1:
+            return False
+    except (requests.exceptions.JSONDecodeError, requests.exceptions.ReadTimeout):
+        return False
+
+    return True