WebDAV: Content.file_extension usage for files names

tracim/migration/versions/59fc98c3c965_delete_content_file_name_column.py

+"""delete content file name column
+
+Revision ID: 59fc98c3c965
+Revises: 15305f71bfda
+Create Date: 2016-11-25 14:55:22.176175
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '59fc98c3c965'
+down_revision = '15305f71bfda'
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade():
+    op.drop_column('content_revisions', 'file_name')
+
+
+def downgrade():
+    op.add_column(
+        'content_revisions',
+        sa.Column(
+            'file_name',
+            sa.VARCHAR(length=255),
+            server_default=sa.text("''::character varying"),
+            autoincrement=False,
+            nullable=False
+        ),
+    )

tracim/tracim/lib/content.py

 # -*- coding: utf-8 -*-
+import os
+
 from operator import itemgetter
 
 __author__ = 'damien'
 
         return revision
 
-    def get_one_by_label_and_parent(self, content_label: str, content_parent: Content = None,
-                                    workspace: Workspace = None) -> Content:
+    def get_one_by_label_and_parent(
+            self,
+            content_label: str,
+            content_parent: Content=None,
+    ) -> Content:
         """
         This method let us request the database to obtain a Content with its name and parent
         :param content_label: Either the content's label or the content's filename if the label is None
         :param workspace: The workspace's content
         :return The corresponding Content
         """
-        assert content_label is not None# DYN_REMOVE
+        query = self._base_query(content_parent.workspace)
+        parent_id = content_parent.content_id if content_parent else None
+        query = query.filter(Content.parent_id == parent_id)
 
-        resultset = self._base_query(workspace)
+        file_name, file_extension = os.path.splitext(content_label)
 
-        parent_id = content_parent.content_id if content_parent else None
+        return query.filter(
+            or_(
+                and_(
+                    Content.type == ContentType.File,
+                    Content.label == file_name,
+                    Content.file_extension == file_extension,
+                ),
+                and_(
+                    Content.type == ContentType.Thread,
+                    Content.label == file_name,
+                ),
+                and_(
+                    Content.type == ContentType.Page,
+                    Content.label == file_name,
+                ),
+                and_(
+                    Content.type == ContentType.Folder,
+                    Content.label == content_label,
+                ),
+            )
+        ).one()
 
-        resultset = resultset.filter(Content.parent_id == parent_id)
+    def get_one_by_label_and_parent_labels(
+            self,
+            content_label: str,
+            workspace: Workspace,
+            content_parent_labels: [str]=None,
+    ):
+        """
+        TODO: Assurer l'unicité des path dans Tracim
+        TODO: Ecrire tous les cas de tests par rapports aux
+         cas d'erreurs possible (duplications de labels, etc) et TROUVES
+        Return content with it's label, workspace and parents labels (optional)
+        :param content_label: label of content (label or file_name)
+        :param workspace: workspace containing all of this
+        :param content_parent_labels: Ordered list of labels representing path
+            of folder (without workspace label).
+        E.g.: ['foo', 'bar'] for complete path /Workspace1/foo/bar folder
+        :return: Found Content
+        """
+        query = self._base_query(workspace)
+        file_name, file_extension = os.path.splitext(content_label)
+        parent_folder = None
 
-        return resultset.filter(or_(
-            Content.label == content_label,
-            Content.file_name == content_label,
-            Content.label == re.sub(r'\.[^.]+$', '', content_label)
-        )).one()
+        # Grab content parent folder if parent path given
+        if content_parent_labels:
+            parent_folder = self.get_folder_with_workspace_path_labels(
+                content_parent_labels,
+                workspace,
+            )
 
-    def get_one_by_label_and_parent_label(self, content_label: str, content_parent_label: [str]=None, workspace: Workspace=None):
-        assert content_label is not None  # DYN_REMOVE
-        resultset = self._base_query(workspace)
+        # Build query for found content by label
+        content_query = query.filter(or_(
+            and_(
+                Content.type == ContentType.File,
+                Content.label == file_name,
+                Content.file_extension == file_extension,
+            ),
+            and_(
+                Content.type == ContentType.Thread,
+                Content.label == file_name,
+            ),
+            and_(
+                Content.type == ContentType.Page,
+                Content.label == file_name,
+            ),
+            and_(
+                Content.type == ContentType.Folder,
+                Content.label == content_label,
+            ),
+        ))
 
-        res =  resultset.filter(or_(
-            Content.label == content_label,
-            Content.file_name == content_label,
-            Content.label == re.sub(r'\.[^.]+$', '', content_label)
-        )).all()
+        # Modify query to apply parent folder filter if any
+        if parent_folder:
+            content_query = content_query.filter(
+                Content.parent_id == parent_folder.content_id,
+                Content.workspace_id == workspace.workspace_id,
+            )
 
-        if content_parent_label:
-            tmp = dict()
-            for content in res:
-                tmp[content] = content.parent
+        # Return the content
+        return content_query\
+            .order_by(
+                Content.revision_id.desc(),
+            )\
+            .one()
 
-            for parent_label in reversed(content_parent_label):
-                a = []
-                tmp = {content: parent.parent for content, parent in tmp.items()
-                       if parent and parent.label == parent_label}
+    def get_folder_with_workspace_path_labels(
+            self,
+            path_labels: [str],
+            workspace: Workspace,
+    ) -> Content:
+        """
+        Return a Content folder for given relative path.
+        TODO BS 20161124: Not safe if web interface allow folder duplicate names
+        :param path_labels: List of labels representing path of folder
+        (without workspace label).
+        E.g.: ['foo', 'bar'] for complete path /Workspace1/foo/bar folder
+        :param workspace: workspace of folders
+        :return: Content folder
+        """
+        query = self._base_query(workspace)
+        folder = None
+
+        for label in path_labels:
+            # Filter query on label
+            folder_query = query \
+                .filter(
+                    Content.type == ContentType.Folder,
+                    Content.label == label,
+                    Content.workspace_id == workspace.workspace_id,
+                )
 
-                if len(tmp) == 1:
-                    content, last_parent = tmp.popitem()
-                    return content
-                elif len(tmp) == 0:
-                    return None
+            # Search into parent folder (if already deep)
+            if folder:
+                folder_query\
+                    .filter(
+                        Content.parent_id == folder.content_id,
+                    )
 
-            for content, parent_content in tmp.items():
-                if not parent_content:
-                    return content
+            # Get thirst corresponding folder
+            folder = folder_query \
+                .order_by(Content.revision_id.desc()) \
+                .one()
 
-            return None
-        return res[0]
+        return folder
 
     def get_all(self, parent_id: int=None, content_type: str=ContentType.Any, workspace: Workspace=None) -> [Content]:
         assert parent_id is None or isinstance(parent_id, int) # DYN_REMOVE

tracim/tracim/lib/daemons.py

 
         config['provider_mapping'] = {
             config['root_path']: Provider(
-                show_archived=config['show_archived'],
-                show_deleted=config['show_deleted'],
-                show_history=config['show_history'],
+                # TODO: Test to Re enabme archived and deleted
+                show_archived=False,  # config['show_archived'],
+                show_deleted=False,  # config['show_deleted'],
+                show_history=False,  # config['show_history'],
                 manage_locks=config['manager_locks']
             )
         }

tracim/tracim/lib/notifications.py

             subject = self._global_config.EMAIL_NOTIFICATION_CONTENT_UPDATE_SUBJECT
             subject = subject.replace(EST.WEBSITE_TITLE, self._global_config.WEBSITE_TITLE.__str__())
             subject = subject.replace(EST.WORKSPACE_LABEL, main_content.workspace.label.__str__())
-            subject = subject.replace(EST.CONTENT_LABEL, main_content.label.__str__() if main_content.label else main_content.file_name.__str__())
+            subject = subject.replace(EST.CONTENT_LABEL, main_content.label.__str__())
             subject = subject.replace(EST.CONTENT_STATUS_LABEL, main_content.get_status().label.__str__())
 
             message = MIMEMultipart('alternative')
                 content_intro = _('<span id="content-intro-username">{}</span> added a file entitled:').format(actor.display_name)
                 if content.description:
                     content_text = content.description
-                elif content.label:
-                    content_text = '<span id="content-body-only-title">{}</span>'.format(content.label)
                 else:
-                    content_text = '<span id="content-body-only-title">{}</span>'.format(content.file_name)
-
+                    content_text = '<span id="content-body-only-title">{}</span>'.format(content.label)
 
             elif ContentType.Page == content.type:
                 content_intro = _('<span id="content-intro-username">{}</span> added a page entitled:').format(actor.display_name)

tracim/tracim/lib/webdav/sql_dav_provider.py

 
 import re
 from os.path import basename, dirname, normpath
+from sqlalchemy.orm.exc import NoResultFound
 from tracim.lib.webdav.utils import transform_to_bdd
 
 from wsgidav.dav_provider import DAVProvider
 
         content_api = ContentApi(
             user,
-            show_archived=self._show_archive,
-            show_deleted=self._show_delete
+            show_archived=False,  # self._show_archive,
+            show_deleted=False,  # self._show_delete
         )
 
         content = self.get_content_from_path(
         if parent_path == root_path or workspace is None:
             return workspace is not None
 
-        content_api = ContentApi(user, show_archived=True, show_deleted=True)
+        # TODO bastien: Arnaud avait mis a True, verif le comportement
+        # lorsque l'on explore les dossiers archive et deleted
+        content_api = ContentApi(user, show_archived=False, show_deleted=False)
 
         revision_id = re.search(r'/\.history/[^/]+/\((\d+) - [a-zA-Z]+\) ([^/].+)$', path)
 
         parents = [transform_to_bdd(x) for x in parents]
 
         try:
-            return content_api.get_one_by_label_and_parent_label(
-                transform_to_bdd(basename(path)),
-                parents,
-                workspace
+            return content_api.get_one_by_label_and_parent_labels(
+                content_label=transform_to_bdd(basename(path)),
+                content_parent_labels=parents,
+                workspace=workspace,
             )
-        except:
+        except NoResultFound:
             return None
 
     def get_content_from_revision(self, revision: ContentRevisionRO, api: ContentApi) -> Content:
         try:
             return api.get_one(revision.content_id, ContentType.Any)
-        except:
+        except NoResultFound:
             return None
 
     def get_parent_from_path(self, path, api: ContentApi, workspace) -> Content:
     def get_workspace_from_path(self, path: str, api: WorkspaceApi) -> Workspace:
         try:
             return api.get_one_by_label(transform_to_bdd(path.split('/')[1]))
-        except:
+        except NoResultFound:
             return None

tracim/tracim/lib/webdav/sql_resources.py

         try:
             # When undeleting/unarchiving we except a content with the new name to not exist, thus if we
             # don't get an error and the database request send back a result, we stop the action
-            self.content_api.get_one_by_label_and_parent(self._new_name, self.content.parent, self.content.workspace)
+            self.content_api.get_one_by_label_and_parent(self._new_name, self.content.parent)
             raise DAVError(HTTP_FORBIDDEN)
         except NoResultFound:
             with new_revision(self.content):
         Will create the new name, either by adding '- deleted the [date]' after the name when archiving/deleting or
         removing this string when undeleting/unarchiving
         """
