Archives
/
calibre-web
mirror of https://github.com/janeczku/calibre-web
2
0
Fork 0
Code Issues Releases Wiki Activity

Refactored admin.py

Browse Source
pull/1409/head
Ozzieisaacs 4 years ago
parent 718d50a037
commit e8aedeac36
1 changed files with 286 additions and 256 deletions
Show Stats Download Patch File Download Diff File

366
cps/admin.py
Unescape Escape View File

@ -429,7 +429,6 @@ def delete_restriction(res_type):
return ""
#@admi.route("/ajax/listrestriction/<int:type>/<int:user_id>", defaults={'user_id': '0'})
@admi.route("/ajax/listrestriction/<int:res_type>")
@login_required
@admin_required
@ -475,6 +474,7 @@ def list_restriction(res_type):
response.headers["Content-Type"] = "application/json; charset=utf-8"
return response
@admi.route("/config", methods=["GET", "POST"])
@unconfigured
def basic_configuration():
@ -484,19 +484,23 @@ def basic_configuration():
return _configuration_result()
def _configuration_update_helper():
reboot_required = False
db_change = False
to_save = request.form.to_dict()
def _config_int(to_save, x):
return config.set_from_dictionary(to_save, x, int)
_config_string = lambda x: config.set_from_dictionary(to_save, x, lambda y: y.strip() if y else y)
_config_int = lambda x: config.set_from_dictionary(to_save, x, int)
_config_checkbox = lambda x: config.set_from_dictionary(to_save, x, lambda y: y == "on", False)
_config_checkbox_int = lambda x: config.set_from_dictionary(to_save, x, lambda y: 1 if (y == "on") else 0, 0)
db_change |= _config_string("config_calibre_dir")
def _config_checkbox(to_save, x):
return config.set_from_dictionary(to_save, x, lambda y: y == "on", False)
# Google drive setup
def _config_checkbox_int(to_save, x):
return config.set_from_dictionary(to_save, x, lambda y: 1 if (y == "on") else 0, 0)
def _config_string(to_save, x):
return config.set_from_dictionary(to_save, x, lambda y: y.strip() if y else y)
def _configuration_gdrive_helper(to_save):
if not os.path.isfile(gdriveutils.SETTINGS_YAML):
config.config_use_google_drive = False
@ -515,45 +519,56 @@ def _configuration_update_helper():
# always show google drive settings, but in case of error deny support
config.config_use_google_drive = (not gdriveError) and ("config_use_google_drive" in to_save)
if _config_string("config_google_drive_folder"):
if _config_string(to_save, "config_google_drive_folder"):
gdriveutils.deleteDatabaseOnChange()
return gdriveError
reboot_required |= _config_int("config_port")
reboot_required |= _config_string("config_keyfile")
if config.config_keyfile and not os.path.isfile(config.config_keyfile):
return _configuration_result(_('Keyfile Location is not Valid, Please Enter Correct Path'), gdriveError)
reboot_required |= _config_string("config_certfile")
if config.config_certfile and not os.path.isfile(config.config_certfile):
return _configuration_result(_('Certfile Location is not Valid, Please Enter Correct Path'), gdriveError)
_config_checkbox_int("config_uploading")
_config_checkbox_int("config_anonbrowse")
_config_checkbox_int("config_public_reg")
reboot_required |= _config_checkbox_int("config_kobo_sync")
_config_checkbox_int("config_kobo_proxy")
def _configuration_oauth_helper(to_save):
active_oauths = 0
for element in oauthblueprints:
if to_save["config_" + str(element['id']) + "_oauth_client_id"] != element['oauth_client_id'] \
or to_save["config_" + str(element['id']) + "_oauth_client_secret"] != element['oauth_client_secret']:
reboot_required = True
element['oauth_client_id'] = to_save["config_" + str(element['id']) + "_oauth_client_id"]
element['oauth_client_secret'] = to_save["config_" + str(element['id']) + "_oauth_client_secret"]
if to_save["config_" + str(element['id']) + "_oauth_client_id"] \
and to_save["config_" + str(element['id']) + "_oauth_client_secret"]:
active_oauths += 1
element["active"] = 1
else:
element["active"] = 0
ub.session.query(ub.OAuthProvider).filter(ub.OAuthProvider.id == element['id']).update(
{"oauth_client_id": to_save["config_" + str(element['id']) + "_oauth_client_id"],
"oauth_client_secret": to_save["config_" + str(element['id']) + "_oauth_client_secret"],
"active": element["active"]})
_config_string("config_calibre")
_config_string("config_converterpath")
_config_string("config_kepubifypath")
def _configuration_logfile_helper(gdriveError, to_save):
reboot_required = False
reboot_required |= _config_int(to_save, "config_log_level")
reboot_required |= _config_string(to_save, "config_logfile")
if not logger.is_valid_logfile(config.config_logfile):
return _configuration_result(_('Logfile Location is not Valid, Please Enter Correct Path'), gdriveError)
reboot_required |= _config_int("config_login_type")
reboot_required |= _config_checkbox_int(to_save, "config_access_log")
reboot_required |= _config_string(to_save, "config_access_logfile")
if not logger.is_valid_logfile(config.config_access_logfile):
return _configuration_result(_('Access Logfile Location is not Valid, Please Enter Correct Path'), gdriveError)
return reboot_required
#LDAP configurator,
if config.config_login_type == constants.LOGIN_LDAP:
reboot_required |= _config_string("config_ldap_provider_url")
reboot_required |= _config_int("config_ldap_port")
reboot_required |= _config_int("config_ldap_authentication")
reboot_required |= _config_string("config_ldap_dn")
reboot_required |= _config_string("config_ldap_serv_username")
reboot_required |= _config_string("config_ldap_user_object")
reboot_required |= _config_string("config_ldap_group_object_filter")
reboot_required |= _config_string("config_ldap_group_members_field")
reboot_required |= _config_checkbox("config_ldap_openldap")
reboot_required |= _config_int("config_ldap_encryption")
reboot_required |= _config_string("config_ldap_cert_path")
_config_string("config_ldap_group_name")
def _configuration_ldap_helper(gdriveError, to_save):
reboot_required = False
reboot_required |= _config_string(to_save, "config_ldap_provider_url")
reboot_required |= _config_int(to_save, "config_ldap_port")
reboot_required |= _config_int(to_save, "config_ldap_authentication")
reboot_required |= _config_string(to_save, "config_ldap_dn")
reboot_required |= _config_string(to_save, "config_ldap_serv_username")
reboot_required |= _config_string(to_save, "config_ldap_user_object")
reboot_required |= _config_string(to_save, "config_ldap_group_object_filter")
reboot_required |= _config_string(to_save, "config_ldap_group_members_field")
reboot_required |= _config_checkbox(to_save, "config_ldap_openldap")
reboot_required |= _config_int(to_save, "config_ldap_encryption")
reboot_required |= _config_string(to_save, "config_ldap_cert_path")
_config_string(to_save, "config_ldap_group_name")
if "config_ldap_serv_password" in to_save and to_save["config_ldap_serv_password"] != "":
reboot_required |= 1
config.set_from_dictionary(to_save, "config_ldap_serv_password", base64.b64encode, encode='UTF-8')
@ -592,61 +607,74 @@ def _configuration_update_helper():
if config.config_ldap_cert_path and not os.path.isdir(config.config_ldap_cert_path):
return _configuration_result(_('LDAP Certificate Location is not Valid, Please Enter Correct Path'),
gdriveError)
return reboot_required
def _configuration_update_helper():
reboot_required = False
db_change = False
to_save = request.form.to_dict()
db_change |= _config_string(to_save, "config_calibre_dir")
# Google drive setup
gdriveError = _configuration_gdrive_helper(to_save)
reboot_required |= _config_int(to_save, "config_port")
reboot_required |= _config_string(to_save, "config_keyfile")
if config.config_keyfile and not os.path.isfile(config.config_keyfile):
return _configuration_result(_('Keyfile Location is not Valid, Please Enter Correct Path'), gdriveError)
reboot_required |= _config_string(to_save, "config_certfile")
if config.config_certfile and not os.path.isfile(config.config_certfile):
return _configuration_result(_('Certfile Location is not Valid, Please Enter Correct Path'), gdriveError)
_config_checkbox_int(to_save, "config_uploading")
_config_checkbox_int(to_save, "config_anonbrowse")
_config_checkbox_int(to_save, "config_public_reg")
reboot_required |= _config_checkbox_int(to_save, "config_kobo_sync")
_config_checkbox_int(to_save, "config_kobo_proxy")
_config_string(to_save, "config_calibre")
_config_string(to_save, "config_converterpath")
_config_string(to_save, "config_kepubifypath")
reboot_required |= _config_int(to_save, "config_login_type")
#LDAP configurator,
if config.config_login_type == constants.LOGIN_LDAP:
reboot_required |= _configuration_ldap_helper(to_save, gdriveError)
# Remote login configuration
_config_checkbox("config_remote_login")
_config_checkbox(to_save, "config_remote_login")
if not config.config_remote_login:
ub.session.query(ub.RemoteAuthToken).filter(ub.RemoteAuthToken.token_type==0).delete()
# Goodreads configuration
_config_checkbox("config_use_goodreads")
_config_string("config_goodreads_api_key")
_config_string("config_goodreads_api_secret")
_config_checkbox(to_save, "config_use_goodreads")
_config_string(to_save, "config_goodreads_api_key")
_config_string(to_save, "config_goodreads_api_secret")
if services.goodreads_support:
services.goodreads_support.connect(config.config_goodreads_api_key,
config.config_goodreads_api_secret,
config.config_use_goodreads)
_config_int("config_updatechannel")
_config_int(to_save, "config_updatechannel")
# Reverse proxy login configuration
_config_checkbox("config_allow_reverse_proxy_header_login")
_config_string("config_reverse_proxy_login_header_name")
_config_checkbox(to_save, "config_allow_reverse_proxy_header_login")
_config_string(to_save, "config_reverse_proxy_login_header_name")
# GitHub OAuth configuration
# OAuth configuration
if config.config_login_type == constants.LOGIN_OAUTH:
active_oauths = 0
for element in oauthblueprints:
if to_save["config_" + str(element['id']) + "_oauth_client_id"] != element['oauth_client_id'] \
or to_save["config_" + str(element['id']) + "_oauth_client_secret"] != element['oauth_client_secret']:
reboot_required = True
element['oauth_client_id'] = to_save["config_" + str(element['id']) + "_oauth_client_id"]
element['oauth_client_secret'] = to_save["config_" + str(element['id']) + "_oauth_client_secret"]
if to_save["config_"+str(element['id'])+"_oauth_client_id"] \
and to_save["config_"+str(element['id'])+"_oauth_client_secret"]:
active_oauths += 1
element["active"] = 1
else:
element["active"] = 0
ub.session.query(ub.OAuthProvider).filter(ub.OAuthProvider.id == element['id']).update(
{"oauth_client_id":to_save["config_"+str(element['id'])+"_oauth_client_id"],
"oauth_client_secret":to_save["config_"+str(element['id'])+"_oauth_client_secret"],
"active":element["active"]})
reboot_required |= _config_int("config_log_level")
reboot_required |= _config_string("config_logfile")
if not logger.is_valid_logfile(config.config_logfile):
return _configuration_result(_('Logfile Location is not Valid, Please Enter Correct Path'), gdriveError)
reboot_required |= _config_checkbox_int("config_access_log")
reboot_required |= _config_string("config_access_logfile")
if not logger.is_valid_logfile(config.config_access_logfile):
return _configuration_result(_('Access Logfile Location is not Valid, Please Enter Correct Path'), gdriveError)
_configuration_oauth_helper(to_save)
reboot_required |= _configuration_logfile_helper(to_save, gdriveError)
# Rarfile Content configuration
_config_string("config_rarfile_location")
_config_string(to_save, "config_rarfile_location")
unrar_status = helper.check_unrar(config.config_rarfile_location)
if unrar_status:
return _configuration_result(unrar_status, gdriveError)
@ -660,7 +688,6 @@ def _configuration_update_helper():
return _configuration_result('%s' % e, gdriveError)
if db_change:
# reload(db)
if not db.setup_db(config):
return _configuration_result(_('DB Location is not Valid, Please Enter Correct Path'), gdriveError)
@ -698,16 +725,7 @@ def _configuration_result(error_flash=None, gdriveError=None):
title=_(u"Basic Configuration"), page="config")
@admi.route("/admin/user/new", methods=["GET", "POST"])
@login_required
@admin_required
def new_user():
content = ub.User()
languages = speaking_language()
translations = [LC('en')] + babel.list_translations()
kobo_support = feature_support['kobo'] and config.config_kobo_sync
if request.method == "POST":
to_save = request.form.to_dict()
def _handle_new_user(to_save, content,languages, translations, kobo_support):
content.default_language = to_save["default_language"]
# content.mature_content = "Show_mature_content" in to_save
content.locale = to_save.get("locale", content.locale)
@ -724,9 +742,9 @@ def new_user():
registered_oauth=oauth_check, kobo_support=kobo_support,
title=_(u"Add new user"))
content.password = generate_password_hash(to_save["password"])
existing_user = ub.session.query(ub.User).filter(func.lower(ub.User.nickname) == to_save["nickname"].lower())\
existing_user = ub.session.query(ub.User).filter(func.lower(ub.User.nickname) == to_save["nickname"].lower()) \
.first()
existing_email = ub.session.query(ub.User).filter(ub.User.email == to_save["email"].lower())\
existing_email = ub.session.query(ub.User).filter(ub.User.email == to_save["email"].lower()) \
.first()
if not existing_user and not existing_email:
content.nickname = to_save["nickname"]
@ -754,79 +772,9 @@ def new_user():
except IntegrityError:
ub.session.rollback()
flash(_(u"Found an existing account for this e-mail address or nickname."), category="error")
else:
content.role = config.config_default_role
content.sidebar_view = config.config_default_show
return render_title_template("user_edit.html", new_user=1, content=content, translations=translations,
languages=languages, title=_(u"Add new user"), page="newuser",
kobo_support=kobo_support, registered_oauth=oauth_check)
@admi.route("/admin/mailsettings")
@login_required
@admin_required
def edit_mailsettings():
content = config.get_mail_settings()
return render_title_template("email_edit.html", content=content, title=_(u"Edit E-mail Server Settings"),
page="mailset")
@admi.route("/admin/mailsettings", methods=["POST"])
@login_required
@admin_required
def update_mailsettings():
to_save = request.form.to_dict()
log.debug("update_mailsettings %r", to_save)
_config_string = lambda x: config.set_from_dictionary(to_save, x, lambda y: y.strip() if y else y)
_config_int = lambda x: config.set_from_dictionary(to_save, x, int)
_config_string("mail_server")
_config_int("mail_port")
_config_int("mail_use_ssl")
_config_string("mail_login")
_config_string("mail_password")
_config_string("mail_from")
config.save()
if to_save.get("test"):
if current_user.email:
result = send_test_mail(current_user.email, current_user.nickname)
if result is None:
flash(_(u"Test e-mail successfully send to %(kindlemail)s", kindlemail=current_user.email),
category="success")
else:
flash(_(u"There was an error sending the Test e-mail: %(res)s", res=result), category="error")
else:
flash(_(u"Please configure your e-mail address first..."), category="error")
else:
flash(_(u"E-mail server settings updated"), category="success")
return edit_mailsettings()
@admi.route("/admin/user/<int:user_id>", methods=["GET", "POST"])
@login_required
@admin_required
def edit_user(user_id):
content = ub.session.query(ub.User).filter(ub.User.id == int(user_id)).first() # type: ub.User
if not content:
flash(_(u"User not found"), category="error")
return redirect(url_for('admin.admin'))
downloads = list()
languages = speaking_language()
translations = babel.list_translations() + [LC('en')]
kobo_support = feature_support['kobo'] and config.config_kobo_sync
for book in content.downloads:
downloadbook = db.session.query(db.Books).filter(db.Books.id == book.book_id).first()
if downloadbook:
downloads.append(downloadbook)
else:
ub.delete_download(book.book_id)
# ub.session.query(ub.Downloads).filter(book.book_id == ub.Downloads.book_id).delete()
# ub.session.commit()
if request.method == "POST":
to_save = request.form.to_dict()
def _handle_edit_user(to_save, content,languages, translations, kobo_support, downloads):
if "delete" in to_save:
if ub.session.query(ub.User).filter(ub.User.role.op('&')(constants.ROLE_ADMIN) == constants.ROLE_ADMIN,
ub.User.id != content.id).count():
@ -881,7 +829,7 @@ def edit_user(user_id):
return render_title_template("user_edit.html",
translations=translations,
languages=languages,
mail_configured = config.get_mail_server_configured(),
mail_configured=config.get_mail_server_configured(),
kobo_support=kobo_support,
new_user=0,
content=content,
@ -913,6 +861,88 @@ def edit_user(user_id):
except IntegrityError:
ub.session.rollback()
flash(_(u"An unknown error occured."), category="error")
@admi.route("/admin/user/new", methods=["GET", "POST"])
@login_required
@admin_required
def new_user():
content = ub.User()
languages = speaking_language()
translations = [LC('en')] + babel.list_translations()
kobo_support = feature_support['kobo'] and config.config_kobo_sync
if request.method == "POST":
to_save = request.form.to_dict()
_handle_new_user(to_save, content, languages, translations, kobo_support)
else:
content.role = config.config_default_role
content.sidebar_view = config.config_default_show
return render_title_template("user_edit.html", new_user=1, content=content, translations=translations,
languages=languages, title=_(u"Add new user"), page="newuser",
kobo_support=kobo_support, registered_oauth=oauth_check)
@admi.route("/admin/mailsettings")
@login_required
@admin_required
def edit_mailsettings():
content = config.get_mail_settings()
return render_title_template("email_edit.html", content=content, title=_(u"Edit E-mail Server Settings"),
page="mailset")
@admi.route("/admin/mailsettings", methods=["POST"])
@login_required
@admin_required
def update_mailsettings():
to_save = request.form.to_dict()
log.debug("update_mailsettings %r", to_save)
_config_string(to_save, "mail_server")
_config_int(to_save, "mail_port")
_config_int(to_save, "mail_use_ssl")
_config_string(to_save, "mail_login")
_config_string(to_save, "mail_password")
_config_string(to_save, "mail_from")
config.save()
if to_save.get("test"):
if current_user.email:
result = send_test_mail(current_user.email, current_user.nickname)
if result is None:
flash(_(u"Test e-mail successfully send to %(kindlemail)s", kindlemail=current_user.email),
category="success")
else:
flash(_(u"There was an error sending the Test e-mail: %(res)s", res=result), category="error")
else:
flash(_(u"Please configure your e-mail address first..."), category="error")
else:
flash(_(u"E-mail server settings updated"), category="success")
return edit_mailsettings()
@admi.route("/admin/user/<int:user_id>", methods=["GET", "POST"])
@login_required
@admin_required
def edit_user(user_id):
content = ub.session.query(ub.User).filter(ub.User.id == int(user_id)).first() # type: ub.User
if not content:
flash(_(u"User not found"), category="error")
return redirect(url_for('admin.admin'))
downloads = list()
languages = speaking_language()
translations = babel.list_translations() + [LC('en')]
kobo_support = feature_support['kobo'] and config.config_kobo_sync
for book in content.downloads:
downloadbook = db.session.query(db.Books).filter(db.Books.id == book.book_id).first()
if downloadbook:
downloads.append(downloadbook)
else:
ub.delete_download(book.book_id)
if request.method == "POST":
to_save = request.form.to_dict()
_handle_edit_user(to_save, content, languages, translations, kobo_support, downloads)
return render_title_template("user_edit.html",
translations=translations,
languages=languages,

Write Preview
Loading…
Cancel
Save