IRVE : amélioration du rapport, utilisation de messages explicites plutôt que des exceptions pour les erreurs connues (#5544)

Author: vdegove

apps/transport/lib/irve/consolidation.ex

@@ -84,6 +84,16 @@ defmodule Transport.IRVE.Consolidation do
     process_resource(resource)
   rescue
     error ->
+      Logger.error(
+        "IRVE consolidation: unexpected error for resource #{resource.resource_id} (#{resource.url})\n" <>
+          Exception.format(:error, error, __STACKTRACE__)
+      )
+
+      Sentry.capture_exception(error,
+        stacktrace: __STACKTRACE__,
+        extra: %{resource_id: resource.resource_id, url: resource.url}
+      )
+
       {:error_occurred, error, resource}
   end
 
@@ -105,28 +115,18 @@ defmodule Transport.IRVE.Consolidation do
         |> Map.put(:estimated_pdc_count, estimated_pdc_count)
         |> Map.put(:file_extension, extension)
 
-      # The code is convoluted mostly because we didn't go far enough on the validator work.
-      # The validator will ultimately stop raising exceptions, and will instead return structures.
-      # But currently if a cheap check fails, an exception is thrown, and we would lose the estimated PDC count,
-      # something which is essential to report on for our current work.
-      try do
-        # Raise if the producer is not an organization. This check is not in the validator itself:
-        # 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.
-        ensure_producer_is_org!(resource)
-
-        validation_result = Transport.IRVE.Validator.validate(path, extension)
-        file_valid? = validation_result |> Transport.IRVE.Validator.full_file_valid?()
-
-        if file_valid? do
-          {Transport.IRVE.DatabaseImporter.try_write_to_db(path, resource), resource}
-        else
-          {:not_compliant_with_schema, resource}
-        end
-      rescue
-        error ->
-          {:error_occurred, error, resource}
+      # This producer_is_org_check is not in the validator itself:
+      # 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.
+      with :producer_is_an_organization <- producer_is_org(resource),
+           %{valid: true} <- Transport.IRVE.Validator.validate_and_summarize(path, extension),
+           import_status <- Transport.IRVE.DatabaseImporter.try_write_to_db(path, resource) do
+        {import_status, resource}
+      else
+        :producer_not_an_organization -> {:producer_not_an_organization, resource}
+        %{file_level_errors: [_ | _] = errors} -> {:file_level_errors, resource, errors}
+        %{file_level_errors: []} -> {:not_compliant_with_schema, resource}
       end
     end)
   end
@@ -208,31 +208,38 @@ defmodule Transport.IRVE.Consolidation do
   # regular workflow: process the file then delete it afterwards, no matter what, to ensure
   # the files do not stack up on the production disk.
   def with_maybe_cached_download_on_disk(resource, file_path, extension, false = _use_permanent_disk_cache, work_fn) do
-    download!(resource.resource_id, resource.url, file_path)
-    # NOTE: we need to pass the original extension (provided in the URL) because some heuristics use it afterwards.
-    # but the caching mechanism stores everything under the same `.dat` extension (so the file path is not enough
-    # to keep the extension around)
-    work_fn.(file_path, extension)
+    case download(resource.resource_id, resource.url, file_path) do
+      # NOTE: we need to pass the original extension (provided in the URL) because some heuristics use it afterwards.
+      # but the caching mechanism stores everything under the same `.dat` extension (so the file path is not enough
+      # to keep the extension around)
+      :ok -> work_fn.(file_path, extension)
+      {:error, message} -> {:download_failed, resource, message}
+    end
   after
     File.rm!(file_path)
   end
 
   # variant for dev work, where it is important to support permanent disk caching (fully offline, no etag)
   def with_maybe_cached_download_on_disk(resource, file_path, extension, true = _use_permanent_disk_cache, work_fn) do
-    if !File.exists?(file_path), do: download!(resource.resource_id, resource.url, file_path)
-    work_fn.(file_path, extension)
+    download_result =
+      if File.exists?(file_path), do: :ok, else: download(resource.resource_id, resource.url, file_path)
+
+    case download_result do
+      :ok -> work_fn.(file_path, extension)
+      {:error, message} -> {:download_failed, resource, message}
+    end
   end
 
-  def download!(resource_id, url, file) do
+  # Returns `:ok` on success, or `{:error, message}` for a non-200 response.
+  # Timeouts / transport errors still raise for now (caught upstream as `:error_occurred`).
+  def download(resource_id, url, file) do
     Logger.info("Processing resource #{resource_id} (url=#{url})")
     %{status: status} = Transport.HTTPClient.get!(url, compressed: false, decode_body: false, into: File.stream!(file))
 
