diff --git a/lib/microwaveprop/radio/csv_import.ex b/lib/microwaveprop/radio/csv_import.ex index ff5c80d9..65333223 100644 --- a/lib/microwaveprop/radio/csv_import.ex +++ b/lib/microwaveprop/radio/csv_import.ex @@ -1,23 +1,112 @@ defmodule Microwaveprop.Radio.CsvImport do - @moduledoc false + @moduledoc """ + Parse a CSV of contacts into a preview (validated + de-duplicated) and then + commit those rows into the database. + + Upload flow: + + {:ok, preview} = CsvImport.preview(csv_string, submitter_email) + # show preview.valid / preview.invalid / preview.duplicates to the user + {:ok, result} = CsvImport.commit(preview.valid) + + Two rows count as duplicates of each other if they share: + + * the same pair of callsigns (direction-agnostic), + * the same grids for those callsigns, + * the same band, and + * timestamps within 60 minutes of each other. + + Dedup runs both against existing DB contacts and against earlier rows in + the same CSV (first row wins, subsequent matches are flagged). + """ + + import Ecto.Query, only: [from: 2] alias Microwaveprop.Radio + alias Microwaveprop.Radio.Contact + alias Microwaveprop.Repo alias Microwaveprop.Workers.ContactWeatherEnqueueWorker + @dedup_window_seconds 3600 + @expected_columns 7 @columns ~w(station1 station2 grid1 grid2 band mode qso_timestamp) @doc """ - Imports contacts from a CSV string. The first line is treated as a header. - Each valid row is inserted via `Radio.create_contact/1` and enrichment is enqueued. - The submitter_email is added to each row's attributes. - - Returns: - - `{:ok, %{imported: [%Contact{}, ...], errors: [{row_num, [error_string]}]}}` - - `{:error, :empty_csv}` if the input is empty or whitespace-only - - `{:error, :no_data_rows}` if there are no data rows after the header + Parse-and-insert in one shot. Kept for backward compatibility — does NOT + perform duplicate detection. New code should use `preview/2` + `commit/1`. """ def import(csv_string, submitter_email) do + with {:ok, rows} <- parse_csv(csv_string, submitter_email) do + {valid, invalid} = split_parsed(rows) + {:ok, commit_result} = commit(valid) + invalid_errors = Enum.map(invalid, fn %{row_num: n, messages: m} -> {n, m} end) + + {:ok, + %{ + imported: commit_result.imported, + errors: invalid_errors ++ commit_result.errors + }} + end + end + + @doc """ + Parses the CSV, validates each row with the submission changeset, and + classifies every row as valid / invalid / duplicate without inserting. + + Returns: + + * `{:ok, preview}` where `preview` is a map with keys + `:valid`, `:invalid`, `:duplicates`, `:total_rows`, `:submitter_email`. + * `{:error, :empty_csv}` if there is nothing to process. + * `{:error, :no_data_rows}` if only a header is present. + + Valid rows are maps of the form `%{row_num, attrs, timestamp}`. Duplicate + rows add a `:reason` field indicating whether the conflict was found in + the database (`:existing_contact`) or earlier in the uploaded file + (`:earlier_in_upload`). + """ + def preview(csv_string, submitter_email) do + with {:ok, rows} <- parse_csv(csv_string, submitter_email) do + {valid, invalid} = split_parsed(rows) + existing = load_existing_for_dedup(valid) + {unique, duplicates} = dedupe(valid, existing) + + {:ok, + %{ + valid: unique, + invalid: invalid, + duplicates: duplicates, + total_rows: length(rows), + submitter_email: submitter_email + }} + end + end + + @doc """ + Inserts the rows returned by `preview/2` in `:valid`. Each is passed to + `Radio.create_contact/1` and its enrichment is enqueued on success. + Returns `{:ok, %{imported: [...], errors: [{row_num, messages}, ...]}}`. + """ + def commit(valid_rows) when is_list(valid_rows) do + result = + Enum.reduce(valid_rows, %{imported: [], errors: []}, fn row, acc -> + case Radio.create_contact(row.attrs) do + {:ok, contact} -> + ContactWeatherEnqueueWorker.enqueue_for_contact(contact) + %{acc | imported: [contact | acc.imported]} + + {:error, %Ecto.Changeset{} = changeset} -> + %{acc | errors: [{row.row_num, changeset_error_strings(changeset)} | acc.errors]} + end + end) + + {:ok, %{imported: Enum.reverse(result.imported), errors: Enum.reverse(result.errors)}} + end + + # -- parsing --------------------------------------------------------------- + + defp parse_csv(csv_string, submitter_email) do numbered_lines = csv_string |> String.replace("\r\n", "\n") @@ -34,28 +123,14 @@ defmodule Microwaveprop.Radio.CsvImport do {:error, :no_data_rows} [{_header, _} | data_lines] -> - import_data_lines(data_lines, submitter_email) + rows = Enum.map(data_lines, fn {line, row_num} -> parse_row(line, row_num, submitter_email) end) + {:ok, rows} end end defp blank?(line), do: String.trim(line) == "" - defp import_data_lines(data_lines, submitter_email) do - result = - Enum.reduce(data_lines, %{imported: [], errors: []}, fn {line, row_num}, acc -> - case parse_and_import_row(line, submitter_email) do - {:ok, contact} -> - %{acc | imported: [contact | acc.imported]} - - {:error, error_messages} -> - %{acc | errors: [{row_num, error_messages} | acc.errors]} - end - end) - - {:ok, %{imported: Enum.reverse(result.imported), errors: Enum.reverse(result.errors)}} - end - - defp parse_and_import_row(line, submitter_email) do + defp parse_row(line, row_num, submitter_email) do fields = parse_csv_fields(line) if length(fields) == @expected_columns do @@ -68,24 +143,132 @@ defmodule Microwaveprop.Radio.CsvImport do case normalize_timestamp(attrs["qso_timestamp"]) do {:ok, iso_timestamp} -> attrs = Map.put(attrs, "qso_timestamp", iso_timestamp) + changeset = Contact.submission_changeset(%Contact{}, attrs) - case Radio.create_contact(attrs) do - {:ok, contact} -> - ContactWeatherEnqueueWorker.enqueue_for_contact(contact) - {:ok, contact} - - {:error, changeset} -> - {:error, changeset_error_strings(changeset)} + if changeset.valid? do + {:ok, datetime} = parse_iso_datetime(iso_timestamp) + {:parsed, %{row_num: row_num, attrs: attrs, timestamp: datetime}} + else + {:invalid, %{row_num: row_num, messages: changeset_error_strings(changeset)}} end {:error, msg} -> - {:error, [msg]} + {:invalid, %{row_num: row_num, messages: [msg]}} end else - {:error, ["expected #{@expected_columns} columns, got #{length(fields)}"]} + {:invalid, %{row_num: row_num, messages: ["expected #{@expected_columns} columns, got #{length(fields)}"]}} end end + defp split_parsed(rows) do + rows + |> Enum.reduce({[], []}, fn + {:parsed, row}, {valid, invalid} -> {[row | valid], invalid} + {:invalid, row}, {valid, invalid} -> {valid, [row | invalid]} + end) + |> then(fn {valid, invalid} -> {Enum.reverse(valid), Enum.reverse(invalid)} end) + end + + # -- dedup ----------------------------------------------------------------- + + defp load_existing_for_dedup([]), do: %{} + + defp load_existing_for_dedup(valid_rows) do + bands = valid_rows |> Enum.map(& &1.attrs["band"]) |> Enum.uniq() |> Enum.map(&to_decimal/1) + timestamps = Enum.map(valid_rows, & &1.timestamp) + min_ts = timestamps |> Enum.min(DateTime) |> DateTime.add(-@dedup_window_seconds, :second) + max_ts = timestamps |> Enum.max(DateTime) |> DateTime.add(@dedup_window_seconds, :second) + + from(c in Contact, + where: + c.band in ^bands and c.qso_timestamp >= ^min_ts and c.qso_timestamp <= ^max_ts and + c.flagged_invalid == false, + select: %{ + station1: c.station1, + station2: c.station2, + grid1: c.grid1, + grid2: c.grid2, + band: c.band, + qso_timestamp: c.qso_timestamp + } + ) + |> Repo.all() + |> Enum.reduce(%{}, fn row, acc -> + key = dedup_key(row.station1, row.grid1, row.station2, row.grid2, row.band) + Map.update(acc, key, [row.qso_timestamp], &[row.qso_timestamp | &1]) + end) + end + + defp dedupe(rows, db_map) do + # in_batch_map accumulates keys from rows already accepted in this CSV so + # we can distinguish DB duplicates from earlier-row duplicates when we + # report them. + {accepted, _in_batch, duplicates} = + Enum.reduce(rows, {[], %{}, []}, fn row, {accepted, in_batch, dups} -> + key = dedup_key_from_attrs(row.attrs) + + cond do + conflict_in_map?(db_map, key, row.timestamp) -> + {accepted, in_batch, [Map.put(row, :reason, :existing_contact) | dups]} + + conflict_in_map?(in_batch, key, row.timestamp) -> + {accepted, in_batch, [Map.put(row, :reason, :earlier_in_upload) | dups]} + + true -> + new_batch = Map.update(in_batch, key, [row.timestamp], &[row.timestamp | &1]) + {[row | accepted], new_batch, dups} + end + end) + + {Enum.reverse(accepted), Enum.reverse(duplicates)} + end + + defp conflict_in_map?(map, key, timestamp) do + case Map.fetch(map, key) do + :error -> false + {:ok, timestamps} -> Enum.any?(timestamps, &within_window?(&1, timestamp)) + end + end + + defp within_window?(t1, t2) do + abs(DateTime.diff(t1, t2, :second)) <= @dedup_window_seconds + end + + defp dedup_key_from_attrs(attrs) do + dedup_key(attrs["station1"], attrs["grid1"], attrs["station2"], attrs["grid2"], attrs["band"]) + end + + defp dedup_key(s1, g1, s2, g2, band) do + a = normalize_callsite(s1, g1) + b = normalize_callsite(s2, g2) + [c1, c2] = Enum.sort([a, b]) + {c1, c2, normalize_band(band)} + end + + defp normalize_callsite(call, grid) do + {upcase_trim(call), upcase_trim(grid)} + end + + defp upcase_trim(nil), do: "" + defp upcase_trim(value), do: value |> to_string() |> String.trim() |> String.upcase() + + defp normalize_band(%Decimal{} = b), do: Decimal.to_integer(b) + defp normalize_band(b) when is_integer(b), do: b + defp normalize_band(b) when is_binary(b), do: b |> String.trim() |> String.to_integer() + + defp to_decimal(%Decimal{} = d), do: d + defp to_decimal(v) when is_integer(v), do: Decimal.new(v) + defp to_decimal(v) when is_binary(v), do: v |> String.trim() |> Decimal.new() + + defp parse_iso_datetime(iso) do + case DateTime.from_iso8601(iso) do + {:ok, dt, _} -> {:ok, dt} + _ -> :error + end + end + + # -- timestamps ------------------------------------------------------------ + # Regex patterns for timestamp parsing, ordered most-specific first. # Accepts many hand-entered formats. All times assumed UTC. # Each returns {year, month, day, hour, minute, second} or nil. @@ -172,6 +355,8 @@ defmodule Microwaveprop.Radio.CsvImport do {:ok, "#{y}-#{m}-#{d}T#{h}:#{mi}:#{s}Z"} end + # -- csv field parser ------------------------------------------------------ + # RFC 4180 CSV field parser. Handles quoted fields with commas and escaped quotes. defp parse_csv_fields(line) do line diff --git a/lib/microwaveprop_web/live/submit_live.ex b/lib/microwaveprop_web/live/submit_live.ex index e27773a1..a09ba88e 100644 --- a/lib/microwaveprop_web/live/submit_live.ex +++ b/lib/microwaveprop_web/live/submit_live.ex @@ -32,6 +32,7 @@ defmodule MicrowavepropWeb.SubmitLive do mode_options: @mode_options, submitted_at: nil, active_tab: :single, + csv_preview: nil, csv_result: nil, csv_email: "" )} @@ -92,7 +93,7 @@ defmodule MicrowavepropWeb.SubmitLive do case uploaded_contents do [content] -> - handle_csv_import(content, email, socket) + handle_csv_preview(content, email, socket) [] -> {:noreply, put_flash(socket, :error, "Please select a CSV file")} @@ -100,14 +101,33 @@ defmodule MicrowavepropWeb.SubmitLive do end end - def handle_event("reset_csv", _params, socket) do - {:noreply, assign(socket, csv_result: nil)} + def handle_event("confirm_csv", _params, socket) do + case socket.assigns.csv_preview do + %{valid: valid_rows} when valid_rows != [] -> + {:ok, commit_result} = CsvImport.commit(valid_rows) + + {:noreply, + socket + |> assign(csv_preview: nil, csv_result: commit_result) + |> put_flash(:info, "Imported #{length(commit_result.imported)} contacts.")} + + _ -> + {:noreply, put_flash(socket, :error, "Nothing to import.")} + end end - defp handle_csv_import(content, email, socket) do - case CsvImport.import(content, email) do - {:ok, result} -> - {:noreply, assign(socket, csv_result: result, csv_email: email)} + def handle_event("cancel_csv", _params, socket) do + {:noreply, assign(socket, csv_preview: nil, csv_result: nil)} + end + + def handle_event("reset_csv", _params, socket) do + {:noreply, assign(socket, csv_preview: nil, csv_result: nil)} + end + + defp handle_csv_preview(content, email, socket) do + case CsvImport.preview(content, email) do + {:ok, preview} -> + {:noreply, assign(socket, csv_preview: preview, csv_result: nil, csv_email: email)} {:error, :empty_csv} -> {:noreply, put_flash(socket, :error, "CSV file is empty")} @@ -181,10 +201,13 @@ defmodule MicrowavepropWeb.SubmitLive do <%= if @active_tab == :single do %> <.single_contact_form form={@form} band_options={@band_options} mode_options={@mode_options} /> <% else %> - <%= if @csv_result do %> - <.csv_results result={@csv_result} /> - <% else %> - <.csv_upload_form uploads={@uploads} /> + <%= cond do %> + <% @csv_result -> %> + <.csv_results result={@csv_result} /> + <% @csv_preview -> %> + <.csv_preview preview={@csv_preview} /> + <% true -> %> + <.csv_upload_form uploads={@uploads} /> <% end %> <% end %> @@ -311,6 +334,189 @@ defmodule MicrowavepropWeb.SubmitLive do """ end + defp csv_preview(assigns) do + assigns = + assign(assigns, + valid_count: length(assigns.preview.valid), + invalid_count: length(assigns.preview.invalid), + duplicate_count: length(assigns.preview.duplicates), + valid_sample: Enum.take(assigns.preview.valid, 20) + ) + + ~H""" +
+ Processed {@preview.total_rows} {if @preview.total_rows == 1, do: "row", else: "rows"}. + Nothing has been inserted yet — confirm below to import. +
+| Row | +Errors | +
|---|---|
| Row {row.row_num} | +
+ {msg}
+ |
+
100} class="text-xs opacity-60 mt-1"> + Showing first 100 of {@invalid_count} invalid rows. +
++ Same two callsigns at the same grids on the same band within an hour. +
+| Row | +Station 1 | +Station 2 | +Band | +Timestamp | +Source | +
|---|---|---|---|---|---|
| Row {row.row_num} | +{row.attrs["station1"]} {row.attrs["grid1"]} | +{row.attrs["station2"]} {row.attrs["grid2"]} | +{row.attrs["band"]} | ++ {Calendar.strftime(row.timestamp, "%Y-%m-%d %H:%M UTC")} + | +{duplicate_source(row.reason)} | +
100} class="text-xs opacity-60 mt-1"> + Showing first 100 of {@duplicate_count} duplicates. +
+| Row | +Station 1 | +Station 2 | +Band | +Mode | +Timestamp | +
|---|---|---|---|---|---|
| Row {row.row_num} | +{row.attrs["station1"]} {row.attrs["grid1"]} | +{row.attrs["station2"]} {row.attrs["grid2"]} | +{row.attrs["band"]} | +{row.attrs["mode"]} | ++ {Calendar.strftime(row.timestamp, "%Y-%m-%d %H:%M UTC")} + | +
length(@valid_sample)} class="text-xs opacity-60 mt-1"> + Showing first {length(@valid_sample)} of {@valid_count} valid rows. +
+