mirrors/calibre-web
1
0
Fork 0
mirror of https://github.com/janeczku/calibre-web synced 2024-11-25 02:57:22 +00:00
Code Issues Releases Wiki Activity

Make drive letters available in file picker

Browse Source
This commit is contained in:
Ozzie Isaacs 2022-10-03 15:17:17 +02:00
parent e22ecda137
commit 7eef44f73c
1 changed files with 77 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

90
cps/admin.py Executable file → Normal file
View File

@ -26,11 +26,12 @@ import base64
import json import json
import operator import operator
import time import time
import sys
import string
from datetime import datetime, timedelta from datetime import datetime, timedelta
from datetime import time as datetime_time from datetime import time as datetime_time
from functools import wraps from functools import wraps
from flask import Blueprint, flash, redirect, url_for, abort, request, make_response, send_from_directory, g, Response from flask import Blueprint, flash, redirect, url_for, abort, request, make_response, send_from_directory, g, Response
from flask_login import login_required, current_user, logout_user, confirm_login from flask_login import login_required, current_user, logout_user, confirm_login
from flask_babel import gettext as _ from flask_babel import gettext as _
@ -52,7 +53,6 @@ from .services.worker import WorkerThread
from .babel import get_available_translations, get_available_locale, get_user_locale_language from .babel import get_available_translations, get_available_locale, get_user_locale_language
from . import debug_info from . import debug_info
log = logger.create() log = logger.create()
feature_support = { feature_support = {
@ -63,16 +63,18 @@ feature_support = {
'gmail': bool(services.gmail), 'gmail': bool(services.gmail),
'scheduler': schedule.use_APScheduler, 'scheduler': schedule.use_APScheduler,
'gdrive': gdrive_support 'gdrive': gdrive_support
} }
try: try:
import rarfile # pylint: disable=unused-import import rarfile # pylint: disable=unused-import
feature_support['rar'] = True feature_support['rar'] = True
except (ImportError, SyntaxError): except (ImportError, SyntaxError):
feature_support['rar'] = False feature_support['rar'] = False
try: try:
from .oauth_bb import oauth_check, oauthblueprints from .oauth_bb import oauth_check, oauthblueprints
feature_support['oauth'] = True feature_support['oauth'] = True
except ImportError as err: except ImportError as err:
log.debug('Cannot import Flask-Dance, login with Oauth will not work: %s', err) log.debug('Cannot import Flask-Dance, login with Oauth will not work: %s', err)
@ -80,7 +82,6 @@ except ImportError as err:
oauthblueprints = [] oauthblueprints = []
oauth_check = {} oauth_check = {}
admi = Blueprint('admin', __name__) admi = Blueprint('admin', __name__)
@ -159,6 +160,7 @@ def shutdown():
show_text['text'] = _(u'Unknown command') show_text['text'] = _(u'Unknown command')
return json.dumps(show_text), 400 return json.dumps(show_text), 400
@admi.route("/metadata_backup", methods=["POST"]) @admi.route("/metadata_backup", methods=["POST"])
@login_required @login_required
@admin_required @admin_required
@ -272,9 +274,9 @@ def calibreweb_alive():
@login_required @login_required
@admin_required @admin_required
def view_configuration(): def view_configuration():
read_column = calibre_db.session.query(db.CustomColumns)\ read_column = calibre_db.session.query(db.CustomColumns) \
.filter(and_(db.CustomColumns.datatype == 'bool', db.CustomColumns.mark_for_delete == 0)).all() .filter(and_(db.CustomColumns.datatype == 'bool', db.CustomColumns.mark_for_delete == 0)).all()
restrict_columns = calibre_db.session.query(db.CustomColumns)\ restrict_columns = calibre_db.session.query(db.CustomColumns) \
.filter(and_(db.CustomColumns.datatype == 'text', db.CustomColumns.mark_for_delete == 0)).all() .filter(and_(db.CustomColumns.datatype == 'text', db.CustomColumns.mark_for_delete == 0)).all()
languages = calibre_db.speaking_language() languages = calibre_db.speaking_language()
translations = get_available_locale() translations = get_available_locale()
@ -293,11 +295,11 @@ def edit_user_table():
languages = calibre_db.speaking_language() languages = calibre_db.speaking_language()
translations = get_available_locale() translations = get_available_locale()
all_user = ub.session.query(ub.User) all_user = ub.session.query(ub.User)
tags = calibre_db.session.query(db.Tags)\ tags = calibre_db.session.query(db.Tags) \
.join(db.books_tags_link)\ .join(db.books_tags_link) \
.join(db.Books)\ .join(db.Books) \
.filter(calibre_db.common_filters()) \ .filter(calibre_db.common_filters()) \
.group_by(text('books_tags_link.tag'))\ .group_by(text('books_tags_link.tag')) \
.order_by(db.Tags.name).all() .order_by(db.Tags.name).all()
if config.config_restricted_column: if config.config_restricted_column:
custom_values = calibre_db.session.query(db.cc_classes[config.config_restricted_column]).all() custom_values = calibre_db.session.query(db.cc_classes[config.config_restricted_column]).all()
@ -477,12 +479,12 @@ def edit_list_user(param):
[constants.ROLE_ADMIN, constants.ROLE_PASSWD, constants.ROLE_EDIT_SHELFS]: [constants.ROLE_ADMIN, constants.ROLE_PASSWD, constants.ROLE_EDIT_SHELFS]:
raise Exception(_("Guest can't have this role")) raise Exception(_("Guest can't have this role"))
# check for valid value, last on checks for power of 2 value # check for valid value, last on checks for power of 2 value
if value > 0 and value <= constants.ROLE_VIEWER and (value & value-1 == 0 or value == 1): if value > 0 and value <= constants.ROLE_VIEWER and (value & value - 1 == 0 or value == 1):
if vals['value'] == 'true': if vals['value'] == 'true':
user.role |= value user.role |= value
elif vals['value'] == 'false': elif vals['value'] == 'false':
if value == constants.ROLE_ADMIN: if value == constants.ROLE_ADMIN:
if not ub.session.query(ub.User).\ if not ub.session.query(ub.User). \
filter(ub.User.role.op('&')(constants.ROLE_ADMIN) == constants.ROLE_ADMIN, filter(ub.User.role.op('&')(constants.ROLE_ADMIN) == constants.ROLE_ADMIN,
ub.User.id != user.id).count(): ub.User.id != user.id).count():
return Response( return Response(
@ -499,7 +501,7 @@ def edit_list_user(param):
if user.name == "Guest" and value == constants.SIDEBAR_READ_AND_UNREAD: if user.name == "Guest" and value == constants.SIDEBAR_READ_AND_UNREAD:
raise Exception(_("Guest can't have this view")) raise Exception(_("Guest can't have this view"))
# check for valid value, last on checks for power of 2 value # check for valid value, last on checks for power of 2 value
if value > 0 and value <= constants.SIDEBAR_LIST and (value & value-1 == 0 or value == 1): if value > 0 and value <= constants.SIDEBAR_LIST and (value & value - 1 == 0 or value == 1):
if vals['value'] == 'true': if vals['value'] == 'true':
user.sidebar_view |= value user.sidebar_view |= value
elif vals['value'] == 'false': elif vals['value'] == 'false':
@ -652,7 +654,7 @@ def edit_domain(allow):
@admin_required @admin_required
def add_domain(allow): def add_domain(allow):
domain_name = request.form.to_dict()['domainname'].replace('*', '%').replace('?', '_').lower() domain_name = request.form.to_dict()['domainname'].replace('*', '%').replace('?', '_').lower()
check = ub.session.query(ub.Registration).filter(ub.Registration.domain == domain_name)\ check = ub.session.query(ub.Registration).filter(ub.Registration.domain == domain_name) \
.filter(ub.Registration.allow == allow).first() .filter(ub.Registration.allow == allow).first()
if not check: if not check:
new_domain = ub.Registration(domain=domain_name, allow=allow) new_domain = ub.Registration(domain=domain_name, allow=allow)
@ -871,15 +873,15 @@ def delete_restriction(res_type, user_id):
@admin_required @admin_required
def list_restriction(res_type, user_id): def list_restriction(res_type, user_id):
if res_type == 0: # Tags as template if res_type == 0: # Tags as template
restrict = [{'Element': x, 'type': _('Deny'), 'id': 'd'+str(i)} restrict = [{'Element': x, 'type': _('Deny'), 'id': 'd' + str(i)}
for i, x in enumerate(config.list_denied_tags()) if x != ''] for i, x in enumerate(config.list_denied_tags()) if x != '']
allow = [{'Element': x, 'type': _('Allow'), 'id': 'a'+str(i)} allow = [{'Element': x, 'type': _('Allow'), 'id': 'a' + str(i)}
for i, x in enumerate(config.list_allowed_tags()) if x != ''] for i, x in enumerate(config.list_allowed_tags()) if x != '']
json_dumps = restrict + allow json_dumps = restrict + allow
elif res_type == 1: # CustomC as template elif res_type == 1: # CustomC as template
restrict = [{'Element': x, 'type': _('Deny'), 'id': 'd'+str(i)} restrict = [{'Element': x, 'type': _('Deny'), 'id': 'd' + str(i)}
for i, x in enumerate(config.list_denied_column_values()) if x != ''] for i, x in enumerate(config.list_denied_column_values()) if x != '']
allow = [{'Element': x, 'type': _('Allow'), 'id': 'a'+str(i)} allow = [{'Element': x, 'type': _('Allow'), 'id': 'a' + str(i)}
for i, x in enumerate(config.list_allowed_column_values()) if x != ''] for i, x in enumerate(config.list_allowed_column_values()) if x != '']
json_dumps = restrict + allow json_dumps = restrict + allow
elif res_type == 2: # Tags per user elif res_type == 2: # Tags per user
@ -887,9 +889,9 @@ def list_restriction(res_type, user_id):
usr = ub.session.query(ub.User).filter(ub.User.id == user_id).first() usr = ub.session.query(ub.User).filter(ub.User.id == user_id).first()
else: else:
usr = current_user usr = current_user
restrict = [{'Element': x, 'type': _('Deny'), 'id': 'd'+str(i)} restrict = [{'Element': x, 'type': _('Deny'), 'id': 'd' + str(i)}
for i, x in enumerate(usr.list_denied_tags()) if x != ''] for i, x in enumerate(usr.list_denied_tags()) if x != '']
allow = [{'Element': x, 'type': _('Allow'), 'id': 'a'+str(i)} allow = [{'Element': x, 'type': _('Allow'), 'id': 'a' + str(i)}
for i, x in enumerate(usr.list_allowed_tags()) if x != ''] for i, x in enumerate(usr.list_allowed_tags()) if x != '']
json_dumps = restrict + allow json_dumps = restrict + allow
elif res_type == 3: # CustomC per user elif res_type == 3: # CustomC per user
@ -897,9 +899,9 @@ def list_restriction(res_type, user_id):
usr = ub.session.query(ub.User).filter(ub.User.id == user_id).first() usr = ub.session.query(ub.User).filter(ub.User.id == user_id).first()
else: else:
usr = current_user usr = current_user
restrict = [{'Element': x, 'type': _('Deny'), 'id': 'd'+str(i)} restrict = [{'Element': x, 'type': _('Deny'), 'id': 'd' + str(i)}
for i, x in enumerate(usr.list_denied_column_values()) if x != ''] for i, x in enumerate(usr.list_denied_column_values()) if x != '']
allow = [{'Element': x, 'type': _('Allow'), 'id': 'a'+str(i)} allow = [{'Element': x, 'type': _('Allow'), 'id': 'a' + str(i)}
for i, x in enumerate(usr.list_allowed_column_values()) if x != ''] for i, x in enumerate(usr.list_allowed_column_values()) if x != '']
json_dumps = restrict + allow json_dumps = restrict + allow
else: else:
@ -965,7 +967,7 @@ def prepare_tags(user, action, tags_name, id_list):
raise Exception(_("Tag not found")) raise Exception(_("Tag not found"))
new_tags_list = [x.name for x in tags] new_tags_list = [x.name for x in tags]
else: else:
tags = calibre_db.session.query(db.cc_classes[config.config_restricted_column])\ tags = calibre_db.session.query(db.cc_classes[config.config_restricted_column]) \
.filter(db.cc_classes[config.config_restricted_column].id.in_(id_list)).all() .filter(db.cc_classes[config.config_restricted_column].id.in_(id_list)).all()
new_tags_list = [x.value for x in tags] new_tags_list = [x.value for x in tags]
saved_tags_list = user.__dict__[tags_name].split(",") if len(user.__dict__[tags_name]) else [] saved_tags_list = user.__dict__[tags_name].split(",") if len(user.__dict__[tags_name]) else []
@ -978,6 +980,19 @@ def prepare_tags(user, action, tags_name, id_list):
return ",".join(saved_tags_list) return ",".join(saved_tags_list)
def get_drives(current):
drive_letters = []
for d in string.ascii_uppercase:
if os.path.exists('{}:'.format(d)) and current[0].lower() != d.lower():
drive = "{}:\\".format(d)
data = {"name": drive, "fullpath": drive}
data["sort"] = "_" + data["fullpath"].lower()
data["type"] = "dir"
data["size"] = ""
drive_letters.append(data)
return drive_letters
def pathchooser(): def pathchooser():
browse_for = "folder" browse_for = "folder"
folder_only = request.args.get('folder', False) == "true" folder_only = request.args.get('folder', False) == "true"
@ -985,40 +1000,41 @@ def pathchooser():
path = os.path.normpath(request.args.get('path', "")) path = os.path.normpath(request.args.get('path', ""))
if os.path.isfile(path): if os.path.isfile(path):
oldfile = path old_file = path
path = os.path.dirname(path) path = os.path.dirname(path)
else: else:
oldfile = "" old_file = ""
absolute = False absolute = False
if os.path.isdir(path): if os.path.isdir(path):
# if os.path.isabs(path):
cwd = os.path.realpath(path) cwd = os.path.realpath(path)
absolute = True absolute = True
# else:
# cwd = os.path.relpath(path)
else: else:
cwd = os.getcwd() cwd = os.getcwd()
cwd = os.path.normpath(os.path.realpath(cwd)) cwd = os.path.normpath(os.path.realpath(cwd))
parentdir = os.path.dirname(cwd) parent_dir = os.path.dirname(cwd)
if not absolute: if not absolute:
if os.path.realpath(cwd) == os.path.realpath("/"): if os.path.realpath(cwd) == os.path.realpath("/"):
cwd = os.path.relpath(cwd) cwd = os.path.relpath(cwd)
else: else:
cwd = os.path.relpath(cwd) + os.path.sep cwd = os.path.relpath(cwd) + os.path.sep
parentdir = os.path.relpath(parentdir) + os.path.sep parent_dir = os.path.relpath(parent_dir) + os.path.sep
if os.path.realpath(cwd) == os.path.realpath("/"): files = []
parentdir = "" if os.path.realpath(cwd) == os.path.realpath("/") \
or (sys.platform == "win32" and os.path.realpath(cwd)[1:] == os.path.realpath("/")[1:]):
# we are in root
parent_dir = ""
if sys.platform == "win32":
files = get_drives(cwd)
try: try:
folders = os.listdir(cwd) folders = os.listdir(cwd)
except Exception: except Exception:
folders = [] folders = []
files = []
for f in folders: for f in folders:
try: try:
data = {"name": f, "fullpath": os.path.join(cwd, f)} data = {"name": f, "fullpath": os.path.join(cwd, f)}
@ -1051,9 +1067,9 @@ def pathchooser():
context = { context = {
"cwd": cwd, "cwd": cwd,
"files": files, "files": files,
"parentdir": parentdir, "parentdir": parent_dir,
"type": browse_for, "type": browse_for,
"oldfile": oldfile, "oldfile": old_file,
"absolute": absolute, "absolute": absolute,
} }
return json.dumps(context) return json.dumps(context)
@ -1277,7 +1293,7 @@ def update_mailsettings():
_config_int(to_save, "mail_port") _config_int(to_save, "mail_port")
_config_int(to_save, "mail_use_ssl") _config_int(to_save, "mail_use_ssl")
_config_string(to_save, "mail_password") _config_string(to_save, "mail_password")
_config_int(to_save, "mail_size", lambda y: int(y)*1024*1024) _config_int(to_save, "mail_size", lambda y: int(y) * 1024 * 1024)
config.mail_server = to_save.get('mail_server', "").strip() config.mail_server = to_save.get('mail_server', "").strip()
config.mail_from = to_save.get('mail_from', "").strip() config.mail_from = to_save.get('mail_from', "").strip()
config.mail_login = to_save.get('mail_login', "").strip() config.mail_login = to_save.get('mail_login', "").strip()
@ -1317,7 +1333,7 @@ def edit_scheduledtasks():
duration_field = list() duration_field = list()
for n in range(24): for n in range(24):
time_field.append((n, format_time(datetime_time(hour=n), format="short",))) time_field.append((n, format_time(datetime_time(hour=n), format="short", )))
for n in range(5, 65, 5): for n in range(5, 65, 5):
t = timedelta(hours=n // 60, minutes=n % 60) t = timedelta(hours=n // 60, minutes=n % 60)
duration_field.append((n, format_timedelta(t, threshold=.97))) duration_field.append((n, format_timedelta(t, threshold=.97)))