Использование GenStage и Flow для создания системы рекомендаций товаров

File.stream!("path/to/some/file")
|> Flow.from_enumerable()
|> Flow.flat_map(&String.split(&1, " "))
|> Flow.partition()
|> Flow.reduce(fn -> %{} end, fn word, acc ->
  Map.update(acc, word, 1, & &1 + 1)
end)
|> Enum.to_list()

defmodule Learn.Indexer do
  @moduledoc """
  Index mentions of resources from an authoritative source.
  """

  alias Learn.Indexer

  alias Learn.Indexer.{
    Mention,
    Post,
    Ranking,
    Resource,
  }

  @doc """
  Rank the given resources based on their mentions in topics matching the given keywords
  """
  @spec rank(list(String.t), list(Resource.t)) :: list(Ranking.t)
  def rank(keywords, resources, opts \\ []) do
    keywords
    |> Flow.from_enumerable()
    |> Flow.flat_map(&Indexer.search(&1, opts))
    |> Flow.uniq()
    |> Flow.partition()
    |> Flow.flat_map(&Indexer.list_posts(&1, opts))
    |> Flow.partition(key: {:key, :id})
    |> Flow.uniq_by(&(&1.id))
    |> Flow.map(&Indexer.parse_html_content/1)
    |> Flow.map(&Indexer.parse_sentences/1)
    |> Flow.flat_map(&Indexer.extract_mentions(&1, resources))
    |> Flow.map(&Indexer.sentiment_analysis/1)
    |> Flow.partition(key: {:key, :resource})
    |> Flow.group_by(&(&1.resource))
    |> Flow.map(fn {resource, mentions} -> {resource, Indexer.aggregate_mentions_by_author(mentions)} end)
    |> Flow.map(fn {resource, recommendations} -> Indexer.rank_recommendations(resource, recommendations) end)
    |> Enum.to_list()
    |> Enum.sort_by(&(&1.score), &>=/2)
  end
end

defmodule Learn.Indexer.Sources.ElixirForum do
  use HTTPoison.Base

  alias Learn.Indexer.Cache.HttpRequestCache
  alias Learn.Indexer.Sources.ElixirForum

  @endpoint "https://elixirforum.com"

  defp process_url(path) do
    @endpoint <> path
  end

  def process_response_body(body) do
    body
    |> Poison.decode!
  end

  def cached_request!(url, opts) do
    HttpRequestCache.cached("elixirforum.com/" <> url, fn ->
      rate_limit_access(fn ->
        ElixirForum.get!(url).body
      end, opts)
    end)
  end

  defp rate_limit_access(request, opts \\ []) do
    scale = Keyword.get(opts, :scale, 1_000)
    limit = Keyword.get(opts, :limit, 1)

    case ExRated.check_rate(@endpoint, scale, limit) do
      {:ok, _} ->
        request.()

      {:error, _} ->
        :timer.sleep(1_000)
        rate_limit_access(request, opts)
    end
  end

  defmodule Search do
    @expected_fields ~w(posts topics)

    def query(q, opts) do
      "/search.json?" <> URI.encode_query(q: q)
      |> ElixirForum.cached_request!(opts)
      |> Map.take(@expected_fields)
      |> Enum.map(fn({k, v}) -> {String.to_existing_atom(k), v} end)
    end
  end
end

ElixirForum.Search.query("books", [scale: 1_000, limit: 1])

defmodule Learn.Indexer.Cache.HttpRequestCache do
  use GenServer
  require Logger

  defmodule State do
    defstruct [
      cache_dir: nil
    ]
  end

  def start_link(cache_dir) do
    GenServer.start_link(__MODULE__, %State{cache_dir: cache_dir}, name: __MODULE__)
  end

  def init(%State{cache_dir: cache_dir} = state) do
    File.mkdir_p!(cache_dir)
    {:ok, state}
  end

  def cached(key, operation) do
    case read(key) do
      {:ok, value} -> value
      {:error, :not_found} ->
        value = operation.()
        :ok = cache(key, value)
        value
    end
  end

  def read(key) do
    GenServer.call(__MODULE__, {:read, key})
  end

  def cache(key, value) do
    GenServer.call(__MODULE__, {:cache, key, value})
  end

  def handle_call({:read, key}, _from, %State{} = state) do
    path = cached_path(key, state)

    reply = case File.read(path) do
      {:ok, data} -> {:ok, Poison.decode!(data)}
      {:error, :enoent} -> {:error, :not_found}
      {:error, _} = reply -> reply
    end

    {:reply, reply, state}
  end

  def handle_call({:cache, key, value}, _from, %State{} = state) do
    path = cached_path(key, state)

    File.write!(path, Poison.encode!(value), [:write])

    {:reply, :ok, state}
  end

  defp cached_path(key, %State{cache_dir: cache_dir}) do
    key = String.slice(key, 0, 255)
    path = Path.join(cache_dir, key)
    ensure_dir(path)
    path
  end

  defp ensure_dir(path) do
    path
    |> Path.dirname()
    |> File.mkdir_p!()
  end
