Crée et importe des réutilisations de data.gouv.fr (#4478)

Author: AntoineAugusti

apps/transport/lib/db/reuse.ex

@@ -0,0 +1,125 @@
+defmodule DB.Reuse do
+  @moduledoc """
+  Represents data.gouv.fr reuses.
+  """
+  use TypedEctoSchema
+  use Ecto.Schema
+  import Ecto.Changeset
+  import Ecto.Query
+
+  typed_schema "reuse" do
+    field(:datagouv_id, :string)
+    field(:title, :string)
+    field(:slug, :string)
+    field(:url, :string)
+    field(:type, :string)
+    field(:description, :string)
+    field(:remote_url, :string)
+    field(:organization, :string)
+    field(:organization_id, :string)
+    field(:owner, :string)
+    field(:owner_id, :string)
+    field(:image, :string)
+    field(:featured, :boolean)
+    field(:archived, :boolean)
+    field(:topic, :string)
+    field(:tags, {:array, :string})
+    field(:metric_discussions, :integer)
+    field(:metric_datasets, :integer)
+    field(:metric_followers, :integer)
+    field(:metric_views, :integer)
+    field(:created_at, :utc_datetime_usec)
+    field(:last_modified, :utc_datetime_usec)
+
+    many_to_many(:datasets, DB.Dataset, join_through: "reuse_dataset", on_replace: :delete)
+  end
+
+  def changeset(model, attrs) do
+    model
+    |> cast(attrs, [
+      :title,
+      :slug,
+      :url,
+      :type,
+      :description,
+      :remote_url,
+      :organization,
+      :organization_id,
+      :owner,
+      :owner_id,
+      :image,
+      :topic,
+      :created_at,
+      :last_modified
+    ])
+    |> transform_datagouv_id(attrs)
+    |> transform_metric_keys(attrs)
+    |> transform_archived(attrs)
+    |> transform_featured(attrs)
+    |> transform_tags(attrs)
+    |> validate_required([
+      :datagouv_id,
+      :title,
+      :slug,
+      :url,
+      :type,
+      :description,
+      :remote_url,
+      :featured,
+      :archived,
+      :topic,
+      :tags,
+      :metric_discussions,
+      :metric_datasets,
+      :metric_followers,
+      :metric_views,
+      :created_at,
+      :last_modified
+    ])
+    |> cast_datasets(attrs)
+  end
+
+  defp cast_datasets(%Ecto.Changeset{} = changeset, %{"datasets" => datasets}) do
+    datagouv_ids = (datasets || "") |> String.split(",")
+
+    datasets =
+      DB.Dataset.base_query()
+      |> where([dataset: d], d.datagouv_id in ^datagouv_ids)
+      |> select([dataset: d], [:id])
+      |> DB.Repo.all()
+
+    changeset |> put_assoc(:datasets, datasets)
+  end
+
+  defp transform_archived(%Ecto.Changeset{} = changeset, params) do
+    transform_bool(changeset, :archived, params)
+  end
+
+  defp transform_featured(%Ecto.Changeset{} = changeset, params) do
+    transform_bool(changeset, :featured, params)
+  end
+
+  def transform_bool(%Ecto.Changeset{} = changeset, key, params) do
+    case Map.get(params, to_string(key)) do
+      value when is_binary(value) -> put_change(changeset, key, String.downcase(value) == "true")
+      value -> put_change(changeset, key, value)
+    end
+  end
+
+  defp transform_tags(%Ecto.Changeset{} = changeset, %{"tags" => tags}) do
+    put_change(changeset, :tags, String.split(tags || "", ","))
+  end
+
+  defp transform_datagouv_id(%Ecto.Changeset{} = changeset, %{"id" => id}) do
+    changeset |> put_change(:datagouv_id, id) |> delete_change(:id)
+  end
+
+  defp transform_metric_keys(%Ecto.Changeset{} = changeset, attributes) do
+    attributes
+    |> Enum.filter(fn {k, _v} -> String.starts_with?(k, "metric.") end)
+    |> Enum.map(fn {k, v} ->
+      {k |> String.replace(".", "_") |> String.to_existing_atom(), String.to_integer(v)}
+    end)
+    |> Enum.reduce(changeset, fn {k, v}, changeset -> put_change(changeset, k, v) end)
+  end
+end

apps/transport/lib/jobs/import_reuses_job.ex

@@ -0,0 +1,59 @@
+defmodule Transport.Jobs.ImportReusesJob do
+  @moduledoc """
+  Import reuses from data.gouv.fr when it uses at least a dataset referenced
+  on our platform.
+
+  See https://www.data.gouv.fr/fr/datasets/catalogue-des-donnees-de-data-gouv-fr/
+  """
+  use Oban.Worker, max_attempts: 3
+  import Ecto.Query
+
+  # reuses.csv export
+  @csv_url "https://www.data.gouv.fr/fr/datasets/r/970aafa0-3778-4d8b-b9d1-de937525e379"
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{}) do
+    DB.Repo.transaction(fn ->
+      truncate_reuses()
+      import_all_reuses()
+    end)
+
+    :ok
+  end
+
+  defp truncate_reuses, do: DB.Repo.delete_all(DB.Reuse)
+
+  defp import_all_reuses do
+    datagouv_ids = dataset_datagouv_ids()
+    %{status: 200, body: body} = http_client().get!(@csv_url, decode_body: false)
+
+    [body]
+    |> CSV.decode!(headers: true, separator: ?;, escape_max_lines: 1_000)
+    |> Stream.reject(fn %{"datasets" => datasets} = attributes ->
+      empty_optional_fields?(attributes) or orphan_reuse?(datasets, datagouv_ids)
+    end)
+    |> Enum.each(fn record ->
+      %DB.Reuse{} |> DB.Reuse.changeset(record) |> DB.Repo.insert!()
+    end)
+  end
+
+  defp orphan_reuse?(datasets, datagouv_ids) do
+    datasets |> String.split(",") |> MapSet.new() |> MapSet.disjoint?(datagouv_ids)
+  end
+
+  defp dataset_datagouv_ids do
+    DB.Dataset.base_query()
+    |> select([dataset: d], d.datagouv_id)
+    |> DB.Repo.all()
+    |> MapSet.new()
+  end
+
+  defp empty_optional_fields?(attributes) do
+    attributes
+    |> Map.take(["remote_url", "description", "datasets"])
+    |> Map.values()
+    |> Enum.all?(&(&1 == ""))
+  end
+
+  defp http_client, do: Transport.Req.impl()
+end

