1 /***********************************************************************
2  * Copyright (c) 2013-2025 General Atomics Integrated Intelligence, Inc.
3  * All rights reserved. This program and the accompanying materials
4  * are made available under the terms of the Apache License, Version 2.0
5  * which accompanies this distribution and is available at
6  * https://www.apache.org/licenses/LICENSE-2.0
7  ***********************************************************************/
8 
9 package org.locationtech.geomesa.convert.text
10 
11 import com.typesafe.config.Config
12 import org.apache.commons.csv.CSVFormat
13 import org.apache.commons.io.IOUtils
14 import org.geotools.api.feature.simple.SimpleFeatureType
15 import org.locationtech.geomesa.convert.EvaluationContext
16 import org.locationtech.geomesa.convert.Modes.{ErrorMode, ParseMode}
17 import org.locationtech.geomesa.convert.text.DelimitedTextConverter._
18 import org.locationtech.geomesa.convert.text.DelimitedTextConverterFactory.{DelimitedTextConfigConvert, DelimitedTextOptionsConvert}
19 import org.locationtech.geomesa.convert2
20 import org.locationtech.geomesa.convert2.AbstractConverter.BasicField
21 import org.locationtech.geomesa.convert2.AbstractConverterFactory.{BasicFieldConvert, ConverterConfigConvert, ConverterOptionsConvert, PrimitiveConvert}
22 import org.locationtech.geomesa.convert2.TypeInference.{FunctionTransform, PathWithValues, TypeWithPath}
23 import org.locationtech.geomesa.convert2.transforms.Expression
24 import org.locationtech.geomesa.convert2.transforms.Expression.{LiteralNull, TryExpression}
25 import org.locationtech.geomesa.convert2.validators.SimpleFeatureValidator
26 import org.locationtech.geomesa.convert2.{AbstractConverterFactory, TypeInference}
27 import org.locationtech.geomesa.utils.geotools.{ObjectType, SimpleFeatureTypes}
28 import org.locationtech.geomesa.utils.io.WithClose
29 import org.locationtech.geomesa.utils.text.TextTools
30 import pureconfig.error.ConfigReaderFailures
31 import pureconfig.{ConfigObjectCursor, ConfigReader}
32 
33 import java.io.{InputStream, StringReader}
34 import java.nio.charset.{Charset, StandardCharsets}
35 import scala.util.{Failure, Try}
36 
37 class DelimitedTextConverterFactory
38     extends AbstractConverterFactory[DelimitedTextConverter, DelimitedTextConfig, BasicField, DelimitedTextOptions](
39       DelimitedTextConverterFactory.TypeToProcess, DelimitedTextConfigConvert, BasicFieldConvert, DelimitedTextOptionsConvert) {
40 
41   import scala.collection.JavaConverters._
42 
43   override def infer(
44       is: InputStream,
45       sft: Option[SimpleFeatureType],
46       path: Option[String]): Option[(SimpleFeatureType, Config)] =
47     infer(is, sft, path.map(EvaluationContext.inputFileParam).getOrElse(Map.empty)).toOption
48 
49   override def infer(
50       is: InputStream,
51       sft: Option[SimpleFeatureType],
52       hints: Map[String, AnyRef]): Try[(SimpleFeatureType, Config)] = {
53     val tryLines = Try {
54       val sampleSize = AbstractConverterFactory.inferSampleSize
55       IOUtils.lineIterator(is, StandardCharsets.UTF_8.displayName).asScala.take(sampleSize).toSeq
56     }
57     tryLines.flatMap { lines =>
58       // if only a single line, assume that it isn't actually delimited text
59       if (lines.lengthCompare(2) < 0) { Failure(new RuntimeException("Not enough lines in input data")) } else {
60         val attempts = Iterator.single(magic(lines, sft)) ++
61             DelimitedTextConverter.inferences.iterator.map(infer(_, lines, sft))
62         convert2.multiTry(attempts, new RuntimeException("Could not parse input as delimited text"))
63       }
64     }
65   }
66 
67   private def infer(
68       format: CSVFormat,
69       lines: Seq[String],
70       sft: Option[SimpleFeatureType]): Try[(SimpleFeatureType, Config)] = {
71     // : Seq[List[String]]
72     val rows = lines.flatMap { line =>
73       if (TextTools.isWhitespace(line)) { None } else {
74         Try(format.parse(new StringReader(line)).iterator().next.iterator.asScala.toList).toOption
75       }
76     }
77     val counts = rows.map(_.length).distinct
78     // try to verify that we actually have a delimited file
79     // ensure that some lines parsed, that there were at most 3 different col counts, and that there were at least 2 cols
80     if (counts.isEmpty || counts.lengthCompare(3) > 0 || counts.max < 2) {
81       val f = if (format == Formats.Tabs) { "tsv" } else if (format == Formats.Quoted) { "quoted csv" }  else { "csv" }
82       return Failure(new RuntimeException(s"Not a valid delimited text file using format: $f"))
83     }
84     Try {
85       val names = sft match {
86         case Some(s) =>
87           s.getAttributeDescriptors.asScala.map(_.getLocalName)
88 
89         case None =>
90           val values = rows.head.map(v => PathWithValues("", Seq(v)))
91           if (TypeInference.infer(values, Left("")).types.forall(_.inferredType.typed == ObjectType.STRING)) {
92             // assume the first row is headers
93             rows.head
94           } else {
95             Seq.empty
96           }
97       }
98       val nameIter = names.iterator
99       val pathsAndValues = Seq.tabulate(counts.max) { col =>
100         val values = rows.drop(1).map(vals => if (vals.lengthCompare(col) > 0) { vals(col) } else { null })
101         PathWithValues(if (nameIter.hasNext) { nameIter.next } else { "" }, values)
102       }
103       val inferred =
104         TypeInference.infer(pathsAndValues,
105           sft.toRight(s"inferred-${if (format == Formats.Tabs) { "tsv" } else { "csv" }}"))
106       val converterConfig = DelimitedTextConfig(typeToProcess, None, formats.find(_._2 == format).get._1,
107         Some(Expression("md5(string2bytes($0))")), Map.empty, Map.empty)
108 
109       val fields = {
110         var i = 0 // 0 is the whole record
111         inferred.types.map { case TypeWithPath(_, inferredType) =>
112           i += 1
113           BasicField(inferredType.name, Some(Expression(inferredType.transform(i))))
114         }
115       }
116 
117       val options = DelimitedTextOptions(None, CharNotSpecified, CharNotSpecified, None,
118         SimpleFeatureValidator.default, Seq.empty, ParseMode.Default, ErrorMode(), StandardCharsets.UTF_8)
119 
120       val config = configConvert.to(converterConfig)
121           .withFallback(fieldConvert.to(fields.toSeq))
122           .withFallback(optsConvert.to(options))
123           .toConfig
124 
125       (inferred.sft, config)
126     }
127   }
128 
129   private def magic(lines: Seq[String], sft: Option[SimpleFeatureType]): Try[(SimpleFeatureType, Config)] = {
130     if (!lines.head.startsWith("id,")) { Failure(new RuntimeException("Not a geomesa delimited export")) } else {
131       Try {
132         val spec = WithClose(Formats.QuotedMinimal.parse(new StringReader(lines.head.drop(3)))) { result =>
133           result.iterator().next.asScala.mkString(",")
134         }
135         val schema = SimpleFeatureTypes.createType("", spec)
136 
137         val converterConfig =
138           DelimitedTextConfig(typeToProcess, None, formats.find(_._2 == Formats.QuotedMinimal).get._1,
139             Some(Expression("$1")), Map.empty, Map.empty)
140 
141         var i = 1 // 0 is the whole record, 1 is the id
142         val fields = schema.getAttributeDescriptors.asScala.map { d =>
143           val bindings = ObjectType.selectType(d)
144           val transform = bindings.head match {
145             case ObjectType.STRING   => TypeInference.IdentityTransform
146             case ObjectType.INT      => TypeInference.CastToInt
147             case ObjectType.LONG     => TypeInference.CastToLong
148             case ObjectType.FLOAT    => TypeInference.CastToFloat
149             case ObjectType.DOUBLE   => TypeInference.CastToDouble
150             case ObjectType.BOOLEAN  => TypeInference.CastToBoolean
151             case ObjectType.DATE     => FunctionTransform("datetime(", ")")
152             case ObjectType.UUID     => TypeInference.IdentityTransform
153             case ObjectType.LIST     => FunctionTransform(s"parseList('${bindings(1)}',", ")")
154             case ObjectType.MAP      => FunctionTransform(s"parseMap('${bindings(1)}->${bindings(2)}',", ")")
155             case ObjectType.BYTES    => TypeInference.IdentityTransform
156             case ObjectType.GEOMETRY =>
157               bindings.drop(1).headOption.getOrElse(ObjectType.GEOMETRY) match {
158                 case ObjectType.POINT               => FunctionTransform("point(", ")")
159                 case ObjectType.LINESTRING          => FunctionTransform("linestring(", ")")
160                 case ObjectType.POLYGON             => FunctionTransform("polygon(", ")")
161                 case ObjectType.MULTIPOINT          => FunctionTransform("multipoint(", ")")
162                 case ObjectType.MULTILINESTRING     => FunctionTransform("multilinestring(", ")")
163                 case ObjectType.MULTIPOLYGON        => FunctionTransform("multipolygon(", ")")
164                 case ObjectType.GEOMETRY_COLLECTION => FunctionTransform("geometrycollection(", ")")
165                 case _                              => FunctionTransform("geometry(", ")")
166               }
167 
168             case _ => throw new IllegalStateException(s"Unexpected binding: $bindings")
169           }
170           i += 1
171           BasicField(d.getLocalName, Some(TryExpression(Expression(transform.apply(i)), LiteralNull)))
172         }
173 
174         val options = DelimitedTextOptions(Some(1), CharNotSpecified, CharNotSpecified, None,
175           SimpleFeatureValidator.default, Seq.empty, ParseMode.Default, ErrorMode(), StandardCharsets.UTF_8)
176 
177         val config = configConvert.to(converterConfig)
178             .withFallback(fieldConvert.to(fields.toSeq))
179             .withFallback(optsConvert.to(options))
180             .toConfig
181 
182         (sft.getOrElse(schema), config)
183       }
184     }
185   }
186 }
187 
188 object DelimitedTextConverterFactory {
189 
190   val TypeToProcess = "delimited-text"
191 
192   object DelimitedTextConfigConvert extends ConverterConfigConvert[DelimitedTextConfig] {
193 
194     override protected def decodeConfig(
195         cur: ConfigObjectCursor,
196         `type`: String,
197         idField: Option[Expression],
198         caches: Map[String, Config],
199         userData: Map[String, Expression]): Either[ConfigReaderFailures, DelimitedTextConfig] = {
200       for {
201         name   <- converterName(cur).right
202         format <- cur.atKey("format").right.flatMap(_.asString).right
203       } yield {
204         DelimitedTextConfig(`type`, name, format, idField, caches, userData)
205       }
206     }
207 
208     override protected def encodeConfig(
209         config: DelimitedTextConfig,
210         base: java.util.Map[String, AnyRef]): Unit = {
211       base.put("format", config.format)
212     }
213   }
214 
215   object DelimitedTextOptionsConvert extends ConverterOptionsConvert[DelimitedTextOptions] {
216     override protected def decodeOptions(
217         cur: ConfigObjectCursor,
218         validators: Seq[String],
219         reporters: Seq[Config],
220         parseMode: ParseMode,
221         errorMode: ErrorMode,
222         encoding: Charset): Either[ConfigReaderFailures, DelimitedTextOptions] = {
223       def option[T](key: String, reader: ConfigReader[T]): Either[ConfigReaderFailures, Option[T]] = {
224         val value = cur.atKeyOrUndefined(key)
225         if (value.isUndefined) { Right(None) } else { reader.from(value).right.map(Option.apply) }
226       }
227 
228       def optionalChar(key: String): Either[ConfigReaderFailures, OptionalChar] = {
229         val value = cur.atKeyOrUndefined(key)
230         if (value.isUndefined) { Right(CharNotSpecified) } else {
231           PrimitiveConvert.charConfigReader.from(value) match {
232             case Right(c) => Right(CharEnabled(c))
233             case Left(failures) =>
234               if (PrimitiveConvert.stringConfigReader.from(value).right.exists(_.isEmpty)) {
235                 Right(CharDisabled)
236               } else {
237                 Left(failures)
238               }
239           }
240         }
241       }
242 
243       for {
244         skipLines <- option("skip-lines", PrimitiveConvert.intConfigReader).right
245         quote     <- optionalChar("quote").right
246         escape    <- optionalChar("escape").right
247         delimiter <- option("delimiter", PrimitiveConvert.charConfigReader).right
248       } yield {
249         DelimitedTextOptions(skipLines, quote, escape, delimiter, validators, reporters, parseMode, errorMode, encoding)
250       }
251     }
252 
253     override protected def encodeOptions(options: DelimitedTextOptions, base: java.util.Map[String, AnyRef]): Unit = {
254       options.skipLines.foreach(o => base.put("skip-lines", Int.box(o)))
255       options.quote.foreach(o => base.put("quote", Char.box(o)))
256       options.escape.foreach(o => base.put("escape", Char.box(o)))
257       options.delimiter.foreach(o => base.put("delimiter", Char.box(o)))
258     }
259   }
260 }