<commit_before>(ns akvo.lumen.lib.transformation.split-column
  (:require [akvo.lumen.lib.transformation.engine :as engine]
            [akvo.lumen.lib.dataset.utils :refer (find-column)]
            [akvo.lumen.util :as util]
            [akvo.lumen.db.transformation.engine :as db.tx.engine]
            [akvo.lumen.db.transformation :as db.transformation]
            [clojure.java.jdbc :as jdbc]
            [akvo.lumen.postgres :as postgres]
            [clojure.string :as string]
            [clojure.tools.logging :as log]
            [clojure.walk :as walk])
  (:import [java.util.regex Pattern]))

(defn pattern-analysis [re-pattern* column-values]
  (reduce
   (fn [store value]
     (let [freqs (frequencies (re-seq re-pattern* value))]
       (-> (reduce (fn [c [k v]]
                     (let [path [:analysis k]]
                       (-> c
                           (assoc-in path
                                     {:max-coincidences-in-one-row (max v (get-in c (conj path :max-coincidences-in-one-row) 0))
                                      :total-row-coincidences      (inc (get-in c (conj path :total-row-coincidences) 0))
                                      :total-column-coincidences   (+ v (get-in c (conj path :total-column-coincidences) 0))}))))
                   store freqs)
           (update :rows inc))))
   {:rows 0}
   column-values))

(defn selected-column [args]
  (-> args :selectedColumn))

(defn col-name [args]
  (-> (selected-column args) :columnName))

(defn pattern* [args]
  (-> args :pattern))

(defn new-column-name [args]
  (-> args :newColumnName))

(defmethod engine/valid? "core/split-column"
  [op-spec]
  (let [{:keys [onError op args] :as op-spec} (walk/keywordize-keys op-spec)]
    (and (util/valid-column-name? (col-name args))
         (pattern* args)
         (new-column-name args))))

(defn- add-name-to-new-columns
  [columns new-columns]
  (let [next-column-index (engine/next-column-index columns)
        indexes (map engine/derivation-column-name (iterate inc next-column-index))]
    (map #(assoc % :columnName %2 :id %2) new-columns indexes)))

(defn columns-to-extract [prefix number-new-rows selected-column columns]
  (let [selected-column (find-column (walk/keywordize-keys columns) (:columnName selected-column))
        base-column (dissoc selected-column :type :columnName)
        new-columns (map #(assoc base-column :title (str prefix "-" %) :type "text")
                         (range 1 (inc number-new-rows)))]
    (add-name-to-new-columns columns new-columns)))

(defn- update-row [conn table-name row-id vals-map]
  (let [r (string/join "," (doall (map (fn [[k v]]
                                         (str k "=" (postgres/adapt-string-value v))) vals-map)))
        sql (str  "update " table-name " SET "  r " where rnum=" row-id)]
    (log/debug :sql sql)
    (jdbc/execute! conn sql)))

(defn split [cell-value re-pattern* new-rows-count]
  (let [default-values (take new-rows-count (repeat ""))]
    (if cell-value
      (as-> (string/split cell-value re-pattern*) values
        (cond
          (nil? values)         default-values
          (empty? values)       default-values
          (== 1 (count values)) default-values
          :else                 (apply conj default-values (reverse values))))
      default-values)))

(defmethod engine/apply-operation "core/split-column"
  [{:keys [tenant-conn]} table-name columns op-spec]
  (jdbc/with-db-transaction [tenant-conn tenant-conn]
    (let [{:keys [onError op args]} (walk/keywordize-keys op-spec)
          column-name               (col-name args)
          pattern                   (pattern* args)<commit_msg>[#2486] Adapt split-column tx<commit_after>(ns akvo.lumen.lib.transformation.split-column
  (:require [akvo.lumen.lib.transformation.engine :as engine]
            [akvo.lumen.lib.dataset.utils :refer (find-column)]
            [akvo.lumen.util :as util]
            [akvo.lumen.db.transformation.engine :as db.tx.engine]
            [akvo.lumen.db.transformation :as db.transformation]
            [clojure.java.jdbc :as jdbc]
            [akvo.lumen.postgres :as postgres]
            [clojure.string :as string]
            [clojure.tools.logging :as log]
            [clojure.walk :as walk])
  (:import [java.util.regex Pattern]))

(defn pattern-analysis [re-pattern* column-values]
  (reduce
   (fn [store value]
     (let [freqs (frequencies (re-seq re-pattern* value))]
       (-> (reduce (fn [c [k v]]
                     (let [path [:analysis k]]
                       (-> c
                           (assoc-in path
                                     {:max-coincidences-in-one-row (max v (get-in c (conj path :max-coincidences-in-one-row) 0))
                                      :total-row-coincidences      (inc (get-in c (conj path :total-row-coincidences) 0))
                                      :total-column-coincidences   (+ v (get-in c (conj path :total-column-coincidences) 0))}))))
                   store freqs)
           (update :rows inc))))
   {:rows 0}
   column-values))

(defn selected-column [args]
  (-> args :selectedColumn))

(defn col-name [args]
  (-> (selected-column args) :columnName))

(defn pattern* [args]
  (-> args :pattern))

(defn new-column-name [args]
  (-> args :newColumnName))

(defmethod engine/valid? "core/split-column"
  [op-spec]
  (let [{:keys [onError op args] :as op-spec} (walk/keywordize-keys op-spec)]
    (and (util/valid-column-name? (col-name args))
         (pattern* args)
         (new-column-name args))))

(defn- add-name-to-new-columns
  [columns new-columns]
  (let [next-column-index (engine/next-column-index columns)
        indexes (map engine/derivation-column-name (iterate inc next-column-index))]
    (map #(assoc % :columnName %2 :id %2) new-columns indexes)))

(defn columns-to-extract [prefix number-new-rows selected-column columns]
  (let [selected-column (find-column (walk/keywordize-keys columns) (:columnName selected-column))
        base-column (dissoc selected-column :type :columnName :groupId :groupName)
        new-columns (map #(assoc base-column :title (str prefix "-" %) :type "text")
                         (range 1 (inc number-new-rows)))]
    (add-name-to-new-columns columns new-columns)))

(defn- update-row [conn table-name row-id vals-map]
  (let [r (string/join "," (doall (map (fn [[k v]]
                                         (str k "=" (postgres/adapt-string-value v))) vals-map)))
        sql (str  "update " table-name " SET "  r " where rnum=" row-id)]
    (log/debug :sql sql)
    (jdbc/execute! conn sql)))

(defn split [cell-value re-pattern* new-rows-count]
  (let [default-values (take new-rows-count (repeat ""))]
    (if cell-value
      (as-> (string/split cell-value re-pattern*) values
        (cond
          (nil? values)         default-values
          (empty? values)       default-values
          (== 1 (count values)) default-values
          :else                 (apply conj default-values (reverse values))))
      default-values)))

(defmethod engine/apply-operation "core/split-column"
  [{:keys [tenant-conn]} table-name columns op-spec]
  (jdbc/with-db-transaction [tenant-conn tenant-conn]
    (let [{:keys [onError op args]} (walk/keywordize-keys op-spec)
          column-name               (col-name args)
          pattern                   (pattern* args)
{'fraction_non_alphanumeric': 0.14653304442036835, 'fraction_numerical': 0.0027085590465872156, 'mean_word_length': 2.1790787774429616, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4151104', 'n_tokens_mistral': 2399, 'n_tokens_neox': 2332, 'n_words': 576}