end

defmodule Learn.Indexer.Application do
  @moduledoc false

  use Application

  def start(_type, _args) do
    import Supervisor.Spec, warn: false

    children = [
      worker(Learn.Indexer.Cache.HttpRequestCache, ["fixture/http_cache"])
    ]

    opts = [strategy: :one_for_one, name: Learn.Indexer.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

defmodule Learn.Indexer.Sources.ElixirForumTest do
  use ExUnit.Case, async: false
  use ExVCR.Mock, adapter: ExVCR.Adapter.Hackney

  alias Learn.Indexer.Sources.ElixirForum

  setup_all do
    HTTPoison.start
    :ok
  end

  describe "search" do
    test "for \"books\"" do
      use_cassette "elixirforum.com/search.json?&q=book", match_requests_on: [:query] do
        response = ElixirForum.Search.query("book", [])

        assert length(response[:posts]) > 0
        assert length(response[:topics]) > 0
      end
    end
  end
end

defmodule Learn.Indexer.Stages.ParseHtmlContent do
  @moduledoc """
  Parse the HTML post content into paragraphs of text.
  """

  alias Learn.Indexer.{
    Content,
    Post,
  }

  def execute(%Post{content: content} = post) do
    %Post{post |
      content: parse_content_html(content),
    }
  end

  defp parse_content_html(%Content{html: html} = content) do
    paragraphs =
      html
      |> Floki.filter_out("aside")
      |> Floki.filter_out("pre")
      |> Floki.find("p")
      |> Enum.map(&Floki.text/1)
      |> Enum.flat_map(fn p -> String.split(p, "\n") end)
      |> Enum.reject(fn p -> p == "" end)

    %Content{content |
      paragraphs: paragraphs,
    }
  end
end

defmodule Learn.Indexer.Stages.ParseSentences do
  @moduledoc """
  Parse the paragraphs of text into sentences.
  """

  alias Learn.Indexer.{
    Content,
    Post,
  }

  def execute(%Post{content: content} = post) do
    %Post{post |
      content: parse_sentences(content),
    }
  end

  # chunks the given paragraphs into sentences.
  defp parse_sentences(%Content{paragraphs: paragraphs} = content) do
    sentences =
      paragraphs
      |> Enum.map(&Essence.Chunker.sentences/1)
      |> Enum.flat_map(fn sentences -> sentences end)

    %Content{content |
      sentences: sentences,
    }
  end
end

defp mentioned?(sentence, %Resource{name: name}) do
  contains?(tokenize_downcase(sentence), tokenize_downcase(name))
end

def tokenize_downcase(text), do: text |> String.downcase |> Essence.Tokenizer.tokenize

defp contains?(source_tokens, search_tokens) do
  source_tokens
  |> Stream.chunk(length(search_tokens), 1)
  |> Enum.any?(fn chunk -> chunk == search_tokens end)
end

defmodule Learn.Indexer.Stages.SentimentAnalysis do
  @moduledoc """
  Analyse the mention sentence for its sentiment (positive, neutral, or negative).

  Uses the AFINN-111 word list.
  """

  alias Learn.Indexer.{
    Mention,
  }

  @spec execute(Mention.t) :: Mention.t
  def execute(%Mention{} = mention) do
    %Mention{mention |
      sentiment_score: sentiment(mention),
    }
  end

  @override_words %{"free" => 0}

  defp sentiment(%Mention{sentence: sentence}) do
    Sentient.analyze(sentence, @override_words)
  end
end

score = (upvotes + α) / (upvotes + downvotes + β)

defmodule Learn.Indexer.Stages.RankRecommendations do
  @moduledoc """
  Combine recommendations for the same resource into a single ranking and score
  """

  alias Learn.Indexer.{
    Ranking,
    Recommendation,
    Resource,
  }

  @spec execute(Resource.t, list(Recommendation.t)) :: Ranking.t
  def execute(resource, recommendations) do
    %Ranking{
      resource: resource,
      recommendations: recommendations,
      score: calculate_score(recommendations),
    }
  end

  # calculate score by using Laplace smoothing on positive, neutral and negative mentions
  defp calculate_score([]), do: 0
  defp calculate_score(recommendations) do
    recommendations
    |> Enum.reduce({0, 0, 0}, &Recommendation.score/2)
    |> score
  end

  defp score({negative, neutral, positive}) do
    upvotes = neutral + (positive * 2)
    downvotes = negative

    (upvotes + 1) / (upvotes + downvotes + 2)
  end
end

def index_mentions(keywords, resources, opts \\ []) do
  keywords
  |> Learn.Indexer.rank(resources, opts)
  |> record_rankings(resources)
end

defmodule Learn.Recommendations.Queries.RecommendedResources do
  import Ecto.Query

  alias Learn.Recommendations.{
    Resource,
    Score,
  }

  def new do
    from r in Resource,
    left_join: s in assoc(r, :score),
    order_by: [asc: s.rank, asc: r.title],
    preload: [:score]
  end

  def by_experience(query, level) do
    from [r, s] in query,
    where: r.experience_level == ^level
  end

  def by_language(query, language) do
    from [r, s] in query,
    where: r.programming_language == ^language
  end
end

defmodule Learn.Web.ResourceController do
  use Learn.Web.Web, :controller

  alias Learn.Recommendations.Repo,
  alias Learn.Recommendations.Queries.RecommendedResources

  def index(conn, _params) do
    resources =
      RecommendedResources.new
      |> RecommendedResources.by_experience("beginner")
      |> RecommendedResources.by_language("Elixir")
      |> Repo.all()

    render conn, "index.html", resources: resources
  end
end

defmodule Learn.Progress do
  @moduledoc """
  Progress stats collector, courtesy of http://teamon.eu/2016/measuring-visualizing-genstage-flow-with-gnuplot/
  """

  use GenServer

  @timeres :millisecond

  # Progress.start_link [:a, :b, :c]
  def start_link(scopes \\ []) do
    GenServer.start_link(__MODULE__, scopes, name: __MODULE__)
  end

  def stop do
    GenServer.stop(__MODULE__)
  end

  # increment counter for given scope by `n`
  #     Progress.incr(:my_scope)
  #     Progress.incr(:my_scope, 10)
  def incr(scope, n \\ 1) do
    GenServer.cast __MODULE__, {:incr, scope, n}
  end

  def init(scopes) do
    File.mkdir_p!("fixture/trace")

    # open "progress-{scope}.log" file for every scope
    files = Enum.map(scopes, fn scope ->
      {scope, File.open!("fixture/trace/progress-#{scope}.log", [:write])}
    end)

    # keep current counter for every scope
    counts = Enum.map(scopes, fn scope -> {scope, 0} end)

    # save current time
    time = :os.system_time(@timeres)

    # write first data point for every scope with current time and value 0
    # this helps to keep the graph starting nicely at (0,0) point
    Enum.each(files, fn {_, io} -> write(io, time, 0) end)

    {:ok, {time, files, counts}}
  end

  def handle_cast({:incr, scope, n}, {time, files, counts}) do
    # update counter
    {value, counts} = Keyword.get_and_update!(counts, scope, &({&1+n, &1+n}))

    # write new data point
    write(files[scope], time, value)

    {:noreply, {time, files, counts}}
  end

  defp write(file, time, value) do
    time = :os.system_time(@timeres) - time
    IO.write(file, "#{time}\t#{value}\n")
  end
end

defmodule Learn.Indexer do
  @moduledoc """
  Index mentions of resources from an authoritative source.
  """

  use Learn.ProgressDecorator

  @doc """
  Search for topics matching a given query
  """
  @decorate progress
  def search(query, opts \\ []), do: SearchKeyword.execute(query, opts)
end

defmodule Learn.ProgressDecorator do
  use Decorator.Define, [progress: 0]

  alias Learn.Progress

  def progress(body, context) do
    quote do
      reply = unquote(body)

      case reply do
        list when is_list(list) -> Progress.incr(unquote(context.name), length(list))
        _ -> Progress.incr(unquote(context.name), 1)
      end

      reply
    end
  end
end

defmodule Learn.Indexer do
  @moduledoc """
  Index mentions of resources from an authoritative source.
  """

  @doc """
  Rank the given resources based on their mentions in topics matching the given keywords
  """
  @spec rank(list(String.t), list(Resource.t)) :: list(Ranking.t)
  def rank(keywords, resources, opts \\ []) do
    Progress.start_link([
      :search,
      :list_posts,
      :parse_html_content,
      :parse_sentences,
      :extract_mentions,
      :sentiment_analysis,
      :aggregate_mentions_by_author,
      :rank_recommendations,
    ])

    rankings =
      keywords
      |> Flow.from_enumerable(max_demand: 1, stages: 1)
      |> Flow.flat_map(&Indexer.search(&1, opts))
      |> Flow.uniq()
      |> Flow.partition(max_demand: 5)
      |> Flow.flat_map(&Indexer.list_posts(&1, opts))
      |> Flow.partition(key: {:key, :id}, max_demand: 5)
      |> Flow.uniq_by(&(&1.id))
      |> Flow.map(&Indexer.parse_html_content/1)
      |> Flow.map(&Indexer.parse_sentences/1)
      |> Flow.flat_map(&Indexer.extract_mentions(&1, resources))
      |> Flow.map(&Indexer.sentiment_analysis/1)
      |> Flow.partition(key: {:key, :resource}, max_demand: 5)
      |> Flow.group_by(&(&1.resource))
      |> Flow.map(fn {resource, mentions} -> {resource, Indexer.aggregate_mentions_by_author(mentions)} end)
      |> Flow.map(fn {resource, recommendations} -> Indexer.rank_recommendations(resource, recommendations) end)
      |> Enum.to_list()
      |> Enum.sort_by(&(&1.score), &>=/2)

    Progress.stop()

    rankings
  end
end

# plot.gp
set terminal png font "Arial,10" size 700,500
set output "progress.png"

set title "Elixir Flow processing progress over time"
set xlabel "Time (ms)"

set ylabel "Progress (%)"
set y2label "Rankings"
set ytics nomirror
set yrange [0:100]
set format y '%2.0f%%'
set y2tics

set key top left # put labels in top-left corner

# limit x range to 35.000 ms instead of dynamic one - needed when generating graphs that will be later compared visually
set xrange [0:35000]

plot  "trace/progress-search.log"                         using ($1):($2/1249*100)  with steps  axes x1y1 ls 1 title "Search topics",\
      "trace/progress-search.log"                         using ($1):($2/1249*100)  with points axes x1y1 ls 1 notitle,\
      "trace/progress-list_posts.log"                     using ($1):($2/14974*100) with lines  axes x1y1 ls 2 title "List posts",\
      "trace/progress-parse_html_content.log"             using ($1):($2/6780*100)  with lines  axes x1y1 ls 3 title "Parse HTML",\
      "trace/progress-parse_sentences.log"                using ($1):($2/6780*100)  with lines  axes x1y1 ls 4 title "Parse sentences",\
      "trace/progress-extract_mentions.log"               using ($1):($2/515*100)   with lines  axes x1y1 ls 5 title "Extract mentions",\
      "trace/progress-sentiment_analysis.log"             using ($1):($2/515*100)   with lines  axes x1y1 ls 6 title "Sentiment analysis",\
      "trace/progress-aggregate_mentions_by_author.log"   using ($1):($2/314*100)   with lines  axes x1y1 ls 7 title "Aggregate mentions by author",\
      "trace/progress-rank_recommendations.log"                                     with steps  axes x1y2 ls 8 title "Rank",\
      "trace/progress-rank_recommendations.log"                                     with points axes x1y2 ls 8 notitle