# ============================================================================
#
# Copyright (C) 2007-2016 Conceptive Engineering bvba.
# www.conceptive.be / info@conceptive.be
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# * Neither the name of Conceptive Engineering nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
# ============================================================================
import logging
logger = logging.getLogger( 'camelot.core.files.storage' )
import six
from ..qt import Qt, QtGui
from camelot.core.conf import settings
from camelot.core.exception import UserException
from camelot.core.utils import ugettext
[docs]class StoredFile( object ):
"""Helper class for the File field type.
Stored file objects can be used within the GUI thread, as none of
its methods should block.
"""
def __init__( self, storage, name ):
"""
:param storage: the storage in which the file is stored
:param name: the key by which the file is known in the storage"""
self.storage = storage
self.name = name
@property
[docs] def verbose_name( self ):
"""The name of the file, as it is to be displayed in the GUI"""
return self.name
def __getstate__( self ):
"""Returns the key of the file. To support pickling stored files
in the database in a :class:`camelot.model.memento.Memento`
object"""
return dict( name = self.name )
def __unicode__( self ):
return self.verbose_name
[docs]class StoredImage( StoredFile ):
"""Helper class for the Image field type Class linking an image and the
location and filename where the image is stored"""
def __init__( self, storage, name ):
super(StoredImage, self).__init__( storage, name )
self._thumbnails = dict()
[docs] def checkout_image( self ):
"""Checkout the image from the storage, this function is only to be
used in the model thread.
:return: a QImage
"""
p = self.storage.checkout( self )
image = QtGui.QImage(p)
if image.isNull():
return QtGui.QImage(':/image_not_found.png')
else:
return image
[docs] def checkout_thumbnail( self, width, height ):
"""Checkout a thumbnail for this image from the storage, this function
is only to be used in the model thread
:param width: the requested width of the thumbnail
:return: a QImage
"""
key = (width, height)
try:
thumbnail_image = self._thumbnails[key]
return thumbnail_image
except KeyError:
pass
original_image = self.checkout_image()
thumbnail_image = original_image.scaled( width, height, Qt.KeepAspectRatio )
self._thumbnails[key] = thumbnail_image
return thumbnail_image
[docs]class Storage( object ):
"""Helper class that opens and saves StoredFile objects
The default implementation stores files in the settings.CAMELOT_MEDIA_ROOT
directory. The storage object should only be used within the model thread,
as all of it's methods might block.
The methods of this class don't verify if they are called on the model
thread, because these classes can be used server side or in a non-gui
script as well.
"""
def __init__( self, upload_to = '',
stored_file_implementation = StoredFile,
root = None ):
"""
:param upload_to: the sub directory in which to put files
:param stored_file_implementation: the subclass of StoredFile to be used when
checking out files from the storage
:param root: the root directory in which to put files, this may be a callable that
takes no arguments. if root is a callable, it will be called in the model thread
to get the actual root of the media store.
The actual files will be put in root + upload to. If None is given as root,
the settings.CAMELOT_MEDIA_ROOT will be taken as the root directory.
"""
self._root = root
self._subfolder = upload_to
self._upload_to = None
self.stored_file_implementation = stored_file_implementation
#
# don't do anything here that might reduce the startup time, like verifying the
# availability of the storage, since the path might be on a slow network share
#
@property
def upload_to(self):
if self._upload_to == None:
root = self._root or settings.CAMELOT_MEDIA_ROOT
import os
if six.callable( root ):
root = root()
self._upload_to = os.path.join( root, self._subfolder )
return self._upload_to
[docs] def available(self):
"""Verify if the storage is available
:return: True if the storage is available, False otherwise
"""
import os
try:
if not os.path.exists( self.upload_to ):
os.makedirs( self.upload_to )
return True
except Exception as e:
logger.warn( 'Could not access or create path %s, files will be unreachable' % self.upload_to, exc_info = e )
[docs] def writeable(self):
"""Verify if the storage is available and writeable
:return: True if the storage is writeable, False otherwise
"""
import os
if self.available():
return os.access(self.upload_to, os.W_OK)
[docs] def exists( self, name ):
"""True if a file exists given some name"""
if self.available():
import os
return os.path.exists( self.path( name ) )
return False
[docs] def list(self, prefix='*', suffix='*'):
"""Lists all files with a given prefix and or suffix available in this storage
:return: a iterator of StoredFile objects
"""
import glob
import os
return (StoredFile(self, os.path.basename(name) ) for name in glob.glob( os.path.join( self.upload_to, u'%s*%s'%(prefix, suffix) ) ) )
[docs] def path( self, name ):
"""The local filesystem path where the file can be opened using Python standard open"""
import os
return os.path.join( self.upload_to, name )
def _create_tempfile( self, suffix, prefix ):
import tempfile
# @todo suffix and prefix should be cleaned, because the user might be
# able to get directory separators in here or something related
try:
return tempfile.mkstemp( suffix = suffix, prefix = prefix, dir = self.upload_to, text = 'b' )
except EnvironmentError as e:
if not self.available():
raise UserException( text = ugettext('The directory %s does not exist')%(self.upload_to),
resolution = ugettext( 'Contact your system administrator' ) )
if not self.writeable():
raise UserException( text = ugettext('You have no write permissions for %s')%(self.upload_to),
resolution = ugettext( 'Contact your system administrator' ) )
raise UserException( text = ugettext('Unable to write file to %s')%(self.upload_to),
resolution = ugettext( 'Contact your system administrator' ),
detail = ugettext('OS Error number : %s \nError : %s \nPrefix : %s \nSuffix : %s')%( e.errno,
e.strerror,
prefix,
suffix ) )
[docs] def checkin( self, local_path, filename=None ):
"""Check the file pointed to by local_path into the storage, and
return a StoredFile
:param local_path: the path to the local file that needs to be checked in
:param filename: a hint for the filename to be given to the checked in file, if None
is given, the filename from the local path will be taken.
The stored file is not guaranteed to have the filename asked, since the
storage might not support this filename, or another file might be named
like that. In each case the storage will choose the filename.
"""
self.available()
import shutil
import os
to_path = os.path.join( self.upload_to, filename or os.path.basename( local_path ) )
if os.path.exists(to_path):
# only if the default to_path exists, we'll give it a new name
root, extension = os.path.splitext( filename or os.path.basename( local_path ) )
( handle, to_path ) = self._create_tempfile( extension, root )
os.close( handle )
logger.debug( u'copy file from %s to %s', local_path, to_path )
shutil.copy( local_path, to_path )
return self.stored_file_implementation( self, os.path.basename( to_path ) )
[docs] def checkin_stream( self, prefix, suffix, stream ):
"""Check the datastream in as a file into the storage
:param prefix: the prefix to use for generating a file name
:param suffix: the suffix to use for generating a filen name, eg '.png'
:return: a StoredFile
This method can also be used in combination with the StringIO module::
import StringIO
stream = StringIO.StringIO()
# write everything to the stream
stream.write( 'bla bla bla' )
# prepare the stream for reading
stream.seek( 0 )
stored_file = storage.checkin_stream( 'document', '.txt', stream )
"""
self.available()
import os
( handle, to_path ) = self._create_tempfile( suffix, prefix )
logger.debug(u'checkin stream to %s'%to_path)
file = os.fdopen( handle, 'wb' )
file.write( stream.read() )
file.flush()
file.close()
return self.stored_file_implementation( self, os.path.basename( to_path ) )
[docs] def checkout( self, stored_file ):
"""Check the file pointed to by the local_path out of the storage and return
a local filesystem path where the file can be opened"""
self.available()
import os
return os.path.join( self.upload_to, stored_file.name )
[docs] def checkout_stream( self, stored_file ):
"""Check the file stored_file out of the storage as a datastream
:return: a file object
"""
self.available()
import os
return open( os.path.join( self.upload_to, stored_file.name ), 'rb' )
def delete( self, name ):
pass