-        new_name = self.content.get_label()
+        new_name = self.content.get_label_as_file()
         extension = ''
 
         # if the content has no label, the last .ext is important
             # the purpose is to display .history only if there's at least one content's type that has a history
             if content.type != ContentType.Folder:
                 self._file_count += 1
-            retlist.append(content.get_label())
+            retlist.append(content.get_label_as_file())
 
         return retlist
 
         children = self.content_api.get_all(False, ContentType.Any, self.workspace)
 
         for content in children:
-            content_path = '%s/%s' % (self.path, transform_to_display(content.get_label()))
+            content_path = '%s/%s' % (self.path, transform_to_display(content.get_label_as_file()))
 
             if content.type == ContentType.Folder:
                 members.append(Folder(content_path, self.environ, self.workspace, content))
         return mktime(self.content.created.timetuple())
 
     def getDisplayName(self) -> str:
-        return transform_to_display(self.content.get_label())
+        return transform_to_display(self.content.get_label_as_file())
 
     def getLastModified(self) -> float:
         return mktime(self.content.updated.timetuple())
         )
 
         for content in visible_children:
-            content_path = '%s/%s' % (self.path, transform_to_display(content.get_label()))
+            content_path = '%s/%s' % (self.path, transform_to_display(content.get_label_as_file()))
 
             if content.type == ContentType.Folder:
                 members.append(Folder(content_path, self.environ, self.workspace, content))
         )
 
         return HistoryFileFolder(
-            path='%s/%s' % (self.path, content.get_label()),
+            path='%s/%s' % (self.path, content.get_label_as_file()),
             environ=self.environ,
             content=content)
 
                 self._is_deleted and content.is_deleted or
                 not (content.is_archived or self._is_archived or content.is_deleted or self._is_deleted))\
                     and content.type != ContentType.Folder:
