sort() - Code Metrics - Inspection of "browsepy as a command, update readme" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — dev-4.1 ( 57d16a...5679a9 )

by Felipe A.

created 2016-11-17 10:40 UTC

sort() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
dl	0
loc	27
rs	7.5384
c	0
b	0
f	0

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

import logging
import os
import os.path
import json
import base64

from flask import Flask, Response, request, render_template, redirect, \
                  url_for, send_from_directory, stream_with_context, \
                  make_response
from werkzeug.exceptions import NotFound

from .__meta__ import __app__, __version__, __license__, __author__  # noqa
from .manager import PluginManager
from .file import Node, OutsideRemovableBase, OutsideDirectoryBase, \
                  secure_filename
from . import compat

__basedir__ = os.path.abspath(os.path.dirname(compat.fsdecode(__file__)))

logger = logging.getLogger(__name__)

app = Flask(
    __name__,
    static_url_path='/static',
    static_folder=os.path.join(__basedir__, "static"),
    template_folder=os.path.join(__basedir__, "templates")
    )
app.config.update(
    directory_base=compat.getcwd(),
    directory_start=compat.getcwd(),
    directory_remove=None,
    directory_upload=None,
    directory_tar_buffsize=262144,
    directory_downloadable=True,
    use_binary_multiples=True,
    plugin_modules=[],
    plugin_namespaces=(
        'browsepy.plugin',
        'browsepy_',
        '',
        ),
    )
app.jinja_env.add_extension('browsepy.extensions.HTMLCompress')

if "BROWSEPY_SETTINGS" in os.environ:
    app.config.from_envvar("BROWSEPY_SETTINGS")

plugin_manager = PluginManager(app)


def iter_cookie_browse_sorting():
    '''
    Get sorting-cookie data of current request.

    :yields: tuple of path and sorting property
    :ytype: 2-tuple of strings
    '''
    try:
        data = request.cookies.get('browse-sorting', 'e30=').encode('ascii')
        for path, prop in json.loads(base64.b64decode(data).decode('utf-8')):
            yield path, prop
    except (ValueError, TypeError, KeyError) as e:
        logger.exception(e)


def get_cookie_browse_sorting(path, default):
    '''
    Get sorting-cookie data for path of current request.

    :returns: sorting property
    :rtype: string
    '''
    for cpath, cprop in iter_cookie_browse_sorting():
        if path == cpath:
            return cprop
    return default


def browse_sortkey_reverse(prop):
    '''
    Get sorting function for browse

    :returns: tuple with sorting gunction and reverse bool
    :rtype: tuple of a dict and a bool
    '''
    if prop.startswith('-'):
        prop = prop[1:]
        reverse = True
    elif prop.startswith('+'):
        prop = prop[1:]
        reverse = False
    else:
        reverse = False

    if prop == 'text':
        return (
            lambda x: (
                x.is_directory == reverse,
                x.link.text.lower() if x.link and x.link.text else x.name
                ),
            reverse
            )
    if prop == 'size':
        return (
            lambda x: (
                x.is_directory == reverse,
                x.stats.st_size
                ),
            reverse
            )
    return (
        lambda x: (
            x.is_directory == reverse,
            getattr(x, prop, None)
            ),
        reverse
        )


def stream_template(template_name, **context):
    '''
    Some templates can be huge, this function returns an streaming response,
    sending the content in chunks and preventing from timeout.

    :param template_name: template
    :param **context: parameters for templates.
    :yields: HTML strings
    '''
    app.update_template_context(context)
    template = app.jinja_env.get_template(template_name)
    stream = template.generate(context)
    return Response(stream_with_context(stream))


@app.context_processor
def template_globals():
    return {
        'manager': app.extensions['plugin_manager'],
        'len': len,
        }


