<commit_before>(ns akvo.lumen.lib.import.flow
  (:require [akvo.commons.psql-util :as pg]
            [akvo.lumen.lib.import.common :as import]
            [akvo.lumen.component.flow :as c.flow]
            [akvo.lumen.protocols :as p]
            [akvo.lumen.lib.import.flow-common :as flow-common]
            [akvo.lumen.lib.import.flow-v2 :as v2]
            [akvo.lumen.lib.import.flow-v3 :as v3]
            [akvo.lumen.lib.import.flow-v4 :as v4]
            [clojure.tools.logging :as log]))

(defn- read-flow-urls [flow-api]
  {:internal (:internal-url flow-api)
   :url (:url flow-api)})

(defn adapter [file version rows-cols col*]
  col*)

(defmethod import/dataset-importer "AKVO_FLOW"
  [{:strs [instance surveyId formId token email version] :as spec}
   {:keys [flow-api environment] :as config}]
  (let [version (if version version 1)
        headers-fn #((:internal-api-headers flow-api) {:email email :token token})
        survey (delay (flow-common/survey-definition (:internal-url flow-api)
                                                     headers-fn
                                                     instance
                                                     surveyId))]
    (reify
      java.io.Closeable
      (close [this])
      p/DatasetImporter
      (columns [this]
        (try
          (let [res (cond
                  (<= version 2) (v2/dataset-columns (flow-common/form @survey formId) environment)
                  (<= version 3) (v3/dataset-columns (flow-common/form @survey formId) environment))]
            (adapter (format "%s-%s-%s" instance surveyId formId) 3 :cols res))
          (catch Throwable e
            (if-let [ex-d (ex-data e)]
              (do
                (log/error e)
                (let [message (if (= 403 (:status ex-d))
                                (str "You don't have access to instance: " instance)
                                (or (:cause e) (str "Null cause from instance: " instance)))]
                  (throw (ex-info message
                                  (assoc ex-d
                                         :instance instance
                                         :flow-urls (read-flow-urls flow-api))))))
              (throw e)))))
      (records [this]
        (try
          (let [res (cond
                      (<= version 2) (v2/form-data headers-fn @survey formId)
                      (<= version 3) (v3/form-data headers-fn instance @survey formId))]
            (adapter (format "%s-%s-%s" instance surveyId formId) 3 :rows res))
          (catch Throwable e
            (if-let [ex-d (ex-data e)]
              (throw (ex-info (or (:cause e) (str "Null cause from instance: " instance))
                              (assoc ex-d
                                     :instance instance
                                     :flow-urls (read-flow-urls flow-api))))
              (throw e))))))))