-    unless status == 200 do
-      raise "Error processing resource (#{resource_id}) (http_status=#{status})"
-    end
+    if status == 200, do: :ok, else: {:error, "http_status=#{status}"}
   end
 
-  defp ensure_producer_is_org!(%{dataset_organisation_id: "???"}), do: raise("producer is not an organization")
-
-  defp ensure_producer_is_org!(_row), do: :ok
+  defp producer_is_org(%{dataset_organisation_id: org_id}) do
+    if org_id == "???", do: :producer_not_an_organization, else: :producer_is_an_organization
+  end
 end

apps/transport/lib/irve/report_item.ex

@@ -20,11 +20,19 @@ defmodule Transport.IRVE.ReportItem do
   ]
 
   def from_result({:error_occurred, error, resource}) do
-    new(resource, :error_occurred, error)
+    new(resource, :error_occurred, Exception.message(error), inspect(error.__struct__))
+  end
+
+  def from_result({:download_failed, resource, message}) do
+    new(resource, :download_failed, message, nil)
+  end
+
+  def from_result({:file_level_errors, resource, file_level_errors}) do
+    new(resource, :file_level_errors, Enum.join(file_level_errors, "\n"), nil)
   end
 
   def from_result({status, resource}) do
-    new(resource, status, nil)
+    new(resource, status, nil, nil)
   end
 
   def to_map(%__MODULE__{} = report_row) do
@@ -33,7 +41,7 @@ defmodule Transport.IRVE.ReportItem do
     |> Map.update!(:status, &to_string/1)
   end
 
-  defp new(resource, status, error) do
+  defp new(resource, status, error_message, error_type) do
     %__MODULE__{
       dataset_id: resource.dataset_id,
       resource_id: resource.resource_id,
@@ -43,15 +51,9 @@ defmodule Transport.IRVE.ReportItem do
       datagouv_last_modified: resource.datagouv_last_modified,
       status: status,
       estimated_pdc_count: resource[:estimated_pdc_count],
-      error_message: maybe_error_message(error),
-      error_type: maybe_error_type(error),
+      error_message: error_message,
+      error_type: error_type,
       file_extension: resource[:file_extension]
     }
   end
-
-  defp maybe_error_message(nil), do: nil
-  defp maybe_error_message(error), do: Exception.message(error)
-
-  defp maybe_error_type(nil), do: nil
-  defp maybe_error_type(error), do: error.__struct__ |> inspect()
 end

apps/transport/lib/irve/static_probes.ex

@@ -3,28 +3,34 @@ defmodule Transport.IRVE.Static.Probes do
   This module groups functions related to IRVE-specific CSV handling.
   """
 
-  def run_cheap_blocking_checks(body, extension) do
-    if Transport.ZipProbe.likely_zip_content?(body) do
-      raise("the content is likely to be a zip file, not uncompressed CSV data")
-    end
+  def file_level_errors(body, extension) do
+    cond do
+      Transport.ZipProbe.likely_zip_content?(body) ->
+        ["the content is likely to be a zip file, not uncompressed CSV data"]
 
-    if String.downcase(extension) not in ["", ".csv"] do
-      raise("the content is likely not a CSV file (extension is #{extension})")
-    end
+      String.downcase(extension) not in ["", ".csv"] ->
+        ["the content is likely not a CSV file (extension is #{extension})"]
 
-    if probably_v1_schema(body) do
-      raise("looks like a v1 irve")
-    end
+      probably_v1_schema(body) ->
+        ["looks like a v1 irve"]
 
-    if !has_id_pdc_itinerance(body) do
-      raise("content has no id_pdc_itinerance in first line")
-    end
+      !has_id_pdc_itinerance(body) ->
+        ["content has no id_pdc_itinerance in first line"]
 
-    header_separator = hint_header_separator(body)
+      true ->
+        separator_errors(body)
+    end
+  end
 
-    unless header_separator in [";", ","] do
-      raise("unsupported column separator #{header_separator}")
+  # `hint_header_separator/1` raises when it cannot find a separator at all
+  # (e.g. single-column file); we catch that so this stays non-raising.
+  defp separator_errors(body) do
+    case hint_header_separator(body) do
+      sep when sep in [";", ","] -> []
+      sep -> ["unsupported column separator #{sep}"]
     end
+  rescue
+    error -> [Exception.message(error)]
   end
 
   @doc """

apps/transport/lib/irve/validator/validator.ex