@app.route('/sort/<string:property>', defaults={"path": ""})
@app.route('/sort/<string:property>/<path:path>')
def sort(property, path):
    try:
        directory = Node.from_urlpath(path)
    except OutsideDirectoryBase:
        return NotFound()

    if not directory.is_directory:
        return NotFound()

    data = [
        (cpath, cprop)
        for cpath, cprop in iter_cookie_browse_sorting()
        if cpath != path
        ]
    data.append((path, property))
    raw_data = base64.b64encode(json.dumps(data).encode('utf-8'))

    # prevent cookie becoming too large
    while len(raw_data) > 3975:  # 4000 - len('browse-sorting=""; Path=/')
        data.pop(0)
        raw_data = base64.b64encode(json.dumps(data).encode('utf-8'))

    response = redirect(url_for(".browse", path=directory.urlpath))
    response.set_cookie('browse-sorting', raw_data)
    return response


@app.route("/browse", defaults={"path": ""})
@app.route('/browse/<path:path>')
def browse(path):
    sort_property = get_cookie_browse_sorting(path, 'text')
    sort_fnc, sort_reverse = browse_sortkey_reverse(sort_property)

    try:
        directory = Node.from_urlpath(path)
        if directory.is_directory:
            return stream_template(
                'browse.html',
                file=directory,
                sort_property=sort_property,
                sort_fnc=sort_fnc,
                sort_reverse=sort_reverse
                )
    except OutsideDirectoryBase:
        pass
    return NotFound()


@app.route('/open/<path:path>', endpoint="open")
def open_file(path):
    try:
        file = Node.from_urlpath(path)
        if file.is_file:
            return send_from_directory(file.parent.path, file.name)
    except OutsideDirectoryBase:
        pass
    return NotFound()


@app.route("/download/file/<path:path>")
def download_file(path):
    try:
        file = Node.from_urlpath(path)
        if file.is_file:
            return file.download()
    except OutsideDirectoryBase:
        pass
    return NotFound()


@app.route("/download/directory/<path:path>.tgz")
def download_directory(path):
    try:
        directory = Node.from_urlpath(path)
        if directory.is_directory:
            return directory.download()
    except OutsideDirectoryBase:
        pass
    return NotFound()


@app.route("/remove/<path:path>", methods=("GET", "POST"))
def remove(path):
    try:
        file = Node.from_urlpath(path)
    except OutsideDirectoryBase:
        return NotFound()
    if request.method == 'GET':
        if not file.can_remove:
            return NotFound()
        return render_template('remove.html', file=file)
    parent = file.parent
    if parent is None:
        # base is not removable
        return NotFound()

    try:
        file.remove()
    except OutsideRemovableBase:
        return NotFound()

    return redirect(url_for(".browse", path=parent.urlpath))


@app.route("/upload", defaults={'path': ''}, methods=("POST",))
@app.route("/upload/<path:path>", methods=("POST",))
def upload(path):
    try:
        directory = Node.from_urlpath(path)
    except OutsideDirectoryBase:
        return NotFound()

    if not directory.is_directory or not directory.can_upload:
        return NotFound()

    for v in request.files.listvalues():
        for f in v:
            filename = secure_filename(f.filename)
            if filename:
                filename = directory.choose_filename(filename)
                filepath = os.path.join(directory.path, filename)
                f.save(filepath)
    return redirect(url_for(".browse", path=directory.urlpath))


@app.route("/")
def index():
    path = app.config["directory_start"] or app.config["directory_base"]
    try:
        urlpath = Node(path).urlpath
    except OutsideDirectoryBase:
        return NotFound()
    return browse(urlpath)


@app.after_request
def page_not_found(response):
    if response.status_code == 404:
        return make_response((render_template('404.html'), 404))
    return response


@app.errorhandler(404)
def page_not_found_error(e):
    return render_template('404.html'), 404


@app.errorhandler(500)
def internal_server_error(e):  # pragma: no cover
    logger.exception(e)
    return getattr(e, 'message', 'Internal server error'), 500


