<commit_before>(ns fscrawler-tika-convert.fsworker
  (:gen-class)
  (:require [clojure.tools.logging :as logging]
            [clj-logging-config.log4j :as log-config])
  (:require [fscrawler-tika-convert.reap :as reap]
            [fscrawler-tika-convert.mqhelper :as mqhelper]
            [fscrawler-tika-convert.routing-key :as rk]
            [fscrawler-tika-convert.misc :as misc])
  (:require [langohr.basic :as lb]
            [langohr.shutdown :as lshutdown]
            [langohr.exchange  :as le]
            [langohr.core :as rmq]
            [langohr.queue :as lq]
            [langohr.channel :as lch]
            [langohr.consumers :as lcons])
  ;; (:require [me.raynes.fs :as fs])
  (:require [clojure.tools.cli :as cli])
  (:require [clojure.data.json :as json])
  (:require [clojure.string :as string])
  (:require [com.brainbot.stat :as stat]
            [com.brainbot.vfs :as vfs])
  (:require [clojure.stacktrace :as trace])
  (:require [com.brainbot.iniconfig :as ini])
  (:require [tika])
  (:import java.io.File)
  (:import [com.rabbitmq.client Address ConnectionFactory Connection Channel ShutdownListener])
  (:import [java.util.concurrent Executors]))


(defn get-permissions-for-entry
  [fs directory {relpath :relpath, {type :type} :stat, error :error :as entry}]
  (if (and (= type "file") (not error))
    (assoc entry "permissions" (vfs/get-permissions fs (vfs/join fs [directory relpath])))
    entry))


(defn get-permissions
  [fs directory entries]
  (map (partial get-permissions-for-entry fs directory)
       entries))

(defn entry-is-type?
  [type entry]
  (= (get-in entry [:stat :type]) type))

(def entry-is-directory? (partial entry-is-type? "directory"))
(def entry-is-file? (partial entry-is-type? "file"))


(defn handle-msg-get_permissions
  [fs ch {:keys [delivery-tag exchange routing-key] :as meta} ^bytes payload]
  (let [body (json/read-json (String. payload "UTF-8"))
        directory (:directory body)
        entries-with-permissions (get-permissions fs directory (:entries body))]

    (let [listdir-cmd (rk/routing-key-string-with-command routing-key "listdir")]
      (doseq [entry (filter (fn [entry]
                              (and (entry-is-directory? entry)
                                   (not (:error entry))))
                            entries-with-permissions)]
        (let [subdirectory-path (vfs/join fs [directory (:relpath entry)])]
          (println "publish listdir" subdirectory-path listdir-cmd)
          (lb/publish ch exchange listdir-cmd
                      (json/write-str {:path subdirectory-path})))))


    (lb/publish ch exchange (rk/routing-key-string-with-command routing-key "update_directory")
                (json/write-str (assoc body :entries entries-with-permissions)))

    (lb/ack ch delivery-tag)))


(defn handle-msg-listdir
  [fs ch {:keys [delivery-tag exchange routing-key] :as meta} ^bytes payload]
  (let [body (json/read-json (String. payload "UTF-8"))
        path (:path body)
        entries (vfs/cmd-listdir fs path)]
    (println "handle-msg-listdir" path)
    (lb/publish ch
                exchange
                (rk/routing-key-string-with-command routing-key "get_permissions")
                (json/write-str {:directory path :entries entries})))
  (lb/ack ch delivery-tag))


(defn publish-some-message
  [conn]
  (let [ch (lch/open conn)
        queue-name (misc/initialize-rabbitmq-structures ch "extract_content" "nextbot" "fscrawler:test")]
    (doseq [i (range 15)]
      (println "[main] Publishing...")
      (lb/publish ch "nextbot" queue-name "Hello!" :content-type "text/plain" :type "greetings.hi"))
    (lch/close ch)
    queue-name))


(def command->msg-handler
  {"listdir" handle-msg-listdir
   "get_permissions" handle-msg-get_permissions})


(defn build-handle-connection
  [filesystems]
  (fn [conn]
    (logging/info "initializing connection")
    (doseq [{:keys [fsid] :as fs} filesystems
            [command handle-msg] (seq command->msg-handler)]
      (mqhelper/channel-loop
       conn
       (fn [ch]
         (let [qname (misc/initialize-rabbitmq-structures ch command "nextbot" fsid)]
           (logging/info "starting consumer for" qname)
           (lb/qos ch 1)
           (lcons/subscribe ch qname (partial handle-msg fs))))))))


