metabase.upload.types

(ns metabase.upload.types
  (:refer-clojure :exclude [make-hierarchy])
  (:require
   [clojure.string :as str]
   [metabase.upload.parsing :as upload-parsing]
   [metabase.util :as u]
   [metabase.util.malli :as mu]
   [metabase.util.ordered-hierarchy :as ordered-hierarchy :refer [make-hierarchy]]))

(def h
  (make-hierarchy
   [::text
    [::varchar-255
     [::boolean ::*boolean-int*]
     [::float
      ;; A number value with a decimal separator, but a zero fractional component.
      [::*float-or-int*
       [::int
        ;; A value that could be legally parsed as either a boolean OR an integer
        ::*boolean-int*
        ::auto-incrementing-int-pk]]]
     [::datetime ::date]
     ::offset-datetime]]))

(def ^:private abstract->concrete
  {::*boolean-int*  ::boolean
   ::*float-or-int* ::float})

(def ^:private allowed-promotions
  {::int #{::float}})

(def ^:private column-type->coercible-value-types
  {::int #{::*float-or-int*}})

(defn- coerce?
  [column-type value-type]
  (contains? (column-type->coercible-value-types column-type) value-type))

(def value-types
  (ordered-hierarchy/sorted-tags h))

(def column-types
  (into #{} (remove abstract->concrete) value-types))

(defn- column-type?
  [value-type]
  (contains? column-types value-type))

(defn concretize
  [existing-type value-type]
  (cond
    ;; If the type is concrete, there is nothing to do.
    (column-type? value-type) value-type
    ;; If we know nothing about the value type, treat it as an arbitrary string.
    (nil? value-type) ::text
    ;; If configured, coerce the value to the existing type
    (coerce? existing-type value-type) existing-type
    ;; Otherwise, project it to its canonical concretion.
    :else (abstract->concrete value-type)))

(defn- with-parens
  {:style/indent [:form]}
  [number-regex]
  (re-pattern (str "(" number-regex ")|(\\(" number-regex "\\))")))

(defn- with-currency
  {:style/indent [:form]}
  [number-regex]
  ;; currency signs can be all over: $2, -$2, $-2, 2€
  (re-pattern (str upload-parsing/currency-regex "?\\s*-?"
                   upload-parsing/currency-regex "?"
                   number-regex
                   "\\s*" upload-parsing/currency-regex "?")))

(defn- int-regex
  [number-separators]
  (with-parens
   (with-currency
    (case number-separators
      ("." ".,") #"\d[\d,]*"
      ",." #"\d[\d.]*"
      ", " #"\d[\d \u00A0]*"
      ".’" #"\d[\d’]*"))))

(defn- float-or-int-regex
  [number-separators]
  (with-parens
   (with-currency
    (case number-separators
      ("." ".,") #"\d[\d,]*(\.0+)?"
      ",." #"\d[\d.]*(\,[0]+)?"
      ", " #"\d[\d \u00A0]*(\,[0.]+)?"
      ".’" #"\d[\d’]*(\.[0.]+)?"))))

(defn- float-regex
  [number-separators]
  (with-parens
   (with-currency
    (case number-separators
      ("." ".,") #"\d[\d,]*(\.\d+)?"
      ",." #"\d[\d.]*(\,[\d]+)?"
      ", " #"\d[\d \u00A0]*(\,[\d.]+)?"
      ".’" #"\d[\d’]*(\.[\d.]+)?"))))

(defmacro does-not-throw?
  [body]
  `(try
     ~body
     true
     (catch Throwable _e#
       false)))

(defn- date-string? [s]
  (does-not-throw? (upload-parsing/parse-local-date s)))

(defn- datetime-string? [s]
  (does-not-throw? (upload-parsing/parse-local-datetime s)))

(defn- offset-datetime-string? [s]
  (does-not-throw? (upload-parsing/parse-offset-datetime s)))

(defn- boolean-string? [s]
  (boolean (re-matches #"(?i)true|t|yes|y|1|false|f|no|n|0" s)))

(defn- boolean-int-string? [s]
  (contains? #{"0" "1"} s))

(defn- varchar-255? [s]
  (<= (count s) 255))

(defn- regex-matcher [regex]
  (fn [s]
    (boolean (re-matches regex s))))

(def ^:private non-inferable-types
  #{::auto-incrementing-int-pk})

(def ^:private type->check-schema
  (into [:map] (map #(vector % [:=> [:cat :string] :boolean])
                    (remove non-inferable-types value-types))))

(mu/defn- settings->type->check :- type->check-schema
  [{:keys [number-separators] :as _settings}]
  (let [int-string?   (regex-matcher (int-regex number-separators))
        float-or-int? (regex-matcher (float-or-int-regex number-separators))
        float-string? (regex-matcher (float-regex number-separators))]
    {::*boolean-int*   boolean-int-string?
     ::boolean         boolean-string?
     ::offset-datetime offset-datetime-string?
     ::date            date-string?
     ::datetime        datetime-string?
     ::int             int-string?
     ::*float-or-int*  float-or-int?
     ::float           float-string?
     ::varchar-255     varchar-255?
     ::text            (constantly true)}))

(defn- value->type
  [type->check value]
  (when-not (str/blank? value)
    (let [trimmed (str/trim value)]
      (->> (remove non-inferable-types value-types)
           (filter #((type->check %) trimmed))
           first))))

(defn- relax-type
  [type->check current-type value]
  (cond (nil? value) current-type
        (nil? current-type) (value->type type->check value)
        :else (let [trimmed (str/trim value)]
                (if (str/blank? trimmed)
                  current-type
                  (->> (cons current-type (ancestors h current-type))
                       (filter #((type->check %) trimmed))
                       first)))))

(defn type-relaxer
  [settings]
  (let [relax (partial relax-type (settings->type->check settings))]
    (fn [value-types row]
      ;; It's important to realize this lazy sequence, because otherwise we can build a huge stack and overflow.
      (vec (u/map-all relax value-types row)))))

(mu/defn column-types-from-rows :- [:sequential (into [:enum] column-types)]
  "Given the types of the existing columns (if there are any), and rows to be added, infer the best supporting types."
  [settings existing-types rows]
  (->> (reduce (type-relaxer settings) existing-types rows)
       (u/map-all concretize existing-types)))

(defn base-type->upload-type
  [base-type]
  (when base-type
    (condp #(isa? %2 %1) base-type
      :type/Float                  ::float
      :type/BigInteger             ::int
      :type/Integer                ::int
      :type/Boolean                ::boolean
      :type/DateTimeWithTZ         ::offset-datetime
      :type/DateTime               ::datetime
      :type/Date                   ::date
      :type/Text                   ::text)))

(defn- promotable?
  [current-type inferred-type]
  (when-let [allowed? (allowed-promotions current-type)]
    (allowed? inferred-type)))

(defn new-type
  [current-type inferred-type]
  (cond
    ;; No restriction on new columns
    (nil? current-type) inferred-type
    ;; No changes required if inferred type matches
    (= current-type inferred-type) current-type
    :else
    ;; Keep the existing type unless a promotion is allowed.
    (if (promotable? current-type inferred-type)
      inferred-type
      current-type)))