mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-11-03 04:43:11 +08:00
Nicola Soranzo
0dfe73a54c
Remove the need to call `Binary.register_unsniffable_binary_ext()` for each unsniffable binary datatype. Fix https://github.com/galaxyproject/galaxy/issues/3441 , where the upload of files of a datatype defined in datatypes_conf.xml as subclass of an unsniffable binary datatype ended up with "The uploaded binary file contains inappropriate content" because it was not possible to register the subclassed datatype as unsniffable. Also remove unused `stop_err()` function in upload.py .
409 lines
19 KiB
Python
409 lines
19 KiB
Python
#!/usr/bin/env python
|
|
# Processes uploads from the user.
|
|
|
|
# WARNING: Changes in this tool (particularly as related to parsing) may need
|
|
# to be reflected in galaxy.web.controllers.tool_runner and galaxy.tools
|
|
from __future__ import print_function
|
|
|
|
import errno
|
|
import gzip
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import zipfile
|
|
from json import dumps, loads
|
|
|
|
from six.moves.urllib.request import urlopen
|
|
|
|
from galaxy import util
|
|
from galaxy.datatypes import sniff
|
|
from galaxy.datatypes.binary import Binary
|
|
from galaxy.datatypes.registry import Registry
|
|
from galaxy.util.checkers import (
|
|
check_binary,
|
|
check_bz2,
|
|
check_gzip,
|
|
check_html,
|
|
check_zip
|
|
)
|
|
|
|
if sys.version_info < (3, 3):
|
|
import bz2file as bz2
|
|
else:
|
|
import bz2
|
|
|
|
assert sys.version_info[:2] >= (2, 7)
|
|
|
|
|
|
def file_err(msg, dataset, json_file):
|
|
json_file.write(dumps(dict(type='dataset',
|
|
ext='data',
|
|
dataset_id=dataset.dataset_id,
|
|
stderr=msg)) + "\n")
|
|
# never remove a server-side upload
|
|
if dataset.type in ('server_dir', 'path_paste'):
|
|
return
|
|
try:
|
|
os.remove(dataset.path)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def safe_dict(d):
|
|
"""
|
|
Recursively clone json structure with UTF-8 dictionary keys
|
|
http://mellowmachines.com/blog/2009/06/exploding-dictionary-with-unicode-keys-as-python-arguments/
|
|
"""
|
|
if isinstance(d, dict):
|
|
return dict([(k.encode('utf-8'), safe_dict(v)) for k, v in d.items()])
|
|
elif isinstance(d, list):
|
|
return [safe_dict(x) for x in d]
|
|
else:
|
|
return d
|
|
|
|
|
|
def parse_outputs(args):
|
|
rval = {}
|
|
for arg in args:
|
|
id, files_path, path = arg.split(':', 2)
|
|
rval[int(id)] = (path, files_path)
|
|
return rval
|
|
|
|
|
|
def add_file(dataset, registry, json_file, output_path):
|
|
data_type = None
|
|
line_count = None
|
|
converted_path = None
|
|
stdout = None
|
|
link_data_only = dataset.get('link_data_only', 'copy_files')
|
|
run_as_real_user = in_place = dataset.get('in_place', True)
|
|
purge_source = dataset.get('purge_source', True)
|
|
# in_place is True if there is no external chmod in place,
|
|
# however there are other instances where modifications should not occur in_place:
|
|
# when a file is added from a directory on the local file system (ftp import folder or any other path).
|
|
if dataset.type in ('server_dir', 'path_paste', 'ftp_import'):
|
|
in_place = False
|
|
check_content = dataset.get('check_content' , True)
|
|
auto_decompress = dataset.get('auto_decompress', True)
|
|
try:
|
|
ext = dataset.file_type
|
|
except AttributeError:
|
|
file_err('Unable to process uploaded file, missing file_type parameter.', dataset, json_file)
|
|
return
|
|
|
|
if dataset.type == 'url':
|
|
try:
|
|
page = urlopen(dataset.path) # page will be .close()ed by sniff methods
|
|
temp_name = sniff.stream_to_file(page, prefix='url_paste', source_encoding=util.get_charset_from_http_headers(page.headers))
|
|
except Exception as e:
|
|
file_err('Unable to fetch %s\n%s' % (dataset.path, str(e)), dataset, json_file)
|
|
return
|
|
dataset.path = temp_name
|
|
# See if we have an empty file
|
|
if not os.path.exists(dataset.path):
|
|
file_err('Uploaded temporary file (%s) does not exist.' % dataset.path, dataset, json_file)
|
|
return
|
|
if not os.path.getsize(dataset.path) > 0:
|
|
file_err('The uploaded file is empty', dataset, json_file)
|
|
return
|
|
# Is dataset content supported sniffable binary?
|
|
if check_binary(dataset.path):
|
|
# Sniff the data type
|
|
guessed_ext = sniff.guess_ext(dataset.path, registry.sniff_order)
|
|
# Set data_type only if guessed_ext is a binary datatype
|
|
datatype = registry.get_datatype_by_extension(guessed_ext)
|
|
if isinstance(datatype, Binary):
|
|
data_type = guessed_ext
|
|
ext = guessed_ext
|
|
if not data_type:
|
|
root_datatype = registry.get_datatype_by_extension(dataset.file_type)
|
|
if getattr(root_datatype, 'compressed', False):
|
|
data_type = 'compressed archive'
|
|
ext = dataset.file_type
|
|
else:
|
|
# See if we have a gzipped file, which, if it passes our restrictions, we'll uncompress
|
|
is_gzipped, is_valid = check_gzip(dataset.path, check_content=check_content)
|
|
if is_gzipped and not is_valid:
|
|
file_err('The gzipped uploaded file contains inappropriate content', dataset, json_file)
|
|
return
|
|
elif is_gzipped and is_valid and auto_decompress:
|
|
if link_data_only == 'copy_files':
|
|
# We need to uncompress the temp_name file, but BAM files must remain compressed in the BGZF format
|
|
CHUNK_SIZE = 2 ** 20 # 1Mb
|
|
fd, uncompressed = tempfile.mkstemp(prefix='data_id_%s_upload_gunzip_' % dataset.dataset_id, dir=os.path.dirname(output_path), text=False)
|
|
gzipped_file = gzip.GzipFile(dataset.path, 'rb')
|
|
while 1:
|
|
try:
|
|
chunk = gzipped_file.read(CHUNK_SIZE)
|
|
except IOError:
|
|
os.close(fd)
|
|
os.remove(uncompressed)
|
|
file_err('Problem decompressing gzipped data', dataset, json_file)
|
|
return
|
|
if not chunk:
|
|
break
|
|
os.write(fd, chunk)
|
|
os.close(fd)
|
|
gzipped_file.close()
|
|
# Replace the gzipped file with the decompressed file if it's safe to do so
|
|
if not in_place:
|
|
dataset.path = uncompressed
|
|
else:
|
|
shutil.move(uncompressed, dataset.path)
|
|
os.chmod(dataset.path, 0o644)
|
|
dataset.name = dataset.name.rstrip('.gz')
|
|
data_type = 'gzip'
|
|
if not data_type:
|
|
# See if we have a bz2 file, much like gzip
|
|
is_bzipped, is_valid = check_bz2(dataset.path, check_content)
|
|
if is_bzipped and not is_valid:
|
|
file_err('The gzipped uploaded file contains inappropriate content', dataset, json_file)
|
|
return
|
|
elif is_bzipped and is_valid and auto_decompress:
|
|
if link_data_only == 'copy_files':
|
|
# We need to uncompress the temp_name file
|
|
CHUNK_SIZE = 2 ** 20 # 1Mb
|
|
fd, uncompressed = tempfile.mkstemp(prefix='data_id_%s_upload_bunzip2_' % dataset.dataset_id, dir=os.path.dirname(output_path), text=False)
|
|
bzipped_file = bz2.BZ2File(dataset.path, 'rb')
|
|
while 1:
|
|
try:
|
|
chunk = bzipped_file.read(CHUNK_SIZE)
|
|
except IOError:
|
|
os.close(fd)
|
|
os.remove(uncompressed)
|
|
file_err('Problem decompressing bz2 compressed data', dataset, json_file)
|
|
return
|
|
if not chunk:
|
|
break
|
|
os.write(fd, chunk)
|
|
os.close(fd)
|
|
bzipped_file.close()
|
|
# Replace the bzipped file with the decompressed file if it's safe to do so
|
|
if not in_place:
|
|
dataset.path = uncompressed
|
|
else:
|
|
shutil.move(uncompressed, dataset.path)
|
|
os.chmod(dataset.path, 0o644)
|
|
dataset.name = dataset.name.rstrip('.bz2')
|
|
data_type = 'bz2'
|
|
if not data_type:
|
|
# See if we have a zip archive
|
|
is_zipped = check_zip(dataset.path)
|
|
if is_zipped and auto_decompress:
|
|
if link_data_only == 'copy_files':
|
|
CHUNK_SIZE = 2 ** 20 # 1Mb
|
|
uncompressed = None
|
|
uncompressed_name = None
|
|
unzipped = False
|
|
z = zipfile.ZipFile(dataset.path)
|
|
for name in z.namelist():
|
|
if name.endswith('/'):
|
|
continue
|
|
if unzipped:
|
|
stdout = 'ZIP file contained more than one file, only the first file was added to Galaxy.'
|
|
break
|
|
fd, uncompressed = tempfile.mkstemp(prefix='data_id_%s_upload_zip_' % dataset.dataset_id, dir=os.path.dirname(output_path), text=False)
|
|
if sys.version_info[:2] >= (2, 6):
|
|
zipped_file = z.open(name)
|
|
while 1:
|
|
try:
|
|
chunk = zipped_file.read(CHUNK_SIZE)
|
|
except IOError:
|
|
os.close(fd)
|
|
os.remove(uncompressed)
|
|
file_err('Problem decompressing zipped data', dataset, json_file)
|
|
return
|
|
if not chunk:
|
|
break
|
|
os.write(fd, chunk)
|
|
os.close(fd)
|
|
zipped_file.close()
|
|
uncompressed_name = name
|
|
unzipped = True
|
|
else:
|
|
# python < 2.5 doesn't have a way to read members in chunks(!)
|
|
try:
|
|
with open(uncompressed, 'wb') as outfile:
|
|
outfile.write(z.read(name))
|
|
uncompressed_name = name
|
|
unzipped = True
|
|
except IOError:
|
|
os.close(fd)
|
|
os.remove(uncompressed)
|
|
file_err('Problem decompressing zipped data', dataset, json_file)
|
|
return
|
|
z.close()
|
|
# Replace the zipped file with the decompressed file if it's safe to do so
|
|
if uncompressed is not None:
|
|
if not in_place:
|
|
dataset.path = uncompressed
|
|
else:
|
|
shutil.move(uncompressed, dataset.path)
|
|
os.chmod(dataset.path, 0o644)
|
|
dataset.name = uncompressed_name
|
|
data_type = 'zip'
|
|
if not data_type:
|
|
if check_binary(dataset.path) or registry.is_extension_unsniffable_binary(dataset.file_type):
|
|
# We have a binary dataset, but it is not Bam, Sff or Pdf
|
|
data_type = 'binary'
|
|
parts = dataset.name.split(".")
|
|
if len(parts) > 1:
|
|
ext = parts[-1].strip().lower()
|
|
is_ext_unsniffable_binary = registry.is_extension_unsniffable_binary(ext)
|
|
if check_content and not is_ext_unsniffable_binary:
|
|
file_err('The uploaded binary file contains inappropriate content', dataset, json_file)
|
|
return
|
|
elif is_ext_unsniffable_binary and dataset.file_type != ext:
|
|
err_msg = "You must manually set the 'File Format' to '%s' when uploading %s files." % (ext, ext)
|
|
file_err(err_msg, dataset, json_file)
|
|
return
|
|
if not data_type:
|
|
# We must have a text file
|
|
if check_content and check_html(dataset.path):
|
|
file_err('The uploaded file contains inappropriate HTML content', dataset, json_file)
|
|
return
|
|
if data_type != 'binary':
|
|
if link_data_only == 'copy_files' and data_type not in ('gzip', 'bz2', 'zip'):
|
|
# Convert universal line endings to Posix line endings if to_posix_lines is True
|
|
# and the data is not binary or gzip-, bz2- or zip-compressed.
|
|
if dataset.to_posix_lines:
|
|
tmpdir = output_adjacent_tmpdir(output_path)
|
|
tmp_prefix = 'data_id_%s_convert_' % dataset.dataset_id
|
|
if dataset.space_to_tab:
|
|
line_count, converted_path = sniff.convert_newlines_sep2tabs(dataset.path, in_place=in_place, tmp_dir=tmpdir, tmp_prefix=tmp_prefix)
|
|
else:
|
|
line_count, converted_path = sniff.convert_newlines(dataset.path, in_place=in_place, tmp_dir=tmpdir, tmp_prefix=tmp_prefix)
|
|
if dataset.file_type == 'auto':
|
|
ext = sniff.guess_ext(dataset.path, registry.sniff_order)
|
|
else:
|
|
ext = dataset.file_type
|
|
data_type = ext
|
|
# Save job info for the framework
|
|
if ext == 'auto' and data_type == 'binary':
|
|
ext = 'data'
|
|
if ext == 'auto' and dataset.ext:
|
|
ext = dataset.ext
|
|
if ext == 'auto':
|
|
ext = 'data'
|
|
datatype = registry.get_datatype_by_extension(ext)
|
|
if dataset.type in ('server_dir', 'path_paste') and link_data_only == 'link_to_files':
|
|
# Never alter a file that will not be copied to Galaxy's local file store.
|
|
if datatype.dataset_content_needs_grooming(dataset.path):
|
|
err_msg = 'The uploaded files need grooming, so change your <b>Copy data into Galaxy?</b> selection to be ' + \
|
|
'<b>Copy files into Galaxy</b> instead of <b>Link to files without copying into Galaxy</b> so grooming can be performed.'
|
|
file_err(err_msg, dataset, json_file)
|
|
return
|
|
if link_data_only == 'copy_files' and converted_path:
|
|
# Move the dataset to its "real" path
|
|
try:
|
|
shutil.move(converted_path, output_path)
|
|
except OSError as e:
|
|
# We may not have permission to remove converted_path
|
|
if e.errno != errno.EACCES:
|
|
raise
|
|
elif link_data_only == 'copy_files':
|
|
if purge_source and not run_as_real_user:
|
|
# if the upload tool runs as a real user the real user
|
|
# can't move dataset.path as this path is owned by galaxy.
|
|
shutil.move(dataset.path, output_path)
|
|
else:
|
|
shutil.copy(dataset.path, output_path)
|
|
# Write the job info
|
|
stdout = stdout or 'uploaded %s file' % data_type
|
|
info = dict(type='dataset',
|
|
dataset_id=dataset.dataset_id,
|
|
ext=ext,
|
|
stdout=stdout,
|
|
name=dataset.name,
|
|
line_count=line_count)
|
|
if dataset.get('uuid', None) is not None:
|
|
info['uuid'] = dataset.get('uuid')
|
|
json_file.write(dumps(info) + "\n")
|
|
if link_data_only == 'copy_files' and datatype and datatype.dataset_content_needs_grooming(output_path):
|
|
# Groom the dataset content if necessary
|
|
datatype.groom_dataset_content(output_path)
|
|
|
|
|
|
def add_composite_file(dataset, json_file, output_path, files_path):
|
|
if dataset.composite_files:
|
|
os.mkdir(files_path)
|
|
for name, value in dataset.composite_files.items():
|
|
value = util.bunch.Bunch(**value)
|
|
if dataset.composite_file_paths[value.name] is None and not value.optional:
|
|
file_err('A required composite data file was not provided (%s)' % name, dataset, json_file)
|
|
break
|
|
elif dataset.composite_file_paths[value.name] is not None:
|
|
dp = dataset.composite_file_paths[value.name]['path']
|
|
isurl = dp.find('://') != -1 # todo fixme
|
|
if isurl:
|
|
try:
|
|
temp_name = sniff.stream_to_file(urlopen(dp), prefix='url_paste')
|
|
except Exception as e:
|
|
file_err('Unable to fetch %s\n%s' % (dp, str(e)), dataset, json_file)
|
|
return
|
|
dataset.path = temp_name
|
|
dp = temp_name
|
|
if not value.is_binary:
|
|
tmpdir = output_adjacent_tmpdir(output_path)
|
|
tmp_prefix = 'data_id_%s_convert_' % dataset.dataset_id
|
|
if dataset.composite_file_paths[value.name].get('space_to_tab', value.space_to_tab):
|
|
sniff.convert_newlines_sep2tabs(dp, tmp_dir=tmpdir, tmp_prefix=tmp_prefix)
|
|
else:
|
|
sniff.convert_newlines(dp, tmp_dir=tmpdir, tmp_prefix=tmp_prefix)
|
|
shutil.move(dp, os.path.join(files_path, name))
|
|
# Move the dataset to its "real" path
|
|
shutil.move(dataset.primary_file, output_path)
|
|
# Write the job info
|
|
info = dict(type='dataset',
|
|
dataset_id=dataset.dataset_id,
|
|
stdout='uploaded %s file' % dataset.file_type)
|
|
json_file.write(dumps(info) + "\n")
|
|
|
|
|
|
def output_adjacent_tmpdir(output_path):
|
|
""" For temp files that will ultimately be moved to output_path anyway
|
|
just create the file directly in output_path's directory so shutil.move
|
|
will work optimially.
|
|
"""
|
|
return os.path.dirname(output_path)
|
|
|
|
|
|
def __main__():
|
|
|
|
if len(sys.argv) < 4:
|
|
print('usage: upload.py <root> <datatypes_conf> <json paramfile> <output spec> ...', file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
output_paths = parse_outputs(sys.argv[4:])
|
|
json_file = open('galaxy.json', 'w')
|
|
|
|
registry = Registry()
|
|
registry.load_datatypes(root_dir=sys.argv[1], config=sys.argv[2])
|
|
|
|
for line in open(sys.argv[3], 'r'):
|
|
dataset = loads(line)
|
|
dataset = util.bunch.Bunch(**safe_dict(dataset))
|
|
try:
|
|
output_path = output_paths[int(dataset.dataset_id)][0]
|
|
except Exception:
|
|
print('Output path for dataset %s not found on command line' % dataset.dataset_id, file=sys.stderr)
|
|
sys.exit(1)
|
|
if dataset.type == 'composite':
|
|
files_path = output_paths[int(dataset.dataset_id)][1]
|
|
add_composite_file(dataset, json_file, output_path, files_path)
|
|
else:
|
|
add_file(dataset, registry, json_file, output_path)
|
|
|
|
# clean up paramfile
|
|
# TODO: this will not work when running as the actual user unless the
|
|
# parent directory is writable by the user.
|
|
try:
|
|
os.remove(sys.argv[3])
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
if __name__ == '__main__':
|
|
__main__()
|