Elixir program that fetches all Eve Online market data and writes to csv

orders.ex

Mix.install([{:req, "~> 0.5"}])

defmodule OrdersParser do
  defmacrop drop(input, str) do
    quote do
      <<unquote(str), rest::binary>> = unquote(input)
      rest
    end
  end

  defmacrop field(input, field_name) do
    quote do
      <<?", unquote(field_name), ?", ?:, rest::binary>> = unquote(input)
      rest
    end
  end

  def parse_orders(<<"[", input::binary>>), do: parse_orders(input, [])

  defp parse_orders(<<"]">>, orders), do: orders

  defp parse_orders(input, orders) do
    {input, order} = parse_order(input)
    parse_orders(input, [order | orders])
  end

  defp parse_order(input) do
    input = input |> drop("{")

    {input, duration} = input |> field("duration") |> int()
    {input, is_buy_order} = input |> drop(",") |> field("is_buy_order") |> bool()
    {input, issued} = input |> drop(",") |> field("issued") |> str()
    {input, location_id} = input |> drop(",") |> field("location_id") |> int()
    {input, min_volume} = input |> drop(",") |> field("min_volume") |> int()
    {input, order_id} = input |> drop(",") |> field("order_id") |> int()
    {input, price} = input |> drop(",") |> field("price") |> float()
    {input, range} = input |> drop(",") |> field("range") |> str()
    {input, system_id} = input |> drop(",") |> field("system_id") |> int()
    {input, type_id} = input |> drop(",") |> field("type_id") |> int()
    {input, volume_remain} = input |> drop(",") |> field("volume_remain") |> int()
    {input, volume_total} = input |> drop(",") |> field("volume_total") |> int()

    input = input |> drop("}") |> maybe_drop(",")

    order = [
      duration,
      ",",
      is_buy_order,
      ",",
      issued,
      ",",
      location_id,
      ",",
      min_volume,
      ",",
      order_id,
      ",",
      price,
      ",",
      range,
      ",",
      system_id,
      ",",
      type_id,
      ",",
      volume_remain,
      ",",
      volume_total,
      "\n"
    ]

    {input, order}
  end

  defp maybe_drop(input, str) do
    case input do
      <<^str::binary, rest::binary>> -> rest
      _ -> input
    end
  end

  defp bool(<<"false", rest::binary>>), do: {rest, "false"}
  defp bool(<<"true", rest::binary>>), do: {rest, "true"}

  defp float(input, acc \\ [])

  defp float(<<n::utf8, rest::binary>>, acc) when n in ?0..?9 or n == ?.,
    do: float(rest, acc ++ [n])

  defp float(input, acc), do: {input, acc}

  defp int(input, acc \\ [])
  defp int(<<n::utf8, rest::binary>>, acc) when n in ?0..?9, do: int(rest, acc ++ [n])
  defp int(input, acc), do: {input, acc}

  defp str(input), do: input |> drop("\"") |> str([])
  defp str(<<?", rest::binary>>, acc), do: {rest, acc}
  defp str(<<c::utf8, rest::binary>>, acc) when c != ?", do: str(rest, acc ++ [c])
end

headers =
  ~w[ duration is_buy_order issued location_id min_volume order_id price range system_id type_id volume_remain volume_total ]

fetch_region_page_last =
  fn region_id ->
    page_last =
      "https://esi.evetech.net/v1/markets/#{region_id}/orders/"
      |> Req.head!()
      |> then(& &1.headers["x-pages"])
      |> hd()
      |> String.to_integer()

    {region_id, page_last}
  end

fetch_region_page =
  fn {region_id, page} ->
    "https://esi.evetech.net/v1/markets/#{region_id}/orders?page=#{page}"
    |> Req.get!()
    |> then(& &1.body)
    # simpler, but slower csv row encoding
    |> Stream.map(fn item ->
      headers |> Enum.map(&to_string(item[&1])) |> Enum.join(",")
    end)
    |> Enum.join("\n")
  end

fetch_region_page_faster =
  fn {region_id, page} ->
    "https://esi.evetech.net/v1/markets/#{region_id}/orders?page=#{page}"
    |> Req.get!()
    |> then(& &1.body)
    # hard coded row csv encoding shaves off ~0.6s
    |> Enum.map(fn %{
                     "duration" => duration,
                     "is_buy_order" => is_buy_order,
                     "issued" => issued,
                     "location_id" => location_id,
                     "min_volume" => min_volume,
                     "order_id" => order_id,
                     "price" => price,
                     "range" => range,
                     "system_id" => system_id,
                     "type_id" => type_id,
                     "volume_remain" => volume_remain,
                     "volume_total" => volume_total
                   } ->
      [
        Integer.to_string(duration),
        ",",
        Atom.to_string(is_buy_order),
        ",",
        issued,
        ",",
        Integer.to_string(location_id),
        ",",
        Integer.to_string(min_volume),
        ",",
        Integer.to_string(order_id),
        ",",
        Float.to_string(price),
        ",",
        range,
        ",",
        Integer.to_string(system_id),
        ",",
        Integer.to_string(type_id),
        ",",
        Integer.to_string(volume_remain),
        ",",
        Integer.to_string(volume_total),
        "\n"
      ]
    end)
    |> IO.iodata_to_binary()
  end

fetch_region_page_fastest =
  fn {region_id, page} ->
    "https://esi.evetech.net/v1/markets/#{region_id}/orders?page=#{page}"
    |> Req.get!(decode_body: false)
    |> then(& &1.body)
    |> OrdersParser.parse_orders()
    |> IO.iodata_to_binary()
  end

File.open("orders2.csv", [:write], fn file ->
  :ok = IO.puts(file, headers |> Enum.join(","))

  "https://esi.evetech.net/v1/universe/regions"
  |> Req.get!()
  |> then(& &1.body)
  |> Task.async_stream(fetch_region_page_last, max_concurrency: 100, timeout: 120_000)
  |> Stream.flat_map(fn {:ok, {region_id, page_last}} ->
    1..page_last |> Enum.map(&{region_id, &1})
  end)
  |> Task.async_stream(fetch_region_page_fastest, max_concurrency: 100, timeout: 120_000)
  |> Enum.each(fn {:ok, region_page_data} -> :ok = IO.puts(file, region_page_data) end)
end)