mirrors/calibre-web
1
0
Fork 0
mirror of https://github.com/janeczku/calibre-web synced 2024-12-25 09:30:31 +00:00
Code Issues Releases Wiki Activity

reenable startup logging

Browse Source 
Bugfixes from refactoring and merge
This commit is contained in:
Ozzie Isaacs 2022-04-26 14:44:55 +02:00
parent e7464f2694
commit 2e007a160e
14 changed files with 1516 additions and 1526 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
cps/__init__.py
View File

@ -27,16 +27,15 @@ from flask import Flask
from .MyLoginManager import MyLoginManager
from flask_principal import Principal
from . import logger
from .cli import CliParameter
from .constants import CONFIG_DIR
from .reverseproxy import ReverseProxied
from .server import WebServer
from .dep_check import dependency_check
from . import services
from .updater import Updater
from .babel import babel, BABEL_TRANSLATIONS
from .babel import babel
from . import config_sql
from . import logger
from . import cache_buster
from . import ub, db
@ -157,8 +156,8 @@ def create_app():
web_server.init_app(app, config)
babel.init_app(app)
BABEL_TRANSLATIONS.update(str(item) for item in babel.list_translations())
BABEL_TRANSLATIONS.add('en')
from . import services
if services.ldap:
services.ldap.init_app(app, config)

957
cps/admin.py
View File

File diff suppressed because it is too large Load Diff

18
cps/babel.py
View File

@ -1,4 +1,4 @@
from babel import Locale as LC
from babel import Locale
from babel import negotiate_locale
from flask_babel import Babel
from babel.core import UnknownLocaleError
@ -9,7 +9,7 @@ from . import logger
log = logger.create()
babel = Babel()
BABEL_TRANSLATIONS = set()
@babel.localeselector
def get_locale():
@ -23,8 +23,18 @@ def get_locale():
if request.accept_languages:
for x in request.accept_languages.values():
try:
preferred.append(str(LC.parse(x.replace('-', '_'))))
preferred.append(str(Locale.parse(x.replace('-', '_'))))
except (UnknownLocaleError, ValueError) as e:
log.debug('Could not parse locale "%s": %s', x, e)
return negotiate_locale(preferred or ['en'], BABEL_TRANSLATIONS)
return negotiate_locale(preferred or ['en'], get_available_translations())
def get_user_locale_language(user_language):
return Locale.parse(user_language).get_language_name(get_locale())
def get_available_locale():
return [Locale('en')] + babel.list_translations()
def get_available_translations():
return set(str(item) for item in get_available_locale())

1627
cps/editbooks.py
View File

File diff suppressed because it is too large Load Diff

80
cps/helper.py
View File