-                ret.append(content.get_label())
+                ret.append(content.get_label_as_file())
 
         return ret
 
         for content in children:
             if content.is_archived == self._is_archived and content.is_deleted == self._is_deleted:
                 members.append(HistoryFileFolder(
-                    path='%s/%s' % (self.path, content.get_label()),
+                    path='%s/%s' % (self.path, content.get_label_as_file()),
                     environ=self.environ,
                     content=content))
 
         )
 
         return self.provider.getResourceInst(
-            path='%s/%s' % (self.path, transform_to_display(content.get_label())),
+            path='%s/%s' % (self.path, transform_to_display(content.get_label_as_file())),
             environ=self.environ
             )
 
 
         for content in children:
             if content.is_deleted:
-                retlist.append(content.get_label())
+                retlist.append(content.get_label_as_file())
 
                 if content.type != ContentType.Folder:
                     self._file_count += 1
 
         for content in children:
             if content.is_deleted:
-                content_path = '%s/%s' % (self.path, transform_to_display(content.get_label()))
+                content_path = '%s/%s' % (self.path, transform_to_display(content.get_label_as_file()))
 
                 if content.type == ContentType.Folder:
                     members.append(Folder(content_path, self.environ, self.workspace, content))
         )
 
         return self.provider.getResourceInst(
-            path=self.path + '/' + transform_to_display(content.get_label()),
+            path=self.path + '/' + transform_to_display(content.get_label_as_file()),
             environ=self.environ
         )
 
 
         for content in self.content_api.get_all_with_filter(
                 self.content if self.content is None else self.content.id, ContentType.Any):