apps/transport/priv/repo/migrations/20250305142322_create_reuse.exs

@@ -0,0 +1,30 @@
+defmodule DB.Repo.Migrations.CreateReuse do
+  use Ecto.Migration
+
+  def change do
+    create table(:reuse) do
+      add(:datagouv_id, :string)
+      add(:title, :string)
+      add(:slug, :string)
+      add(:url, :string, size: 500)
+      add(:type, :string)
+      add(:description, :text)
+      add(:remote_url, :string, size: 1_000)
+      add(:organization, :string)
+      add(:organization_id, :string)
+      add(:owner, :string)
+      add(:owner_id, :string)
+      add(:image, :string)
+      add(:featured, :boolean, default: false)
+      add(:archived, :boolean, default: false)
+      add(:topic, :string)
+      add(:tags, {:array, :string})
+      add(:metric_discussions, :integer)
+      add(:metric_datasets, :integer)
+      add(:metric_followers, :integer)
+      add(:metric_views, :integer)
+      add(:created_at, :utc_datetime_usec)
+      add(:last_modified, :utc_datetime_usec)
+    end
+  end
+end

apps/transport/priv/repo/migrations/20250305143006_create_reuse_dataset.exs

@@ -0,0 +1,10 @@
+defmodule DB.Repo.Migrations.CreateReuseDataset do
+  use Ecto.Migration
+
+  def change do
+    create table(:reuse_dataset, primary_key: false) do
+      add(:reuse_id, references(:reuse, on_delete: :delete_all))
+      add(:dataset_id, references(:dataset, on_delete: :delete_all))
+    end
+  end
+end

