stream = IO.binstream(fp, :line) |> Stream.filter(fn l -> l !== "\n" end)

    stream = IO.binstream(fp, :line)

    |> Stream.map(fn word -> String.normalize(word, :nfd) end)
    |> Stream.map(fn word ->
      String.replace(word, ~r/[!#$%&()*+,.:;<=>?@\^_`{|}~"\[\]«»\&„“]/, "")

    |> Stream.chunk_while(
      [],
      fn elem, acc ->
        case elem do
          "\n" ->
            {:cont, Enum.reverse(acc), []}
          _ ->
            {:cont,
             [
               elem |> String.trim() |> String.split() |> Enum.join(" ")
               | acc
             ]}
        end
      end,
      fn
        [] -> {:cont, []}
        acc -> {:cont, Enum.reverse(acc), []}
      end
    )
    |> Stream.filter(fn p -> p != [] end)
    |> Stream.map(fn
      [l] ->
        l
      p ->
        Enum.join(p, " ")

    |> Stream.map(fn word -> String.normalize(word, :nfc) end)
    |> Stream.flat_map(&String.split/1)
    |> Stream.map(&String.downcase/1)

  end
  def words(str) do
    str
    |> String.codepoints()
    |> Stream.filter(fn s ->
      s not in [
        "!",
        "#",
        "$",
        "%",
        "&",
        "(",
        ")",
        "*",
        "+",
        ",",
        ".",
        ":",
        ";",
        "<",
        "=",
        ">",
        "?",
        "@",
        "^",
        "_",
        "`",
        "{",
        "|",
        "}",
        "~",
        "\"",
        "[",
        "]",
        "«",
        "»",
        "&",
        "„",
        "“"
      ]
    end)
    |> Enum.join()
    |> String.downcase()
    |> String.split()
    |> Stream.map(fn w -> String.trim_leading(w, "-") end)
    |> Enum.filter(fn w -> w != "" end)

    defstruct [:book_id, :prev_word]

    defstruct [:paragraph_id, :prev_word]

    |> Stream.scan(%State{book_id: book_id}, &save/2)

    |> Task.async_stream(fn p -> process_paragraph(p, book_id) end,
      timeout: :infinity
    )
    |> Stream.run()
  end
  defp process_paragraph(paragraph, book_id) do
    paragraph_id = DB.save_paragraph(book_id, paragraph)
    paragraph
    |> Esperantisto.Text.words()
    |> Stream.scan(%State{paragraph_id: paragraph_id}, &save_word/2)

  defp save(word, %State{book_id: book_id, prev_word: prev_word} = acc) do
    DB.save_word(book_id, word)

  defp save_word(word, %State{paragraph_id: paragraph_id, prev_word: prev_word} = acc) do
    DB.save_word(paragraph_id, word)

  def save_paragraph(book_id, paragraph) do
    result =
      Bolt.Sips.query!(
        Bolt.Sips.conn(),
        """
        MATCH (book)
        WHERE id(book) = $id
        WITH book
        MERGE (book) -[:HAS_PARAGRAPH]-> (p:Paragraph {text: $paragraph})
        RETURN id(p) as paragraph_id
        """,
        %{id: book_id, paragraph: paragraph},
        timeout: 25000
      )

  def save_word(book_id, word) do

    [id] = result["paragraph_id"]
    id
  end
  def save_word(paragraph_id, word) do

      MATCH (book)
      WHERE id(book) = $id
      WITH book

      MATCH (paragraph:Paragraph)
      WHERE id(paragraph) = $id
      WITH paragraph

      MERGE (word) -[:OF_BOOK]-> (book)

      MERGE (word) -[:OF_PARAGRAH]-> (paragraph)

      %{id: book_id, word: word},

      %{id: paragraph_id, word: word},