-            retlist.append(content.get_label())
+            retlist.append(content.get_label_as_file())
 
             if content.type != ContentType.Folder:
                 self._file_count += 1
 
         for content in children:
             if content.is_archived:
-                content_path = '%s/%s' % (self.path, transform_to_display(content.get_label()))
+                content_path = '%s/%s' % (self.path, transform_to_display(content.get_label_as_file()))
 
                 if content.type == ContentType.Folder:
                     members.append(Folder(content_path, self.environ, self.workspace, content))
         return "<DAVCollection: HistoryFileFolder (%s)" % self.content.file_name
 
     def getDisplayName(self) -> str:
-        return self.content.get_label()
+        return self.content.get_label_as_file()
 
     def createCollection(self, name):
         raise DAVError(HTTP_FORBIDDEN)
                 content_revision=revision)
         else:
             return HistoryOtherFile(
-                path='%s%s' % (left_side, transform_to_display(revision.get_label())),
+                path='%s%s' % (left_side, transform_to_display(revision.get_label_as_file())),
                 environ=self.environ,
                 content=self.content,
                 content_revision=revision)
         # but i wasn't able to set this property so you'll have to look into it >.>
         # self.setPropertyValue('Win32FileAttributes', '00000021')
 
-    def getPreferredPath(self):
-        fix_txt = '.txt' if self.getContentType() == 'text/plain' else mimetypes.guess_extension(self.getContentType())
-        if not fix_txt:
-            fix_txt = ''
-        if self.content and self.path and (self.content.label == '' or self.path.endswith(fix_txt)):
-            return self.path
-        else:
-            return self.path + fix_txt
-
     def __repr__(self) -> str:
         return "<DAVNonCollection: File (%d)>" % self.content.revision_id
 
         return mktime(self.content.created.timetuple())
 
     def getDisplayName(self) -> str:
