ForgeFiles/forgefiles/model/files.py

#       Licensed to the Apache Software Foundation (ASF) under one
#       or more contributor license agreements.  See the NOTICE file
#       distributed with this work for additional information
#       regarding copyright ownership.  The ASF 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.

''' This is the Collection module for the Files plugin.
Upload, UploadFolder & UploadFile are the collections'''

from datetime import datetime

from urllib.parse import quote
import re
import typing

from ming import schema as S
from ming.odm import Mapper
from ming.odm import FieldProperty, ForeignIdProperty, RelationProperty
from bson import ObjectId

from tg import tmpl_context as c

from allura.model.artifact import VersionedArtifact
from allura.model.auth import AlluraUserProperty, User
from allura.model.session import project_orm_session, artifact_orm_session
from allura.model.filesystem import File
from allura.model.timeline import ActivityObject
from allura.lib import helpers as h

if typing.TYPE_CHECKING:
    from ming.odm.mapper import Query


README_RE = re.compile(r'^README(\.[^.]*)?$', re.IGNORECASE)


class Upload(VersionedArtifact, ActivityObject):

    ''' The Upload collection for files. '''

    class __mongometa__:
        name = 'upload'
        session = project_orm_session

    query: 'Query[Upload]'

    type_s = 'Upload'

    _id = FieldProperty(S.ObjectId)
    filename = FieldProperty(str)
    filetype = FieldProperty(str)
    project_id = ForeignIdProperty('Project', if_missing=lambda: c.project._id)
    created_date = FieldProperty(datetime, if_missing=datetime.utcnow)
    project = RelationProperty('Project', via='project_id')
    file_url = None

    @classmethod
    def attachment_class(cls):
        return UploadFiles

    @property
    def activity_name(self):
        return self.filename

    @property
    def activity_url(self):
        return self.file_url


class UploadFolder(VersionedArtifact, ActivityObject):

    ''' The UploadFolder collection is for the Folders. Every folder is an object of this class.'''

    class __mongometa__:
        name = 'upload_folder'
        session = project_orm_session

    query: 'Query[UploadFolder]'

    type_s = 'UploadFolder'

    _id = FieldProperty(S.ObjectId)
    folder_name = FieldProperty(str)
    project_id = ForeignIdProperty('Project', if_missing=lambda: c.project._id)
    parent_folder_id = ForeignIdProperty('UploadFolder')
    created_date = FieldProperty(datetime, if_missing=datetime.utcnow)
    author_id = ForeignIdProperty('User', if_missing=lambda: c.user._id)
    parent_folder = RelationProperty('UploadFolder', via='parent_folder_id')
    project = RelationProperty('Project', via='project_id')
    path = FieldProperty(str)
    published = FieldProperty(bool, if_missing=False)
    remarks = FieldProperty(str)
    disabled = FieldProperty(bool, if_missing=False)
    folder_ids = FieldProperty([str])
    file_ids = FieldProperty([str])
    author = RelationProperty(User, via='author_id')

    def url(self):
        parent_folder = self.parent_folder
        if parent_folder:
            string = ''
            while parent_folder:
                string += parent_folder.folder_name + '/'
                parent_folder = parent_folder.parent_folder
            list_obj = string.rsplit('/')[::-1]
            string = '/'.join(list_obj)
            string = string.lstrip('/')
            url_str = c.app.url + string + '/' + self.folder_name
            self.path = string + '/' + self.folder_name
        else:
            url_str = c.app.url + self.folder_name
            self.path = self.folder_name
        return quote(url_str)

    def folder_id(self):
        return str(self._id)

    @property
    def activity_name(self):
        return self.folder_name

    @property
    def activity_url(self):
        parent_folder = self.parent_folder
        if parent_folder:
            return parent_folder.url()
        else:
            return c.app.url


class UploadFiles(File):

    '''The UploadFiles collection is for the Files. Any file which is uploaded is treated as an Upload object'''

    thumbnail_size = (255, 255)
    ArtifactType = Upload

    class __mongometa__:
        name = 'upload_files'
        session = project_orm_session
        indexes = [
            ('app_config_id', 'parent_folder_id', 'filename'),
            ('app_config_id', 'linked_to_download', 'disabled'),
            ('app_config_id', 'filename', 'path'),
        ]

        def before_save(data):
            _session = artifact_orm_session._get()
            skip_last_updated = getattr(_session, 'skip_last_updated', False)
            data['mod_date'] = datetime.utcnow()
            if c.project and not skip_last_updated:
                c.project.last_updated = datetime.utcnow()

    query: 'Query[UploadFiles]'

    artifact_id = FieldProperty(S.ObjectId)
    app_config_id = FieldProperty(S.ObjectId)
    type = FieldProperty(str)
    project_id = FieldProperty(S.ObjectId)
    parent_folder_id = ForeignIdProperty('UploadFolder')
    created_date = FieldProperty(datetime, if_missing=datetime.utcnow)
    mod_date = FieldProperty(datetime, if_missing=datetime.utcnow)
    author_id: ObjectId = AlluraUserProperty(if_missing=lambda: c.user._id)
    parent_folder = RelationProperty('UploadFolder', via='parent_folder_id')
    linked_to_download = FieldProperty(bool, if_missing=False)
    path = FieldProperty(str)
    disabled = FieldProperty(bool, if_missing=False)
    author = RelationProperty(User, via='author_id')

    @property
    def artifact(self):
        '''Returns the Artifact object'''

        return self.ArtifactType.query.get(_id=self.artifact_id)

    def url(self):
        '''Returns the URL of the uploaded file'''

        parent_folder = self.parent_folder
        if parent_folder:
            string = ''
            while parent_folder:
                string += parent_folder.folder_name + '/'
                parent_folder = parent_folder.parent_folder
            list_obj = string.rsplit('/')[::-1]
            string = '/'.join(list_obj)
            string = string.lstrip('/')
            url_str = c.app.url + string + '/' + self.filename
            self.path = string + '/' + self.filename
        else:
            url_str = c.app.url + self.filename
            self.path = self.filename
        return quote(url_str)

    def readme(self):
        'returns (filename, unicode text) if a readme file is found'
        if README_RE.match(self.filename):
            name = self.filename
            obj_content = self.rfile().read(self.rfile().length)
            return (self.filename, h.really_unicode(obj_content))
        return None, None

    @classmethod
    def save_attachment(cls, filename, fp, content_type=None, **kwargs):
        filename = h.really_unicode(filename)
        original_meta = dict(type="project_file", app_config_id=c.app.config._id, project_id=c.project._id)
        original_meta.update(kwargs)
        fp.seek(0)
        return cls.from_stream(
            filename, fp, content_type=content_type,
            **original_meta)


Mapper.compile_all()