Fix detection of compressed CSV files.
@@ -575,7 +575,7 @@ class KgtkReader(KgtkBase, ClosableIter[typing.List[str]]):
very_verbose: bool = attr.ib(validator=attr.validators.instance_of(bool), default=False)
COMPRESSED_FILE_EXTENSIONS: typing.List[str] = [ ".bz2", ".gz", ".lz4", ".xz" ]
- ALL_CSV_FILE_EXTENSIONS: typing.List[str] = [ ".csv" ] + [ ".csv" + x for x in [ ".bz2", ".gz", ".lz4", ".xz" ] ]
+ CSV_FILE_EXTENSION: str = ".csv"
@classmethod
def _default_options(
@@ -645,13 +645,17 @@ class KgtkReader(KgtkBase, ClosableIter[typing.List[str]]):
(options, value_options) = cls._default_options(options, value_options)
# Supply the default input_format:
- input_format: str
+ input_format: tr = KgtkReaderOptions.INPUT_FORMAT_KGTK
if options.input_format is not None:
input_format = options.input_format
- elif file_path is not None and file_path.suffix in cls.ALL_CSV_FILE_EXTENSIONS:
+ elif file_path is not None:
+ if len(file_path.suffixes) == 1:
+ if file_path.suffix == cls.CSV_FILE_EXTENSION:
input_format = KgtkReaderOptions.INPUT_FORMAT_CSV
- else:
- input_format = KgtkReaderOptions.INPUT_FORMAT_KGTK
+ elif len(file_path.suffixes) > 1:
+ if file_path.suffixes[-1] in cls.COMPRESSED_FILE_EXTENSIONS and file_path.suffixes[-2] == cls.CSV_FILE_EXTENSION:
+ input_format = KgtkReaderOptions.INPUT_FORMAT_CSV
+
if verbose:
print("input format: %s" % input_format, file=error_file, flush=True)
