class AuthMethodView(MethodView):
    decorators = (login_required, ) class File(ModelTimeMixin, db.Model):
    __tablename__ = 'file'

    FILE_TYPE = ("IMAGE", "CSS", "JS")
    FILE_IMAGE = "IMAGE"

    name = db.Column(db.String(108), nullable=False)
    file_type = db.Column(db.String(108), nullable=False)
    hash = db.Column(db.String(1024), nullable=False)
    size = db.Column(db.Integer, nullable=False, default=0)

    path_id = db.Column(
        db.Integer,
        db.ForeignKey('filepath.id', ondelete="CASCADE"),
        nullable=False)
    path = db.relationship(
        FilePath,
        backref=db.backref(
            'files', cascade='all,delete-orphan', lazy='dynamic'),
        lazy='joined',
        uselist=False)

    @property
    def abspath(self):
        return os.path.join(
            self.path.abspath,
            self.name,
        )

    @property
    def relpath(self):
        return os.path.join(
            self.path.relpath,
            self.name,
        )

    @property
    def url(self):
        args = dict(filename=self.relpath, _external=True)
        if config.HTTPS:
            args.update(**dict(_scheme="https"))
        if self.file_type.startswith("image"):
            args.update(type="mini")
        return url_for("storage.show", **args)

    @property
    def is_dir(self):
        return False

    def save(self):
        self.name = self.name.strip("/")
        if "/" in self.name:
            s = self.name.split("/")
            filepath = FilePath.query.filter_by(id=self.path_id).first()
            filepath = filepath.bucket.get_root_path("/".join(s[:-1]), True)
            self.name = s[-1]
            self.path_id = filepath.id
        return super(File, self).save()

    def copy(self, newpath):
        f = File(
            name=self.name,
            file_type=self.file_type,
            hash=self.hash,
            path_id=newpath.id,
        )
        f.save()
        return f

    def move(self, newpath):
        self.path_id = newpath
        self.save()
        return self

    def rename(self, newname):
        newname = secure_filename(newname)
        self.name = newname
        self.save()
        return self

    def __repr__(self):
        return '<File %r>' % self.name

    def __str__(self):
        return self.name class FilePathSerializer(Serializer):
    class Meta:
        exclude = [
            "parent_id",
            "parent_path",
            "files",
            "bucket",
            "child_paths",
            "bucket_id",
        ]
        extra = ["size", "is_dir", "fullname"] class FileSerializer(Serializer):
    class Meta:
        exclude = ["path", "hash", "path_id"]
        extra = ["url", "is_dir"] def file_is_allowed(filename):
    return '.' in filename and filename.rsplit(
        '.', 1)[1].lower() in config.UPLOAD_ALLOWED_EXTENSIONS def gen_hash(file_data):
    sha = sha512()
    sha.update(file_data.read())
    return sha.hexdigest() def gen_size(file_data):
    if file_data.content_length:
        return file_data.content_length

    try:
        pos = file_data.tell()
        file_data.seek(0, 2)  # seek to end
        size = file_data.tell()
        file_data.seek(pos)  # back to original position
        return size
    except (AttributeError, IOError):
        pass
    return 0 def secure_filename(filename):
    if isinstance(filename, text_type):
        from unicodedata import normalize
        filename = normalize('NFKD', filename).encode('utf-8', 'ignore')
        if not PY2:
            filename = filename.decode('utf-8')
    for sep in os.path.sep, os.path.altsep:
        if sep:
            filename = filename.replace(sep, ' ')
    filename = str(_filename_gbk_strip_re.sub('', '_'.join(
        filename.split()))).strip('._')

    if os.name == 'nt' and filename and \
       filename.split('.')[0].upper() in _windows_device_files:
        filename = '_' + filename
    return filename def check_params(keys, req=None):
    '''
    only check is not NULL
    '''
    def _check_params(func):
        @wraps(func)
        def decorator(*args, **kwargs):
            if req is not None:
                request_data = req
            else:
                request_data = request.data
            for key in keys:
                if not request_data.get(key):
                    return HTTP.BAD_REQUEST(message='{0} required'.format(key))
            return func(*args, **kwargs)

        return decorator

    return _check_params def filter_maybe(request_data, columns, params=None):
    if params is None:
        params = dict()
    is_dict = isinstance(columns, dict)
    for column in columns:
        value = request_data.get(column)
        if not value:
            continue
        key = column if not is_dict else columns.get(column, column)

        if key in ["created_at__gte", "created_at__lte"]:
            value = datetime.strptime(value, '%Y-%m-%d')
        params.update({key: value})
    return params def is_true(value):
    if isinstance(value, str):
        return value == "1" or value == "True" or value == "true"
    return bool(value)
import json
import os
from flask import request
from flask_babel import gettext as _
from flask_maple.response import HTTP
from maple.utils import AuthMethodView
from maple.storage.db import File
from maple.storage.serializer import FilePathSerializer, FileSerializer
from maple.storage.util import (
    file_is_allowed, gen_hash, gen_size, secure_filename)
from maple.utils import check_params, filter_maybe, is_true
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# ********************************************************************************
# Copyright © 2019 jianglin
# File Name: file.py
# Author: jianglin
# Email: mail@honmaple.com
# Created: 2019-05-13 16:36:36 (CST)
# Last Update: Monday 2019-09-23 17:11:23 (CST)
#          By:
# Description:
# ********************************************************************************



class FileListView(AuthMethodView):
    def get(self, bucket):
        data = request.data
        user = request.user
        page, number = self.pageinfo

        bucket = user.buckets.filter_by(
            name=bucket).get_or_404("bucket not found")

        params = filter_maybe(
            data, {
                "name": "name__contains",
                "type": "file_type"
            })
        filepath = bucket.get_root_path(data.get("path", "/"))
        if not filepath:
            return HTTP.BAD_REQUEST(message="path not found")

        paths = filepath.child_paths.paginate(page, number)
        files = filepath.files.filter_by(**params).paginate(page, number)
        return HTTP.OK(
            data=FilePathSerializer(paths).data + FileSerializer(files).data)

    def post(self, bucket):
        data = request.data
        user = request.user
        bucket = user.buckets.filter_by(
            name=bucket).get_or_404("bucket not found")

        filepath = bucket.get_root_path(data.get("path", "/"), True)
        files = request.files.getlist('files')
        if files:
            return self.post_with_multi(filepath, files)
        return self.post_with_one(filepath)

    def post_with_one(self, filepath, f=None):
        if not f:
            f = request.files.get("file")
        if not f:
            return HTTP.BAD_REQUEST(message="file is null")

        force = is_true(request.data.get("force"))

        filename = secure_filename(f.filename)
        if not file_is_allowed(filename):
            msg = _("%(name)s unsupported extension", name=filename)
            return HTTP.BAD_REQUEST(message=msg)

        file_type = f.content_type
        hash = gen_hash(f)

        size = gen_size(f)