@ -29,11 +29,9 @@ from tempfile import gettempdir
import requests
import unidecode
from flask import send_from_directory, make_response, redirect, abort, url_for
from flask_babel import gettext as _
from flask_babel import lazy_gettext as N_
from flask_babel import format_datetime, get_locale
from flask_login import current_user
from sqlalchemy.sql.expression import true, false, and_, or_, text, func
from sqlalchemy.exc import InvalidRequestError, OperationalError
@ -42,7 +40,6 @@ from werkzeug.security import generate_password_hash
from markupsafe import escape
from urllib.parse import quote
try:
import advocate
from advocate.exceptions import UnacceptableAddressException
@ -52,14 +49,13 @@ except ImportError:
advocate = requests
UnacceptableAddressException = MissingSchema = BaseException
from . import calibre_db, cli
from . import calibre_db, cli_param
from .tasks.convert import TaskConvert
from . import logger, config, db, ub, fs
from . import gdriveutils as gd
from .constants import STATIC_DIR as _STATIC_DIR, CACHE_TYPE_THUMBNAILS, THUMBNAIL_TYPE_COVER, THUMBNAIL_TYPE_SERIES
from .subproc_wrapper import process_wait
from .services.worker import WorkerThread, STAT_WAITING, STAT_FAIL, STAT_STARTED, STAT_FINISH_SUCCESS, STAT_ENDED, \
STAT_CANCELLED
from .services.worker import WorkerThread
from .tasks.mail import TaskEmail
from .tasks.thumbnail import TaskClearCoverThumbnailCache, TaskGenerateCoverThumbnails
@ -76,10 +72,10 @@ except (ImportError, RuntimeError) as e:
# Convert existing book entry to new format
def convert_book_format(book_id, calibrepath, old_book_format, new_book_format, user_id, kindle_mail=None):
def convert_book_format(book_id, calibre_path, old_book_format, new_book_format, user_id, kindle_mail=None):
book = calibre_db.get_book(book_id)
data = calibre_db.get_book_format(book.id, old_book_format)
file_path = os.path.join(calibrepath, book.path, data.name)
file_path = os.path.join(calibre_path, book.path, data.name)
if not data:
error_message = _(u"%(format)s format not found for book id: %(book)d", format=old_book_format, book=book_id)
log.error("convert_book_format: %s", error_message)
@ -144,20 +140,20 @@ def send_registration_mail(e_mail, user_name, default_password, resend=False):
def check_send_to_kindle_with_converter(formats):
bookformats = list()
book_formats = list()
if 'EPUB' in formats and 'MOBI' not in formats:
bookformats.append({'format': 'Mobi',
'convert': 1,
'text': _('Convert %(orig)s to %(format)s and send to Kindle',
orig='Epub',
format='Mobi')})
book_formats.append({'format': 'Mobi',
'convert': 1,
'text': _('Convert %(orig)s to %(format)s and send to Kindle',
orig='Epub',
format='Mobi')})
if 'AZW3' in formats and 'MOBI' not in formats:
bookformats.append({'format': 'Mobi',
'convert': 2,
'text': _('Convert %(orig)s to %(format)s and send to Kindle',
orig='Azw3',
format='Mobi')})
return bookformats
book_formats.append({'format': 'Mobi',
'convert': 2,
'text': _('Convert %(orig)s to %(format)s and send to Kindle',
orig='Azw3',
format='Mobi')})
return book_formats
def check_send_to_kindle(entry):
@ -165,26 +161,26 @@ def check_send_to_kindle(entry):
returns all available book formats for sending to Kindle
"""
formats = list()
bookformats = list()
book_formats = list()
if len(entry.data):
for ele in iter(entry.data):
if ele.uncompressed_size < config.mail_size:
formats.append(ele.format)
if 'MOBI' in formats:
bookformats.append({'format': 'Mobi',
'convert': 0,
'text': _('Send %(format)s to Kindle', format='Mobi')})
book_formats.append({'format': 'Mobi',
'convert': 0,
'text': _('Send %(format)s to Kindle', format='Mobi')})
if 'PDF' in formats:
bookformats.append({'format': 'Pdf',
'convert': 0,
'text': _('Send %(format)s to Kindle', format='Pdf')})
book_formats.append({'format': 'Pdf',
'convert': 0,
'text': _('Send %(format)s to Kindle', format='Pdf')})
if 'AZW' in formats:
bookformats.append({'format': 'Azw',
'convert': 0,
'text': _('Send %(format)s to Kindle', format='Azw')})
book_formats.append({'format': 'Azw',
'convert': 0,
'text': _('Send %(format)s to Kindle', format='Azw')})
if config.config_converterpath:
bookformats.extend(check_send_to_kindle_with_converter(formats))
return bookformats
book_formats.extend(check_send_to_kindle_with_converter(formats))
return book_formats
else:
log.error(u'Cannot find book entry %d', entry.id)
return None
@ -194,12 +190,12 @@ def check_send_to_kindle(entry):
# list with supported formats
def check_read_formats(entry):
extensions_reader = {'TXT', 'PDF', 'EPUB', 'CBZ', 'CBT', 'CBR', 'DJVU'}
bookformats = list()
book_formats = list()
if len(entry.data):
for ele in iter(entry.data):
if ele.format.upper() in extensions_reader:
bookformats.append(ele.format.lower())
return bookformats
book_formats.append(ele.format.lower())
return book_formats
# Files are processed in the following order/priority:
@ -229,23 +225,11 @@ def send_mail(book_id, book_format, convert, kindle_mail, calibrepath, user_id):
return _(u"The requested file could not be read. Maybe wrong permissions?")
def shorten_component(s, by_what):
l = len(s)
if l < by_what:
return s
l = (l - by_what)//2
if l <= 0:
return s
return s[:l] + s[-l:]
def get_valid_filename(value, replace_whitespace=True, chars=128):
"""
Returns the given string converted to a string that can be used for a clean
filename. Limits num characters to 128 max.
"""
if value[-1:] == u'.':
value = value[:-1]+u'_'
value = value.replace("/", "_").replace(":", "_").strip('\0')
@ -814,7 +798,7 @@ def get_series_thumbnail(series_id, resolution):
# saves book cover from url
def save_cover_from_url(url, book_path):
try:
if cli.allow_localhost:
if cli_param.allow_localhost:
img = requests.get(url, timeout=(10, 200), allow_redirects=False) # ToDo: Error Handling
elif use_advocate:
img = advocate.get(url, timeout=(10, 200), allow_redirects=False) # ToDo: Error Handling

76
cps/kobo_auth.py
View File

@ -71,47 +71,8 @@ from flask_babel import gettext as _
from . import logger, config, calibre_db, db, helper, ub, lm
from .render_template import render_title_template
log = logger.create()
def register_url_value_preprocessor(kobo):
@kobo.url_value_preprocessor
# pylint: disable=unused-variable
def pop_auth_token(__, values):
g.auth_token = values.pop("auth_token")
def disable_failed_auth_redirect_for_blueprint(bp):
lm.blueprint_login_views[bp.name] = None
def get_auth_token():
if "auth_token" in g:
return g.get("auth_token")
else:
return None
def requires_kobo_auth(f):
@wraps(f)
def inner(*args, **kwargs):
auth_token = get_auth_token()
if auth_token is not None:
user = (
ub.session.query(ub.User)
.join(ub.RemoteAuthToken)
.filter(ub.RemoteAuthToken.auth_token == auth_token).filter(ub.RemoteAuthToken.token_type==1)
.first()
)
if user is not None:
login_user(user)
return f(*args, **kwargs)
log.debug("Received Kobo request without a recognizable auth token.")
return abort(401)
return inner
kobo_auth = Blueprint("kobo_auth", __name__, url_prefix="/kobo_auth")
@ -165,3 +126,40 @@ def delete_auth_token(user_id):
.filter(ub.RemoteAuthToken.token_type==1).delete()
return ub.session_commit()
def disable_failed_auth_redirect_for_blueprint(bp):
lm.blueprint_login_views[bp.name] = None
def get_auth_token():
if "auth_token" in g:
return g.get("auth_token")
else:
return None
def register_url_value_preprocessor(kobo):
@kobo.url_value_preprocessor
# pylint: disable=unused-variable
def pop_auth_token(__, values):
g.auth_token = values.pop("auth_token")
def requires_kobo_auth(f):
@wraps(f)
def inner(*args, **kwargs):
auth_token = get_auth_token()
if auth_token is not None:
user = (
ub.session.query(ub.User)
.join(ub.RemoteAuthToken)
.filter(ub.RemoteAuthToken.auth_token == auth_token).filter(ub.RemoteAuthToken.token_type==1)
.first()
)
if user is not None:
login_user(user)
return f(*args, **kwargs)
log.debug("Received Kobo request without a recognizable auth token.")
return abort(401)
return inner

8
cps/main.py
View File

@ -20,11 +20,7 @@ import sys
from . import create_app
from .jinjia import jinjia
from .shelf import shelf
from .remotelogin import remotelogin
from .search_metadata import meta
from .error_handler import init_errorhandler
from .tasks_status import tasks
try:
from kobo import kobo, get_kobo_activated
@ -50,6 +46,10 @@ def main():
from .editbooks import editbook
from .about import about
from .search import search
from .search_metadata import meta
from .shelf import shelf
from .tasks_status import tasks
from .error_handler import init_errorhandler
from . import web_server
init_errorhandler()

28
cps/opds.py
View File

@ -56,20 +56,6 @@ def requires_basic_auth_if_no_ano(f):
return decorated
class FeedObject:
def __init__(self, rating_id, rating_name):
self.rating_id = rating_id
self.rating_name = rating_name
@property
def id(self):
return self.rating_id
@property
def name(self):
return self.rating_name
@opds.route("/opds/")
@opds.route("/opds")
@requires_basic_auth_if_no_ano
@ -468,6 +454,20 @@ def feed_unread_books():
return render_xml_template('feed.xml', entries=result, pagination=pagination)
class FeedObject:
def __init__(self, rating_id, rating_name):
self.rating_id = rating_id
self.rating_name = rating_name
@property
def id(self):
return self.rating_id
@property
def name(self):
return self.rating_name
def feed_search(term):
if term:
entries, __, ___ = calibre_db.get_search_results(term, config=config)

1
cps/search.py
View File

@ -406,7 +406,6 @@ def render_search_results(term, offset=None, order=None, limit=None):
offset,
order,
limit,
False,
*join)
return render_title_template('search.html',
searchterm=term,

5
cps/services/__init__.py
View File

@ -18,11 +18,10 @@
from .. import logger
log = logger.create()
try: from . import goodreads_support
try:
from . import goodreads_support
except ImportError as err:
log.debug("Cannot import goodreads, showing authors-metadata will not work: %s", err)
goodreads_support = None

219
cps/shelf.py
View File

@ -33,27 +33,9 @@ from . import calibre_db, config, db, logger, ub
from .render_template import render_title_template
from .usermanagement import login_required_if_no_ano
shelf = Blueprint('shelf', __name__)
log = logger.create()
def check_shelf_edit_permissions(cur_shelf):
if not cur_shelf.is_public and not cur_shelf.user_id == int(current_user.id):
log.error("User {} not allowed to edit shelf: {}".format(current_user.id, cur_shelf.name))
return False
if cur_shelf.is_public and not current_user.role_edit_shelfs():
log.info("User {} not allowed to edit public shelves".format(current_user.id))
return False
return True
def check_shelf_view_permissions(cur_shelf):
if cur_shelf.is_public:
return True
if current_user.is_anonymous or cur_shelf.user_id != current_user.id:
log.error("User is unauthorized to view non-public shelf: {}".format(cur_shelf.name))
return False
return True
shelf = Blueprint('shelf', __name__)
@shelf.route("/shelf/add/<int:shelf_id>/<int:book_id>", methods=["POST"])
@ -238,96 +220,6 @@ def edit_shelf(shelf_id):
return create_edit_shelf(shelf, page_title=_(u"Edit a shelf"), page="shelfedit", shelf_id=shelf_id)
# if shelf ID is set, we are editing a shelf
def create_edit_shelf(shelf, page_title, page, shelf_id=False):
sync_only_selected_shelves = current_user.kobo_only_shelves_sync
# calibre_db.session.query(ub.Shelf).filter(ub.Shelf.user_id == current_user.id).filter(ub.Shelf.kobo_sync).count()
if request.method == "POST":
to_save = request.form.to_dict()
if not current_user.role_edit_shelfs() and to_save.get("is_public") == "on":
flash(_(u"Sorry you are not allowed to create a public shelf"), category="error")
return redirect(url_for('web.index'))
is_public = 1 if to_save.get("is_public") == "on" else 0
if config.config_kobo_sync:
shelf.kobo_sync = True if to_save.get("kobo_sync") else False
if shelf.kobo_sync:
ub.session.query(ub.ShelfArchive).filter(ub.ShelfArchive.user_id == current_user.id).filter(
ub.ShelfArchive.uuid == shelf.uuid).delete()
ub.session_commit()
shelf_title = to_save.get("title", "")
if check_shelf_is_unique(shelf, shelf_title, is_public, shelf_id):
shelf.name = shelf_title
shelf.is_public = is_public
if not shelf_id:
shelf.user_id = int(current_user.id)
ub.session.add(shelf)
shelf_action = "created"
flash_text = _(u"Shelf %(title)s created", title=shelf_title)
else:
shelf_action = "changed"
flash_text = _(u"Shelf %(title)s changed", title=shelf_title)
try:
ub.session.commit()
log.info(u"Shelf {} {}".format(shelf_title, shelf_action))
flash(flash_text, category="success")
return redirect(url_for('shelf.show_shelf', shelf_id=shelf.id))
except (OperationalError, InvalidRequestError) as ex:
ub.session.rollback()
log.error_or_exception(ex)
log.error_or_exception("Settings Database error: {}".format(ex))
flash(_(u"Database error: %(error)s.", error=ex.orig), category="error")
except Exception as ex:
ub.session.rollback()
log.error_or_exception(ex)
flash(_(u"There was an error"), category="error")
return render_title_template('shelf_edit.html',
shelf=shelf,
title=page_title,
page=page,
kobo_sync_enabled=config.config_kobo_sync,
sync_only_selected_shelves=sync_only_selected_shelves)
def check_shelf_is_unique(shelf, title, is_public, shelf_id=False):
if shelf_id:
ident = ub.Shelf.id != shelf_id
else:
ident = true()
if is_public == 1:
is_shelf_name_unique = ub.session.query(ub.Shelf) \
.filter((ub.Shelf.name == title) & (ub.Shelf.is_public == 1)) \
.filter(ident) \
.first() is None
if not is_shelf_name_unique:
log.error("A public shelf with the name '{}' already exists.".format(title))
flash(_(u"A public shelf with the name '%(title)s' already exists.", title=title),
category="error")
else:
is_shelf_name_unique = ub.session.query(ub.Shelf) \
.filter((ub.Shelf.name == title) & (ub.Shelf.is_public == 0) &
(ub.Shelf.user_id == int(current_user.id))) \
.filter(ident) \
.first() is None
if not is_shelf_name_unique:
log.error("A private shelf with the name '{}' already exists.".format(title))
flash(_(u"A private shelf with the name '%(title)s' already exists.", title=title),
category="error")
return is_shelf_name_unique
def delete_shelf_helper(cur_shelf):
if not cur_shelf or not check_shelf_edit_permissions(cur_shelf):
return False
shelf_id = cur_shelf.id
ub.session.delete(cur_shelf)
ub.session.query(ub.BookShelf).filter(ub.BookShelf.shelf == shelf_id).delete()
ub.session.add(ub.ShelfArchive(uuid=cur_shelf.uuid, user_id=cur_shelf.user_id))
ub.session_commit("successfully deleted Shelf {}".format(cur_shelf.name))
return True
@shelf.route("/shelf/delete/<int:shelf_id>", methods=["POST"])
@login_required
def delete_shelf(shelf_id):
@ -392,6 +284,115 @@ def order_shelf(shelf_id):
abort(404)
def check_shelf_edit_permissions(cur_shelf):
if not cur_shelf.is_public and not cur_shelf.user_id == int(current_user.id):
log.error("User {} not allowed to edit shelf: {}".format(current_user.id, cur_shelf.name))
return False
if cur_shelf.is_public and not current_user.role_edit_shelfs():
log.info("User {} not allowed to edit public shelves".format(current_user.id))
return False
return True
def check_shelf_view_permissions(cur_shelf):
if cur_shelf.is_public:
return True
if current_user.is_anonymous or cur_shelf.user_id != current_user.id:
log.error("User is unauthorized to view non-public shelf: {}".format(cur_shelf.name))
return False
return True
# if shelf ID is set, we are editing a shelf
def create_edit_shelf(shelf, page_title, page, shelf_id=False):
sync_only_selected_shelves = current_user.kobo_only_shelves_sync
# calibre_db.session.query(ub.Shelf).filter(ub.Shelf.user_id == current_user.id).filter(ub.Shelf.kobo_sync).count()
if request.method == "POST":
to_save = request.form.to_dict()
if not current_user.role_edit_shelfs() and to_save.get("is_public") == "on":
flash(_(u"Sorry you are not allowed to create a public shelf"), category="error")
return redirect(url_for('web.index'))
is_public = 1 if to_save.get("is_public") == "on" else 0
if config.config_kobo_sync:
shelf.kobo_sync = True if to_save.get("kobo_sync") else False
if shelf.kobo_sync:
ub.session.query(ub.ShelfArchive).filter(ub.ShelfArchive.user_id == current_user.id).filter(
ub.ShelfArchive.uuid == shelf.uuid).delete()
ub.session_commit()
shelf_title = to_save.get("title", "")
if check_shelf_is_unique(shelf_title, is_public, shelf_id):
shelf.name = shelf_title
shelf.is_public = is_public
if not shelf_id:
shelf.user_id = int(current_user.id)
ub.session.add(shelf)
shelf_action = "created"
flash_text = _(u"Shelf %(title)s created", title=shelf_title)
else:
shelf_action = "changed"
flash_text = _(u"Shelf %(title)s changed", title=shelf_title)
try:
ub.session.commit()
log.info(u"Shelf {} {}".format(shelf_title, shelf_action))
flash(flash_text, category="success")
return redirect(url_for('shelf.show_shelf', shelf_id=shelf.id))
except (OperationalError, InvalidRequestError) as ex:
ub.session.rollback()
log.error_or_exception(ex)
log.error_or_exception("Settings Database error: {}".format(ex))
flash(_(u"Database error: %(error)s.", error=ex.orig), category="error")
except Exception as ex:
ub.session.rollback()
log.error_or_exception(ex)
flash(_(u"There was an error"), category="error")
return render_title_template('shelf_edit.html',
shelf=shelf,
title=page_title,
page=page,
kobo_sync_enabled=config.config_kobo_sync,
sync_only_selected_shelves=sync_only_selected_shelves)
def check_shelf_is_unique(title, is_public, shelf_id=False):
if shelf_id:
ident = ub.Shelf.id != shelf_id
else:
ident = true()
if is_public == 1:
is_shelf_name_unique = ub.session.query(ub.Shelf) \
.filter((ub.Shelf.name == title) & (ub.Shelf.is_public == 1)) \
.filter(ident) \
.first() is None
if not is_shelf_name_unique:
log.error("A public shelf with the name '{}' already exists.".format(title))
flash(_(u"A public shelf with the name '%(title)s' already exists.", title=title),
category="error")
else:
is_shelf_name_unique = ub.session.query(ub.Shelf) \
.filter((ub.Shelf.name == title) & (ub.Shelf.is_public == 0) &
(ub.Shelf.user_id == int(current_user.id))) \
.filter(ident) \
.first() is None
if not is_shelf_name_unique:
log.error("A private shelf with the name '{}' already exists.".format(title))
flash(_(u"A private shelf with the name '%(title)s' already exists.", title=title),
category="error")
return is_shelf_name_unique
def delete_shelf_helper(cur_shelf):
if not cur_shelf or not check_shelf_edit_permissions(cur_shelf):
return False
shelf_id = cur_shelf.id
ub.session.delete(cur_shelf)
ub.session.query(ub.BookShelf).filter(ub.BookShelf.shelf == shelf_id).delete()
ub.session.add(ub.ShelfArchive(uuid=cur_shelf.uuid, user_id=cur_shelf.user_id))
ub.session_commit("successfully deleted Shelf {}".format(cur_shelf.name))
return True
def change_shelf_order(shelf_id, order):
result = calibre_db.session.query(db.Books).outerjoin(db.books_series_link,
db.Books.id == db.books_series_link.c.book)\

4
cps/tasks_status.py
View File

@ -34,7 +34,7 @@ log = logger.create()
@tasks.route("/ajax/emailstat")
@login_required
def get_email_status_json():
tasks = WorkerThread.getInstance().tasks
tasks = WorkerThread.get_instance().tasks
return jsonify(render_task_status(tasks))
@ -42,7 +42,7 @@ def get_email_status_json():
@login_required
def get_tasks_status():
# if current user admin, show all email, otherwise only own emails
tasks = WorkerThread.getInstance().tasks
tasks = WorkerThread.get_instance().tasks
answer = render_task_status(tasks)
return render_title_template('tasks.html', entries=answer, title=_(u"Tasks"), page="tasks")

2
cps/templates/tasks.html
View File

@ -5,7 +5,7 @@
{% block body %}
<div class="discover">
<h2>{{_('Tasks')}}</h2>
<table class="table table-no-bordered" id="tasktable" data-url="{{ url_for('task.get_email_status_json') }}" data-sort-name="starttime" data-sort-order="asc" data-locale="{{ g.user.locale }}">
<table class="table table-no-bordered" id="tasktable" data-url="{{ url_for('tasks.get_email_status_json') }}" data-sort-name="starttime" data-sort-order="asc" data-locale="{{ g.user.locale }}">
<thead>
<tr>
{% if g.user.role_admin() %}

8
cps/web.py
View File

@ -72,10 +72,10 @@ except ImportError:
from functools import wraps
#try:
# from natsort import natsorted as sort
#except ImportError:
# sort = sorted # Just use regular sort then, may cause issues with badly named pages in cbz/cbr files
try:
from natsort import natsorted as sort
except ImportError:
sort = sorted # Just use regular sort then, may cause issues with badly named pages in cbz/cbr files
@app.after_request