Suppression de la consolidation "brute" IRVE et renommage de la consolidation transport (#5529)

Author: vdegove

…ansport/lib/irve/simple_consolidation.ex apps/transport/lib/irve/consolidation.exapps/transport/lib/irve/simple_consolidation.ex renamed to apps/transport/lib/irve/consolidation.ex apps/transport/lib/irve/simple_consolidation.ex renamed to apps/transport/lib/irve/consolidation.ex

@@ -1,4 +1,4 @@
-defmodule Transport.IRVE.SimpleConsolidation do
+defmodule Transport.IRVE.Consolidation do
   @moduledoc """
   This module:
   - takes the list of relevant IRVE resources from data.gouv.fr,
@@ -38,7 +38,7 @@ defmodule Transport.IRVE.SimpleConsolidation do
       # If a task times out, we get {:exit, :timeout} instead of {:ok, result} and the following line will crash.
       # This is intentional, we want to be aware of such timeouts.
       |> Stream.map(fn {:ok, result} -> result end)
-      |> Stream.map(&Transport.IRVE.SimpleReportItem.from_result/1)
+      |> Stream.map(&Transport.IRVE.ReportItem.from_result/1)
       |> maybe_log_items(debug)
       |> Enum.into([])
 
@@ -54,7 +54,7 @@ defmodule Transport.IRVE.SimpleConsolidation do
 
     write_consolidated_file(consolidated_df, @consolidated_file_base_name, destination)
 
-    Logger.info("IRVE simple consolidation process completed.")
+    Logger.info("IRVE consolidation process completed.")
     {:ok, report}
   end
 
@@ -75,7 +75,7 @@ defmodule Transport.IRVE.SimpleConsolidation do
 
   def resource_list do
     Transport.IRVE.Extractor.datagouv_resources()
-    |> Transport.IRVE.RawStaticConsolidation.exclude_irrelevant_resources()
+    |> Transport.IRVE.Extractor.exclude_irrelevant_resources()
     |> Enum.sort_by(fn r -> [r.dataset_id, r.resource_id] end)
   end
 
@@ -114,7 +114,7 @@ defmodule Transport.IRVE.SimpleConsolidation do
         # it’s not linked to the file content/format, but to how it is published on data.gouv.fr.
         # it is done after downloading the file in order to be able to report on the potential
         # loss of PDC count.
-        Transport.IRVE.RawStaticConsolidation.ensure_producer_is_org!(resource)
+        ensure_producer_is_org!(resource)
 
         validation_result = Transport.IRVE.Validator.validate(path, extension)
         file_valid? = validation_result |> Transport.IRVE.Validator.full_file_valid?()
@@ -134,7 +134,7 @@ defmodule Transport.IRVE.SimpleConsolidation do
   def generate_report(report_rows, destination: destination) do
     report_df =
       report_rows
-      |> Enum.map(&Transport.IRVE.SimpleReportItem.to_map/1)
+      |> Enum.map(&Transport.IRVE.ReportItem.to_map/1)
       |> Explorer.DataFrame.new()
       # `select` orders columns in the provided order
       # (https://github.com/elixir-explorer/explorer/issues/1126)
@@ -231,4 +231,8 @@ defmodule Transport.IRVE.SimpleConsolidation do
       raise "Error processing resource (#{resource_id}) (http_status=#{status})"
     end
   end
+
+  defp ensure_producer_is_org!(%{dataset_organisation_id: "???"}), do: raise("producer is not an organization")
+
+  defp ensure_producer_is_org!(_row), do: :ok
 end

apps/transport/lib/irve/database_importer.ex

@@ -56,7 +56,7 @@ defmodule Transport.IRVE.DatabaseImporter do
       ) do
     content =
       File.read!(file_path)
-      |> Transport.IRVE.RawStaticConsolidation.ensure_utf8()
+      |> Transport.IRVE.Transcoder.ensure_utf8()
 
     rows_stream =
       content |> Transport.IRVE.Processing.read_as_data_frame() |> Explorer.DataFrame.to_rows_stream()

apps/transport/lib/irve/extractor.ex

@@ -11,6 +11,11 @@ defmodule Transport.IRVE.Extractor do
 
   @static_irve_datagouv_url "https://www.data.gouv.fr/api/1/datasets/?schema=etalab/schema-irve-statique"
 
+  # needed to filter out the existing, data-gouv provided consolidation
+  @datagouv_organization_id Application.compile_env!(:transport, :datagouvfr_publisher_id)
+  # Filter also our own consolidation!
+  @transport_organization_id Application.compile_env!(:transport, :datagouvfr_transport_publisher_id)
+
   @doc """
   Fetches the list of all `schema-irve-statique` resources from data gouv, using parallelized pagination.
 
@@ -29,6 +34,13 @@ defmodule Transport.IRVE.Extractor do
     |> Enum.into([])
   end
 
+  def exclude_irrelevant_resources(stream) do
+    stream
+    # exclude data gouv generated consolidation
+    |> Enum.reject(fn r -> r.dataset_organisation_id == @datagouv_organization_id end)
+    |> Enum.reject(fn r -> r.dataset_organisation_id == @transport_organization_id end)
+  end
+
   @doc """
   Retrieve one page of resource results from data gouv.
   """

apps/transport/lib/irve/raw_report_item.ex

@@ -1,4 +1,4 @@
-defmodule Transport.IRVE.SimpleReportItem do
+defmodule Transport.IRVE.ReportItem do
   @moduledoc """
   Quick & dirty structure to massage the processing outcome of a single
   IRVE file into a structure with all the same keys.