<commit_before>(ns stravinski.core
  (:require [stravinski.streamer :as streamer]
            [riemann.client :as riemann])
  (:import
   [ java.util Properties ])
  (:gen-class))

(def stats-agent (agent 0))
(def errors? (promise))
(def success? (promise))

(defn load-config-file
  "this loads a config file from the classpath"
  [file-name]
  (let [file-reader (.. (Thread/currentThread)
                        (getContextClassLoader)
                        (getResourceAsStream file-name))
        props (Properties.)]
    (.load props file-reader)
    (into {} props)))

(def ^:dynamic *config* (load-config-file "creds.config"))

(defn assert-get
  "get a value from the config, otherwise throw an exception detailing the problem"
  [key-name]
  (or (get *config* key-name) 
      (throw (Exception. (format "please define %s in the resources/test.config file" key-name)))))

(def riemann-conn (atom nil))

(defn riemann-submit [tweet]
  (if-let [hashtags (:hashtags (:entities tweet))]
    (doseq [hashtag hashtags]
      (riemann.client/send-event @riemann-conn {:service "tweet.hashtag"
                                                :description (:text hashtag)
                                               })))

  (if-let [followers (:followers_count (:user tweet))]
    (riemann.client/send-event @riemann-conn {:service "tweet.followers"
                                              :metric followers }))

  (riemann.client/send-event @riemann-conn {:service "tweet.meta"
                                            :metric (:events_sent tweet)
                                            })

  (riemann.client/send-event @riemann-conn {:service "tweet.text"
                                            :description (:text tweet)
                                            :metric (count (:text tweet))
                                            })
  )
  

(defn start-feeding [f track-params]
  (let [creds-map {:app-consumer-key (assert-get "app.consumer.key")
                   :app-consumer-secret (assert-get "app.consumer.secret")
                   :user-access-token (assert-get "user.access.token")
                   :user-access-token-secret (assert-get "user.access.token.secret")
                   :filter {:track  (assert-get "twitter.filter.track")}}
        processor-fn (partial (fn [f to-deliver-err to-deliver-succ counter tweet]
                                (try 
                                  (send (var-get counter) inc)
                                  ((var-get f) (assoc  tweet :events_sent @(var-get counter)))
                                  (catch Exception e (deliver to-deliver-err (.getMessage e))))
                                (deliver to-deliver-succ tweet))
                              #'riemann-submit #'errors? #'success? #'stats-agent)]
    (if (nil? @riemann-conn)
      (reset! riemann-conn (riemann.client/tcp-client :host (assert-get "riemann.host")
                                                      :port (read-string (assert-get "riemann.port")))))
    (f processor-fn creds-map track-params)))
<commit_msg>Include processing of geo location<commit_after>(ns stravinski.core
  (:require [stravinski.streamer :as streamer]
            [riemann.client :as riemann]
            [clojure.string :as str])
  (:import
   [ java.util Properties ])
  (:gen-class))

(def stats-agent (agent 0))
(def errors? (promise))
(def success? (promise))

(defn mappify [mymap]
  (apply hash-map (interleave (map str (range (count mymap))) mymap)))

(defn dottify-reduce [root-map str-acc]
  (reduce-kv (fn [init k v]
               (let [new-str-acc (if (empty? str-acc)
                                   (name k)
                                   (str/join "." [str-acc (name k)]))]
                 (conj init (cond
                              (map? v) #(dottify-reduce v new-str-acc)
                              (vector? v) #(dottify-reduce (mappify v) new-str-acc)
                              :else {(keyword new-str-acc) (str v)})))) [] root-map))

(defn dottify
  ([root-map] (dottify [] [#(dottify-reduce root-map "")]))
  ([acc exec-calls]
   (if (empty? exec-calls)
     acc
     (let [new-buckets (mapcat (fn [f] (f)) exec-calls)]
       (recur (into acc (filter (complement fn?) new-buckets))
              (filter fn? new-buckets))))))


(defn load-config-file
  "this loads a config file from the classpath"
  [file-name]
  (let [file-reader (.. (Thread/currentThread)
                        (getContextClassLoader)
                        (getResourceAsStream file-name))
        props (Properties.)]
    (.load props file-reader)
    (into {} props)))

(def ^:dynamic *config* (load-config-file "creds.config"))

(defn assert-get
  "get a value from the config, otherwise throw an exception detailing the problem"
  [key-name]
  (or (get *config* key-name) 
      (throw (Exception. (format "please define %s in the resources/test.config file" key-name)))))

(def riemann-conn (atom nil))

(defn riemann-submit [tweet]
  (if-let [hashtags (:hashtags (:entities tweet))]
    (doseq [hashtag hashtags]
      (riemann.client/send-event @riemann-conn {:service "tweet.hashtag"
                                                :description (:text hashtag)
                                               })))

  (if-let [followers (:followers_count (:user tweet))]
    (riemann.client/send-event @riemann-conn {:service "tweet.followers"
                                              :metric followers }))

  (let [x (apply merge (-> tweet
                           (assoc :service "tweet.full")
                           (dissoc :entities)
                           (dissoc :extended_entities)
                           dottify
                           ))
        lx (if (empty? (:geo.coordinates.0 x))
             x
             (assoc x :location (str/join "," [(:geo.coordinates.0 x) (:geo.coordinates.1 x)])))]
    (riemann.client/send-event @riemann-conn lx))
)

(defn start-feeding [f track-params]
  (let [creds-map {:app-consumer-key (assert-get "app.consumer.key")
                   :app-consumer-secret (assert-get "app.consumer.secret")
                   :user-access-token (assert-get "user.access.token")
                   :user-access-token-secret (assert-get "user.access.token.secret")
                   :filter {:track  (assert-get "twitter.filter.track")}}
        processor-fn (partial (fn [f to-deliver-err to-deliver-succ counter tweet]
                                (try 
                                  (send (var-get counter) inc)
                                  ((var-get f) (assoc  tweet :events_sent @(var-get counter)))
                                  (catch Exception e (deliver to-deliver-err (.getMessage e))))
                                (deliver to-deliver-succ tweet))
                              #'riemann-submit #'errors? #'success? #'stats-agent)]
    (if (nil? @riemann-conn)
      (reset! riemann-conn (riemann.client/tcp-client :host (assert-get "riemann.host")
                                                      :port (read-string (assert-get "riemann.port")))))
    (binding [*out* *out*]
      (f processor-fn creds-map track-params))))

{'fraction_non_alphanumeric': 0.13276796885861253, 'fraction_numerical': 0.0006951202558042541, 'mean_word_length': 1.6654316413486476, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 34, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7425950', 'n_tokens_mistral': 2180, 'n_tokens_neox': 2032, 'n_words': 517}