Correction du dag CMA

dags/cma.py

@@ -14,6 +14,7 @@
             "https://data.artisanat.fr/api/explore/v2.1/catalog/datasets/reparacteurs/records"
         ),
         "reparacteurs": True,
+        "source_code": "CMA - Chambre des métiers et de l'artisanat",
         "column_mapping": {
             "name": "nom",
             "reparactor_description": "description",

dags/config/db_mapping.json

@@ -68,7 +68,8 @@
       "materiel informatique"
     ],
     "téléphones, smartphones": ["smartphone, tablette et console"],
-    "vêtements, textiles": ["vetement", "linge de maison"]
+    "vêtements, textiles": ["vetement", "linge de maison"],
+    "autres":[]
   },
   "sous_categories_qualirepar": {
     "tv" : "ecran",

dags/utils/dag_eo_utils.py

@@ -275,14 +275,17 @@ def serialize_to_json(**kwargs):
 
 
 def read_acteur(**kwargs):
+    source_code = kwargs["params"].get("source_code", [])
     df_data_from_api = kwargs["ti"].xcom_pull(task_ids="fetch_data_from_api")
     df_sources = kwargs["ti"].xcom_pull(task_ids="read_source")
 
-    df_data_from_api["source_id"] = df_data_from_api["ecoorganisme"].apply(
-        lambda x: mapping_utils.get_id_from_code(x, df_sources)
-    )
-
-    unique_source_ids = df_data_from_api["source_id"].unique()
+    if source_code:
+        unique_source_ids = [mapping_utils.get_id_from_code(source_code, df_sources)]
+    else:
+        df_data_from_api["source_id"] = df_data_from_api["ecoorganisme"].apply(
+            lambda x: mapping_utils.get_id_from_code(x, df_sources)
+        )
+        unique_source_ids = df_data_from_api["source_id"].unique()
 
     pg_hook = PostgresHook(postgres_conn_id="qfdmo-django-db")
     engine = pg_hook.get_sqlalchemy_engine()
@@ -433,6 +436,7 @@ def create_actors(**kwargs):
 
     params = kwargs["params"]
     reparacteurs = params.get("reparacteurs", False)
+    source_code = params.get("source_code")
     label_bonus_reparation = params.get("label_bonus_reparation")
     column_mapping = params.get("column_mapping", {})
     column_to_drop = params.get("column_to_drop", [])
@@ -456,7 +460,9 @@ def create_actors(**kwargs):
     for k, val in column_to_replace.items():
         df[k] = val
     if reparacteurs:
-        df = mapping_utils.process_reparacteurs(df, df_sources, df_acteurtype)
+        df = mapping_utils.process_reparacteurs(
+            df, df_sources, df_acteurtype, source_code
+        )
     else:
         df = mapping_utils.process_actors(df)
 

dags/utils/mapping_utils.py

@@ -116,11 +116,9 @@ def transform_float(x):
         return None
 
 
-def process_reparacteurs(df, df_sources, df_acteurtype):
+def process_reparacteurs(df, df_sources, df_acteurtype, source_code):
     df["produitsdechets_acceptes"] = df.apply(combine_categories, axis=1)
-    df["source_id"] = get_id_from_code(
-        "CMA - Chambre des métiers et de l'artisanat", df_sources
-    )
+    df["source_id"] = get_id_from_code(source_code, df_sources)
     # TODO : on pourrait gérer en configuration les colonne qui sont pareil pour tous
     df["labels_etou_bonus"] = "reparacteur"
     df["type_de_point_de_collecte"] = None