(defn worker-run-section
  [iniconfig section]
  (let [rmq-settings (misc/rmq-settings-from-config iniconfig)
        filesystems (vfs/make-filesystems-from-iniconfig iniconfig section)]<commit_msg>use simple message handlers in fscrawler<commit_after>(ns fscrawler-tika-convert.fsworker
  (:gen-class)
  (:require [clojure.tools.logging :as logging]
            [clj-logging-config.log4j :as log-config])
  (:require [fscrawler-tika-convert.reap :as reap]
            [fscrawler-tika-convert.mqhelper :as mqhelper]
            [fscrawler-tika-convert.routing-key :as rk]
            [fscrawler-tika-convert.misc :as misc])
  (:require [langohr.basic :as lb]
            [langohr.shutdown :as lshutdown]
            [langohr.exchange  :as le]
            [langohr.core :as rmq]
            [langohr.queue :as lq]
            [langohr.channel :as lch]
            [langohr.consumers :as lcons])
  ;; (:require [me.raynes.fs :as fs])
  (:require [clojure.tools.cli :as cli])
  (:require [clojure.string :as string])
  (:require [com.brainbot.stat :as stat]
            [com.brainbot.vfs :as vfs])
  (:require [clojure.stacktrace :as trace])
  (:require [com.brainbot.iniconfig :as ini])
  (:require [tika])
  (:import java.io.File)
  (:import [com.rabbitmq.client Address ConnectionFactory Connection Channel ShutdownListener])
  (:import [java.util.concurrent Executors]))


(defn get-permissions-for-entry
  [fs directory {relpath :relpath, {type :type} :stat, error :error :as entry}]
  (if (and (= type "file") (not error))
    (assoc entry "permissions" (vfs/get-permissions fs (vfs/join fs [directory relpath])))
    entry))


(defn get-permissions
  [fs directory entries]
  (map (partial get-permissions-for-entry fs directory)
       entries))

(defn entry-is-type?
  [type entry]
  (= (get-in entry [:stat :type]) type))

(def entry-is-directory? (partial entry-is-type? "directory"))
(def entry-is-file? (partial entry-is-type? "file"))


(defn simple-get_permissions
  [fs {:keys [directory entries] :as body} {publish :publish}]
  (let [entries-with-permissions (get-permissions fs directory entries)]
    (doseq [entry (filter (fn [entry]
                            (and (entry-is-directory? entry)
                                 (not (:error entry))))
                          entries-with-permissions)]
      (let [subdirectory-path (vfs/join fs [directory (:relpath entry)])]
        (publish "listdir" {:path subdirectory-path})))
    (publish "update_directory" (assoc body :entries entries-with-permissions))))


(defn simple-listdir
  [fs {path :path :as body} {publish :publish}]
  (publish "get_permissions" {:directory path
                              :entries (vfs/cmd-listdir fs path)}))





(defn publish-some-message
  [conn]
  (let [ch (lch/open conn)
        queue-name (misc/initialize-rabbitmq-structures ch "extract_content" "nextbot" "fscrawler:test")]
    (doseq [i (range 15)]
      (println "[main] Publishing...")
      (lb/publish ch "nextbot" queue-name "Hello!" :content-type "text/plain" :type "greetings.hi"))
    (lch/close ch)
    queue-name))


(def command->msg-handler
  {"listdir" simple-listdir
   "get_permissions" simple-get_permissions})


(defn build-handle-connection
  [filesystems]
  (fn [conn]
    (logging/info "initializing connection")
    (doseq [{:keys [fsid] :as fs} filesystems
            [command handle-msg] (seq command->msg-handler)]
      (mqhelper/channel-loop
       conn
       (fn [ch]
         (let [qname (misc/initialize-rabbitmq-structures ch command "nextbot" fsid)]
           (logging/info "starting consumer for" qname)
           (lb/qos ch 1)
           (lcons/subscribe ch qname (mqhelper/make-handler (partial handle-msg fs)))))))))


(defn worker-run-section
  [iniconfig section]
  (let [rmq-settings (misc/rmq-settings-from-config iniconfig)
        filesystems (vfs/make-filesystems-from-iniconfig iniconfig section)]
{'fraction_non_alphanumeric': 0.14139444173573867, 'fraction_numerical': 0.0012189176011701609, 'mean_word_length': 2.8812677388836327, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5615154', 'n_tokens_mistral': 2804, 'n_tokens_neox': 2671, 'n_words': 666}