@@ -3,6 +3,8 @@ defmodule Transport.IRVE.Validator do
   Central entry point for IRVE file validation (currently working on `DataFrame`).
   """
 
+  @unexpected_error_prefix "Unexpected error: "
+
   def compute_validation(%Explorer.DataFrame{} = df) do
     schema = Transport.IRVE.StaticIRVESchema.schema_content()
 
@@ -12,27 +14,6 @@ defmodule Transport.IRVE.Validator do
     |> Transport.IRVE.Validator.DataFrameValidation.setup_computed_warning_columns()
   end
 
-  @doc """
-  Validate an IRVE file located at `path`, returning a DataFrame with validation results.
-  This wrapper includes some pre-processing steps before actual validation.
-  These preprocessing steps do not output any warning and are silent,
-  so files that are not strictly valid may be considered as valid without any notice by this function.
-  If you want to call a strict validator (no preprocessing), use `compute_validation/1` instead.
-  """
-  def validate(path, extension \\ ".csv") do
-    # NOTE: for now, load the body in memory, because refactoring to get full streaming
-    # is too involved for the current sprint deadline.
-    body = File.read!(path)
-    Transport.IRVE.Static.Probes.run_cheap_blocking_checks(body, extension)
-    # TODO: accumulate warnings
-    body = Transport.IRVE.Transcoder.ensure_utf8(body)
-    # TODO: accumulate warnings
-
-    body
-    |> Transport.IRVE.Processing.read_as_uncasted_data_frame()
-    |> compute_validation()
-  end
-
   @doc """
   Says from the dataframe output of compute_validation/1 if all rows are valid.
   """
@@ -72,33 +53,58 @@ defmodule Transport.IRVE.Validator do
   end
 
   @doc """
-  Combines `validate/2` and `summarize/1` into a single call, catching any file-level error
-  (bad encoding, wrong format, missing columns, etc.) instead of raising.
+  The main entry point for IRVE validation: turns a file into a `%Summary{}`.
+  Used by on demand validation as well as the consolidation pipeline.
+
+  If you want to call a strict validator (no preprocessing, no file-level error handling),
+  or want to have access to line by line details, use `compute_validation/1` instead,
+  as this function doesn’t output the whole validation report, only a summary.
 
-  Returns the same map as `summarize/1` with an additional `file_level_error` key:
-  - `nil` when validation ran successfully (the file could still be invalid row-by-row)
-  - an error message string when a hard file-level error was caught
+  Known file-level problems (zip, wrong format, v1 schema, …) are detected up-front by
+  `Transport.IRVE.Static.Probes.file_level_errors/2` and reported in `file_level_errors`.
+  In this case, `valid_row_count`, `invalid_row_count`, `column_errors`, and
+  `error_samples` are all `nil`/empty as the file was not processed at all.
 