-        return self.content.get_label()
+        return self.content.file_name
 
     def getLastModified(self) -> float:
         return mktime(self.content.updated.timetuple())
         return FakeFileStream(
             content=self.content,
             content_api=self.content_api,
-            file_name=self.content.get_label(),
+            file_name=self.content.get_label_as_file(),
             workspace=self.content.workspace,
             path=self.path
         )
             self.path += '.html'
 
     def getDisplayName(self) -> str:
-        return self.content.get_label()
+        return self.content.get_label_as_file()
 
     def getPreferredPath(self):
         return self.path
 
     def getDisplayName(self) -> str:
         left_side = '(%d - %s) ' % (self.content_revision.revision_id, self.content_revision.revision_type)
-        return '%s%s' % (left_side, transform_to_display(self.content_revision.get_label()))
+        return '%s%s' % (left_side, transform_to_display(self.content_revision.get_label_as_file()))
 
     def getContent(self):
         filestream = compat.BytesIO()

tracim/tracim/model/__init__.py

 
     @classmethod
     def remove_from_updatable(cls, revision: 'ContentRevisionRO') -> None:
-        cls._updatable_revisions.remove(revision)
+        if revision in cls._updatable_revisions:
+            cls._updatable_revisions.remove(revision)
 
     @classmethod
     def is_updatable(cls, revision: 'ContentRevisionRO') -> bool:

tracim/tracim/model/data.py

 
 import datetime as datetime_root
 import json
+import os
 from datetime import datetime
 
 import tg
 
     label = Column(Unicode(1024), unique=False, nullable=False)
     description = Column(Text(), unique=False, nullable=False, default='')
-    file_name = Column(Unicode(255),  unique=False, nullable=False, default='')
+    file_extension = Column(
+        Unicode(255),
+        unique=False,
+        nullable=False,
+        server_default='',
+    )
     file_mimetype = Column(Unicode(255),  unique=False, nullable=False, default='')
     file_content = deferred(Column(LargeBinary(), unique=False, nullable=True))
     properties = Column('properties', Text(), unique=False, nullable=False, default='')
 
     """ List of column copied when make a new revision from another """
     _cloned_columns = (
-        'content_id', 'created', 'description', 'file_content', 'file_mimetype', 'file_name', 'is_archived',
-        'is_deleted', 'label', 'node', 'owner', 'owner_id', 'parent', 'parent_id', 'properties', 'revision_type',
-        'status', 'type', 'updated', 'workspace', 'workspace_id', 'is_temporary',
+        'content_id',
+        'created',
+        'description',
+        'file_content',
+        'file_mimetype',
+        'file_extension',
+        'is_archived',
+        'is_deleted',
+        'label',
+        'node',
+        'owner',
+        'owner_id',
+        'parent',
+        'parent_id',
+        'properties',
+        'revision_type',
+        'status',
+        'type',
+        'updated',
+        'workspace',
+        'workspace_id',
+        'is_temporary',
     )
 
     # Read by must be used like this:
             RevisionReadStatus(user=k, view_datetime=v)
     )
 
