Graham McIntire
44b169b234
Uploads now show a review page with three summary cards (valid, duplicates, invalid), tables for invalid and duplicate rows, and a sample of the first 20 valid rows. Nothing is written to the database until the user clicks "Looks good — submit N contacts". Duplicate detection treats two rows as the same contact if they share the same pair of (callsign, grid) tuples (direction-agnostic), the same band, and timestamps within 60 minutes. Dedup runs against both existing DB contacts (single band/time-scoped query) and earlier rows in the same upload, and the preview labels which side matched. CsvImport grows preview/2 + commit/1; the old import/2 is kept as a thin backward-compat wrapper that skips dedup. Added 11 new CsvImport tests and reworked the SubmitLive CSV tests for the two-step flow.
413 lines
14 KiB
Elixir
413 lines
14 KiB
Elixir
defmodule Microwaveprop.Radio.CsvImport do
|
|
@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 """
|
|
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")
|
|
|> String.split("\n")
|
|
|> Enum.with_index(1)
|
|
|
|
non_blank = Enum.reject(numbered_lines, fn {line, _num} -> blank?(line) end)
|
|
|
|
case non_blank do
|
|
[] ->
|
|
{:error, :empty_csv}
|
|
|
|
[{_header, _}] ->
|
|
{:error, :no_data_rows}
|
|
|
|
[{_header, _} | data_lines] ->
|
|
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 parse_row(line, row_num, submitter_email) do
|
|
fields = parse_csv_fields(line)
|
|
|
|
if length(fields) == @expected_columns do
|
|
attrs =
|
|
@columns
|
|
|> Enum.zip(fields)
|
|
|> Map.new()
|
|
|> Map.put("submitter_email", submitter_email)
|
|
|
|
case normalize_timestamp(attrs["qso_timestamp"]) do
|
|
{:ok, iso_timestamp} ->
|
|
attrs = Map.put(attrs, "qso_timestamp", iso_timestamp)
|
|
changeset = Contact.submission_changeset(%Contact{}, attrs)
|
|
|
|
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} ->
|
|
{:invalid, %{row_num: row_num, messages: [msg]}}
|
|
end
|
|
else
|
|
{: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.
|
|
defp timestamp_parsers do
|
|
[
|
|
# ISO-ish: 2024-06-15T14:30:00Z, 2024-06-15 14:30, 2024/06/15 14:30:00, etc.
|
|
# Accepts T or space separator, optional seconds, optional timezone suffix
|
|
{~r"^(\d{4})[-/](\d{1,2})[-/](\d{1,2})[T\s]+(\d{1,2}):(\d{2})(?::(\d{2}))?(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})?$"i,
|
|
fn
|
|
[_, y, m, d, h, mi, s] when byte_size(s) > 0 -> {y, m, d, h, mi, s}
|
|
[_, y, m, d, h, mi | _] -> {y, m, d, h, mi, "00"}
|
|
end},
|
|
# Date only: 2024-06-15, 2024/06/15
|
|
{~r"^(\d{4})[-/](\d{1,2})[-/](\d{1,2})$", fn [_, y, m, d] -> {y, m, d, "00", "00", "00"} end},
|
|
# US with AM/PM: 6/15/2024 2:30 PM, 6-15-2024 2:30PM, 06.15.2024 02:30 am
|
|
{~r"^(\d{1,2})[-/.](\d{1,2})[-/.](\d{4})\s+(\d{1,2}):(\d{2})(?::(\d{2}))?\s*(AM|PM)$"i,
|
|
fn
|
|
[_, m, d, y, h, mi, s, ampm] when byte_size(s) > 0 ->
|
|
{y, m, d, to_string(parse_12h(String.to_integer(h), String.upcase(ampm))), mi, s}
|
|
|
|
[_, m, d, y, h, mi, _, ampm] ->
|
|
{y, m, d, to_string(parse_12h(String.to_integer(h), String.upcase(ampm))), mi, "00"}
|
|
end},
|
|
# US 24h: 6/15/2024 14:30, 06-15-2024 14:30:00, 6.15.2024 14:30
|
|
{~r"^(\d{1,2})[-/.](\d{1,2})[-/.](\d{4})\s+(\d{1,2}):(\d{2})(?::(\d{2}))?$",
|
|
fn
|
|
[_, m, d, y, h, mi, s] when byte_size(s) > 0 -> {y, m, d, h, mi, s}
|
|
[_, m, d, y, h, mi | _] -> {y, m, d, h, mi, "00"}
|
|
end},
|
|
# US date only: 6/15/2024, 06-15-2024
|
|
{~r"^(\d{1,2})[-/.](\d{1,2})[-/.](\d{4})$", fn [_, m, d, y] -> {y, m, d, "00", "00", "00"} end},
|
|
# Compact: 20240615T143000, 20240615 1430, 20240615T1430
|
|
{~r"^(\d{4})(\d{2})(\d{2})[T\s]?(\d{2})(\d{2})(\d{2})?$",
|
|
fn
|
|
[_, y, m, d, h, mi, s] when byte_size(s) > 0 -> {y, m, d, h, mi, s}
|
|
[_, y, m, d, h, mi | _] -> {y, m, d, h, mi, "00"}
|
|
end}
|
|
]
|
|
end
|
|
|
|
@doc """
|
|
Parses various date/time formats into ISO 8601 UTC strings.
|
|
All times are assumed UTC.
|
|
"""
|
|
def normalize_timestamp(raw) do
|
|
trimmed = String.trim(raw)
|
|
|
|
# Try ISO 8601 with Elixir's built-in parser first
|
|
case DateTime.from_iso8601(trimmed) do
|
|
{:ok, dt, _} ->
|
|
{:ok, DateTime.to_iso8601(dt)}
|
|
|
|
_ ->
|
|
try_regex_parsers(trimmed)
|
|
end
|
|
end
|
|
|
|
defp try_regex_parsers(trimmed) do
|
|
timestamp_parsers()
|
|
|> Enum.find_value(fn {regex, extractor} ->
|
|
case Regex.run(regex, trimmed) do
|
|
nil -> nil
|
|
match -> format_iso(extractor.(match))
|
|
end
|
|
end)
|
|
|> case do
|
|
{:ok, _} = ok -> ok
|
|
nil -> {:error, "qso_timestamp could not be parsed: #{trimmed}"}
|
|
end
|
|
end
|
|
|
|
defp parse_12h(12, "AM"), do: 0
|
|
defp parse_12h(12, "PM"), do: 12
|
|
defp parse_12h(h, "PM"), do: h + 12
|
|
defp parse_12h(h, "AM"), do: h
|
|
|
|
defp format_iso({y, m, d, h, mi, s}) do
|
|
y = String.pad_leading(to_string(y), 4, "0")
|
|
m = String.pad_leading(to_string(m), 2, "0")
|
|
d = String.pad_leading(to_string(d), 2, "0")
|
|
h = String.pad_leading(to_string(h), 2, "0")
|
|
mi = String.pad_leading(to_string(mi), 2, "0")
|
|
s = String.pad_leading(to_string(s), 2, "0")
|
|
{: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
|
|
|> String.trim()
|
|
|> do_parse_fields([], "")
|
|
|> Enum.reverse()
|
|
|> Enum.map(&String.trim/1)
|
|
end
|
|
|
|
defp do_parse_fields("", acc, current), do: [current | acc]
|
|
|
|
defp do_parse_fields(<<"\"", rest::binary>>, acc, "") do
|
|
parse_quoted_field(rest, acc, "")
|
|
end
|
|
|
|
defp do_parse_fields(<<",", rest::binary>>, acc, current) do
|
|
do_parse_fields(rest, [current | acc], "")
|
|
end
|
|
|
|
defp do_parse_fields(<<c, rest::binary>>, acc, current) do
|
|
do_parse_fields(rest, acc, current <> <<c>>)
|
|
end
|
|
|
|
defp parse_quoted_field(<<"\"\"", rest::binary>>, acc, current) do
|
|
parse_quoted_field(rest, acc, current <> "\"")
|
|
end
|
|
|
|
defp parse_quoted_field(<<"\"", rest::binary>>, acc, current) do
|
|
# End of quoted field — skip to next comma or end
|
|
case rest do
|
|
<<",", rest2::binary>> -> do_parse_fields(rest2, [current | acc], "")
|
|
"" -> [current | acc]
|
|
_ -> do_parse_fields(rest, [current | acc], "")
|
|
end
|
|
end
|
|
|
|
defp parse_quoted_field(<<c, rest::binary>>, acc, current) do
|
|
parse_quoted_field(rest, acc, current <> <<c>>)
|
|
end
|
|
|
|
defp parse_quoted_field("", acc, current), do: [current | acc]
|
|
|
|
defp changeset_error_strings(changeset) do
|
|
changeset
|
|
|> Ecto.Changeset.traverse_errors(fn {message, opts} ->
|
|
Regex.replace(~r"%{(\w+)}", message, fn _, key ->
|
|
opts |> Keyword.get(String.to_existing_atom(key), key) |> to_string()
|
|
end)
|
|
end)
|
|
|> Enum.flat_map(fn {field, messages} ->
|
|
Enum.map(messages, &"#{field} #{&1}")
|
|
end)
|
|
end
|
|
end
|