Exporter les Acteurs selon la licence à appliquer à la source

qfdmo/admin/acteur.py

@@ -1,4 +1,4 @@
-from typing import Any
+from typing import Any, List
 
 import orjson
 from django import forms
@@ -566,20 +566,25 @@ class OpenSourceDisplayedActeurResource(resources.ModelResource):
 
     limit = 0
     offset = 0
+    licenses = []
 
-    def __init__(self, limit=0, offset=0, **kwargs):
+    def __init__(
+        self, limit: int = 0, offset: int = 0, licenses: List[str] = [], **kwargs
+    ):
         self.limit = limit
         self.offset = offset
+        self.licenses = licenses
         super().__init__(**kwargs)
 
     uuid = fields.Field(column_name="Identifiant", attribute="uuid", readonly=True)
-    sources = fields.Field(
-        column_name="Contributeurs", attribute="sources", readonly=True
-    )
+    sources = fields.Field(column_name="Paternité", attribute="sources", readonly=True)
 
     def dehydrate_sources(self, acteur):
         sources = ["Longue Vie Aux Objets", "ADEME"]
-        sources.extend([f"{source.libelle}" for source in acteur.sources.all()])
+        acteur_sources = acteur.sources.all()
+        if self.licenses:
+            acteur_sources = acteur_sources.filter(licence__in=self.licenses)
+        sources.extend([f"{source.libelle}" for source in acteur_sources])
         seen = set()
         deduplicated_sources = []
         for source in sources:
@@ -682,10 +687,22 @@ def dehydrate_propositions_services(self, acteur):
     )
 
     def get_queryset(self):
+
         queryset = super().get_queryset()
