<commit_before>(ns fowles.fetcher
  (:refer-clojure :exclude [partition])
  (:require ;; [com.keminglabs.zmq-async.core :refer [register-socket!]]
            [clojure.string :as str]
            [org.httpkit.client :as http]
            [clojure.core.async
             :refer [chan go partition map< split >! <!]
             :as async]))

;;-------------------------------------------

;; HTTP client

(def api-key "<KEY>")

(defn csv [strs]
  (str/join "," strs))

(defn hmap->query-string [hmap]
  (->> hmap
       clojure.walk/stringify-keys
       (map (fn [[k v]] (str k "=" v)))
       (str/join "&")))

;; https://developers.google.com/youtube/v3/docs/videos/list?hl=ca
(defn mk-video-uri [api-key video-ids parts]
  (let [args {:key api-key
              :id (csv video-ids)
              :part (csv parts)}]
    (str "https://www.googleapis.com/youtube/v3/videos?"
         (hmap->query-string args))))


;; Use `partition` to to create a channel
;; which will create collections of items from another channel.
;; Push individual video-ids onto A, partition them onto B.

;; Channels
;;   1. video-id   -> [video-id]                 (`partition`)
;;   2. [video-id] -> uri                        (`map<`)
;;   3. uri        -> response-promise           
;;   4. response-promise -> good-response | bad-response   (`split`)


(def IDS_PER_QUERY 2)
(defn query-count [seq]
  (/ (count seq) IDS_PER_QUERY))

(defn mk-pipeline-chan
  [api-key video-ids parts]
  (->> (async/to-chan video-ids)
       (async/partition IDS_PER_QUERY)
       (async/map< #(mk-video-uri api-key % parts))
       ;; http://http-kit.org/client.html
       (async/map< #(deref (http/get %)))))

;; this doesn't work
;; async/close!))

;; (doseq [vid-id video-ids]
;; (go (async/>! ch vid-id)))

(defn -main []
  (let [video-ids ["7lCDEYXw3mM" "MjtOzLfebgY" "6QIw1BQIvT4" "2xJWQPdG7jE"]
        parts     ["snippet" "contentDetails" "statistics" "status"]
        ch        (mk-pipeline-chan api-key video-ids parts)]
    (go (dotimes [_ (query-count video-ids)]
          (println (async/<! ch))))
    (Thread/sleep 10000)
))<commit_msg>using altsgit add src/fowles/fetcher.clj<commit_after>(ns fowles.fetcher
  (:refer-clojure :exclude [partition])
  (:require ;; [com.keminglabs.zmq-async.core :refer [register-socket!]]
            [clojure.string :as str]
            [org.httpkit.client :as http]
            [clojure.core.async
             :refer [chan go partition map< split >! <! take! alts!!]
             :as async]))

;;-------------------------------------------

;; HTTP client

(def api-key "<KEY>")

(defn csv [strs]
  (str/join "," strs))

(defn hmap->query-string [hmap]
  (->> hmap
       clojure.walk/stringify-keys
       (map (fn [[k v]] (str k "=" v)))
       (str/join "&")))

;; https://developers.google.com/youtube/v3/docs/videos/list?hl=ca
(defn mk-video-uri [api-key video-ids parts]
  (let [args {:key api-key
              :id (csv video-ids)
              :part (csv parts)}]
    (str "https://www.googleapis.com/youtube/v3/videos?"
         (hmap->query-string args))))


(def IDS_PER_QUERY 2)

;; Channels
;;   1. video-id   -> [video-id]                 (`partition`)
;;   2. [video-id] -> uri                        (`map<`)
;;   3. uri        -> response-promise           
;;   4. response-promise -> good-response | bad-response   (`split`)

(defn mk-pipeline-chan
  [api-key video-ids parts]
  (->> (async/to-chan video-ids)
       (async/partition IDS_PER_QUERY)
       (async/map< #(mk-video-uri api-key % parts))
       ;; http://http-kit.org/client.html
       (async/map< #(deref (http/get %)))))

(defn query-count [seq]
  (/ (count seq) IDS_PER_QUERY))

(defn -main []
  (let [video-ids ["7lCDEYXw3mM" "MjtOzLfebgY" "6QIw1BQIvT4" "2xJWQPdG7jE"]
        parts     ["snippet" "contentDetails" "statistics" "status"]
        ch        (mk-pipeline-chan api-key video-ids parts)]

    ;; (go (dotimes [_ (query-count video-ids)]
    ;;      (println (async/<! ch))))
    ;;      ;; (async/take! ch println))
    ;; (Thread/sleep 10000)

    (dotimes [_ (query-count video-ids)]
      (let [[v c] (async/alts!! [ch])]
        (println v)))

    (async/close! ch)
))
{'fraction_non_alphanumeric': 0.19771046983067017, 'fraction_numerical': 0.009062723586930598, 'mean_word_length': 2.191780821917808, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 28, 'https://': 4, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7112984', 'n_tokens_mistral': 1574, 'n_tokens_neox': 1508, 'n_words': 330}