 ;;; TOOL METHODS
 
 (defn tool-run [this args]
-  (let [job (hadoop/default-jobconf this)
-        outpath (Path. (hadoop/job-path :pre :altcrawl))
+  (let [job (h/default-jobconf this)
+        outpath (Path. (h/job-path :pre :altcrawl))
         hdfs (FileSystem/get job)]
     (.setJobName job "pre.altcrawl")
-    (FileInputFormat/setInputPaths job (str (hadoop/job-path :seqfiles :altcrawl) "/*.jsr"))
+    (FileInputFormat/setInputPaths job (str (h/job-path :seqfiles :altcrawl) "/*.jsr"))
     (FileOutputFormat/setOutputPath job outpath)
     (.setInputFormat job TextInputFormat)
     (.delete hdfs outpath true)
     (.setMapperClass job (Class/forName "org.altlaw.jobs.pre.altcrawl_mapper"))
     (.setReducerClass job (Class/forName "org.altlaw.jobs.pre.altcrawl_reducer"))
     (.setMapOutputKeyClass job Text)
-    (DistributedCache/addCacheFile (URI. (hadoop/docid-path :seqfiles :ohm1)) job)
+    (DistributedCache/addCacheFile (URI. (h/docid-path :seqfiles :ohm1)) job)
     (JobClient/runJob job))
   0)

          " from " (:request_uri download))))


(defn my-map [download]
  (status-code-dispatch (decode-response-body download)))



;;; MAPPER METHODS

(defn load-jruby []
  (alter-var-root #'*ruby*
                  (fn [_] (.getEngineByName (ScriptEngineManager.) "jruby")))
  (alter-var-root #'*ruby-context*
                  (fn [_] (.getContext *ruby*)))
  (.eval *ruby* (str "require 'org/altlaw/extract/scrape/run_scraper'"))
  (.eval *ruby* (str "$runner = RunScraper.new")))

(defn mapper-configure [this job]
  (let [cached (DistributedCache/getLocalCacheFiles job)
        path (first (filter #(= (.getName %) "docids.tsv.gz") cached))]
    (when (nil? path) (throw (RuntimeException. "No docid.tsv.gz in DistributedCache.")))
    (info "Loading docid map from " path)
    (alter-var-root #'*docid-map*
                    (fn [_] (tsv/load-tsv-map (str path))))
    (load-jruby)))

(defn mapper-map [this wkey wvalue output reporter]
    (let [input (walk/keywordize-keys (read-json-string (str wvalue)))]
      (debug "Called map with file position " wkey
             " and value " (logstr input))
      (binding [*reporter* reporter
                *raw-input* (str wvalue)]
        (doseq [[key value] (my-map input)]
          (.collect output (Text. (str key)) (Text. value))))))




;;; Step 3 (for PDFs): dispatch on presence of Docid

(defn handle-pdf [download]
  (if-let [docid (ids/get-docid "altcrawl" (:request_uri download))]
    (do (debug "handle-pdf got docid " docid " for " (:request_uri download))
        (h/counter "PDF files" "with docid")
        (convert-pdf docid download))
    (do (warn "handle-pdf found no docid for " (:request_uri download))
        (h/counter "PDF files" "no docid"))))


;;; Step 2: dispatch on MIME type of the download
