Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 8 additions & 3 deletions Orange/widgets/data/owcsvimport.py
Original file line number Diff line number Diff line change
Expand Up @@ -1101,9 +1101,14 @@ def _open(path, mode, encoding=None):
arh = zipfile.ZipFile(path, 'r')
filelist = arh.infolist()
if len(filelist) == 1:
filename = filelist[0]
zinfo = arh.getinfo(filename)
f = arh.open(zinfo.filename, 'r')
f = arh.open(filelist[0], 'r')
# patch the f.close to also close the main archive file
f_close = f.close

def close_():
f_close()
arh.close()
f.close = close_
if 't' in mode:
f = io.TextIOWrapper(f, encoding=encoding)
return f
Expand Down
47 changes: 46 additions & 1 deletion Orange/widgets/data/tests/test_owcsvimport.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
# pylint: disable=no-self-use
# pylint: disable=no-self-use,protected-access
import unittest
from unittest import mock
from contextlib import ExitStack
Expand Down Expand Up @@ -306,6 +306,51 @@ class Dialect(csv.excel):
df = owcsvimport.load_csv(io.BytesIO(contents), opts)
assert_array_equal(df.values, np.array([[3.21, 3.37], [4.13, 1000.142]]))

def test_open_compressed(self):
content = 'abc'
for ext in ["txt", "gz", "bz2", "xz", "zip"]:
with named_file('', suffix=f".{ext}") as fname:
with _open_write(fname, "wt", encoding="ascii") as f:
f.write(content)
f.close()

with owcsvimport._open(fname, "rt", encoding="ascii") as f:
self.assertEqual(content, f.read())


def _open_write(path, mode, encoding=None):
# pylint: disable=import-outside-toplevel
if mode not in {'w', 'wb', 'wt'}:
raise ValueError('r')
_, ext = os.path.splitext(path)
ext = ext.lower()
if ext == ".gz":
import gzip
return gzip.open(path, mode, encoding=encoding)
elif ext == ".bz2":
import bz2
return bz2.open(path, mode, encoding=encoding)
elif ext == ".xz":
import lzma
return lzma.open(path, mode, encoding=encoding)
elif ext == ".zip":
import zipfile
arh = zipfile.ZipFile(path, 'w')
filename, _ = os.path.splitext(os.path.basename(path))
f = arh.open(filename, mode="w")
f_close = f.close
# patch the f.close to also close the main archive file

def close_():
f_close()
arh.close()
f.close = close_
if 't' in mode:
f = io.TextIOWrapper(f, encoding=encoding)
return f
else:
return open(path, mode, encoding=encoding)


if __name__ == "__main__":
unittest.main()