Spaces:
Runtime error
Runtime error
"""Process uploads.""" | |
# pylint: disable=invalid-name, unused-import | |
from typing import Union | |
from pathlib import Path | |
import tempfile | |
import cchardet | |
from logzero import logger | |
def process_upload(upload: Union[tempfile._TemporaryFileWrapper, bytes]) -> str: | |
"""Process upload (fileobj or bytes(zip file: io.BytesIO further to zipfile.ZipFile)). | |
gr.inputs.File("file"): upload normal file | |
gr.inputs.File("bytes"): upload zip file | |
""" | |
if isinstance(upload, bytes): | |
logger.warning("Not implemented, yet, for zip file") | |
return "Not implemented, yet, for zip file" | |
try: | |
fpath = Path(upload.name) | |
except Exception as e: | |
logger.error("Path(upload.name) error: %s", e) | |
return str(e) | |
suffixes = [ | |
"", | |
".txt", | |
".text", | |
".md", | |
"tsv", | |
] | |
# check .txt .md ''(no suffix) | |
if fpath.suffix.lower() not in suffixes: | |
logger.warning('suffix: [%s] not in %s', fpath.suffix, suffixes) | |
# return "File type not supported, yet." | |
try: | |
data = Path(upload.name).read_bytes() | |
except Exception as e: | |
logger.error("Unable to read data from %s, errors: %s", fpath, e) | |
data = str(e).encode() | |
# no data, empty file, return "" | |
if not data: | |
logger.info("empty file: %s", upload.name) | |
return "" | |
encoding = cchardet.detect(data).get("encoding") | |
if encoding is not None: | |
try: | |
text = fpath.read_text(encoding=encoding) | |
except Exception as e: | |
logger.error("Unable to retrieve text, error: %s", e) | |
text = str(e) | |
# return f"{upload.name} {type(upload)}\n\n{text}" | |
# return f"{upload.name}\n{text}" | |
return text | |
# not able to cchardet: encoding is None, docx, pdf, epub, zip etc | |
logger.info("Trying docx...to be implemented") | |
# T ODO .docx .epub .mobi .pdf etc. | |
_ = Path(upload.name) | |
msg = f"binary file: {_.stem[:-8]}{_.suffix}" | |
logger.warning("%s", msg) | |
return msg | |
_ = ''' # colab gradio-file-inputs-upload.ipynb | |
# file_to_text/process_file | |
def zip_to_text(file_obj): | |
""" | |
# zf = zipfile.ZipFile('german-recipes-dataset.zip') | |
zf = file_obj | |
namelist = zipfile.ZipFile.namelist(zf); | |
# filename = zf.open(namelist[0]); | |
file_contents = [] | |
for filename in namelist: | |
with zf.open(filename) as fhandle: | |
file_contents.append(fhandle.read().decode()) | |
""" | |
# fileobj is <class 'tempfile._TemporaryFileWrapper'> | |
# gr.inputs.File("bytes") | |
if isinstance(file_obj, bytes): | |
data = file_obj.decode() | |
return f"{type(file_obj)}\n{dir(file_obj)}\n{data}" | |
# "file"/gr.inputs.File("file") file_obj.name: /tmp/READMEzm8hc5ze.md | |
data = Path(file_obj.name).read_bytes() | |
return f"{file_obj.name} {type(file_obj)}\n{dir(file_obj)} \n{data}" | |
# ''' | |