+    @property
+    def file_name(self):
+        return '{0}{1}'.format(
+            self.label,
+            self.file_extension,
+        )
+
     @classmethod
     def new_from(cls, revision: 'ContentRevisionRO') -> 'ContentRevisionRO':
         """
         return ContentStatus(self.status)
 
     def get_label(self) -> str:
-        return self.label if self.label else self.file_name if self.file_name else ''
+        return self.label or self.file_name or ''
 
     def get_last_action(self) -> ActionDescription:
         return ActionDescription(self.revision_type)
 
         return False
 
+    def get_label_as_file(self):
+        file_extension = self.file_extension
+
+        if self.type == ContentType.Thread:
+            file_extension = '.html'
+        elif self.type == ContentType.Page:
+            file_extension = '.html'
+
+        return '{0}{1}'.format(
+            self.label,
+            file_extension,
+        )
+
+
 Index('idx__content_revisions__owner_id', ContentRevisionRO.owner_id)
 Index('idx__content_revisions__parent_id', ContentRevisionRO.parent_id)
 
 
     @hybrid_property
     def file_name(self) -> str:
-        return self.revision.file_name
+        return '{0}{1}'.format(
+            self.revision.label,
+            self.revision.file_extension,
+        )
 
     @file_name.setter
     def file_name(self, value: str) -> None:
-        self.revision.file_name = value
+        file_name, file_extension = os.path.splitext(value)
+        self.revision.label = file_name
+        self.revision.file_extension = file_extension
 
     @file_name.expression
     def file_name(cls) -> InstrumentedAttribute:
-        return ContentRevisionRO.file_name
+        return ContentRevisionRO.file_name + ContentRevisionRO.file_extension
+
+    @hybrid_property
+    def file_extension(self) -> str:
+        return self.revision.file_extension
+
+    @file_extension.setter
+    def file_extension(self, value: str) -> None:
+        self.revision.file_extension = value
+
+    @file_extension.expression
+    def file_extension(cls) -> InstrumentedAttribute:
+        return ContentRevisionRO.file_extension
 
     @hybrid_property
     def file_mimetype(self) -> str:
         return child_nb
 
     def get_label(self):
-        return self.label if self.label else self.file_name if self.file_name else ''
+        return self.label or self.file_name or ''
+
+    def get_label_as_file(self) -> str:
+        """
+        :return: Return content label in file representation context
+        """
+        return self.revision.get_label_as_file()
 
     def get_status(self) -> ContentStatus:
         return ContentStatus(self.status, self.type.__str__())

tracim/tracim/model/serializers.py

 def serialize_version_for_page_or_file(version: ContentRevisionRO, context: Context):
     return DictLikeClass(
         id = version.revision_id,
-        label = version.label if version.label else version.file_name,
+        label = version.label,
         owner = context.toDict(version.owner),
         created = version.created,
         action = context.toDict(version.get_last_action())
 
     result = DictLikeClass(
         id = content.content_id,
-        label = content.label if content.label else content.file_name,
+        label = content.label,
         icon = ContentType.get_icon(content.type),
         status = context.toDict(content.get_status()),
         folder = context.toDict(DictLikeClass(id = content.parent.content_id if content.parent else None)),
     if content.type==ContentType.File:
         result = DictLikeClass(
             id = content.content_id,
-            label = content.label if content.label else content.file_name,
+            label = content.label,
             status = context.toDict(content.get_status()),
             folder = Context(CTX.DEFAULT).toDict(content.parent)
         )
         )
 
         if content.type==ContentType.File:
-            result.label = content.label if content.label else content.file_name
+            result.label = content.label
             result['file'] = DictLikeClass(
                 name = data_container.file_name,
                 size = len(data_container.file_content),
         )
 
         if content.type==ContentType.File:
-            result.label = content.label.__str__() if content.label else content.file_name.__str__()
+            result.label = content.label.__str__()
 
         if not result.label or ''==result.label:
             result.label = 'No title'