Source code for publiforge.lib.packutils

"""Some various utilities for packs."""
# pylint: disable = C0302

from os import walk, sep
from os.path import exists, join, isfile, dirname, relpath, basename, normpath
from os.path import splitext
import tempfile
import zipfile

from lxml import etree
from sqlalchemy import desc
from sqlalchemy.exc import IntegrityError

from pyramid.httpexceptions import HTTPNotFound

from .i18n import _
from .utils import EXCLUDED_FILES, normalize_name, normalize_spaces, decrypt
from .utils import has_permission
from .viewutils import file_download, current_project, task_auto_build
from .xml import PUBLIFORGE_RNG_VERSION, load_xml
from .paging import Paging
from ..models import LABEL_LEN, DBSession
from ..models.users import User
from ..models.roles import Role
from ..models.storages import Storage, StorageUser
from ..models.projects import Project
from ..models.processings import Processing
from ..models.tasks import Task
from ..models.packs import Pack, PackFile, PackEvent

# =============================================================================
[docs]def create_pack(request, filenames, path='.'): """Create a new pack with selected files. :type request: :class:`pyramid.request.Request` :param request: Current request. :param list filenames: Names of files to add to a new pack. :param path: (string, optional) Common path. :rtype: tuple A tuple such as ``(project_id, pack_id``) or ``(None, None)`` if failed. """ label = ', '.join([splitext(basename(k))[0] for k in filenames])[ 0:LABEL_LEN] project_id = request.session['project']['project_id'] if DBSession.query(Pack) \ .filter_by(project_id=project_id, label=label).first(): request.session.flash(_('This pack already exists.'), 'alert') return None, None pack = Pack(project_id, label) for name in filenames: pack.files.append( PackFile('file', normpath(join(path, name)))) DBSession.add(pack) try: DBSession.commit() except IntegrityError: request.session.flash(_('This pack already exists.'), 'alert') return None, None return pack.project_id, pack.pack_id
# =============================================================================
[docs]def pack2task(request, pack, link_type, target_task_id): """Move pack ``pack`` to task with ID ``target_task_id``. :param request: (:class:`pyramid.request.Request` instance) Current request. :param pack: (:class:`~..models.packs.Pack` instance) Pack object. :param link_type: (string) Type of link: ``normal``, ``back``, ``redo`` or ``kept``. :param target_task_id: (integer) Task ID. If ``None``, the first non ``auto`` task is used. """ # Find the new task and the new operator target_task_id, operator_type, operator_id = operator4task( pack, link_type, target_task_id) if not target_task_id: return task = DBSession.query(Task).filter_by( project_id=pack.project_id, task_id=target_task_id).first() if task is None: return # Move pack to task old_pack = (pack.task_id, pack.operator_type, pack.operator_id) pack.task_id = task.task_id pack.operator_type = operator_type \ if operator_type is not None else task.operator_type pack.operator_id = operator_id \ if operator_type is not None else task.operator_id # Add event project = current_project(request) operator = operator_label( request, project, pack.operator_type, pack.operator_id) event = PackEvent( pack.project_id, pack.pack_id, pack.task_id, project['task_labels'][pack.task_id], pack.operator_type, pack.operator_id, operator) DBSession.add(event) DBSession.commit() # Automatic task if pack.operator_type == 'auto' \ and not task_auto_build(request, pack, task): DBSession.query(PackEvent).filter_by( project_id=event.project_id, pack_id=event.pack_id, begin=event.begin).delete() pack.task_id = old_pack[0] pack.operator_type = old_pack[1] pack.operator_id = old_pack[2] DBSession.commit()
# =============================================================================
[docs]def pack_download(request, project_id, pack_id): """Download files of a pack in a ZIP file. :param request: (:class:`pyramid.request.Request` instance) Current request. :param project_id: (string) ID of pack to download. :param pack_id: (string) ID of pack to download. :return: (:class:`pyramid.response.FileResponse` instance or raise a :class:`pyramid.httpexceptions.HTTPNotFound` exception.) """ pack = DBSession.query(Pack).filter_by( project_id=project_id, pack_id=pack_id).first() if pack is None: raise HTTPNotFound(comment=_('Unknown pack!')) storage_root = request.registry.settings['storage.root'] done = set(['pack.xml']) def _add_directory(zip_file, dirpath): """Add all files of a directory.""" for root, ignored_, files in walk(dirpath): for name in files: name = relpath(join(root, name), storage_root) if name not in done: zip_file.write(join(storage_root, name), name) done.add(name) # Create ZIP tmp = tempfile.NamedTemporaryFile( dir=request.registry.settings['temporary_dir']) zip_file = zipfile.ZipFile(tmp, 'w', zipfile.ZIP_DEFLATED) # Add meta data root = etree.Element('publiforge', version=PUBLIFORGE_RNG_VERSION) root.append(pack.xml()) name = zipfile.ZipInfo('pack.xml') name.external_attr = 2175008768 zip_file.writestr( name, etree.tostring( root, encoding='utf-8', xml_declaration=True, pretty_print=True)) # Add files for pfile in pack.files: if pfile.path not in done: path = join(storage_root, pfile.path) if isfile(path): zip_file.write(path, pfile.path) else: _add_directory(zip_file, path) done.add(pfile.path) zip_file.close() name = '' % normalize_name(pack.label) return file_download(request, '', (,), name.decode('utf8'))
# =============================================================================
[docs]def pack_upload_settings(request, project_id, pack_doc=None, pack_id=None): """Import pack settings. :param request: (:class:`pyramid.request.Request` instance) Current request. :param project_id: (string) Current project ID. :param pack_doc: (:class:`lxml.etree.ElementTree` instance, optional) Pack XML DOM. :param pack_id: (integer, optional) Forced pack ID. """ # pylint: disable = E1103 # Read content if pack_doc is None: upload = request.params.get('pack_file') if isinstance(upload, basestring): return if splitext(upload.filename)[1] != '.xml': request.session.flash(_('Incorrect file type!'), 'alert') return pack_doc = load_xml( 'pack.xml', {'publiforge': join(dirname(__file__), '..', 'RelaxNG', 'publiforge.rng')}, upload.file.close() if isinstance(pack_doc, basestring): request.session.flash(pack_doc, 'alert') return # Upload configurations roles = dict([ ('rol%d.%d' % (project_id, k[0]), k[0]) for k in DBSession.query(Role.role_id).filter_by(project_id=project_id)]) tasks = dict([ ('tsk%d.%d' % (project_id, k[0]), k[0]) for k in DBSession.query(Task.task_id).filter_by(project_id=project_id)]) processings = dict([ ('prc%d.%d' % (project_id, k[0]), k[0]) for k in DBSession.query(Processing.processing_id) .filter_by(project_id=project_id)]) for elt in pack_doc.xpath('pack|packs/pack'): pack = Pack.load( project_id, roles, processings, tasks, elt, request.registry.settings['storage.root'], pack_id) if isinstance(pack, basestring): request.session.flash(pack, 'alert')
# =============================================================================
[docs]def pack_upload_content(request, project_id, message, label=None, handler=None): """Import pack content. :param request: (:class:`pyramid.request.Request` instance) Current request. :param project_id: (string) Current project ID. :param message: (string) Message for version control system. :param label: (string, optional) Label of pack to replace. If ``None``, the pack must be a new one. :param handler: (file handler, optional) File handler. :return: (``None`` or :class:`pyramid.i18n.TranslationString`) """ # pylint: disable = too-many-return-statements # Check parameters if handler is None: upload = request.params.get('pack_file') if isinstance(upload, basestring): return None handler = upload.file if not message: error = _('Message is required!') request.session.flash(error, 'alert') return error if not zipfile.is_zipfile(handler): error = _('Incorrect ZIP file!') request.session.flash(error, 'alert') return error zip_file = zipfile.ZipFile(handler, 'r') # Read "pack.xml" pack_doc = load_xml( 'pack.xml', {'publiforge': join(dirname(__file__), '..', 'RelaxNG', 'publiforge.rng')}, 'pack.xml' in zip_file.namelist() and'pack.xml') or '') if isinstance(pack_doc, basestring): request.session.flash(pack_doc, 'alert') zip_file.close() handler.close() return pack_doc # Check content error, pack_id, label, storage_ids = _pack_upload_content_check( request, project_id, pack_doc, label, zip_file) if error: zip_file.close() handler.close() return error # Check storage access, add and commit error = _pack_upload_content_extract( request, pack_doc, storage_ids, zip_file, message) zip_file.close() handler.close() if error: return error # Save settings DBSession.query(Pack).filter_by(label=label).delete() pack_upload_settings(request, project_id, pack_doc, pack_id) return None
# =============================================================================
[docs]def paging_packs(request, project_id): """Return a :class:`~..lib.widget.Paging` object filled with packs of project ``project_id``. :param request: (:class:`pyramid.request.Request` instance) Current request. :param project_id: (integer) Project ID. :return: (tuple) A tuple such as ``(paging, filters)`` where ``paging`` is a :class:`~..lib.widget.Paging` object and ``filters`` a dictionary of filters. """ # Parameters params = Paging.params(request, 'packs', '+pack_id') # Query query = DBSession.query(Pack).filter_by(project_id=project_id) if 'f_label' in params: query = query.filter( Pack.label.ilike('%%%s%%' % params['f_label'])) if 'f_task' in params and params['f_task'] != '*': query = query.filter(Pack.task_id == params['f_task']) # Order by oby = getattr(Pack, params['sort'][1:]) query = query.order_by(desc(oby) if params['sort'][0] == '-' else oby) return Paging(request, 'packs', query), params
# =============================================================================
[docs]def operator4task(pack, link_type, task_id=None): """Find the operator for task ``task_id`` according mode ``link_type``. :param pack: (:class:`~..models.packs.Pack` instance) Pack object. :param link_type: (string) Type of link: ``normal``, ``back``, ``redo`` or ``kept``. :param task_id: (integer, optional) Task ID. If ``None``, the first non ``auto`` task is used. :return: (tuple) A tuple such as ``(task_id, operator_type, operator_id)``. """ # Find the new operator and, possibly, the new task ID operator_type = None operator_id = None if link_type == 'kept' and pack.operator_id is not None: operator_type = pack.operator_type operator_id = pack.operator_id elif link_type in ('back', 'redo', 'kept'): event = DBSession.query( PackEvent.task_id, PackEvent.operator_type, PackEvent.operator_id)\ .filter_by(project_id=pack.project_id, pack_id=pack.pack_id)\ .filter(PackEvent.operator_type != 'auto') if link_type in ('back', 'redo') and task_id: event = event.filter_by(task_id=task_id) event = event.order_by(desc('begin')).first() if event: task_id = task_id if task_id is not None else event[0] operator_type = event[1] operator_id = event[2] return task_id, operator_type, operator_id
# =============================================================================
[docs]def operator_label(request, project, operator_type, operator_id): """Return localized operator label. :param request: (:class:`pyramid.request.Request` instance) Current request. :param project: (dictionary) Current project dictionary. :param operator_type: (string) Operator type. :param operator_id: (integer) Operator ID. :return: (string) """ if operator_type == 'role': label = request.localizer.translate(_( '[Role] ${l}', {'l': project['role_labels'].get(operator_id, '')})) elif operator_type == 'user': label = DBSession.query( user_id=operator_id).first() label = label[0] if label else '' else: label = request.localizer.translate(_('[Automatic]')) return label
# =============================================================================
[docs]def operator_labels(project, with_auto=False): """Return a list of all operators of the project ``project``. :param project: (dictionary) Current project dictionary. :param with_auto: (boolean) If ``True`` add automatic operator in the list. :return: (list) """ operators = Project.team_query(project['project_id'])\ .order_by( operators = [ ('role%d' % k[0], _('[Role] ${l}', {'l': k[1]})) for k in project['role_labels'].items()] \ + [('user%d' % k[0], k[2]) for k in operators] if with_auto: operators = [('autoNone', _('[Automatic]'))] + operators return operators
# ============================================================================= def _pack_upload_content_check(request, project_id, pack_doc, label, zip_file): """Check pack content before importing it. :param request: (:class:`pyramid.request.Request` instance) Current request. :param project_id: (string) Current project ID. :param pack_doc: (:class:`lxml.etree.ElementTree` instance) Pack XML DOM. :param label: (string) Label of pack to replace. If ``None``, the pack must be a new one, if ``''``, the pack must exist. :param zip_file: (file handler) ZIP file handler. :return: (tuple) A tupple such as ``(error, pack_doc, pack_id, label, storage_ids)``. """ # pylint: disable = E1103 # Error function def _error(error): """Return an error message.""" request.session.flash(error, 'alert') return error, None, None, None # Check "pack.xml" item = normalize_spaces(pack_doc.findtext('pack/label'), LABEL_LEN) if label and label != item: return _error(_('Pack labels are different!')) pack = DBSession.query(Pack.pack_id).filter_by( project_id=project_id, label=item).first() if label is None and pack: return _error(_('Pack "${l}" already exists.', {'l': item})) if label == '' and pack is None: return _error(_('Pack "${l}" does not exist.', {'l': item})) label = item # Check ZIP content root = request.registry.settings['storage.root'] storage_ids = set() for item in pack_doc.xpath( 'pack/files/file|pack/resources/resource|pack/templates/template'): item = item.text.strip() if item not in EXCLUDED_FILES and item not in zip_file.namelist() \ and not exists(join(root, item)): return _error(_('Unknown file "${n}".', {'n': item})) if item not in EXCLUDED_FILES and item in zip_file.namelist(): item = item.split(sep)[0] if item not in storage_ids and not exists(join(root, item)): return _error(_('Unknown storage "${n}".', {'n': item})) storage_ids.add(item) return None, pack and pack[0] or None, label, storage_ids # ============================================================================= def _pack_upload_content_extract(request, pack_doc, storage_ids, zip_file, message): """Check access to storage and extract files, add them and commit. :param request: (:class:`pyramid.request.Request` instance) Current request. :param pack_doc: (:class:`lxml.etree.ElementTree` instance) Pack XML DOM. :param storage_ids: (list) ZIP file handler. :param zip_file: (file handler) ZIP file handler. :param message: (string) Message for version control system. :return: (``None`` or :class:`pyramid.i18n.TranslationString`) """ # Authorization for storages vcs_user = {} for storage_id in storage_ids: storage = DBSession.query(Storage).filter_by(storage_id=storage_id)\ .first() if storage is None: return _('Unknown storage "${n}".', {'n': storage_id}) user = DBSession.query(StorageUser).filter_by( storage_id=storage_id, user_id=request.session['user_id']).first() if not has_permission(request, 'stg_modifier') \ and storage.access != 'open' \ and (not user or not user.perm or user.perm == 'user'): return _('You cannot write into "${n}"!', {'n': storage_id}) if storage.vcs_engine not in ('none', 'local') \ and not (user and user.vcs_user): return _( 'ID and password for "${n}" are missing.', {'n': storage_id}) if storage.vcs_engine != 'none': vcs_user[storage_id] = ( user and user.vcs_user or None, user and decrypt( user.vcs_password, request.registry.settings.get('encryption', '-'))) request.registry['handler'].get_handler(storage_id, storage) # Extract content # pylint: disable = E1103 root = request.registry.settings['storage.root'] for name in pack_doc.xpath( 'pack/files/file|pack/resources/resource|pack/templates/template'): name = name.text.strip() if name in zip_file.namelist() \ and basename(name) not in EXCLUDED_FILES: zip_file.extract(name, root) # Add and commit for storage_id in storage_ids: if storage_id in vcs_user: handler = request.registry['handler'].get_handler(storage_id) handler.add( (vcs_user[storage_id][0], vcs_user[storage_id][1], request.session['name']), '.', message) name, error = handler.progress() if name == 'error': return error handler.cache.clear() return None