apps/transport/test/db/reuse_test.exs

@@ -0,0 +1,33 @@
+defmodule DB.ReuseTest do
+  use ExUnit.Case, async: true
+  import DB.Factory
+
+  setup do
+    Ecto.Adapters.SQL.Sandbox.checkout(DB.Repo)
+  end
+
+  test "changeset" do
+    %DB.Dataset{id: dataset_id} = insert(:dataset, datagouv_id: datagouv_id = "53699569a3a729239d2046eb")
+
+    # Payload from https://tabular-api.data.gouv.fr/api/resources/970aafa0-3778-4d8b-b9d1-de937525e379/data/?page=1&page_size=50&topic__exact=transport_and_mobility
+    data =
+      ~s|{"id": "67c02dfe7172569a69c367e6","title": "Carte nationale des plateaux techniques spécialisés (PTS) pour « évaluer l’aptitude médicale à la conduite » ","slug": "carte-nationale-des-plateaux-techniques-specialises-pts-pour-evaluer-laptitude-medicale-a-la-conduite","url": "http://www.data.gouv.fr/fr/reuses/carte-nationale-des-plateaux-techniques-specialises-pts-pour-evaluer-laptitude-medicale-a-la-conduite/","type": "visualization","description": "Ceci est une description","remote_url": "https://www.securite-routiere.gouv.fr/permis-et-situation-de-handicap/carte-des-plateaux-techniques-de-sante","organization": null,"organization_id": null,"owner": "ilyes-zeroual","owner_id": "67c0216ab1f98413870cc70c","image": "https://static.data.gouv.fr/images/69/f0053e284741c9b6d2a73cc490edb2-500.png","featured": "False","created_at": "2025-02-27T09:18:54.658000","last_modified": "2025-02-27T09:49:33.676000","archived": "False","topic": "transport_and_mobility","tags": "foo,bar","datasets": "54730e00c751df4f2ec2acbe,#{datagouv_id}","metric.discussions": "0","metric.datasets": "2","metric.followers": "0","metric.views": "234"}|
+
+    assert %Ecto.Changeset{
+             valid?: true,
+             changes: %{
+               datagouv_id: "67c02dfe7172569a69c367e6",
+               tags: ["foo", "bar"],
+               metric_views: 234,
+               archived: false,
+               featured: false,
+               created_at: ~U[2025-02-27 09:18:54.658000Z]
+             }
+           } = changeset = DB.Reuse.changeset(%DB.Reuse{}, Jason.decode!(data))
+
+    DB.Repo.insert!(changeset)
+
+    [reuse] = DB.Repo.all(DB.Reuse)
+    assert [%DB.Dataset{id: ^dataset_id}] = reuse |> DB.Repo.preload(:datasets) |> Map.fetch!(:datasets)
+  end
+end

apps/transport/test/transport/jobs/import_reuses_job_test.exs