+
+        queryset = queryset.prefetch_related(
+            "sources",
+            "labels",
+            "proposition_services__sous_categories",
+            "proposition_services__action",
+        )
+
+        # Only Actif
         queryset = queryset.filter(
             statut=ActeurStatus.ACTIF,
-        ).exclude(
+        )
+        # Exclude acteurs only professionals
+        queryset = queryset.exclude(
             public_accueilli__in=[
                 ActeurPublicAccueilli.AUCUN,
                 ActeurPublicAccueilli.PROFESSIONNELS,
@@ -695,16 +712,15 @@ def get_queryset(self):
         queryset = queryset.exclude(
             identifiant_unique__icontains="_reparation_",
         )
-
-        queryset = queryset.prefetch_related(
-            "sources",
-            "labels",
-            "proposition_services__sous_categories",
-            "proposition_services__action",
-        )
+        # Export only acteurs with expected licenses
+        if self.licenses:
+            queryset = queryset.filter(sources__licence__in=self.licenses)
+        queryset = queryset.distinct()
         queryset = queryset.order_by("uuid")
+
         if self.limit:
             return queryset[self.offset : self.offset + self.limit]
+
         return queryset
 
     class Meta:

qfdmo/management/commands/export_displayedacteur.py

@@ -8,17 +8,51 @@
 from django.core.management.base import BaseCommand
 
 from qfdmo.admin import OpenSourceDisplayedActeurResource
+from qfdmo.models.acteur import DataLicense
 
 CHUNK = 1000
 
 
 class Command(BaseCommand):
     help = "Export Ressources using CSV format"
 
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--file",
+            type=str,
+            help="File to export to",
+            default=(f"export_acteur_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"),
+        )
+        parser.add_argument(
+            "--licenses",
+            nargs="+",
+            action="extend",
+            type=str,
+            help=(
+                f"Licenses to export, options : {DataLicense.values}, "
+                f"default: '{DataLicense.OPEN_LICENSE.value}'"
+            ),
+        )
+
     def handle(self, *args, **options):
-        self.stdout.write(f"Exporting Ressources, starting at {datetime.now()}")
-        target_file = datetime.now().strftime(
-            "exports/export_acteur_%Y%m%d_%H%M%S.xlsx"
+        self.stdout.write(
+            self.style.SUCCESS(f"Exporting Ressources, starting at {datetime.now()}")
+        )
+        target_file = "exports/" + options["file"]
+        licenses = options["licenses"]
+        if not licenses:
+            licenses = [DataLicense.OPEN_LICENSE.value]
+        if not all(license in DataLicense.values for license in licenses):
+            self.stdout.write(
+                self.style.ERROR(
+                    f"Invalid licenses, options : {licenses}, "
+                    f"Available values: '{DataLicense.values}'"
+                )
+            )
+            return
+
+        self.stdout.write(
+            self.style.SUCCESS(f"Exporting DisplayedActeur using licenses: {licenses}")
         )
 
         with tempfile.NamedTemporaryFile(mode="w+b", suffix=".xlsx") as tmp_file:
@@ -31,26 +65,27 @@ def handle(self, *args, **options):
 
             offset = 0
             dataset = OpenSourceDisplayedActeurResource(
-                limit=CHUNK, offset=offset
+                limit=CHUNK, offset=offset, licenses=licenses
             ).export()
             sheet.append(dataset.headers)
 
             while dataset.dict:
-                self.stdout.write(f"Exporting {offset} to {offset + CHUNK}")
+                self.stdout.write(
+                    self.style.SUCCESS(f"Exporting {offset} to {offset + CHUNK}")
+                )
                 dataset.headers = None
 
                 for row in dataset.dict:
                     sheet.append(row)
 
                 offset += CHUNK
                 dataset = OpenSourceDisplayedActeurResource(
-                    limit=CHUNK, offset=offset
+                    limit=CHUNK, offset=offset, licenses=licenses
                 ).export()
-
-            self.stdout.write(f"Writing to {target_file}")
+            self.stdout.write(self.style.SUCCESS(f"Writing to {target_file}"))
 
             workbook.save(tmp_file.name)
             tmp_file.seek(0)
             default_storage.save(target_file, ContentFile(tmp_file.read()))
 
-        self.stdout.write(f"Ended at {datetime.now()}")
+        self.stdout.write(self.style.SUCCESS(f"Ended at {datetime.now()}"))

unit_tests/qfdmo/test_open_source_displayedacteur_ressource.py

@@ -1,6 +1,7 @@
 import pytest
 
 from qfdmo.admin.acteur import OpenSourceDisplayedActeurResource
+from qfdmo.models.acteur import DataLicense
 from unit_tests.qfdmo.acteur_factory import (
     DisplayedActeurFactory,
     DisplayedPropositionServiceFactory,
@@ -22,7 +23,7 @@ def test_export_columns(self):
         for row in dataset.dict:
             assert list(row.keys()) == [
                 "Identifiant",
-                "Contributeurs",
+                "Paternité",
                 "Nom",
                 "Nom commercial",
                 "SIREN",
@@ -99,30 +100,30 @@ def test_propositions_de_services(self):
         )
 
     @pytest.mark.parametrize(
-        "source_data, expected_contributeurs",
+        "source_data, expected_other_contributeurs",
         [
             (
                 [],
-                "Longue Vie Aux Objets|ADEME",
+                [],
             ),
             (
                 [
                     {"libelle": "Source 1", "code": "source1"},
                     {"libelle": "Source 2", "code": "source2"},
                 ],
-                "Longue Vie Aux Objets|ADEME|Source 1|Source 2",
+                ["Source 1", "Source 2"],
             ),
             (
                 [
                     {"libelle": "Source 1", "code": "source1"},
                     {"libelle": "Source 2", "code": "source2"},
                     {"libelle": "Source 1", "code": "source3"},
                 ],
-                "Longue Vie Aux Objets|ADEME|Source 1|Source 2",
+                ["Source 1", "Source 2"],
             ),
         ],
     )
-    def test_sources(self, source_data, expected_contributeurs):
+    def test_sources(self, source_data, expected_other_contributeurs):
         displayedacteur = DisplayedActeurFactory()
         sources = [SourceFactory(**data) for data in source_data]
         displayedacteur.sources.set(sources)
@@ -131,8 +132,57 @@ def test_sources(self, source_data, expected_contributeurs):
 
         dataset_dict = dataset.dict
 
-        contributeurs = dataset_dict[0]["Contributeurs"].split("|")
+        contributeurs = dataset_dict[0]["Paternité"].split("|")
         assert contributeurs[0] == "Longue Vie Aux Objets"
         assert contributeurs[1] == "ADEME"
-        other_contributeurs = set(contributeurs[2:])
-        assert sorted(contributeurs[2:]) == sorted(other_contributeurs)
+        assert sorted(contributeurs[2:]) == sorted(expected_other_contributeurs)
+
+    def test_filter_by_licenses(self):
+        source_open_licence = SourceFactory(licence=DataLicense.OPEN_LICENSE.value)
+        source_cc_by_nc_sa = SourceFactory(licence=DataLicense.CC_BY_NC_SA.value)
+        acteur_open_license = DisplayedActeurFactory()
+        acteur_open_license.sources.set([source_open_licence])
+        acteur_multi_licences = DisplayedActeurFactory()
+        acteur_multi_licences.sources.set([source_open_licence, source_cc_by_nc_sa])
+        acteur_cc_license = DisplayedActeurFactory.create()
+        acteur_cc_license.sources.set([source_cc_by_nc_sa])
+
+        dataset = OpenSourceDisplayedActeurResource(
+            licenses=[DataLicense.OPEN_LICENSE.value, DataLicense.CC_BY_NC_SA.value]
+        ).export()
+
+        identifiants = [row["Identifiant"] for row in dataset.dict]
+        assert len(dataset) == 3
+        assert acteur_open_license.uuid in identifiants
+        assert acteur_multi_licences.uuid in identifiants
+        assert acteur_cc_license.uuid in identifiants
+
+        dataset = OpenSourceDisplayedActeurResource(
+            licenses=[DataLicense.OPEN_LICENSE.value]
+        ).export()
+
+        identifiants = [row["Identifiant"] for row in dataset.dict]
+        assert len(dataset) == 2
+        assert acteur_open_license.uuid in identifiants
+        assert acteur_multi_licences.uuid in identifiants
+
+        # test acteur_multi doesn't refer license CC_BY_NC_SA
+        row_multi = next(
+            row
+            for row in dataset.dict
+            if row["Identifiant"] == acteur_multi_licences.uuid
+        )
+        assert row_multi["Identifiant"] == acteur_multi_licences.uuid
+        assert (
+            row_multi["Paternité"]
+            == f"Longue Vie Aux Objets|ADEME|{source_open_licence.libelle}"
+        )
+
+        dataset = OpenSourceDisplayedActeurResource(
+            licenses=[DataLicense.CC_BY_NC_SA.value]
+        ).export()
+
+        identifiants = [row["Identifiant"] for row in dataset.dict]
+        assert len(dataset) == 2
+        assert acteur_multi_licences.uuid in identifiants
+        assert acteur_cc_license.uuid in identifiants