-  On a file-level error, `valid_row_count`, `invalid_row_count`, `column_errors`, and
-  `error_samples` are all `nil`/empty since there is no DataFrame to summarize from.
+  Unexpected errors anywhere in the pipeline are also caught and reported in `file_level_errors`
+  and prefixed with `#{@unexpected_error_prefix}` to tell them apart.
   """
   def validate_and_summarize(path, extension \\ ".csv") do
-    path
-    |> validate(extension)
-    |> summarize()
+    # Known problem: whole file is loaded in memory, no streaming.
+    # We could change that by having probes use only the first lines,
+    # stream the utf8 conversion,
+    # and finally cast to a dataframe in a streaming way (Explorer.DataFrame.from_csv/2 supports streaming).
+    body = File.read!(path)
+
+    # TODO: wrong delimiter is silently fixed, should send a file-level warning instead
+    case Transport.IRVE.Static.Probes.file_level_errors(body, extension) do
+      [] ->
+        body
+        # TODO: send a file level warning instead of silently fixing
+        |> Transport.IRVE.Transcoder.ensure_utf8()
+        |> Transport.IRVE.Processing.read_as_uncasted_data_frame()
+        |> compute_validation()
+        |> summarize()
+
+      file_level_errors ->
+        summary_with_file_level_errors(file_level_errors)
+    end
   rescue
     error ->
-      %Transport.IRVE.Validator.Summary{
-        valid: false,
-        valid_row_count: nil,
-        invalid_row_count: nil,
-        total_row_count: nil,
-        file_level_errors: [Exception.message(error)],
-        column_errors: %{},
-        error_samples: [],
-        warnings: %{},
-        warning_samples: []
-      }
+      summary_with_file_level_errors([@unexpected_error_prefix <> Exception.message(error)])
+  end
+
+  defp summary_with_file_level_errors(file_level_errors) do
+    %Transport.IRVE.Validator.Summary{
+      valid: false,
+      valid_row_count: nil,
+      invalid_row_count: nil,
+      total_row_count: nil,
+      file_level_errors: file_level_errors,
+      column_errors: %{},
+      error_samples: [],
+      warnings: %{},
+      warning_samples: []
+    }
   end
 
   @max_samples_per_group 5

apps/transport/test/transport/irve/consolidation_test.exs

@@ -33,30 +33,32 @@ defmodule Transport.IRVE.ConsolidationTest do
           %{
             "dataset_id" => "another-dataset-id",
             "resource_id" => "another-resource-id",
-            "status" => "error_occurred",
-            "error_type" => "ArgumentError",
+            "status" => "file_level_errors",
+            "error_type" => nil,
             "estimated_pdc_count" => "1",
             "file_extension" => ".csv",
             "url" => "https://static.data.gouv.fr/resources/another-irve-url-2024/data.csv",
             "dataset_title" => "another-dataset-title",
             "datagouv_organization_or_owner" => "another-org",
             "datagouv_last_modified" => "2024-02-29T07:43:59.660000+00:00",
+            # An unexpected error (here a missing required column) is caught by the validator,
+            # marked with the "Unexpected error: " prefix and surfaced in error_message.
             # TODO rework to only compare the part of the message that matters
             "error_message" =>
-              ~s|could not find column name "nom_station". The available columns are: ["accessibilite_pmr", "telephone_operateur", "coordonneesXY", "observations", "date_maj", "num_pdl", "code_insee_commune", "nom_enseigne", "puissance_nominale", "adresse_station", "id_station_itinerance", "siren_amenageur", "contact_operateur", "implantation_station", "date_mise_en_service", "horaires", "id_pdc_itinerance", "nbre_pdc", "raccordement", "id_station_local", "nom_amenageur", "restriction_gabarit", "nom_operateur", "contact_amenageur", "id_pdc_local", "tarification", "condition_acces", "prise_type_ef", "prise_type_2", "prise_type_combo_ccs", "prise_type_chademo", "prise_type_autre", "gratuit", "paiement_acte", "paiement_cb", "paiement_autre", "reservation", "station_deux_roues", "cable_t2_attache", "check_column_nom_amenageur_valid", "check_column_siren_amenageur_valid", "check_column_contact_amenageur_valid", "check_column_nom_operateur_valid", "check_column_contact_operateur_valid", "check_column_telephone_operateur_valid", "check_column_nom_enseigne_valid", "check_column_id_station_itinerance_valid", "check_column_id_station_local_valid"].\nIf you are attempting to interpolate a value, use ^nom_station.|
+              ~s|Unexpected error: could not find column name "nom_station". The available columns are: ["accessibilite_pmr", "telephone_operateur", "coordonneesXY", "observations", "date_maj", "num_pdl", "code_insee_commune", "nom_enseigne", "puissance_nominale", "adresse_station", "id_station_itinerance", "siren_amenageur", "contact_operateur", "implantation_station", "date_mise_en_service", "horaires", "id_pdc_itinerance", "nbre_pdc", "raccordement", "id_station_local", "nom_amenageur", "restriction_gabarit", "nom_operateur", "contact_amenageur", "id_pdc_local", "tarification", "condition_acces", "prise_type_ef", "prise_type_2", "prise_type_combo_ccs", "prise_type_chademo", "prise_type_autre", "gratuit", "paiement_acte", "paiement_cb", "paiement_autre", "reservation", "station_deux_roues", "cable_t2_attache", "check_column_nom_amenageur_valid", "check_column_siren_amenageur_valid", "check_column_contact_amenageur_valid", "check_column_nom_operateur_valid", "check_column_contact_operateur_valid", "check_column_telephone_operateur_valid", "check_column_nom_enseigne_valid", "check_column_id_station_itinerance_valid", "check_column_id_station_local_valid"].\nIf you are attempting to interpolate a value, use ^nom_station.|
           },
           %{
             "dataset_id" => "individual-published-dataset-id",
             "resource_id" => "individual-published-resource-id",
-            "status" => "error_occurred",
-            "error_type" => "RuntimeError",
+            "status" => "producer_not_an_organization",
+            "error_type" => nil,
             "estimated_pdc_count" => "1",
             "file_extension" => ".csv",
             "url" => "https://static.data.gouv.fr/resources/individual-published-irve-url-2024/data.csv",
             "dataset_title" => "individual-published-dataset-title",
             "datagouv_organization_or_owner" => "Guy Who loves IRVE",
             "datagouv_last_modified" => "2024-02-29T07:43:59.660000+00:00",
-            "error_message" => "producer is not an organization"
+            "error_message" => nil
           },
           %{
             "dataset_id" => "the-dataset-id",
@@ -100,7 +102,7 @@ defmodule Transport.IRVE.ConsolidationTest do
         start_path: "consolidation_transport_irve_statique_rapport_#{date}",
         bucket: bucket_name,
         acl: :private,
-        file_content: "f06fd15d5afcd8be10880b049dc45424c6c9475b8ee2071c5ab1b9880638f3d9"
+        file_content: sha256_of(report_content)
       )
 
       Transport.Test.S3TestUtils.s3_mocks_remote_copy_file(