Erlang and Elixir: hackney, oauth, twitter stream

twitter.erl

%#!/usr/bin/env escript
%% -*- erlang -*-
%%! -sasl errlog_type error

%%% Dependencies: hackney, erlang-oauth, jsx
%%% https://dev.twitter.com/docs/auth/authorizing-request
%%% https://dev.twitter.com/docs/api/1.1/post/statuses/filter

-module(twitter).
-author(gdamjan).
-export([start/0, main/1]).

auth_header(Method, Url, Params) ->
    % get these from https://apps.twitter.com/app/new
    ApiKey       = "____",
    ApiSecret    = "____",
    AccessToken  = "____",
    AccessSecret = "____",
    Consumer = {ApiKey, ApiSecret, hmac_sha1},
    SignedParams = oauth:sign(Method, Url, Params, Consumer, AccessToken, AccessSecret),
    SignedParams1 = lists:filter(fun ({K, _}) -> string:str(K, "oauth_") == 1 end, SignedParams),
    SignedParams2 = lists:sort(SignedParams1),
    BinaryParams = [ {list_to_binary(K), hackney_url:urlencode(list_to_binary(V))} || {K, V} <- SignedParams2 ],
    OAuth = hackney_bstr:join([ <<K/binary, "=", $\", V/binary, $\">> || {K, V} <- BinaryParams ], ","),
                                  {<<"Authorization">>, <<"OAuth ", OAuth/binary>>}.

start() ->
    main([]).

main(_) ->
    hackney:start(),
    Tags = [ "#skopjehacklab" ],
    Users = ["2cmk", "gdamjan"],
    {ok, UserIDs} = lookup_screen_names(Users),
    stream(UserIDs, Tags).


lookup_screen_names(Screen_Names) ->
    Url = "https://api.twitter.com/1.1/users/lookup.json",
    Method = "POST",
    Params = [ {"screen_name", string:join(Screen_Names, ",")} ],

    Headers = [auth_header(Method, Url, Params) ],
    io:format("Looking up users...~n"),
    case hackney:post(Url, Headers, {form, Params}, []) of
        {ok, 200, _H, R} ->
            {ok, Body} = hackney:body(R),
            {ok, [ binary:bin_to_list(proplists:get_value(<<"id_str">>, L)) || L <- jsx:decode(Body) ]};
        {ok, 404, _H, R} ->
            {ok, Body} = hackney:body(R),
            io:format(Body);
        Other ->
            io:format("~p~n", [Other])
    end.


stream(Follow, Track) ->
    Url = "https://stream.twitter.com/1.1/statuses/filter.json",
    Method = "POST",
    Params = [
              {"follow", string:join(Follow, ",")},
              {"track", string:join(Track, ",")}
             ],

    Headers = [auth_header(Method, Url, Params)],
    Options = [{recv_timeout, 60000}],
    io:format("Starting stream...~n"),
    {ok, 200, _H, Ref} = hackney:post(Url, Headers, {form, Params}, Options),
    stream_loop(Ref),
    stream(Follow, Track).

stream_loop(Ref) ->
    stream_loop(Ref, fun jsx:decode/1).

stream_loop(Ref, Decoder) ->
    case hackney:stream_body(Ref) of
        {ok, Data} ->
            case Decoder(Data) of
                {incomplete, NewDecoder} ->
                    stream_loop(Ref, NewDecoder);
                Obj ->
                    PrettyJson = jsx:encode(Obj, [{indent, 4}]),
                    io:format("~ts~n", [PrettyJson]),
                    stream_loop(Ref)
            end;
        done ->
            io:format("done"),
            done;
        {error, Reason} ->
            io:format("~p~n", [{error, Reason}])
    end.

twitter.ex

## dependencies for mix.exs
#   defp deps do
#     [
#        { :hackney, github: "benoitc/hackney" },
#        { :jsx, github: "talentdeficit/jsx" },
#        { :oauther, "~> 1.0.0" }
#     ]
#   end
defmodule Twitter do

  def start do
    :hackney.start()
    tags = [ "#skopjehacklab", "#хаклаб" ]
    users = [ "2cmk", "erlbot" ]
    {:ok, user_ids} = lookup_screen_names(users)
    stream(user_ids, tags)
  end

  def auth_header(method, url, params) do
    creds = OAuther.credentials(consumer_key:    "xxxx",
                                consumer_secret: "xxxx",
                                token:           "xxxx",
                                token_secret:    "xxxx")
    params = OAuther.sign(method, url, params, creds)
    {header, req_params} = OAuther.header(params)
  end

  def lookup_screen_names(screen_names) do
    url = "https://api.twitter.com/1.1/users/lookup.json"
    params = [ {"screen_name", Enum.join(screen_names, ",")} ]
    {header, req_params} = auth_header("post", url, params)
    IO.puts("Looking up users...")
    {:ok, 200, _, ref} = :hackney.post(url, [header], {:form, req_params})
    {:ok, body} = :hackney.body(ref)
    ids = for el <- :jsx.decode(body), do: :proplists.get_value("id_str", el)
    {:ok, ids}
  end

  def stream(follow, track) do
    url = "https://stream.twitter.com/1.1/statuses/filter.json"
    options = [recv_timeout: 120000]
    params = [
        {"follow", Enum.join(follow, ",")},
        {"track", Enum.join(track, ",")}
    ]
    {header, req_params} = auth_header("post", url, params)
    IO.puts("Starting stream...")
    #url = "http://localhost:8000"
    {:ok, 200, _, ref} = :hackney.post(url, [header], {:form, req_params}, options)
    stream_loop(ref)
    stream(follow, track)
  end

  def stream_loop(ref) do
    stream_hackney_response(ref)
    |> stream_into_lines
    |> Stream.each(&IO.puts(&1))
    |> stream_decode_json
    |> Stream.each(fn obj -> IO.puts(:jsx.encode([obj], [indent: 4])) end)
    |> Stream.run
  end

  def stream_hackney_response(ref) do
    Stream.resource(
        fn -> ref end,
        fn ref ->
            case :hackney.stream_body(ref) do
                {:ok, data} -> {data, ref}
                _ -> nil
            end
        end,
        fn ref -> :hackney.close(ref) end
    )
  end

  def stream_decode_json(enum) do
    decoder0 = fn x -> :jsx.decode(x, [:stream]) end
    Stream.transform(enum,
        decoder0,
        fn (data, decoder) ->
            {:incomplete, new_decoder} = decoder.(data)
            try do
                json = new_decoder.(:end_stream)
                {json, decoder0}
            rescue
                e in ArgumentError ->
                    {[], new_decoder}
            end
        end
    )
  end

  def stream_into_lines(enum) do
    Stream.transform(enum, "", fn (el, acc) ->
        chunks = String.split(acc <> el, ~r{[\r\n]+})
        if Enum.empty?(chunks) do
            {[], ""}
        else
            {result, [rest]} = Enum.split(chunks, -1)
            result = Enum.filter(result, fn s -> s != "" end)
            {result, rest}
        end
    end)
  end
end

oauth:sign/6 doesn't work with cyrillic tags. will have to reimplement it.
also, wtf is there no binary:trim/1 (strip)