Bcredi miner example

miner.exs

    path
    |> CsvParser.parse(:stream)
    |> Flow.from_enumerable()
    |> Flow.map(&cast_contract/1) # will return {:ok, %Contract{}} | {:error, changeset}
    |> Flow.map(&apply_blacklist/1) # will return {:ok, %Contract{}} | {:error, changeset} | {:filtered, reason}
    |> Flow.partition(key: {:elem, 0})
    |> Flow.group_by(& elem(&1, 0))
    |> Flow.filter(fn
      {:error, changeset} -> log_error(changeset) && false
      {:filtered, reason} -> log_filtered(reason) && false
      {:ok, contract} -> false
    end)
    |> Flow.reduce([], fn {:ok, contract}, acc -> [contract | acc] end)
    |> Enum.to_list()
    |> save_contracts()
    |> transform_contracts() # start risk, pdd, notification updates