| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319 |
- #!/usr/bin/env python
- # Licensed to Cloudera, Inc. under one
- # or more contributor license agreements. See the NOTICE file
- # distributed with this work for additional information
- # regarding copyright ownership. Cloudera, Inc. licenses this file
- # to you under the Apache License, Version 2.0 (the
- # "License"); you may not use this file except in compliance
- # with the License. You may obtain a copy of the License at
- #
- # http://www.apache.org/licenses/LICENSE-2.0
- #
- # Unless required by applicable law or agreed to in writing, software
- # distributed under the License is distributed on an "AS IS" BASIS,
- # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- # See the License for the specific language governing permissions and
- # limitations under the License.
- #
- # Implements simple file system browsing operations.
- #
- # Useful resources:
- # django/views/static.py manages django's internal directory index
- import errno
- import logging
- import mimetypes
- import operator
- import posixpath
- import re
- import shutil
- import stat as stat_module
- import os
- try:
- import json
- except ImportError:
- import simplejson as json
- from datetime import datetime
- from django.contrib import messages
- from django.contrib.auth.models import User, Group
- from django.core import urlresolvers
- from django.template.defaultfilters import stringformat, filesizeformat
- from django.http import Http404, HttpResponse, HttpResponseNotModified
- from django.views.decorators.http import require_http_methods
- from django.views.static import was_modified_since
- from django.utils.functional import curry
- from django.utils.http import http_date, urlquote
- from django.utils.html import escape
- from cStringIO import StringIO
- from gzip import GzipFile
- from avro import datafile, io
- from desktop.lib import i18n, paginator
- from desktop.lib.conf import coerce_bool
- from desktop.lib.django_util import make_absolute, render, render_json, format_preserving_redirect
- from desktop.lib.exceptions_renderable import PopupException
- from filebrowser.conf import MAX_SNAPPY_DECOMPRESSION_SIZE
- from filebrowser.lib.archives import archive_factory
- from filebrowser.lib.rwx import filetype, rwx
- from filebrowser.lib import xxd
- from filebrowser.forms import RenameForm, UploadFileForm, UploadArchiveForm, MkDirForm, EditorForm, TouchForm,\
- RenameFormSet, RmTreeFormSet, ChmodFormSet, ChownFormSet, CopyFormSet, RestoreFormSet,\
- TrashPurgeForm
- from hadoop.core_site import get_trash_interval
- from hadoop.fs.hadoopfs import Hdfs
- from hadoop.fs.exceptions import WebHdfsException
- from django.utils.translation import ugettext as _
- DEFAULT_CHUNK_SIZE_BYTES = 1024 * 4 # 4KB
- MAX_CHUNK_SIZE_BYTES = 1024 * 1024 # 1MB
- DOWNLOAD_CHUNK_SIZE = 64 * 1024 * 1024 # 64MB
- # Defaults for "xxd"-style output.
- # Sentences refer to groups of bytes printed together, within a line.
- BYTES_PER_LINE = 16
- BYTES_PER_SENTENCE = 2
- # The maximum size the file editor will allow you to edit
- MAX_FILEEDITOR_SIZE = 256 * 1024
- logger = logging.getLogger(__name__)
- def index(request):
- # Redirect to home directory by default
- path = request.user.get_home_directory()
- if not request.fs.isdir(path):
- path = '/'
- return view(request, path)
- def _file_reader(fh):
- """Generator that reads a file, chunk-by-chunk."""
- while True:
- chunk = fh.read(DOWNLOAD_CHUNK_SIZE)
- if chunk == '':
- fh.close()
- break
- yield chunk
- def download(request, path):
- """
- Downloads a file.
- This is inspired by django.views.static.serve.
- """
- if not request.fs.exists(path):
- raise Http404(_("File not found: %(path)s") % {'path': escape(path)})
- if not request.fs.isfile(path):
- raise PopupException(_("'%(path)s' is not a file") % {'path': path})
- mimetype = mimetypes.guess_type(path)[0] or 'application/octet-stream'
- stats = request.fs.stats(path)
- mtime = stats['mtime']
- size = stats['size']
- if not was_modified_since(request.META.get('HTTP_IF_MODIFIED_SINCE'), mtime, size):
- return HttpResponseNotModified()
- # TODO(philip): Ideally a with statement would protect from leaks,
- # but tricky to do here.
- fh = request.fs.open(path)
- response = HttpResponse(_file_reader(fh), mimetype=mimetype)
- response["Last-Modified"] = http_date(stats['mtime'])
- response["Content-Length"] = stats['size']
- response["Content-Disposition"] = "attachment"
- return response
- def view(request, path):
- """Dispatches viewing of a path to either index() or fileview(), depending on type."""
- # default_to_home is set in bootstrap.js
- if 'default_to_home' in request.GET:
- home_dir_path = request.user.get_home_directory()
- if request.fs.isdir(home_dir_path):
- return format_preserving_redirect(request, urlresolvers.reverse(view, kwargs=dict(path=home_dir_path)))
- # default_to_home is set in bootstrap.js
- if 'default_to_trash' in request.GET:
- if request.fs.isdir(request.fs.trash_path):
- return format_preserving_redirect(request, urlresolvers.reverse(view, kwargs=dict(path=request.fs.trash_path)))
- try:
- stats = request.fs.stats(path)
- if stats.isDir:
- return listdir_paged(request, path)
- else:
- return display(request, path)
- except (IOError, WebHdfsException), e:
- msg = _("Cannot access: %(path)s.") % {'path': escape(path)}
- if request.user.is_superuser and not request.user == request.fs.superuser:
- msg += _(' Note: you are a Hue admin but not a HDFS superuser (which is "%(superuser)s").') % {'superuser': request.fs.superuser}
- if request.is_ajax():
- exception = {
- 'error': msg
- }
- return render_json(exception)
- else:
- raise PopupException(msg , detail=e)
- def home_relative_view(request, path):
- home_dir_path = request.user.get_home_directory()
- if request.fs.exists(home_dir_path):
- path = '%s%s' % (home_dir_path, path)
- return view(request, path)
- def edit(request, path, form=None):
- """Shows an edit form for the given path. Path does not necessarily have to exist."""
- try:
- stats = request.fs.stats(path)
- except IOError, ioe:
- # A file not found is OK, otherwise re-raise
- if ioe.errno == errno.ENOENT:
- stats = None
- else:
- raise
- # Can't edit a directory
- if stats and stats['mode'] & stat_module.S_IFDIR:
- raise PopupException(_("Cannot edit a directory: %(path)s") % {'path': path})
- # Maximum size of edit
- if stats and stats['size'] > MAX_FILEEDITOR_SIZE:
- raise PopupException(_("File too big to edit: %(path)s") % {'path': path})
- if not form:
- encoding = request.REQUEST.get('encoding') or i18n.get_site_encoding()
- if stats:
- f = request.fs.open(path)
- try:
- try:
- current_contents = unicode(f.read(), encoding)
- except UnicodeDecodeError:
- raise PopupException(_("File is not encoded in %(encoding)s; cannot be edited: %(path)s") % {'encoding': encoding, 'path': path})
- finally:
- f.close()
- else:
- current_contents = u""
- form = EditorForm(dict(path=path, contents=current_contents, encoding=encoding))
- data = dict(
- exists=(stats is not None),
- form=form,
- path=path,
- filename=os.path.basename(path),
- dirname=os.path.dirname(path),
- breadcrumbs = parse_breadcrumbs(path))
- return render("edit.mako", request, data)
- def save_file(request):
- """
- The POST endpoint to save a file in the file editor.
- Does the save and then redirects back to the edit page.
- """
- form = EditorForm(request.POST)
- is_valid = form.is_valid()
- path = form.cleaned_data.get('path')
- if request.POST.get('save') == "Save As":
- if not is_valid:
- return edit(request, path, form=form)
- else:
- data = dict(form=form)
- return render("saveas.mako", request, data)
- if not path:
- raise PopupException("No path specified")
- if not is_valid:
- return edit(request, path, form=form)
- if request.fs.exists(path):
- _do_overwrite_save(request.fs, path,
- form.cleaned_data['contents'],
- form.cleaned_data['encoding'])
- else:
- _do_newfile_save(request.fs, path,
- form.cleaned_data['contents'],
- form.cleaned_data['encoding'])
- messages.info(request, _('Saved %(path)s.') % {'path': os.path.basename(path)})
- """ Changing path to reflect the request path of the JFrame that will actually be returned."""
- request.path = urlresolvers.reverse("filebrowser.views.edit", kwargs=dict(path=path))
- return edit(request, path, form)
- def _do_overwrite_save(fs, path, data, encoding):
- """
- Atomically (best-effort) save the specified data to the given path
- on the filesystem.
- TODO(todd) should this be in some fsutil.py?
- """
- # TODO(todd) Should probably do an advisory permissions check here to
- # see if we're likely to fail (eg make sure we own the file
- # and can write to the dir)
- # First write somewhat-kinda-atomically to a staging file
- # so that if we fail, we don't clobber the old one
- path_dest = path + "._hue_new"
- new_file = fs.open(path_dest, "w")
- try:
- try:
- new_file.write(data.encode(encoding))
- logging.info("Wrote to " + path_dest)
- finally:
- new_file.close()
- except Exception, e:
- # An error occurred in writing, we should clean up
- # the tmp file if it exists, before re-raising
- try:
- fs.remove(path_dest)
- except:
- pass
- raise e
- # Try to match the permissions and ownership of the old file
- cur_stats = fs.stats(path)
- try:
- fs.chmod(path_dest, stat_module.S_IMODE(cur_stats['mode']))
- except:
- logging.warn("Could not chmod new file %s to match old file %s" % (
- path_dest, path), exc_info=True)
- # but not the end of the world - keep going
- try:
- fs.chown(path_dest, cur_stats['user'], cur_stats['group'])
- except:
- logging.warn("Could not chown new file %s to match old file %s" % (
- path_dest, path), exc_info=True)
- # but not the end of the world - keep going
- # Now delete the old - nothing we can do here to recover
- fs.remove(path)
- # Now move the new one into place
- # If this fails, then we have no reason to assume
- # we can do anything to recover, since we know the
- # destination shouldn't already exist (we just deleted it above)
- fs.rename(path_dest, path)
- def _do_newfile_save(fs, path, data, encoding):
- """
- Save data to the path 'path' on the filesystem 'fs'.
- There must not be a pre-existing file at that path.
- """
- new_file = fs.open(path, "w")
- try:
- new_file.write(data.encode(encoding))
- finally:
- new_file.close()
- def parse_breadcrumbs(path):
- breadcrumbs_parts = Hdfs.normpath(path).split('/')
- i = 1
- breadcrumbs = [{'url': '', 'label': '/'}]
- while (i < len(breadcrumbs_parts)):
- breadcrumb_url = breadcrumbs[i - 1]['url'] + '/' + breadcrumbs_parts[i]
- if breadcrumb_url != '/':
- breadcrumbs.append({'url': breadcrumb_url, 'label': breadcrumbs_parts[i]})
- i = i + 1
- return breadcrumbs
- def listdir(request, path, chooser):
- """
- Implements directory listing (or index).
- Intended to be called via view().
- TODO: Remove?
- """
- if not request.fs.isdir(path):
- raise PopupException(_("Not a directory: %(path)s") % {'path': path})
- file_filter = request.REQUEST.get('file_filter', 'any')
- assert file_filter in ['any', 'file', 'dir']
- home_dir_path = request.user.get_home_directory()
- breadcrumbs = parse_breadcrumbs(path)
- data = {
- 'path': path,
- 'file_filter': file_filter,
- 'breadcrumbs': breadcrumbs,
- 'current_dir_path': path,
- # These could also be put in automatically via
- # http://docs.djangoproject.com/en/dev/ref/templates/api/#django-core-context-processors-request,
- # but manually seems cleaner, since we only need it here.
- 'current_request_path': request.path,
- 'home_directory': request.fs.isdir(home_dir_path) and home_dir_path or None,
- 'cwd_set': True,
- 'is_superuser': request.user.username == request.fs.superuser,
- 'groups': request.user.username == request.fs.superuser and [str(x) for x in Group.objects.values_list('name', flat=True)] or [],
- 'users': request.user.username == request.fs.superuser and [str(x) for x in User.objects.values_list('username', flat=True)] or [],
- 'superuser': request.fs.superuser,
- 'show_upload': (request.REQUEST.get('show_upload') == 'false' and (False,) or (True,))[0]
- }
- stats = request.fs.listdir_stats(path)
- # Include parent dir, unless at filesystem root.
- if Hdfs.normpath(path) != posixpath.sep:
- parent_path = request.fs.join(path, "..")
- parent_stat = request.fs.stats(parent_path)
- # The 'path' field would be absolute, but we want its basename to be
- # actually '..' for display purposes. Encode it since _massage_stats expects byte strings.
- parent_stat['path'] = parent_path
- stats.insert(0, parent_stat)
- data['files'] = [_massage_stats(request, stat) for stat in stats]
- if chooser:
- return render('chooser.mako', request, data)
- else:
- return render('listdir.mako', request, data)
- def _massage_page(page):
- return {
- 'number': page.number,
- 'num_pages': page.num_pages(),
- 'previous_page_number': page.previous_page_number(),
- 'next_page_number': page.next_page_number(),
- 'start_index': page.start_index(),
- 'end_index': page.end_index(),
- 'total_count': page.total_count()
- }
- def listdir_paged(request, path):
- """
- A paginated version of listdir.
- Query parameters:
- pagenum - The page number to show. Defaults to 1.
- pagesize - How many to show on a page. Defaults to 15.
- sortby=? - Specify attribute to sort by. Accepts:
- (type, name, atime, mtime, size, user, group)
- Defaults to name.
- descending - Specify a descending sort order.
- Default to false.
- filter=? - Specify a substring filter to search for in
- the filename field.
- """
- if not request.fs.isdir(path):
- raise PopupException("Not a directory: %s" % (path,))
- trash_enabled = get_trash_interval()
- pagenum = int(request.GET.get('pagenum', 1))
- pagesize = int(request.GET.get('pagesize', 30))
- home_dir_path = request.user.get_home_directory()
- breadcrumbs = parse_breadcrumbs(path)
- all_stats = request.fs.listdir_stats(path)
- # Filter first
- filter_str = request.GET.get('filter', None)
- if filter_str:
- filtered_stats = filter(lambda sb: filter_str in sb['name'], all_stats)
- all_stats = filtered_stats
- # Sort next
- sortby = request.GET.get('sortby', None)
- descending_param = request.GET.get('descending', None)
- if sortby is not None:
- if sortby not in ('type', 'name', 'atime', 'mtime', 'user', 'group', 'size'):
- logger.info("Invalid sort attribute '%s' for listdir." %
- (sortby,))
- else:
- all_stats = sorted(all_stats,
- key=operator.attrgetter(sortby),
- reverse=coerce_bool(descending_param))
- # Do pagination
- page = paginator.Paginator(all_stats, pagesize).page(pagenum)
- shown_stats = page.object_list
- # Include parent dir always as second option, unless at filesystem root.
- if Hdfs.normpath(path) != posixpath.sep:
- parent_path = request.fs.join(path, "..")
- parent_stat = request.fs.stats(parent_path)
- # The 'path' field would be absolute, but we want its basename to be
- # actually '..' for display purposes. Encode it since _massage_stats expects byte strings.
- parent_stat['path'] = parent_path
- parent_stat['name'] = ".."
- shown_stats.insert(0, parent_stat)
- # Include same dir always as first option to see stats of the current folder
- current_stat = request.fs.stats(path)
- # The 'path' field would be absolute, but we want its basename to be
- # actually '.' for display purposes. Encode it since _massage_stats expects byte strings.
- current_stat['path'] = path
- current_stat['name'] = "."
- shown_stats.insert(0, current_stat)
- page.object_list = [ _massage_stats(request, s) for s in shown_stats ]
- data = {
- 'path': path,
- 'breadcrumbs': breadcrumbs,
- 'current_request_path': request.path,
- 'files': page.object_list,
- 'page': _massage_page(page),
- 'pagesize': pagesize,
- 'home_directory': request.fs.isdir(home_dir_path) and home_dir_path or None,
- 'trash_enabled': trash_enabled,
- 'sortby': sortby,
- 'descending': descending_param,
- # The following should probably be deprecated
- 'cwd_set': True,
- 'file_filter': 'any',
- 'current_dir_path': path,
- 'is_fs_superuser': request.user.username == request.fs.superuser,
- 'is_superuser': request.user.username == request.fs.superuser,
- 'groups': request.user.username == request.fs.superuser and [str(x) for x in Group.objects.values_list('name', flat=True)] or [],
- 'users': request.user.username == request.fs.superuser and [str(x) for x in User.objects.values_list('username', flat=True)] or [],
- 'superuser': request.fs.superuser
- }
- return render('listdir.mako', request, data)
- def chooser(request, path):
- """
- Returns the html to JFrame that will display a file prompt.
- Dispatches viewing of a path to either index() or fileview(), depending on type.
- """
- # default_to_home is set in bootstrap.js
- home_dir_path = request.user.get_home_directory()
- if 'default_to_home' in request.GET and request.fs.isdir(home_dir_path):
- return listdir(request, home_dir_path, True)
- if request.fs.isdir(path):
- return listdir(request, path, True)
- elif request.fs.isfile(path):
- return display(request, path)
- else:
- raise Http404(_("File not found: %(path)s") % {'path': escape(path)})
- def _massage_stats(request, stats):
- """
- Massage a stats record as returned by the filesystem implementation
- into the format that the views would like it in.
- """
- path = stats['path']
- normalized = Hdfs.normpath(path)
- return {
- 'path': normalized,
- 'name': stats['name'],
- 'stats': stats.to_json_dict(),
- 'mtime': datetime.fromtimestamp(stats['mtime']).strftime('%B %d, %Y %I:%M %p'),
- 'humansize': filesizeformat(stats['size']),
- 'type': filetype(stats['mode']),
- 'rwx': rwx(stats['mode']),
- 'mode': stringformat(stats['mode'], "o"),
- 'url': make_absolute(request, "view", dict(path=urlquote(normalized))),
- }
- def stat(request, path):
- """
- Returns just the generic stats of a file.
- Intended for use via AJAX (and hence doesn't provide
- an HTML view).
- """
- if not request.fs.exists(path):
- raise Http404(_("File not found: %(path)s") % {'path': escape(path)})
- stats = request.fs.stats(path)
- return render_json(_massage_stats(request, stats))
- def display(request, path):
- """
- Implements displaying part of a file.
- GET arguments are length, offset, mode, compression and encoding
- with reasonable defaults chosen.
- Note that display by length and offset are on bytes, not on characters.
- TODO(philip): Could easily built-in file type detection
- (perhaps using something similar to file(1)), as well
- as more advanced binary-file viewing capability (de-serialize
- sequence files, decompress gzipped text files, etc.).
- There exists a python-magic package to interface with libmagic.
- """
- if not request.fs.isfile(path):
- raise PopupException(_("Not a file: '%(path)s'") % {'path': path})
- stats = request.fs.stats(path)
- encoding = request.GET.get('encoding') or i18n.get_site_encoding()
- # I'm mixing URL-based parameters and traditional
- # HTTP GET parameters, since URL-based parameters
- # can't naturally be optional.
- # Need to deal with possibility that length is not present
- # because the offset came in via the toolbar manual byte entry.
- end = request.GET.get("end")
- if end:
- end = int(end)
- begin = request.GET.get("begin", 1)
- if begin:
- # Subtract one to zero index for file read
- begin = int(begin) - 1
- if end:
- offset = begin
- length = end - begin
- if begin >= end:
- raise PopupException(_("First byte to display must be before last byte to display."))
- else:
- length = int(request.GET.get("length", DEFAULT_CHUNK_SIZE_BYTES))
- # Display first block by default.
- offset = int(request.GET.get("offset", 0))
- mode = request.GET.get("mode")
- compression = request.GET.get("compression")
- if mode and mode not in ["binary", "text"]:
- raise PopupException(_("Mode must be one of 'binary' or 'text'."))
- if offset < 0:
- raise PopupException(_("Offset may not be less than zero."))
- if length < 0:
- raise PopupException(_("Length may not be less than zero."))
- if length > MAX_CHUNK_SIZE_BYTES:
- raise PopupException(_("Cannot request chunks greater than %(bytes)d bytes") % {'bytes': MAX_CHUNK_SIZE_BYTES})
- # Do not decompress in binary mode.
- if mode == 'binary':
- compression = 'none'
- # Read out based on meta.
- compression, offset, length, contents =\
- read_contents(compression, path, request.fs, offset, length)
- # Get contents as string for text mode, or at least try
- uni_contents = None
- if not mode or mode == 'text':
- uni_contents = unicode(contents, encoding, errors='replace')
- is_binary = uni_contents.find(i18n.REPLACEMENT_CHAR) != -1
- # Auto-detect mode
- if not mode:
- mode = is_binary and 'binary' or 'text'
- # Get contents as bytes
- if mode == "binary":
- xxd_out = list(xxd.xxd(offset, contents, BYTES_PER_LINE, BYTES_PER_SENTENCE))
- dirname = posixpath.dirname(path)
- # Start with index-like data:
- data = _massage_stats(request, request.fs.stats(path))
- # And add a view structure:
- data["success"] = True
- data["view"] = {
- 'offset': offset,
- 'length': length,
- 'end': offset + len(contents),
- 'dirname': dirname,
- 'mode': mode,
- 'compression': compression,
- 'size': stats['size'],
- 'max_chunk_size': str(MAX_CHUNK_SIZE_BYTES)
- }
- data["filename"] = os.path.basename(path)
- data["editable"] = stats['size'] < MAX_FILEEDITOR_SIZE
- if mode == "binary":
- # This might be the wrong thing for ?format=json; doing the
- # xxd'ing in javascript might be more compact, or sending a less
- # intermediate representation...
- logger.debug("xxd: " + str(xxd_out))
- data['view']['xxd'] = xxd_out
- data['view']['masked_binary_data'] = False
- else:
- data['view']['contents'] = uni_contents
- data['view']['masked_binary_data'] = is_binary
- data['breadcrumbs'] = parse_breadcrumbs(path)
- return render("display.mako", request, data)
- def read_contents(codec_type, path, fs, offset, length):
- """
- Reads contents of a passed path, by appropriately decoding the data.
- Arguments:
- codec_type - The type of codec to use to decode. (Auto-detected if None).
- path - The path of the file to read.
- fs - The FileSystem instance to use to read.
- offset - Offset to seek to before read begins.
- length - Amount of bytes to read after offset.
- Returns: A tuple of codec_type, offset, length and contents read.
- """
- contents = ''
- try:
- fhandle = fs.open(path)
- stats = fs.stats(path)
- # Auto codec detection for [gzip, avro, none]
- # Only done when codec_type is unset
- contents = fhandle.read(3)
- if not codec_type:
- codec_type = 'none'
- if path.endswith('.gz') and detect_gzip(contents):
- codec_type = 'gzip'
- offset = 0
- elif path.endswith('.avro'):
- if detect_avro(contents):
- codec_type = 'avro'
- elif snappy_installed():
- if stats.size > MAX_SNAPPY_DECOMPRESSION_SIZE.get():
- raise PopupException(_('Failed to validate snappy compressed file. File size is greater than allowed max snappy decompression size of %d') % MAX_SNAPPY_DECOMPRESSION_SIZE.get())
- if detect_snappy(contents + fhandle.read()):
- codec_type = 'snappy_avro'
- fhandle.seek(0)
- if codec_type == 'avro' and snappy_installed() and detect_snappy(fhandle.read()):
- fhandle.seek(0)
- codec_type = 'snappy_avro'
- if codec_type == 'gzip':
- contents = _read_gzip(fhandle, path, offset, length, stats)
- elif codec_type == 'avro':
- contents = _read_avro(fhandle, path, offset, length, stats)
- elif codec_type == 'snappy_avro':
- contents = _read_snappy_avro(fhandle, path, offset, length, stats)
- else:
- # for 'none' type.
- contents = _read_simple(fhandle, path, offset, length, stats)
- finally:
- fhandle.close()
- return (codec_type, offset, length, contents)
- def _decompress_snappy(compressed_content):
- try:
- import snappy
- return snappy.decompress(compressed_content)
- except Exception, e:
- raise PopupException(_('Failed to decompress snappy compressed file.'), detail=e)
- def _read_snappy_avro(fhandle, path, offset, length, stats):
- if not snappy_installed():
- raise PopupException(_('Failed to decompress snappy compressed file. Snappy is not installed!'))
- if stats.size > MAX_SNAPPY_DECOMPRESSION_SIZE.get():
- raise PopupException(_('Failed to decompress snappy compressed file. File size is greater than allowed max snappy decompression size of %d') % MAX_SNAPPY_DECOMPRESSION_SIZE.get())
- return _read_avro(StringIO(_decompress_snappy(fhandle.read())), path, offset, length, stats)
- def _read_avro(fhandle, path, offset, length, stats):
- contents = ''
- try:
- fhandle.seek(offset)
- data_file_reader = datafile.DataFileReader(fhandle, io.DatumReader())
- contents_list = []
- read_start = fhandle.tell()
- # Iterate over the entire sought file.
- for datum in data_file_reader:
- read_length = fhandle.tell() - read_start
- if read_length > length and len(contents_list) > 0:
- break
- else:
- datum_str = str(datum) + "\n"
- contents_list.append(datum_str)
- data_file_reader.close()
- contents = "".join(contents_list)
- except:
- logging.warn("Could not read avro file at %s" % path, exc_info=True)
- raise PopupException(_("Failed to read Avro file."))
- return contents
- def _read_gzip(fhandle, path, offset, length, stats):
- contents = ''
- if offset and offset != 0:
- raise PopupException(_("Offsets are not supported with Gzip compression."))
- try:
- contents = GzipFile('', 'r', 0, StringIO(fhandle.read())).read(length)
- except:
- logging.warn("Could not decompress file at %s" % path, exc_info=True)
- raise PopupException(_("Failed to decompress file."))
- return contents
- def _read_simple(fhandle, path, offset, length, stats):
- contents = ''
- try:
- fhandle.seek(offset)
- contents = fhandle.read(length)
- except:
- logging.warn("Could not read file at %s" % path, exc_info=True)
- raise PopupException(_("Failed to read file."))
- return contents
- def detect_gzip(contents):
- '''This is a silly small function which checks to see if the file is Gzip'''
- return contents[:2] == '\x1f\x8b'
- def detect_avro(contents):
- '''This is a silly small function which checks to see if the file is Avro'''
- # Check if the first three bytes are 'O', 'b' and 'j'
- return contents[:3] == '\x4F\x62\x6A'
- def detect_snappy(contents):
- '''
- This is a silly small function which checks to see if the file is Snappy.
- It requires the entire contents of the compressed file.
- This will also return false if snappy decompression if we do not have the library available.
- '''
- try:
- import snappy
- return snappy.isValidCompressed(contents)
- except:
- return False
- def snappy_installed():
- '''Snappy is library that isn't supported by python2.4'''
- try:
- import snappy
- return True
- except:
- return False
- def _calculate_navigation(offset, length, size):
- """
- List of (offset, length, string) tuples for suggested navigation through the file.
- If offset is -1, then this option is already "selected". (Whereas None would
- be the natural pythonic way, Django's template syntax doesn't let us test
- against None (since its truth value is the same as 0).)
- By all means this logic ought to be in the template, but the template
- language is too limiting.
- """
- if offset == 0:
- first, prev = (-1, None, _("First Block")), (-1, None, _("Previous Block"))
- else:
- first, prev = (0, length, _("First Block")), (max(0, offset - length), length, _("Previous Block"))
- if offset + length >= size:
- next, last = (-1, None, _("Next Block")), (-1, None, _("Last Block"))
- else:
- # 1-off Reasoning: if length is the same as size, you want to start at 0.
- next, last = (offset + length, length, _("Next Block")), (max(0, size - length), length, _("Last Block"))
- return first, prev, next, last
- def default_initial_value_extractor(request, parameter_names):
- initial_values = {}
- for p in parameter_names:
- val = request.GET.get(p)
- if val:
- initial_values[p] = val
- return initial_values
- def formset_initial_value_extractor(request, parameter_names):
- """
- Builds a list of data that formsets should use by extending some fields to every object,
- whilst others are assumed to be received in order.
- Formsets should receive data that looks like this: [{'param1': <something>,...}, ...].
- The formsets should then handle construction on their own.
- """
- def _intial_value_extractor(request):
- if not submitted:
- return []
- # Build data with list of in order parameters receive in POST data
- # Size can be inferred from largest list returned in POST data
- data = []
- for param in submitted:
- i = 0
- for val in request.POST.getlist(param):
- if len(data) == i:
- data.append({})
- data[i][param] = val
- i += 1
- # Extend every data object with recurring params
- for kwargs in data:
- for recurrent in recurring:
- kwargs[recurrent] = request.POST.get(recurrent)
- initial_data = data
- return {'initial': initial_data}
- return _intial_value_extractor
- def default_arg_extractor(request, form, parameter_names):
- return [form.cleaned_data[p] for p in parameter_names]
- def formset_arg_extractor(request, formset, parameter_names):
- data = []
- for form in formset.forms:
- data_dict = {}
- for p in parameter_names:
- data_dict[p] = form.cleaned_data[p]
- data.append(data_dict)
- return data
- def default_data_extractor(request):
- return {'data': request.POST.copy()}
- def formset_data_extractor(recurring=[], submitted=[]):
- """
- Builds a list of data that formsets should use by extending some fields to every object,
- whilst others are assumed to be received in order.
- Formsets should receive data that looks like this: [{'param1': <something>,...}, ...].
- The formsets should then handle construction on their own.
- """
- def _data_extractor(request):
- if not submitted:
- return []
- # Build data with list of in order parameters receive in POST data
- # Size can be inferred from largest list returned in POST data
- data = []
- for param in submitted:
- i = 0
- for val in request.POST.getlist(param):
- if len(data) == i:
- data.append({})
- data[i][param] = val
- i += 1
- # Extend every data object with recurring params
- for kwargs in data:
- for recurrent in recurring:
- kwargs[recurrent] = request.POST.get(recurrent)
- initial = list(data)
- return {'initial': initial, 'data': data}
- return _data_extractor
- def generic_op(form_class, request, op, parameter_names, piggyback=None, template="fileop.mako", data_extractor=default_data_extractor, arg_extractor=default_arg_extractor, initial_value_extractor=default_initial_value_extractor, extra_params=None):
- """
- Generic implementation for several operations.
- @param form_class form to instantiate
- @param request incoming request, used for parameters
- @param op callable with the filesystem operation
- @param parameter_names list of form parameters that are extracted and then passed to op
- @param piggyback list of form parameters whose file stats to look up after the operation
- @param data_extractor function that extracts POST data to be used by op
- @param arg_extractor function that extracts args from a given form or formset
- @param initial_value_extractor function that extracts the initial values of a form or formset
- @param extra_params dictionary of extra parameters to send to the template for rendering
- """
- # Use next for non-ajax requests, when available.
- next = request.GET.get("next", request.POST.get("next", None))
- ret = dict({
- 'next': next
- })
- if extra_params is not None:
- ret['extra_params'] = extra_params
- for p in parameter_names:
- val = request.REQUEST.get(p)
- if val:
- ret[p] = val
- if request.method == 'POST':
- form = form_class(**data_extractor(request))
- ret['form'] = form
- if form.is_valid():
- args = arg_extractor(request, form, parameter_names)
- try:
- op(*args)
- except (IOError, WebHdfsException), e:
- msg = _("Cannot perform operation.")
- if request.user.is_superuser and not request.user == request.fs.superuser:
- msg += _(' Note: you are a Hue admin but not a HDFS superuser (which is "%(superuser)s").') \
- % {'superuser': request.fs.superuser}
- raise PopupException(msg, detail=e)
- if next:
- logging.debug("Next: %s" % next)
- # Doesn't need to be quoted: quoting is done by HttpResponseRedirect.
- return format_preserving_redirect(request, next)
- ret["success"] = True
- try:
- if piggyback:
- piggy_path = form.cleaned_data[piggyback]
- ret["result"] = _massage_stats(request, request.fs.stats(piggy_path))
- except Exception, e:
- # Hard to report these more naturally here. These happen either
- # because of a bug in the piggy-back code or because of a
- # race condition.
- logger.exception("Exception while processing piggyback data")
- ret["result_error"] = True
- ret['user'] = request.user
- return render(template, request, ret)
- else:
- # Initial parameters may be specified with get with the default extractor
- initial_values = initial_value_extractor(request, parameter_names)
- formset = form_class(initial=initial_values)
- ret['form'] = formset
- return render(template, request, ret)
- def rename(request):
- def smart_rename(src_path, dest_path):
- """If dest_path doesn't have a directory specified, use same dir."""
- if "#" in dest_path:
- raise PopupException(_("Could not rename folder \"%s\" to \"%s\": Hashes are not allowed in filenames." % (src_path, dest_path)))
- if "/" not in dest_path:
- src_dir = os.path.dirname(src_path)
- dest_path = os.path.join(src_dir, dest_path)
- request.fs.rename(src_path, dest_path)
- return generic_op(RenameForm, request, smart_rename, ["src_path", "dest_path"], None)
- def mkdir(request):
- def smart_mkdir(path, name):
- # Make sure only one directory is specified at a time.
- # No absolute directory specification allowed.
- if posixpath.sep in name or "#" in name:
- raise PopupException(_("Could not name folder \"%s\": Slashes or hashes are not allowed in filenames." % name))
- request.fs.mkdir(os.path.join(path, name))
- return generic_op(MkDirForm, request, smart_mkdir, ["path", "name"], "path")
- def touch(request):
- def smart_touch(path, name):
- # Make sure only the filename is specified.
- # No absolute path specification allowed.
- if posixpath.sep in name:
- raise PopupException(_("Could not name file \"%s\": Slashes are not allowed in filenames." % name))
- request.fs.create(os.path.join(path, name))
- return generic_op(TouchForm, request, smart_touch, ["path", "name"], "path")
- @require_http_methods(["POST"])
- def rmtree(request):
- recurring = []
- params = ["path"]
- def bulk_rmtree(*args, **kwargs):
- original = request.fs.setskiptrash('skip_trash' in request.GET)
- for arg in args:
- request.fs.do_as_user(request.user, request.fs.rmtree, arg['path'])
- request.fs.setskiptrash(original)
- return generic_op(RmTreeFormSet, request, bulk_rmtree, ["path"], None,
- data_extractor=formset_data_extractor(recurring, params),
- arg_extractor=formset_arg_extractor,
- initial_value_extractor=formset_initial_value_extractor)
- @require_http_methods(["POST"])
- def move(request):
- recurring = ['dest_path']
- params = ['src_path']
- def bulk_move(*args, **kwargs):
- for arg in args:
- request.fs.rename(arg['src_path'], arg['dest_path'])
- return generic_op(RenameFormSet, request, bulk_move, ["src_path", "dest_path"], None,
- data_extractor=formset_data_extractor(recurring, params),
- arg_extractor=formset_arg_extractor,
- initial_value_extractor=formset_initial_value_extractor)
- @require_http_methods(["POST"])
- def copy(request):
- recurring = ['dest_path']
- params = ['src_path']
- def bulk_copy(*args, **kwargs):
- for arg in args:
- request.fs.copy(arg['src_path'], arg['dest_path'], recursive=True, owner=request.user)
- return generic_op(CopyFormSet, request, bulk_copy, ["src_path", "dest_path"], None,
- data_extractor=formset_data_extractor(recurring, params),
- arg_extractor=formset_arg_extractor,
- initial_value_extractor=formset_initial_value_extractor)
- @require_http_methods(["POST"])
- def chmod(request):
- recurring = ["sticky", "user_read", "user_write", "user_execute", "group_read", "group_write", "group_execute", "other_read", "other_write", "other_execute"]
- params = ["path"]
- def bulk_chmod(*args, **kwargs):
- op = curry(request.fs.chmod, recursive=request.POST.get('recursive', False))
- for arg in args:
- op(arg['path'], arg['mode'])
- # mode here is abused: on input, it's a string, but when retrieved,
- # it's an int.
- return generic_op(ChmodFormSet, request, bulk_chmod, ['path', 'mode'], "path",
- data_extractor=formset_data_extractor(recurring, params),
- arg_extractor=formset_arg_extractor,
- initial_value_extractor=formset_initial_value_extractor)
- @require_http_methods(["POST"])
- def chown(request):
- # This is a bit clever: generic_op takes an argument (here, args), indicating
- # which POST parameters to pick out and pass to the given function.
- # We update that mapping based on whether or not the user selected "other".
- param_names = ["path", "user", "group"]
- if request.POST.get("user") == "__other__":
- param_names[1] = "user_other"
- if request.POST.get("group") == "__other__":
- param_names[2] = "group_other"
- recurring = ["user", "group", "user_other", "group_other"]
- params = ["path"]
- def bulk_chown(*args, **kwargs):
- op = curry(request.fs.chown, recursive=request.POST.get('recursive', False))
- for arg in args:
- varg = [arg[param] for param in param_names]
- op(*varg)
- return generic_op(ChownFormSet, request, bulk_chown, param_names, "path",
- data_extractor=formset_data_extractor(recurring, params),
- arg_extractor=formset_arg_extractor,
- initial_value_extractor=formset_initial_value_extractor)
- @require_http_methods(["POST"])
- def trash_restore(request):
- recurring = []
- params = ["path"]
- def bulk_restore(*args, **kwargs):
- for arg in args:
- request.fs.do_as_user(request.user, request.fs.restore, arg['path'])
- return generic_op(RestoreFormSet, request, bulk_restore, ["path"], None,
- data_extractor=formset_data_extractor(recurring, params),
- arg_extractor=formset_arg_extractor,
- initial_value_extractor=formset_initial_value_extractor)
- @require_http_methods(["POST"])
- def trash_purge(request):
- return generic_op(TrashPurgeForm, request, request.fs.purge_trash, [], None)
- def upload_file(request):
- """
- A wrapper around the actual upload view function to clean up the temporary file afterwards if it fails.
- Returns JSON.
- e.g. {'status' 0/1, data:'message'...}
- """
- response = {'status': -1, 'data': ''}
- if request.method == 'POST':
- try:
- resp = _upload_file(request)
- response.update(resp)
- except Exception, ex:
- response['data'] = str(ex)
- hdfs_file = request.FILES.get('hdfs_file')
- if hdfs_file:
- hdfs_file.remove()
- else:
- response['data'] = _('A POST request is required.')
- if response['status'] == 0:
- request.info(_('%(destination)s upload succeeded') % {'destination': response['path']})
- else:
- request.error(_('Upload failed: %(data)s') % {'data': response['data']})
- return HttpResponse(json.dumps(response), content_type="text/plain")
- def _upload_file(request):
- """
- Handles file uploaded by HDFSfileUploadHandler.
- The uploaded file is stored in HDFS at its destination with a .tmp suffix.
- We just need to rename it to the destination path.
- """
- form = UploadFileForm(request.POST, request.FILES)
- if form.is_valid():
- uploaded_file = request.FILES['hdfs_file']
- dest = form.cleaned_data['dest']
- if request.fs.isdir(dest) and posixpath.sep in uploaded_file.name:
- raise PopupException(_('Sorry, no "%(sep)s" in the filename %(name)s.' % {'sep': posixpath.sep, 'name': uploaded_file.name}))
- dest = request.fs.join(dest, uploaded_file.name)
- tmp_file = uploaded_file.get_temp_path()
- username = request.user.username
- try:
- # Remove tmp suffix of the file
- request.fs.do_as_user(username, request.fs.rename, tmp_file, dest)
- except IOError, ex:
- already_exists = False
- try:
- already_exists = request.fs.exists(dest)
- except Exception:
- pass
- if already_exists:
- msg = _('Destination %(name)s already exists.') % {'name': dest}
- else:
- msg = _('Copy to %(name)s failed: %(error)s') % {'name': dest, 'error': ex}
- raise PopupException(msg)
- return {
- 'status': 0,
- 'path': dest,
- 'result': _massage_stats(request, request.fs.stats(dest)),
- 'next': request.GET.get("next")
- }
- else:
- raise PopupException(_("Error in upload form: %s") % (form.errors,))
- def upload_archive(request):
- """
- A wrapper around the actual upload view function to clean up the temporary file afterwards.
- Returns JSON.
- e.g. {'status' 0/1, data:'message'...}
- """
- response = {'status': -1, 'data': ''}
- if request.method == 'POST':
- try:
- try:
- resp = _upload_archive(request)
- response.update(resp)
- except Exception, ex:
- response['data'] = str(ex)
- finally:
- hdfs_file = request.FILES.get('hdfs_file')
- if hdfs_file:
- hdfs_file.remove()
- else:
- response['data'] = _('A POST request is required.')
- if response['status'] == 0:
- request.info(_('%(destination)s upload succeeded') % {'destination': response['path']})
- else:
- request.error(_('Upload failed: %(data)s') % {'data': response['data']})
- return HttpResponse(json.dumps(response), content_type="text/plain")
- def _upload_archive(request):
- """
- Handles archive upload.
- The uploaded file is stored in memory.
- We need to extract it and rename it.
- """
- form = UploadArchiveForm(request.POST, request.FILES)
- if form.is_valid():
- uploaded_file = request.FILES['archive']
- # Always a dir
- if request.fs.isdir(form.cleaned_data['dest']) and posixpath.sep in uploaded_file.name:
- raise PopupException(_('Sorry, no "%(sep)s" in the filename %(name)s.' % {'sep': posixpath.sep, 'name': uploaded_file.name}))
- dest = request.fs.join(form.cleaned_data['dest'], uploaded_file.name)
- try:
- # Extract if necessary
- # Make sure dest path is without '.zip' extension
- if dest.endswith('.zip'):
- temp_path = archive_factory(uploaded_file).extract()
- if not temp_path:
- raise PopupException(_('Could not extract contents of file.'))
- # Move the file to where it belongs
- dest = dest[:-4]
- request.fs.copyFromLocal(temp_path, dest)
- shutil.rmtree(temp_path)
- else:
- raise PopupException(_('Could not interpret archive type.'))
- except IOError, ex:
- already_exists = False
- try:
- already_exists = request.fs.exists(dest)
- except Exception:
- pass
- if already_exists:
- msg = _('Destination %(name)s already exists.') % {'name': dest}
- else:
- msg = _('Copy to %(name)s failed: %(error)s') % {'name': dest, 'error': ex}
- raise PopupException(msg)
- return {
- 'status': 0,
- 'path': dest,
- 'result': _massage_stats(request, request.fs.stats(dest)),
- 'next': request.GET.get("next")
- }
- else:
- raise PopupException(_("Error in upload form: %s") % (form.errors,))
- def status(request):
- status = request.fs.status()
- data = {
- # Beware: "messages" is special in the context browser.
- 'msgs': status.get_messages(),
- 'health': status.get_health(),
- 'datanode_report': status.get_datanode_report(),
- 'name': request.fs.name
- }
- return render("status.mako", request, data)
- def location_to_url(location, strict=True):
- """
- If possible, returns a file browser URL to the location.
- Location is a URI, if strict is True.
- Python doesn't seem to have a readily-available URI-comparison
- library, so this is quite hacky.
- """
- if location is None:
- return None
- split_path = Hdfs.urlsplit(location)
- if strict and not split_path[1]:
- # No netloc, not full url
- return None
- return urlresolvers.reverse("filebrowser.views.view", kwargs=dict(path=split_path[2]))
- def truncate(toTruncate, charsToKeep=50):
- """
- Returns a string truncated to 'charsToKeep' length plus ellipses.
- """
- if len(toTruncate) > charsToKeep:
- truncated = toTruncate[:charsToKeep] + "..."
- return truncated
- else:
- return toTruncate
|