Wire resource_status in the consolidation

Author: vdegove

apps/transport/lib/irve/consolidation.ex

@@ -1,7 +1,7 @@
 defmodule Transport.IRVE.Consolidation do
   @moduledoc """
   This module:
-  - takes the list of relevant IRVE resources from data.gouv.fr,
+  - takes the list of relevant IRVE datagouv_resources from data.gouv.fr,
   - downloads each resource file,
   - validates it,
   - and if valid writes the PDCs and file to the database.
@@ -29,8 +29,13 @@ defmodule Transport.IRVE.Consolidation do
     destination = Keyword.get(opts, :destination, :send_to_s3)
     debug = Keyword.get(opts, :debug, false)
 
-    report_rows =
-      resource_list()
+    datagouv_resources = resource_list()
+    datagouv_resource_ids = MapSet.new(datagouv_resources, & &1.resource_id)
+    # Snapshot before processing mutates the DB, so `resource_status` reflects the pre-run state.
+    db_resource_ids = DB.IRVEValidFile.existing_datagouv_resource_ids()
+
+    processed_rows =
+      datagouv_resources
       |> maybe_limit(opts[:limit])
       |> Task.async_stream(
         &process_or_rescue/1,
@@ -45,9 +50,12 @@ defmodule Transport.IRVE.Consolidation do
       # This is intentional, we want to be aware of such timeouts.
       |> Stream.map(fn {:ok, result} -> result end)
       |> Stream.map(&Transport.IRVE.ReportItem.from_result/1)
+      |> Stream.map(&Transport.IRVE.ReportItem.put_resource_status(&1, db_resource_ids))
       |> maybe_log_items(debug)
       |> Enum.into([])
 
+    report_rows = processed_rows ++ orphan_report_rows(db_resource_ids, datagouv_resource_ids)
+
     report = generate_report(report_rows, destination: destination)
 
     consolidated_df =
@@ -85,6 +93,17 @@ defmodule Transport.IRVE.Consolidation do
     |> Enum.sort_by(fn r -> [r.dataset_id, r.resource_id] end)
   end
 
+  # Resources still in the DB but no longer listed on data.gouv.fr, appended as a block after the
+  # processed rows (which are not processed this run, so they carry no consolidation outcome).
+  def orphan_report_rows(db_resource_ids, datagouv_resource_ids) do
+    db_resource_ids
+    |> MapSet.difference(datagouv_resource_ids)
+    |> MapSet.to_list()
+    |> DB.IRVEValidFile.orphan_files()
+    |> Enum.sort_by(&{&1.datagouv_dataset_id, &1.datagouv_resource_id})
+    |> Enum.map(&Transport.IRVE.ReportItem.from_orphan_file/1)
+  end
+
   # safety wrapper that we can use inside `Task.async_stream`
   def process_or_rescue(resource) do
     process_resource(resource)
@@ -149,6 +168,7 @@ defmodule Transport.IRVE.Consolidation do
       |> Explorer.DataFrame.select([
         "dataset_id",
         "resource_id",
+        "resource_status",
         "consolidation_status",
         "error_type",
         "estimated_pdc_count",

apps/transport/lib/irve/report_item.ex

@@ -12,6 +12,7 @@ defmodule Transport.IRVE.ReportItem do
     :dataset_title,
     :datagouv_organization_or_owner,
     :datagouv_last_modified,
+    :resource_status,
     :consolidation_status,
     :error_message,
     :error_type,
@@ -35,12 +36,55 @@ defmodule Transport.IRVE.ReportItem do
     new(resource, status, nil, nil)
   end
 
+  # Presence relationship between data.gouv.fr and our database, from a pre-run snapshot of the
+  # resource ids in the database. Kept separate from the consolidation outcome.
+  def put_resource_status(%__MODULE__{resource_id: resource_id} = report_row, db_resource_ids) do
+    resource_status =
+      if MapSet.member?(db_resource_ids, resource_id),
+        do: :on_datagouv_and_in_db,
+        else: :on_datagouv_not_in_db
+
+    %{report_row | resource_status: resource_status}
+  end
+
+  # A resource still in the database but no longer listed on data.gouv.fr: no download/processing
+  # happened this run, so `consolidation_status` and the download-related fields stay nil.
+  def from_orphan_file(%{
+        datagouv_dataset_id: dataset_id,
+        datagouv_resource_id: resource_id,
+        dataset_title: dataset_title,
+        datagouv_organization_or_owner: datagouv_organization_or_owner,
+        datagouv_last_modified: datagouv_last_modified,
+        pdc_count: pdc_count
+      }) do
+    %__MODULE__{
+      dataset_id: dataset_id,
+      resource_id: resource_id,
+      url: nil,
+      dataset_title: dataset_title,
+      datagouv_organization_or_owner: datagouv_organization_or_owner,
+      datagouv_last_modified: datagouv_last_modified,
+      resource_status: :in_db_deleted_from_datagouv,
+      consolidation_status: nil,
+      estimated_pdc_count: pdc_count,
+      error_message: nil,
+      error_type: nil,
+      file_extension: nil
+    }
+  end
+
   def to_map(%__MODULE__{} = report_row) do
     report_row
     |> Map.from_struct()
-    |> Map.update!(:consolidation_status, &to_string/1)
+    |> Map.update!(:consolidation_status, &stringify/1)
+    |> Map.update!(:resource_status, &stringify/1)
+    |> Map.update!(:datagouv_last_modified, &stringify/1)
   end
 
+  # `nil` (empty CSV cell) for orphan rows; keeps mixed atom/DateTime/string columns uniform.
+  defp stringify(nil), do: nil
+  defp stringify(value), do: to_string(value)
+
   defp new(resource, status, error_message, error_type) do
     %__MODULE__{
       dataset_id: resource.dataset_id,

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

@@ -33,6 +33,7 @@ defmodule Transport.IRVE.ConsolidationTest do
           %{
             "dataset_id" => "another-dataset-id",
             "resource_id" => "another-resource-id",
+            "resource_status" => "on_datagouv_not_in_db",
             "consolidation_status" => "file_level_errors",
             "error_type" => nil,
             "estimated_pdc_count" => "1",
@@ -50,6 +51,7 @@ defmodule Transport.IRVE.ConsolidationTest do
           %{
             "dataset_id" => "individual-published-dataset-id",
             "resource_id" => "individual-published-resource-id",
+            "resource_status" => "on_datagouv_not_in_db",
             "consolidation_status" => "producer_not_an_organization",
             "error_type" => nil,
             "estimated_pdc_count" => "1",
@@ -63,6 +65,7 @@ defmodule Transport.IRVE.ConsolidationTest do
           %{
             "dataset_id" => "the-dataset-id",
             "resource_id" => "the-resource-id",
+            "resource_status" => "on_datagouv_not_in_db",
             "consolidation_status" => "import_successful",
             "error_type" => nil,
             "estimated_pdc_count" => "1",
@@ -79,6 +82,7 @@ defmodule Transport.IRVE.ConsolidationTest do
         |> Explorer.DataFrame.select([
           "dataset_id",
           "resource_id",
+          "resource_status",
           "consolidation_status",
           "error_type",
           "estimated_pdc_count",
@@ -222,6 +226,78 @@ defmodule Transport.IRVE.ConsolidationTest do
       refute File.exists?(System.tmp_dir!() |> Path.join("irve-resource-the-resource-id.dat"))
       refute File.exists?(System.tmp_dir!() |> Path.join("irve-resource-another-resource-id.dat"))
     end
+
+    test "reports the presence delta between data.gouv.fr and the database" do
+      mock_datagouv_resources()
+      mock_resource_downloads()
+
+      on_exit(fn ->
+        ~w(
+          consolidation_transport_irve_statique_rapport.csv
+          consolidation_transport_avec_doublons_irve_statique.csv
+          consolidation_transport_irve_statique.csv
+        )
+        |> Enum.each(&File.rm/1)
+      end)
+
+      # Still in the DB but no longer listed on data.gouv.fr, with 2 stored PDCs.
+      seed_file_with_pdcs("deleted-dataset-id", "deleted-resource-id", 2)
+
+      # Already in the DB (older version) and still listed on data.gouv.fr.
+      DB.Repo.insert!(%DB.IRVEValidFile{
+        datagouv_dataset_id: "the-dataset-id",
+        datagouv_resource_id: "the-resource-id",
+        checksum: "old-checksum"
+      })
+
+      {:ok, report_df} = Transport.IRVE.Consolidation.process(destination: :local_disk)
+
+      rows_by_id =
+        report_df
+        |> Explorer.DataFrame.to_rows()
+        |> Map.new(fn row -> {row["resource_id"], row} end)
+
+      # Had a version in the DB, imported a new one this run.
+      assert rows_by_id["the-resource-id"]["resource_status"] == "on_datagouv_and_in_db"
+      assert rows_by_id["the-resource-id"]["consolidation_status"] == "import_successful"
+
+      # Brand-new resources (nothing in the DB before this run).
+      assert rows_by_id["another-resource-id"]["resource_status"] == "on_datagouv_not_in_db"
+      assert rows_by_id["individual-published-resource-id"]["resource_status"] == "on_datagouv_not_in_db"
+
+      # Invariant: cannot be up-to-date without being in the DB.
+      refute Enum.any?(rows_by_id, fn {_id, row} ->
+               row["resource_status"] == "on_datagouv_not_in_db" and
+                 row["consolidation_status"] == "already_up_to_date"
+             end)
+
+      orphan = rows_by_id["deleted-resource-id"]
+      assert orphan["resource_status"] == "in_db_deleted_from_datagouv"
+      assert is_nil(orphan["consolidation_status"])
+      assert orphan["estimated_pdc_count"] == 2
+      assert orphan["dataset_id"] == "deleted-dataset-id"
+    end
+  end
+
+  defp seed_file_with_pdcs(dataset_id, resource_id, pdc_count) do
+    rows =
+      for i <- 1..pdc_count do
+        DB.Factory.IRVE.generate_row(%{"id_pdc_itinerance" => "FRPAN99E0000000#{i}"})
+      end
+
+    path = System.tmp_dir!() |> Path.join("irve_seed_#{Ecto.UUID.generate()}.csv")
+    File.write!(path, DB.Factory.IRVE.to_csv_body(rows))
+
+    Transport.IRVE.DatabaseImporter.write_to_db(
+      path,
+      dataset_id,
+      resource_id,
+      "#{dataset_id}-title",
+      "#{dataset_id}-org",
+      "2024-01-01T10:00:00+00:00"
+    )
+
+    File.rm(path)
   end
 
   defp mock_datagouv_resources do