1			#!/usr/bin/env python
2			# -- coding: UTF-8 --
3
4			import logging
5			import os
6			import os.path
7			import json
8			import base64
9
10			from flask import Flask, Response, request, render_template, redirect, \
11			url_for, send_from_directory, stream_with_context, \
12			make_response
13			from werkzeug.exceptions import NotFound
14
15			from .__meta__ import __app__, __version__, __license__, __author__ # noqa
16			from .manager import PluginManager
17			from .file import Node, OutsideRemovableBase, OutsideDirectoryBase, \
18			secure_filename
19			from . import compat
20
21			__basedir__ = os.path.abspath(os.path.dirname(compat.fsdecode(__file__)))
22
23			logger = logging.getLogger(__name__)
24
25			app = Flask(
26			__name__,
27			static_url_path='/static',
28			static_folder=os.path.join(__basedir__, "static"),
29			template_folder=os.path.join(__basedir__, "templates")
30			)
31			app.config.update(
32			directory_base=compat.getcwd(),
33			directory_start=compat.getcwd(),
34			directory_remove=None,
35			directory_upload=None,
36			directory_tar_buffsize=262144,
37			directory_downloadable=True,
38			use_binary_multiples=True,
39			plugin_modules=[],
40			plugin_namespaces=(
41			'browsepy.plugin',
42			'browsepy_',
43			'',
44			),
45			)
46			app.jinja_env.add_extension('browsepy.extensions.HTMLCompress')
47
48			if "BROWSEPY_SETTINGS" in os.environ:
49			app.config.from_envvar("BROWSEPY_SETTINGS")
50
51			plugin_manager = PluginManager(app)
52
53
54			def iter_cookie_browse_sorting():
55			'''
56			Get sorting-cookie data of current request.
57
58			:yields: tuple of path and sorting property
59			:ytype: 2-tuple of strings
60			'''
61			try:
62			data = request.cookies.get('browse-sorting', 'e30=').encode('ascii')
63			for path, prop in json.loads(base64.b64decode(data).decode('utf-8')):
64			yield path, prop
65			except (ValueError, TypeError, KeyError) as e:
66			logger.exception(e)
67
68
69			def get_cookie_browse_sorting(path, default):
70			'''
71			Get sorting-cookie data for path of current request.
72
73			:returns: sorting property
74			:rtype: string
75			'''
76			for cpath, cprop in iter_cookie_browse_sorting():
77			if path == cpath:
78			return cprop
79			return default
80
81
82			def browse_sortkey_reverse(prop):
83			'''
84			Get sorting function for browse
85
86			:returns: tuple with sorting gunction and reverse bool
87			:rtype: tuple of a dict and a bool
88			'''
89			if prop.startswith('-'):
90			prop = prop[1:]
91			reverse = True
92			elif prop.startswith('+'):
93			prop = prop[1:]
94			reverse = False
95			else:
96			reverse = False
97
98			if prop == 'text':
99			return (
100			lambda x: (
101			x.is_directory == reverse,
102			x.link.text.lower() if x.link and x.link.text else x.name
103			),
104			reverse
105			)
106			if prop == 'size':
107			return (
108			lambda x: (
109			x.is_directory == reverse,
110			x.stats.st_size
111			),
112			reverse
113			)
114			return (
115			lambda x: (
116			x.is_directory == reverse,
117			getattr(x, prop, None)
118			),
119			reverse
120			)
121
122
123			def stream_template(template_name, **context):
124			'''
125			Some templates can be huge, this function returns an streaming response,
126			sending the content in chunks and preventing from timeout.
127
128			:param template_name: template
129			:param **context: parameters for templates.
130			:yields: HTML strings
131			'''
132			app.update_template_context(context)
133			template = app.jinja_env.get_template(template_name)
134			stream = template.generate(context)
135			return Response(stream_with_context(stream))
136
137
138			@app.context_processor
139			def template_globals():
140			return {
141			'manager': app.extensions['plugin_manager'],
142			'len': len,
143			}
144
145
146			@app.route('/sort/<string:property>', defaults={"path": ""})
147			@app.route('/sort/<string:property>/<path:path>')
148			def sort(property, path):
149			try:
150			directory = Node.from_urlpath(path)
151			except OutsideDirectoryBase:
152			return NotFound()
153
154			if not directory.is_directory:
155			return NotFound()
156
157			data = [
158			(cpath, cprop)
159			for cpath, cprop in iter_cookie_browse_sorting()
160			if cpath != path
161			]
162			data.append((path, property))
163			raw_data = base64.b64encode(json.dumps(data).encode('utf-8'))
164
165			# prevent cookie becoming too large
166			while len(raw_data) > 3975: # 4000 - len('browse-sorting=""; Path=/')
167			data.pop(0)
168			raw_data = base64.b64encode(json.dumps(data).encode('utf-8'))
169
170			response = redirect(url_for(".browse", path=directory.urlpath))
171			response.set_cookie('browse-sorting', raw_data)
172			return response
173
174
175			@app.route("/browse", defaults={"path": ""})
176			@app.route('/browse/<path:path>')
177			def browse(path):
178			sort_property = get_cookie_browse_sorting(path, 'text')
179			sort_fnc, sort_reverse = browse_sortkey_reverse(sort_property)
180
181			try:
182			directory = Node.from_urlpath(path)
183			if directory.is_directory:
184			return stream_template(
185			'browse.html',
186			file=directory,
187			sort_property=sort_property,
188			sort_fnc=sort_fnc,
189			sort_reverse=sort_reverse
190			)
191			except OutsideDirectoryBase:
192			pass
193			return NotFound()
194
195
196			@app.route('/open/<path:path>', endpoint="open")
197			def open_file(path):
198			try:
199			file = Node.from_urlpath(path)
200			if file.is_file:
201			return send_from_directory(file.parent.path, file.name)
202			except OutsideDirectoryBase:
203			pass
204			return NotFound()
205
206
207			@app.route("/download/file/<path:path>")
208			def download_file(path):
209			try:
210			file = Node.from_urlpath(path)
211			if file.is_file:
212			return file.download()
213			except OutsideDirectoryBase:
214			pass
215			return NotFound()
216
217
218			@app.route("/download/directory/<path:path>.tgz")
219			def download_directory(path):
220			try:
221			directory = Node.from_urlpath(path)
222			if directory.is_directory:
223			return directory.download()
224			except OutsideDirectoryBase:
225			pass
226			return NotFound()
227
228
229			@app.route("/remove/<path:path>", methods=("GET", "POST"))
230			def remove(path):
231			try:
232			file = Node.from_urlpath(path)
233			except OutsideDirectoryBase:
234			return NotFound()
235			if request.method == 'GET':
236			if not file.can_remove:
237			return NotFound()
238			return render_template('remove.html', file=file)
239			parent = file.parent
240			if parent is None:
241			# base is not removable
242			return NotFound()
243
244			try:
245			file.remove()
246			except OutsideRemovableBase:
247			return NotFound()
248
249			return redirect(url_for(".browse", path=parent.urlpath))
250
251
252			@app.route("/upload", defaults={'path': ''}, methods=("POST",))
253			@app.route("/upload/<path:path>", methods=("POST",))
254			def upload(path):
255			try:
256			directory = Node.from_urlpath(path)
257			except OutsideDirectoryBase:
258			return NotFound()
259
260			if not directory.is_directory or not directory.can_upload:
261			return NotFound()
262
263			for v in request.files.listvalues():
264			for f in v:
265			filename = secure_filename(f.filename)
266			if filename:
267			filename = directory.choose_filename(filename)
268			filepath = os.path.join(directory.path, filename)
269			f.save(filepath)
270			return redirect(url_for(".browse", path=directory.urlpath))
271
272
273			@app.route("/")
274			def index():
275			path = app.config["directory_start"] or app.config["directory_base"]
276			try:
277			urlpath = Node(path).urlpath
278			except OutsideDirectoryBase:
279			return NotFound()
280			return browse(urlpath)
281
282
283			@app.after_request
284			def page_not_found(response):
285			if response.status_code == 404:
286			return make_response((render_template('404.html'), 404))
287			return response
288
289
290			@app.errorhandler(404)
291			def page_not_found_error(e):
292			return render_template('404.html'), 404
293
294
295			@app.errorhandler(500)
296			def internal_server_error(e): # pragma: no cover
297			logger.exception(e)
298			return getattr(e, 'message', 'Internal server error'), 500
299

ergoithz / browsepy

Push — dev-4.1 ( 57d16a...5679a9 )

sort() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like