@@ -0,0 +1,59 @@
+defmodule Transport.Test.Transport.Jobs.ImportReusesJobTest do
+  use ExUnit.Case, async: true
+  import DB.Factory
+  import Mox
+  use Oban.Testing, repo: DB.Repo
+
+  setup :verify_on_exit!
+
+  setup do
+    Ecto.Adapters.SQL.Sandbox.checkout(DB.Repo)
+  end
+
+  @csv_url "https://www.data.gouv.fr/fr/datasets/r/970aafa0-3778-4d8b-b9d1-de937525e379"
+  @dataset_datagouv_id Ecto.UUID.generate()
+
+  test "perform" do
+    %DB.Dataset{id: dataset_id} = insert(:dataset, datagouv_id: @dataset_datagouv_id)
+
+    # An existing reuse will be deleted when importing all reuses
+    DB.Reuse.changeset(%DB.Reuse{}, sample_reuse(Ecto.UUID.generate()))
+    |> DB.Repo.insert!()
+
+    assert 1 == DB.Repo.all(DB.Reuse) |> Enum.count()
+
+    datagouv_id_1 = Ecto.UUID.generate()
+    datagouv_id_2 = Ecto.UUID.generate()
+
+    setup_csv_response([datagouv_id_1, datagouv_id_2])
+
+    assert :ok == perform_job(Transport.Jobs.ImportReusesJob, %{})
+
+    # We now have 2 reuses and they are associated with a dataset
+    # The orphan reuse (not referencing an existing dataset) has been deleted.
+    reuses = DB.Repo.all(DB.Reuse)
+    assert MapSet.new([datagouv_id_1, datagouv_id_2]) == reuses |> Enum.map(& &1.datagouv_id) |> MapSet.new()
+    assert %DB.Reuse{datasets: [%DB.Dataset{id: ^dataset_id}]} = reuses |> hd() |> DB.Repo.preload(:datasets)
+  end
+
+  defp sample_reuse(datagouv_id, datasets \\ [@dataset_datagouv_id]) do
+    ~s|{"title": "Carte nationale des plateaux techniques spécialisés (PTS) pour « évaluer l’aptitude médicale à la conduite » ","slug": "carte-nationale-des-plateaux-techniques-specialises-pts-pour-evaluer-laptitude-medicale-a-la-conduite","url": "http://www.data.gouv.fr/fr/reuses/carte-nationale-des-plateaux-techniques-specialises-pts-pour-evaluer-laptitude-medicale-a-la-conduite/","type": "visualization","description": "Ceci est une description","remote_url": "https://www.securite-routiere.gouv.fr/permis-et-situation-de-handicap/carte-des-plateaux-techniques-de-sante","organization": null,"organization_id": null,"owner": "ilyes-zeroual","owner_id": "67c0216ab1f98413870cc70c","image": "https://static.data.gouv.fr/images/69/f0053e284741c9b6d2a73cc490edb2-500.png","featured": "False","created_at": "2025-02-27T09:18:54.658000","last_modified": "2025-02-27T09:49:33.676000","archived": "False","topic": "transport_and_mobility","tags": "foo,bar","metric.discussions": "0","metric.datasets": "2","metric.followers": "0","metric.views": "234"}|
+    |> Jason.decode!()
+    |> Map.put("id", datagouv_id)
+    |> Map.put("datasets", Enum.join(datasets, ","))
+  end
+
+  defp setup_csv_response(datagouv_ids) do
+    url = @csv_url
+    orphan_reuse = sample_reuse(Ecto.UUID.generate(), [Ecto.UUID.generate()])
+
+    body =
+      (Enum.map(datagouv_ids, &sample_reuse/1) ++ [orphan_reuse])
+      |> CSV.encode(headers: true, separator: ?;)
+      |> Enum.join("")
+
+    expect(Transport.Req.Mock, :get!, fn ^url, [decode_body: false] ->
+      %Req.Response{status: 200, body: body}
+    end)
+  end
+end

config/runtime.exs

@@ -161,6 +161,7 @@ oban_prod_crontab = [
   {"20 5 * * *", Transport.Jobs.ImportDatasetContactPointsJob},
   # Should be ideally executed after `GBFSMultiValidationDispatcherJob` to use fresh metadata
   {"30 8 * * *", Transport.Jobs.ImportGBFSFeedContactEmailJob},
+  {"20 5 * * *", Transport.Jobs.ImportReusesJob},
   {"30 5 * * *", Transport.Jobs.ImportDatasetMonthlyMetricsJob},
   {"45 5 * * *", Transport.Jobs.ImportResourceMonthlyMetricsJob},
   {"0 8 * * *", Transport.Jobs.WarnUserInactivityJob},