Add support for specifying custom patterns (#10)

Author: Arpan-Kreeti

README.md

@@ -76,7 +76,9 @@ defmodule Csv2sql do
  connect_timeout: :integer,
  pool_size: :integer,
  queue_target: :integer,
- queue_interval: :integer
+ queue_interval: :integer,
+ custom_date_patterns: :string,
+ custom_datetime_patterns: :string
  ]
  )
 
@@ -115,6 +117,8 @@ defmodule Csv2sql do
  pool_size = opts[:pool_size] || 20
  queue_target = opts[:queue_target] || 5000
  queue_interval = opts[:queue_interval] || 1000
+ custom_date_patterns = ["{YYYY}-{0M}-{0D}" | String.split((opts[:custom_date_patterns] || ""), ";")]
+ custom_datetime_patterns = ["{YYYY}-{0M}-{0D} {0h24}:{0m}:{0s}" | String.split((opts[:custom_datetime_patterns] || ""), ";")]
 
  repo_config = [
  username: username,
@@ -148,7 +152,9 @@ defmodule Csv2sql do
  [
  varchar_limit: varchar_limit,
  schema_file_path: schema_file_path,
- schema_infer_chunk_size: schema_infer_chunk_size
+ schema_infer_chunk_size: schema_infer_chunk_size,
+ custom_date_patterns: custom_date_patterns,
+ custom_datetime_patterns: custom_datetime_patterns
  ]},
  {Csv2sql.MainServer,
  [

apps/csv2sql/README.md

@@ -114,6 +114,14 @@ defmodule Csv2sql.Database do
  {val, ""} = Float.parse(val)
  val
 
+ # MYSQL
+ "DATE" ->
+ format_datetime(val, true)
+
+ # MYSQL
+ "DATETIME" ->
+ format_datetime(val, false)
+
  _ ->
  val
  end
@@ -157,4 +165,35 @@ defmodule Csv2sql.Database do
 
  Helpers.print_msg("Create Schema for: #{table_name}")
  end
+
+ # Warning: Timezone information if any will be ignored while parsing datetime
+ defp format_datetime(datetime, is_date) do
+ schema_maker_configs = Application.get_env(:csv2sql, Csv2sql.SchemaMaker)
+
+ is_date
+ |> if(
+ do: schema_maker_configs[:custom_date_patterns],
+ else: schema_maker_configs[:custom_datetime_patterns]
+ )
+ |> Enum.find_value(fn pattern ->
+ case Timex.parse(datetime, pattern) do
+ {:ok, %DateTime{} = datetime} ->
+ to_date_or_datetime_string(datetime, is_date)
+
+ {:ok, %NaiveDateTime{} = native_datetime} ->
+ native_datetime
+ |> DateTime.from_naive!("Etc/UTC")
+ |> to_date_or_datetime_string(is_date)
+
+ {:error, _} ->
+ false
+ end
+ end)
+ end
+
+ defp to_date_or_datetime_string(datetime, true),
+ do: datetime |> DateTime.to_date() |> Date.to_string() |> String.trim_trailing("Z")
+
+ defp to_date_or_datetime_string(datetime, false),
+ do: datetime |> DateTime.to_string() |> String.trim_trailing("Z")
 end

apps/csv2sql/lib/csv2sql.ex

@@ -36,7 +36,7 @@ defmodule Csv2sql.SchemaMaker do
  Map.put(type, :is_empty, type.is_empty && empty)
  else
  is_date = type.is_date && is_date?(item)
- is_timestamp = type.is_timestamp && is_timestamp?(item)
+ is_datetime = type.is_datetime && is_datetime?(item)
  is_integer = type.is_integer && is_integer?(item)
  is_float = type.is_float && is_float?(item)
  is_boolean = type.is_boolean && is_boolean?(item)
@@ -45,7 +45,7 @@ defmodule Csv2sql.SchemaMaker do
  %{
  is_empty: type.is_empty && empty,
  is_date: is_date,
- is_timestamp: is_timestamp,
+ is_datetime: is_datetime,
  is_boolean: is_boolean,
  is_integer: is_integer,
  is_float: is_float,
@@ -99,16 +99,20 @@ defmodule Csv2sql.SchemaMaker do
  varchar_limit = Application.get_env(:csv2sql, Csv2sql.SchemaMaker)[:varchar_limit]
  headers_type_list = List.duplicate(get_type_map(), Enum.count(headers))
 
+ schema_infer_chunk_size =
+ Application.get_env(:csv2sql, Csv2sql.SchemaMaker)[:schema_infer_chunk_size]
+
  db_type = Csv2sql.get_db_type()
 
  types =
  path
  |> File.stream!()
  |> CSV.parse_stream()
- |> Stream.chunk_every(
- Application.get_env(:csv2sql, Csv2sql.SchemaMaker)[:schema_infer_chunk_size]
+ |> Stream.chunk_every(schema_infer_chunk_size)
+ |> Task.async_stream(__MODULE__, :infer_type, [headers_type_list],
+ timeout: :infinity,
+ ordered: false
  )
- |> Task.async_stream(__MODULE__, :infer_type, [headers_type_list], timeout: :infinity)
  |> Enum.reduce(headers_type_list, fn {:ok, result}, acc ->
  # Here we get a list of type maps for each chunk of data
  # We need to merge theses type maps obtained from each chunk
@@ -117,7 +121,7 @@ defmodule Csv2sql.SchemaMaker do
  %{
  is_empty: acc_map.is_empty && result_map.is_empty,
  is_date: acc_map.is_date && result_map.is_date,
- is_timestamp: acc_map.is_timestamp && result_map.is_timestamp,
+ is_datetime: acc_map.is_datetime && result_map.is_datetime,
  is_boolean: acc_map.is_boolean && result_map.is_boolean,
  is_integer: acc_map.is_integer && result_map.is_integer,
  is_float: acc_map.is_float && result_map.is_float,
@@ -154,8 +158,8 @@ defmodule Csv2sql.SchemaMaker do
  defp get_column_types(:mysql, varchar_limit, type) do
  cond do
  type[:is_empty] -> "VARCHAR(#{varchar_limit})"
- type[:is_timestamp] -> "TIMESTAMP"
  type[:is_date] -> "DATE"
+ type[:is_datetime] -> "DATETIME"
  type[:is_boolean] -> "BIT"
  type[:is_integer] -> "INT"
  type[:is_float] -> "DOUBLE"
@@ -179,7 +183,7 @@ defmodule Csv2sql.SchemaMaker do
  %{
  is_empty: true,
  is_date: true,
- is_timestamp: true,
+ is_datetime: true,
  is_boolean: true,
  is_integer: true,
  is_float: true,
@@ -198,11 +202,23 @@ defmodule Csv2sql.SchemaMaker do
  end
 
  defp is_date?(item) do
- Regex.match?(~r/\d\d\d\d-\d\d-\d\d/, item)
+ Application.get_env(:csv2sql, Csv2sql.SchemaMaker)[:custom_date_patterns]
+ |> Enum.any?(fn pattern ->
+ case Timex.parse(item, pattern) do
+ {:ok, _} -> true
+ {:error, _} -> false
+ end
+ end)
  end
 
- defp is_timestamp?(item) do
- Regex.match?(~r/\d\d\d\d-\d\d-\d\d \d\d:\d\d:\d\d/, item)
+ defp is_datetime?(item) do
+ Application.get_env(:csv2sql, Csv2sql.SchemaMaker)[:custom_datetime_patterns]
+ |> Enum.any?(fn pattern ->
+ case Timex.parse(item, pattern) do
+ {:ok, _} -> true
+ {:error, _} -> false
+ end
+ end)
  end
 
  defp is_boolean?(item) do

apps/csv2sql/lib/csv2sql/database.ex

@@ -40,7 +40,8 @@ defmodule Csv2sql.MixProject do
  {:ecto, "~> 3.4"},
  {:ecto_sql, "~> 3.1"},
  {:cli_spinners, "~> 0.1.0"},
- {:sizeable, "~> 1.0"}
+ {:sizeable, "~> 1.0"},
+ {:timex, "~> 3.7"}
  ]
  end
 end

apps/csv2sql/lib/csv2sql/schema_maker.ex

@@ -1,15 +1,27 @@
 %{
+ "certifi": {:hex, :certifi, "2.8.0", "d4fb0a6bb20b7c9c3643e22507e42f356ac090a1dcea9ab99e27e0376d695eba", [:rebar3], [], "hexpm", "6ac7efc1c6f8600b08d625292d4bbf584e14847ce1b6b5c44d983d273e1097ea"},
  "cli_spinners": {:hex, :cli_spinners, "0.1.0", "08d89c6f1840a8927daed48c675ecb2f20c05c855dc2b4b58c2933d393d5e0c9", [:mix], [], "hexpm", "3b8ccad722e518309d8b92230960ca4775a761164514f78a89c4f04d5a25c97f"},
+ "combine": {:hex, :combine, "0.10.0", "eff8224eeb56498a2af13011d142c5e7997a80c8f5b97c499f84c841032e429f", [:mix], [], "hexpm", "1b1dbc1790073076580d0d1d64e42eae2366583e7aecd455d1215b0d16f2451b"},
  "connection": {:hex, :connection, "1.0.4", "a1cae72211f0eef17705aaededacac3eb30e6625b04a6117c1b2db6ace7d5976", [:mix], [], "hexpm", "4a0850c9be22a43af9920a71ab17c051f5f7d45c209e40269a1938832510e4d9"},
  "db_connection": {:hex, :db_connection, "2.2.1", "caee17725495f5129cb7faebde001dc4406796f12a62b8949f4ac69315080566", [:mix], [{:connection, "~> 1.0.2", [hex: :connection, repo: "hexpm", optional: false]}], "hexpm", "2b02ece62d9f983fcd40954e443b7d9e6589664380e5546b2b9b523cd0fb59e1"},
  "decimal": {:hex, :decimal, "1.8.1", "a4ef3f5f3428bdbc0d35374029ffcf4ede8533536fa79896dd450168d9acdf3c", [:mix], [], "hexpm", "3cb154b00225ac687f6cbd4acc4b7960027c757a5152b369923ead9ddbca7aec"},
  "dir_walker": {:hex, :dir_walker, "0.0.8", "5332225074e4887e6e60ca0242af490215f296511ded4df18d554ae25394f727", [:mix], [], "hexpm", "2f4fb16e6427523700df9eb12eece5679ad4459aaefb1ca3cb580184bfc8d173"},
  "ecto": {:hex, :ecto, "3.4.0", "a7a83ab8359bf816ce729e5e65981ce25b9fc5adfc89c2ea3980f4fed0bfd7c1", [:mix], [{:decimal, "~> 1.6 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "5eed18252f5b5bbadec56a24112b531343507dbe046273133176b12190ce19cc"},
  "ecto_sql": {:hex, :ecto_sql, "3.4.1", "3c9136ba138f9b74d31286c73c61232a92bd19385f7c5607bdeb3a4587ef91f5", [:mix], [{:db_connection, "~> 2.2", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.4.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.3.0 or ~> 0.4.0", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.15.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.0", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "9b4be0bffe7b0bdf5393defcae52712f248e70cc2bc0e8ab6ddb03be66371516"},
+ "gettext": {:hex, :gettext, "0.18.2", "7df3ea191bb56c0309c00a783334b288d08a879f53a7014341284635850a6e55", [:mix], [], "hexpm", "f9f537b13d4fdd30f3039d33cb80144c3aa1f8d9698e47d7bcbcc8df93b1f5c5"},
+ "hackney": {:hex, :hackney, "1.18.0", "c4443d960bb9fba6d01161d01cd81173089686717d9490e5d3606644c48d121f", [:rebar3], [{:certifi, "~>2.8.0", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "~>6.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "~>1.0.0", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~>1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:parse_trans, "3.3.1", [hex: :parse_trans, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "~>1.1.0", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}, {:unicode_util_compat, "~>0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "9afcda620704d720db8c6a3123e9848d09c87586dc1c10479c42627b905b5c5e"},
+ "idna": {:hex, :idna, "6.1.1", "8a63070e9f7d0c62eb9d9fcb360a7de382448200fbbd1b106cc96d3d8099df8d", [:rebar3], [{:unicode_util_compat, "~>0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "92376eb7894412ed19ac475e4a86f7b413c1b9fbb5bd16dccd57934157944cea"},
+ "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm", "69b09adddc4f74a40716ae54d140f93beb0fb8978d8636eaded0c31b6f099f16"},
+ "mimerl": {:hex, :mimerl, "1.2.0", "67e2d3f571088d5cfd3e550c383094b47159f3eee8ffa08e64106cdf5e981be3", [:rebar3], [], "hexpm", "f278585650aa581986264638ebf698f8bb19df297f66ad91b18910dfc6e19323"},
  "myxql": {:hex, :myxql, "0.3.4", "41163cfc97b899db0fd9ebb6f38a8dc841298a7fa1c1e84a93b9369e2dbb1815", [:mix], [{:db_connection, "~> 2.0", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.6", [hex: :decimal, repo: "hexpm", optional: false]}, {:geo, "~> 3.3", [hex: :geo, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "fc8d49ba141a46a174b410e7c86906d4d655ff8a1e1608c31d5d1777d106d1dd"},
  "nimble_csv": {:hex, :nimble_csv, "0.7.0", "52f23ce46eee304d063d1716e19e45ea544bd751536bc53e5d41cb7fc0ca9405", [:mix], [], "hexpm", "e7051e7a95b5c4f26512af5805c320ee9185e752d949f048bf318fedef86cccc"},
+ "parse_trans": {:hex, :parse_trans, "3.3.1", "16328ab840cc09919bd10dab29e431da3af9e9e7e7e6f0089dd5a2d2820011d8", [:rebar3], [], "hexpm", "07cd9577885f56362d414e8c4c4e6bdf10d43a8767abb92d24cbe8b24c54888b"},
  "postgrex": {:hex, :postgrex, "0.15.6", "a464c72010a56e3214fe2b99c1a76faab4c2bb0255cabdef30dea763a3569aa2", [:mix], [{:connection, "~> 1.0", [hex: :connection, repo: "hexpm", optional: false]}, {:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "f99268325ac8f66ffd6c4964faab9e70fbf721234ab2ad238c00f9530b8cdd55"},
  "sizeable": {:hex, :sizeable, "1.0.2", "625fe06a5dad188b52121a140286f1a6ae1adf350a942cf419499ecd8a11ee29", [:mix], [], "hexpm", "4bab548e6dfba777b400ca50830a9e3a4128e73df77ab1582540cf5860601762"},
+ "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.6", "cf344f5692c82d2cd7554f5ec8fd961548d4fd09e7d22f5b62482e5aeaebd4b0", [:make, :mix, :rebar3], [], "hexpm", "bdb0d2471f453c88ff3908e7686f86f9be327d065cc1ec16fa4540197ea04680"},
  "stream_split": {:hex, :stream_split, "0.1.4", "ea8073779725101127123bd4176500227e958113090741e38b23ee4efd4452f6", [:mix], [], "hexpm", "fd1a8ea510b237c2006bbd33ab0b7ccc6e3831e16b1d7f142dff8434acdbf15e"},
  "telemetry": {:hex, :telemetry, "0.4.1", "ae2718484892448a24470e6aa341bc847c3277bfb8d4e9289f7474d752c09c7f", [:rebar3], [], "hexpm", "4738382e36a0a9a2b6e25d67c960e40e1a2c95560b9f936d8e29de8cd858480f"},
+ "timex": {:hex, :timex, "3.7.6", "502d2347ec550e77fdf419bc12d15bdccd31266bb7d925b30bf478268098282f", [:mix], [{:combine, "~> 0.10", [hex: :combine, repo: "hexpm", optional: false]}, {:gettext, "~> 0.10", [hex: :gettext, repo: "hexpm", optional: false]}, {:tzdata, "~> 1.0", [hex: :tzdata, repo: "hexpm", optional: false]}], "hexpm", "a296327f79cb1ec795b896698c56e662ed7210cc9eb31f0ab365eb3a62e2c589"},
+ "tzdata": {:hex, :tzdata, "1.1.1", "20c8043476dfda8504952d00adac41c6eda23912278add38edc140ae0c5bcc46", [:mix], [{:hackney, "~> 1.17", [hex: :hackney, repo: "hexpm", optional: false]}], "hexpm", "a69cec8352eafcd2e198dea28a34113b60fdc6cb57eb5ad65c10292a6ba89787"},
+ "unicode_util_compat": {:hex, :unicode_util_compat, "0.7.0", "bc84380c9ab48177092f43ac89e4dfa2c6d62b40b8bd132b1059ecc7232f9a78", [:rebar3], [], "hexpm", "25eee6d67df61960cf6a794239566599b09e17e668d3700247bc498638152521"},
 }