(defmethod import/datagroups-importer "AKVO_FLOW"
  [{:strs [instance surveyId formId token email version] :as spec}
   {:keys [flow-api environment] :as config}]
  (let [headers-fn #((:internal-api-headers flow-api) {:email email :token token})
        survey (delay (flow-common/survey-definition (:internal-url flow-api)
                                                     headers-fn
                                                     instance
                                                     surveyId))]
    (reify
      java.io.Closeable
      (close [this])
      p/DatasetImporter
      (columns [this]
        (try
          (let [columns (v3/dataset-columns (flow-common/form @survey formId) environment)
                instance-id-column (first (filter #(= (:id %) "instance_id") columns))
                data-groups (dissoc (group-by #(select-keys % [:groupId :groupName]) columns)
                                    {:groupName "metadata" :groupId "metadata"})
                new-columns (for [[{:keys [groupId groupName]} _] data-groups]
                              (merge instance-id-column {:groupId groupId
                                                         :groupName groupName
                                                         :key false
                                                         :hidden true}))
                res (apply conj columns new-columns)]
            (adapter (format "%s-%s-%s" instance surveyId formId) 4 :cols res)
            )
          (catch Throwable e
            (if-let [ex-d (ex-data e)]
              (do
                (log/error e)
                (let [message (if (= 403 (:status ex-d))
                                (str "You don't have access to instance: " instance)
                                (or (:cause e) (str "Null cause from instance: " instance)))]
                  (throw (ex-info message
                                  (assoc ex-d
                                         :instance instance
                                         :flow-urls (read-flow-urls flow-api))))))
              (throw e)))))
      (records [this]
        (try
          (let [res (v4/form-data headers-fn instance @survey formId)]
            (adapter (format "%s-%s-%s" instance surveyId formId) 4 :rows res))
          (catch Throwable e
            (if-let [ex-d (ex-data e)]
              (throw (ex-info (or (:cause e) (str "Null cause from instance: " instance))
                              (assoc ex-d
                                     :instance instance
                                     :flow-urls (read-flow-urls flow-api))))
              (throw e))))))))<commit_msg>[#1450] Refactor<commit_after>(ns akvo.lumen.lib.import.flow
  (:require [akvo.commons.psql-util :as pg]
            [akvo.lumen.lib.import.common :as import]
            [akvo.lumen.component.flow :as c.flow]
            [akvo.lumen.protocols :as p]
            [akvo.lumen.lib.import.flow-common :as flow-common]
            [akvo.lumen.lib.import.flow-v2 :as v2]
            [akvo.lumen.lib.import.flow-v3 :as v3]
            [akvo.lumen.lib.import.flow-v4 :as v4]
            [clojure.tools.logging :as log]))

(defn- read-flow-urls [flow-api]
  {:internal (:internal-url flow-api)
   :url (:url flow-api)})

(defn adapter [{:keys [version rows-cols instance survey-id form-id]} data]
  data)

(defmethod import/dataset-importer "AKVO_FLOW"
  [{:strs [instance surveyId formId token email version] :as spec}
   {:keys [flow-api environment] :as config}]
  (let [version (if version version 1)
        headers-fn #((:internal-api-headers flow-api) {:email email :token token})
        survey (delay (flow-common/survey-definition (:internal-url flow-api)
                                                     headers-fn
                                                     instance
                                                     surveyId))]
    (reify
      java.io.Closeable
      (close [this])
      p/DatasetImporter
      (columns [this]
        (try
          (let [data (cond
                       (<= version 2) (v2/dataset-columns (flow-common/form @survey formId) environment)
                       (<= version 3) (v3/dataset-columns (flow-common/form @survey formId) environment))]
            (adapter {:version 3
                      :rows-cols :cols
                      :instance instance
                      :survey-id surveyId
                      :form-id formId} data))
          (catch Throwable e
            (if-let [ex-d (ex-data e)]
              (do
                (log/error e)
                (let [message (if (= 403 (:status ex-d))
                                (str "You don't have access to instance: " instance)
                                (or (:cause e) (str "Null cause from instance: " instance)))]
                  (throw (ex-info message
                                  (assoc ex-d
                                         :instance instance
                                         :flow-urls (read-flow-urls flow-api))))))
              (throw e)))))
      (records [this]
        (try
          (let [data (cond
                       (<= version 2) (v2/form-data headers-fn @survey formId)
                       (<= version 3) (v3/form-data headers-fn instance @survey formId))]
            (adapter {:version 3
                      :rows-cols :rows
                      :instance instance
                      :survey-id surveyId
                      :form-id formId} data))
          (catch Throwable e
            (if-let [ex-d (ex-data e)]
              (throw (ex-info (or (:cause e) (str "Null cause from instance: " instance))
                              (assoc ex-d
                                     :instance instance
                                     :flow-urls (read-flow-urls flow-api))))
              (throw e))))))))

(defmethod import/datagroups-importer "AKVO_FLOW"
  [{:strs [instance surveyId formId token email version] :as spec}
   {:keys [flow-api environment] :as config}]
  (let [headers-fn #((:internal-api-headers flow-api) {:email email :token token})
        survey (delay (flow-common/survey-definition (:internal-url flow-api)
                                                     headers-fn
                                                     instance
                                                     surveyId))]
    (reify
      java.io.Closeable
      (close [this])
      p/DatasetImporter
      (columns [this]
        (try
          (let [columns (v3/dataset-columns (flow-common/form @survey formId) environment)
                instance-id-column (first (filter #(= (:id %) "instance_id") columns))
                data-groups (dissoc (group-by #(select-keys % [:groupId :groupName]) columns)
                                    {:groupName "metadata" :groupId "metadata"})
                new-columns (for [[{:keys [groupId groupName]} _] data-groups]
                              (merge instance-id-column {:groupId groupId
                                                         :groupName groupName
                                                         :key false
                                                         :hidden true}))
                data (apply conj columns new-columns)]
            (adapter {:version 4
                      :rows-cols :cols
                      :instance instance
                      :survey-id surveyId
                      :form-id formId} data))
          (catch Throwable e
            (if-let [ex-d (ex-data e)]
              (do
                (log/error e)
                (let [message (if (= 403 (:status ex-d))
                                (str "You don't have access to instance: " instance)
                                (or (:cause e) (str "Null cause from instance: " instance)))]
                  (throw (ex-info message
                                  (assoc ex-d
                                         :instance instance
                                         :flow-urls (read-flow-urls flow-api))))))
              (throw e)))))
      (records [this]
        (try
          (let [data (v4/form-data headers-fn instance @survey formId)]
              (adapter {:version 4
                      :rows-cols :rows
                      :instance instance
                      :survey-id surveyId
                      :form-id formId} data))
          (catch Throwable e
            (if-let [ex-d (ex-data e)]
              (throw (ex-info (or (:cause e) (str "Null cause from instance: " instance))
                              (assoc ex-d
                                     :instance instance
                                     :flow-urls (read-flow-urls flow-api))))
              (throw e))))))))
{'fraction_non_alphanumeric': 0.11206159110350727, 'fraction_numerical': 0.004961505560307955, 'mean_word_length': 1.180343155538978, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 74, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2355078', 'n_tokens_mistral': 3089, 'n_tokens_neox': 2951, 'n_words': 823}