Skip to content

Commit 49633fe

Browse files
vdegovethbar
vdegove
and
thbar
authored
Consolidation IRVE : élimination de tous les doublons restants (#5365)
Co-authored-by: Thibaut Barrère <thibaut.barrere@gmail.com>
1 parent a05821b commit 49633fe

3 files changed

Lines changed: 149 additions & 7 deletions

File tree

‎apps/transport/lib/irve/database_exporter.ex‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,11 @@ defmodule Transport.IRVE.DatabaseExporter do
7171
]
7272
end
7373

74+
@doc """
75+
Returns the list and order of fields to be exported in the dataframe (and CSV file).
76+
This function is also used for grouping fully identical entries in the deduplication module
77+
as Explorer’s group_by needs a list of columns (or a function).
78+
"""
7479
def export_field_list do
7580
Transport.IRVE.StaticIRVESchema.field_names_list()
7681
|> Enum.concat(["consolidated_longitude", "consolidated_latitude"])

‎apps/transport/lib/irve/deduplicator.ex‎

Lines changed: 62 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -41,27 +41,34 @@ defmodule Transport.IRVE.Deduplicator do
4141
4242
Values of the additional column:
4343
- unique
44+
- removed_because_non_concerne
4445
- kept_because_in_prioritary_dataset
4546
- removed_because_not_in_prioritary_dataset
4647
- kept_because_date_maj_more_recent
4748
- removed_because_date_maj_not_more_recent
4849
- kept_because_resource_more_recent
4950
- removed_because_resource_not_more_recent
51+
- kept_because_exact_duplicate_in_same_file
52+
- removed_because_exact_duplicate_in_same_file
53+
- removed_because_no_rule_applies
5054
"""
5155
def add_duplicates_column(%Explorer.DataFrame{} = df) do
52-
# TODO at one point: deal with non_concerné and such.
53-
Explorer.DataFrame.group_by(df, "id_pdc_itinerance")
56+
df
57+
|> Explorer.DataFrame.group_by("id_pdc_itinerance")
5458
|> remove_non_concerne_rule()
5559
|> unique_rule()
5660
|> in_prioritary_datasets_rule()
5761
|> date_maj_rule()
5862
|> datagouv_last_modified_rule()
63+
|> exact_duplicate_in_same_file_rule()
64+
|> remove_undecided_duplicates_rule()
5965
end
6066

6167
def discard_duplicates(df) do
6268
df
6369
|> Explorer.DataFrame.filter(
6470
deduplication_status in [
71+
"kept_because_exact_duplicate_in_same_file",
6572
"unique",
6673
"kept_because_date_maj_more_recent",
6774
"kept_because_resource_more_recent",
@@ -73,7 +80,10 @@ defmodule Transport.IRVE.Deduplicator do
7380
defp remove_non_concerne_rule(df) do
7481
df
7582
|> Explorer.DataFrame.mutate(
76-
deduplication_status: if(id_pdc_itinerance == "Non concerné", do: "removed_because_non_concerne")
83+
deduplication_status:
84+
if(id_pdc_itinerance == "Non concerné",
85+
do: "removed_because_non_concerne"
86+
)
7787
)
7888
end
7989

@@ -148,21 +158,68 @@ defmodule Transport.IRVE.Deduplicator do
148158
# (that have is_max_date_maj true)
149159
df
150160
|> Explorer.DataFrame.mutate(max_datagouv_last_modified: max(datagouv_last_modified))
161+
|> Explorer.DataFrame.mutate(is_max_datagouv_last_modified: datagouv_last_modified == max_datagouv_last_modified)
162+
|> Explorer.DataFrame.group_by(["id_pdc_itinerance", "is_max_date_maj", "is_max_datagouv_last_modified"])
163+
|> Explorer.DataFrame.mutate(count_max_datagouv_last_modified: count(is_max_datagouv_last_modified))
151164
|> Explorer.DataFrame.mutate(
152165
deduplication_status:
153166
cond do
154167
is_not_nil(deduplication_status) ->
155168
deduplication_status
156169

157-
datagouv_last_modified == max_datagouv_last_modified ->
170+
datagouv_last_modified == max_datagouv_last_modified and count_max_datagouv_last_modified == 1 ->
158171
"kept_because_resource_more_recent"
159172

160-
true ->
173+
not is_max_datagouv_last_modified ->
161174
"removed_because_resource_not_more_recent"
175+
176+
true ->
177+
nil
162178
end
163179
)
164180
|> Explorer.DataFrame.discard("max_datagouv_last_modified")
181+
|> Explorer.DataFrame.discard("count_max_datagouv_last_modified")
165182
|> Explorer.DataFrame.ungroup()
166183
|> Explorer.DataFrame.discard("is_max_date_maj")
184+
|> Explorer.DataFrame.discard("is_max_datagouv_last_modified")
185+
end
186+
187+
defp exact_duplicate_in_same_file_rule(df) do
188+
df
189+
# This is grouping by all columns, thus grouping identical entries (with same file ids) together
190+
|> Explorer.DataFrame.group_by(Transport.IRVE.DatabaseExporter.export_field_list())
191+
|> Explorer.DataFrame.mutate(count_dups: count(id_pdc_itinerance))
192+
|> Explorer.DataFrame.mutate(min_row_index: min(row_index(id_pdc_itinerance)))
193+
|> Explorer.DataFrame.mutate(
194+
deduplication_status:
195+
cond do
196+
is_not_nil(deduplication_status) ->
197+
deduplication_status
198+
199+
count_dups > 1 and row_index(id_pdc_itinerance) == min_row_index ->
200+
"kept_because_exact_duplicate_in_same_file"
201+
202+
count_dups > 1 ->
203+
"removed_because_exact_duplicate_in_same_file"
204+
205+
true ->
206+
nil
207+
end
208+
)
209+
|> Explorer.DataFrame.discard("count_dups")
210+
|> Explorer.DataFrame.discard("min_row_index")
211+
|> Explorer.DataFrame.ungroup()
212+
end
213+
214+
defp remove_undecided_duplicates_rule(df) do
215+
df
216+
|> Explorer.DataFrame.mutate(
217+
deduplication_status:
218+
if(
219+
is_nil(deduplication_status),
220+
do: "removed_because_no_rule_applies",
221+
else: deduplication_status
222+
)
223+
)
167224
end
168225
end

‎apps/transport/test/transport/irve/deduplicator_test.exs‎

Lines changed: 82 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,30 @@ defmodule Transport.IRVE.DeduplicatorTest do
7777
"date_maj" => ~D[2025-10-01],
7878
"deduplication_status" => "removed_because_non_concerne",
7979
"id_pdc_itinerance" => "Non concerné"
80+
},
81+
%{
82+
"datagouv_resource_id" => "with-exact-duplicates-resource",
83+
"date_maj" => ~D[2025-10-01],
84+
"deduplication_status" => "kept_because_exact_duplicate_in_same_file",
85+
"id_pdc_itinerance" => "FRS31DUPLICATE5"
86+
},
87+
%{
88+
"datagouv_resource_id" => "with-exact-duplicates-resource",
89+
"date_maj" => ~D[2025-10-01],
90+
"deduplication_status" => "removed_because_exact_duplicate_in_same_file",
91+
"id_pdc_itinerance" => "FRS31DUPLICATE5"
92+
},
93+
%{
94+
"datagouv_resource_id" => "similar-resource-1",
95+
"date_maj" => ~D[2025-10-01],
96+
"deduplication_status" => "removed_because_no_rule_applies",
97+
"id_pdc_itinerance" => "FRS31DUPLICATE5"
98+
},
99+
%{
100+
"datagouv_resource_id" => "similar-resource-2",
101+
"date_maj" => ~D[2025-10-01],
102+
"deduplication_status" => "removed_because_no_rule_applies",
103+
"id_pdc_itinerance" => "FRS31DUPLICATE5"
80104
}
81105
]
82106
end
@@ -110,6 +134,10 @@ defmodule Transport.IRVE.DeduplicatorTest do
110134
%{
111135
"datagouv_resource_id" => "gireve-resource",
112136
"id_pdc_itinerance" => "FRS31DUPLICATE3"
137+
},
138+
%{
139+
"datagouv_resource_id" => "with-exact-duplicates-resource",
140+
"id_pdc_itinerance" => "FRS31DUPLICATE5"
113141
}
114142
]
115143
end
@@ -247,15 +275,67 @@ defmodule Transport.IRVE.DeduplicatorTest do
247275
]
248276
|> Enum.map(&Map.merge(&1, non_itinerance_resource))
249277

278+
resource_with_exact_duplicates = %{
279+
"datagouv_dataset_id" => "with-exact-duplicates-dataset",
280+
"datagouv_resource_id" => "with-exact-duplicates-resource",
281+
"datagouv_last_modified" => DateTime.new!(~D[2025-11-01], ~T[12:00:00.000], "Etc/UTC")
282+
}
283+
284+
resource_with_exact_duplicates_content =
285+
[
286+
%{
287+
"id_pdc_itinerance" => "FRS31DUPLICATE5",
288+
"date_maj" => ~D[2025-10-01]
289+
},
290+
%{
291+
"id_pdc_itinerance" => "FRS31DUPLICATE5",
292+
"date_maj" => ~D[2025-10-01]
293+
}
294+
]
295+
|> Enum.map(&Map.merge(&1, resource_with_exact_duplicates))
296+
297+
similar_resource_published_at_same_timestamp_1 = %{
298+
"datagouv_dataset_id" => "similar-resource-dataset-1",
299+
"datagouv_resource_id" => "similar-resource-1",
300+
"datagouv_last_modified" => DateTime.new!(~D[2025-11-01], ~T[12:00:00.000], "Etc/UTC")
301+
}
302+
303+
similar_resource_published_at_same_timestamp_2 = %{
304+
"datagouv_dataset_id" => "similar-resource-dataset-2",
305+
"datagouv_resource_id" => "similar-resource-2",
306+
"datagouv_last_modified" => DateTime.new!(~D[2025-11-01], ~T[12:00:00.000], "Etc/UTC")
307+
}
308+
309+
similar_resource_content =
310+
[
311+
%{
312+
"id_pdc_itinerance" => "FRS31DUPLICATE5",
313+
"date_maj" => ~D[2025-10-01]
314+
}
315+
]
316+
250317
data =
251318
resource_2026_02_15_content ++
252319
resource_2026_02_17_content ++
253320
resource_2026_02_01_content ++
254321
reource_2026_02_18_content ++
255322
gireve_resource_content ++
256323
qualicharge_resource_content ++
257-
non_itinerance_resource_content
324+
non_itinerance_resource_content ++
325+
resource_with_exact_duplicates_content ++
326+
Enum.map(similar_resource_content, &Map.merge(&1, similar_resource_published_at_same_timestamp_1)) ++
327+
Enum.map(similar_resource_content, &Map.merge(&1, similar_resource_published_at_same_timestamp_2))
328+
329+
data
330+
|> add_dummy_missing_columns()
331+
|> Explorer.DataFrame.new()
332+
end
333+
334+
defp add_dummy_missing_columns(data) do
335+
dummy_columns =
336+
Transport.IRVE.DatabaseExporter.export_field_list()
337+
|> Map.new(&{&1, nil})
258338

259-
Explorer.DataFrame.new(data)
339+
Enum.map(data, &Map.merge(dummy_columns, &1))
260340
end
261341
end

0 commit comments

Comments
 (0)