CVE-2022-0990 (Python, published 2022-04-04T18:15Z)
Description: value: Server-Side Request Forgery (SSRF) in GitHub repository janeczku/calibre-web prior to 0.6.18.
Severity: MEDIUM (CVSS v3 base score: 9.1)
Weakness: CWE-918 \u2014 Server-Side Request Forgery (SSRF)
The web server receives a URL or similar request from an upstream component and retrieves the contents of this URL, but it does not sufficiently ensure that the request is being sent to the expected destination.
Repository: https://github.com/janeczku/calibre-web
Files changed:
cps.py
cps/__init__.py
cps/admin.py
cps/db.py
cps/editbooks.py
cps/epub.py
cps/helper.py
cps/opds.py
cps/pagination.py
cps/remotelogin.py
cps/templates/book_edit.html
cps/templates/book_table.html
cps/templates/detail.html
cps/templates/layout.html
cps/usermanagement.py
cps/web.py
optional-requirements.txt
requirements.txt
Commit message: Better epub cover parsing with multiple cover-image items
Code cosmetics
renamed variables
refactored xml page generation
refactored prepare author


Vulnerable code:
from cps.editbooks import editbook
    app.register_blueprint(editbook)
    config.store_calibre_uuid(calibre_db, db.Library_Id)
from babel import Locale as LC
from functools import wraps
    allUser = ub.session.query(ub.User).all()
    return render_title_template("admin.html", allUser=allUser, email=email_settings, config=config, commit=commit,
    read_column = calibre_db.session.query(db.Custom_Columns)\
        .filter(and_(db.Custom_Columns.datatype == 'bool', db.Custom_Columns.mark_for_delete == 0)).all()
    restrict_columns = calibre_db.session.query(db.Custom_Columns)\
        .filter(and_(db.Custom_Columns.datatype == 'text', db.Custom_Columns.mark_for_delete == 0)).all()
    translations = [LC('en')] + babel.list_translations()
    translations = babel.list_translations() + [LC('en')]
    allUser = ub.session.query(ub.User)
        allUser = allUser.filter(ub.User.role.op('&')(constants.ROLE_ANONYMOUS) != constants.ROLE_ANONYMOUS)
                                 users=allUser.all(),
            user.default = LC.parse(user.default_language).get_language_name(get_locale())
    locale = babel.list_translations() + [LC('en')]
                                 [constants.ROLE_ADMIN, constants.ROLE_PASSWD, constants.ROLE_EDIT_SHELFS]:
        if not calibre_db.session.query(db.Custom_Columns).filter(db.Custom_Columns.id == column) \
              .filter(and_(db.Custom_Columns.datatype == 'bool', db.Custom_Columns.mark_for_delete == 0)).all():
        if not calibre_db.session.query(db.Custom_Columns).filter(db.Custom_Columns.id == column) \
              .filter(and_(db.Custom_Columns.datatype == 'text', db.Custom_Columns.mark_for_delete == 0)).all():
            or to_save["config_" + str(element['id']) + "_oauth_client_secret"] != element['oauth_client_secret']:
            and to_save["config_" + str(element['id']) + "_oauth_client_secret"]:
        or not config.config_ldap_user_object:
          or config.config_calibre_dir != to_save["config_calibre_dir"]:
        config.store_calibre_uuid(calibre_db, db.Library_Id)
                             and config.config_login_type == constants.LOGIN_LDAP)
    translations = [LC('en')] + babel.list_translations()
    translations = babel.list_translations() + [LC('en')]
import copy
try:
    import unidecode
    use_unidecode = True
except ImportError:
    use_unidecode = False
class Library_Id(Base):
    def formatType(self):
    def __init__(self, text, book):
        self.text = text

        return (self.timestamp.strftime('%Y-%m-%dT%H:%M:%S+00:00') or '')
class Custom_Columns(Base):
class CalibreDB():
            self.initSession(expire_on_commit)
    def initSession(self, expire_on_commit=True):
    def setup_db_cc_classes(self, cc):
                          echo=False,
                          isolation_level="SERIALIZABLE",
                          connect_args={'check_same_thread': False},
                          poolclass=StaticPool)
                database_uuid = local_session().query(Library_Id).one_or_none()
            inst.initSession()
            archived_books = (
                ub.session.query(ub.ArchivedBook)
                    .filter(ub.ArchivedBook.user_id == int(current_user.id))
                    .filter(ub.ArchivedBook.is_archived == True)
                    .all()
            )
                        any(cc_classes[self.config.config_restricted_column].value.in_(pos_cc_list))
                        any(cc_classes[self.config.config_restricted_column].value.in_(neg_cc_list))
                               and_(ub.ReadBook.user_id == int(current_user.id), ub.ReadBook.book_id == Books.id)))
                    query =self.session.query(database, None, ub.ArchivedBook.is_archived)

    def get_search_results(self, term, offset=None, order=None, limit=None, allow_show_archived=False,

        log = logger.create()
        log.error_or_exception(ex)
    pass
editbook = Blueprint('editbook', __name__)
        # check if a element with that name exists
            changed = True
# Modifies different Database objects, first check if elements if elements have to be deleted,
    db_dict = dict([(identifier.type.lower(), identifier) for identifier in db_identifiers ])
@editbook.route("/ajax/delete/<int:book_id>", methods=["POST"])
@editbook.route("/delete/<int:book_id>", defaults={'book_format': ""}, methods=["POST"])
@editbook.route("/delete/<int:book_id>/<string:book_format>", methods=["POST"])
    cc = calibre_db.session.query(db.Custom_Columns). \
        filter(db.Custom_Columns.datatype.notin_(db.cc_exceptions)).all()
def render_delete_book_result(book_format, jsonResponse, warning, book_id):
        if jsonResponse:
            return json.dumps([warning, {"location": url_for("editbook.edit_book", book_id=book_id),
            return redirect(url_for('editbook.edit_book', book_id=book_id))
        if jsonResponse:
def delete_book_from_table(book_id, book_format, jsonResponse):
                    if jsonResponse:
                        return json.dumps([{"location": url_for("editbook.edit_book", book_id=book_id),
                                           "type": "danger",
                                           "format": "",
                                           "message": error}])
                        return redirect(url_for('editbook.edit_book', book_id=book_id))
                    if jsonResponse:
                        warning = {"location": url_for("editbook.edit_book", book_id=book_id),
                                                "type": "warning",
                                                "format": "",
                                                "message": error}
                if jsonResponse:
                    return json.dumps([{"location": url_for("editbook.edit_book", book_id=book_id),
                    return redirect(url_for('editbook.edit_book', book_id=book_id))
        return render_delete_book_result(book_format, jsonResponse, warning, book_id)
    if jsonResponse:
        return json.dumps({"location": url_for("editbook.edit_book", book_id=book_id),
        return redirect(url_for('editbook.edit_book', book_id=book_id))
    cc = calibre_db.session.query(db.Custom_Columns).filter(db.Custom_Columns.datatype.notin_(db.cc_exceptions)).all()
        flash(_(u"Oops! Selected book title is unavailable. File does not exist or is not accessible"), category="error")
    valid_source_formats=list()
    kepub_possible=None
    modif_date = False
        modif_date = True
    return modif_date
    modif_date = False
            modif_date = True
            book.comments.append(db.Comments(text=comments, book=book.id))
            modif_date = True
    return modif_date
def edit_book_languages(languages, book, upload=False, invalid=None):
    if not upload:
    for l in unknown_languages:
        log.error("'%s' is not a valid language", l)
            invalid.append(l)
            raise ValueError(_(u"'%(langname)s' is not a valid language", langname=l))
    if upload and len(input_l) == 1:
    cc = (calibre_db.session.query(db.Custom_Columns)
          .filter(db.Custom_Columns.datatype.notin_(db.cc_exceptions))
          .filter(db.Custom_Columns.id == column_id)
    cc = calibre_db.session.query(db.Custom_Columns).filter(db.Custom_Columns.datatype.notin_(db.cc_exceptions)).all()
def upload_single_file(request, book, book_id):
    if 'btn-upload-format' in request.files:
        requested_file = request.files['btn-upload-format']
            uploadText=_(u"File format %(ext)s added to %(book)s", ext=file_ext.upper(), book=link)
            WorkerThread.add(current_user.name, TaskUpload(uploadText, escape(book.title)))
def upload_cover(request, book):
    if 'btn-upload-cover' in request.files:
        requested_file = request.files['btn-upload-cover']
    # renamed = False
    input_authors = author_name.split('&')
                one_book.author_sort = one_book.author_sort.replace(sorted_renamed_author, sorted_old_author)
@editbook.route("/admin/book/<int:book_id>", methods=['GET', 'POST'])
    modif_date = False
        flash(_(u"Oops! Selected book title is unavailable. File does not exist or is not accessible"), category="error")
        modif_date = True
            modif_date = True
        error = False
                                               renamed_author=renamed)
                            modif_date = True
            modif_date |= edit_book_series_index(to_save["series_index"], book)
            modif_date |= edit_book_comments(Markup(to_save['description']).unescape(), book)
            modif_date |= modification
            modif_date |= edit_book_tags(to_save['tags'], book)
            modif_date |= edit_book_series(to_save["series"], book)
            modif_date |= edit_book_publisher(to_save['publisher'], book)
            modif_date |= edit_book_languages(to_save['languages'], book)
            modif_date |= edit_book_ratings(to_save, book)
            modif_date |= edit_all_cc_data(book_id, book, to_save)
            if modif_date:
def prepare_authors_on_upload(title, authr):
    if title != _(u'Unknown') and authr != _(u'Unknown'):
        entry = calibre_db.check_exists_book(authr, title)
        if entry:
            log.info("Uploaded book probably exists in library")
            flash(_(u"Uploaded book probably exists in the library, consider to change before upload new: ")
                  + Markup(render_title_template('book_exists_flash.html', entry=entry)), category="warning")

def create_book_on_upload(modif_date, meta):
    # combine path and normalize path from windows systems
    modif_date |= modify_database_object(input_authors, db_book.authors, db.Authors, calibre_db.session,
                                         'author')
    modif_date |= edit_book_series_index(meta.series_id, db_book)
    invalid=[]
    modif_date |= edit_book_languages(meta.languages, db_book, upload=True, invalid=invalid)
        for l in invalid:
            flash(_(u"'%(langname)s' is not a valid language", langname=l), category="warning")
    modif_date |= edit_book_tags(meta.tags, db_book)
    modif_date |= edit_book_publisher(meta.publisher, db_book)
    modif_date |= edit_book_series(meta.series, db_book)
@editbook.route("/upload", methods=["POST"])
                modif_date = False
                db_book, input_authors, title_dir, renamed_authors = create_book_on_upload(modif_date, meta)
                # Comments needs book id therefore only possible after flush
                modif_date |= edit_book_comments(Markup(meta.description).unescape(), db_book)
                uploadText = _(u"File %(file)s uploaded", file=link)
                WorkerThread.add(current_user.name, TaskUpload(uploadText, escape(title)))
                        resp = {"location": url_for('editbook.edit_book', book_id=book_id)}
@editbook.route("/admin/book/convert/<int:book_id>", methods=['POST'])
        return redirect(url_for('editbook.edit_book', book_id=book_id))
                    book_format=book_format_to),
                    category="success")
    return redirect(url_for('editbook.edit_book', book_id=book_id))
@editbook.route("/ajax/getcustomenum/<int:c_id>")
    cc = (calibre_db.session.query(db.Custom_Columns)
              .filter(db.Custom_Columns.id == c_id)
              .filter(db.Custom_Columns.datatype.notin_(db.cc_exceptions)).one_or_none())
@editbook.route("/ajax/editbooks/<param>", methods=['POST'])
                                        'newValue': ', '.join([publisher.name for publisher in book.publishers])}),
                                mimetype='application/json')
            sort = book.sort
            ret = Response(json.dumps({'success': True,
                                       'newValue':  ' & '.join([author.replace('|',',') for author in input_authors])}),
                           mimetype='application/json')
                                                message="Book {} archivebit set to: {}".format(book.id, vals['value']))
            book.sort = sort
@editbook.route("/ajax/sort_value/<field>/<int:bookid>")
@editbook.route("/ajax/simulatemerge", methods=['POST'])
@editbook.route("/ajax/mergebooks", methods=['POST'])
            to_name = helper.get_valid_filename(to_book.title, chars=96) + ' - ' + \
                      helper.get_valid_filename(to_book.authors[0].name, chars=96)
                    delete_book_from_table(from_book.id,"", True)
@editbook.route("/ajax/xchange", methods=['POST'])
            modif_date = False
                modif_date = True
                                           renamed_author=renamed)
            if modif_date:
    cfname = tree.xpath('n:rootfiles/n:rootfile/@full-path', namespaces=ns)[0]
    cf = epub_zip.read(cfname)
    coverpath = os.path.dirname(cfname)
    cover_file = parse_epub_cover(ns, tree, epub_zip, coverpath, tmp_file_path)
    if len(cover_section) > 0:
        cover_file = _extract_cover(epub_zip, cover_section[0], cover_path, tmp_file_path)
    else:
            if cover_file:
                break
import unicodedata
from urllib.parse import urlparse
    import unidecode
    use_unidecode = True
    use_unidecode = False
from . import logger, config, get_locale, db, ub, kobo_sync_status
    if 'AZW3' in formats and not 'MOBI' in formats:
    EXTENSIONS_READER = {'TXT', 'PDF', 'EPUB', 'CBZ', 'CBT', 'CBR', 'DJVU'}
            if ele.format.upper() in EXTENSIONS_READER:
            EmailText = _(u"%(book)s send to Kindle", book=link)
                             EmailText, _(u'This e-mail has been sent via Calibre-Web.')))
    if use_unidecode:
        if config.config_unicode_filename:
            value = (unidecode.unidecode(value))
    else:
        value = value.replace(u'§', u'SS')
        value = value.replace(u'ß', u'ss')
        value = unicodedata.normalize('NFKD', value)
        re_slugify = re.compile(r'[\W\s-]', re.UNICODE)
        value = re_slugify.sub('', value)
            readBook = ub.ReadBook(user_id=current_user.id, book_id = book_id)
            readBook.read_status = ub.ReadBook.STATUS_FINISHED
            book = readBook
        except (OperationalError, InvalidRequestError) as e:
            log.error(u"Read status could not set: {}".format(e))
            return _("Read status could not set: {}".format(e.orig))
                except (IOError, OSError) as e:
                    log.error("Deleting book %s failed: %s", book.id, e)
                    return False, _("Deleting book %(id)s failed: %(message)s", id=book.id, message=e)
                    except (IOError, OSError) as e:
                        log.error("Deleting authorpath for book %s failed: %s", book.id, e)
                               + get_valid_filename(new_author.name, chars=42)
                            os.path.normcase(os.path.join(all_new_path,
                                                          all_new_name + '.' + file_format.format.lower())))
                        gFile = gd.getFileFromEbooksFolder(all_new_path,
                                                           file_format.name + '.' + file_format.format.lower())
                        if gFile:
                            gd.moveGdriveFileRemote(gFile, all_new_name + u'.' + file_format.format.lower())
                            gd.updateDatabaseOnEdit(gFile['id'], all_new_name + u'.' + file_format.format.lower())
                gFile = gd.getFileFromEbooksFolder(None, old_author_dir)
                if gFile:
                    gd.moveGdriveFolderRemote(gFile, new_author_rename_dir)
                    except (OSError) as ex:
    error = False
                get_valid_filename(first_author, chars=42) + \
                filename_ext
        gFile = gd.getFileFromEbooksFolder(os.path.dirname(book.path), titledir)
        if gFile:
            gd.moveGdriveFileRemote(gFile, new_titledir)
            gd.updateDatabaseOnEdit(gFile['id'], book.path)     # only child folder affected
        gFile = gd.getFileFromEbooksFolder(os.path.dirname(book.path), new_titledir)
        if gFile:
            gd.moveGdriveFolderRemote(gFile, new_authordir)
            gd.updateDatabaseOnEdit(gFile['id'], book.path)
            else: # path is valid copy only files to new location (merge)
                                        os.path.normcase(os.path.join(new_path + dir_name[len(path):], file)))
        localbook.path = os.path.join(new_authordir, new_titledir).replace('\\','/')
        gFile = gd.getFileFromEbooksFolder(book.path, name)
        gFile = gd.getFileFromEbooksFolder(os.path.dirname(book.path), book.path.split('/')[1])
    if gFile:
        gd.deleteDatabaseEntry(gFile['id'])
        gFile.Trash()
    inpt = [ " ".join(inp.split()) for inp in inpt]
        raise Exception (_(u"This username is already taken"))
        if not cli.allow_localhost:
            # 127.0.x.x, localhost, [::1], [::ffff:7f00:1]
            ip = socket.getaddrinfo(urlparse(url).hostname, 0)[0][4][0]
            if ip.startswith("127.") or ip.startswith('::ffff:7f') or ip == "::1" or ip == "0.0.0.0" or ip == "::":
                log.error("Localhost was accessed for cover upload")
                return False, _("You are not allowed to access localhost for cover uploads")
        img = requests.get(url, timeout=(10, 200), allow_redirects=False)      # ToDo: Error Handling
            gd.uploadFileToEbooksFolder(os.path.join(book_path, 'cover.jpg').replace("\\","/"),
        #startTime = time.time()
        #log.debug('%s', time.time() - startTime)
def check_unrar(unrarLocation):
    if not unrarLocation:
    if not os.path.exists(unrarLocation):
        unrarLocation = [unrarLocation]
        value = process_wait(unrarLocation, pattern='UNRAR (.*) freeware')
    retVal = ""
        retVal = format_unit(runtime.days, 'duration-day', length="long", locale=get_locale()) + ', '
        retVal += '{:d}:{:02d}:{:02d}s'.format(hours, minutes, seconds)
        retVal += '{:2d}:{:02d}s'.format(minutes, seconds)
        retVal += '{:2d}s'.format(seconds)
    return retVal
    tmpcc = calibre_db.session.query(db.Custom_Columns)\
        .filter(db.Custom_Columns.datatype.notin_(db.cc_exceptions)).all()
from tornado.httputil import HTTPServerRequest
    shift = 0
    off = int(request.args.get("offset") or 0)
    entries = calibre_db.session.query(func.upper(func.substr(db.Books.sort, 1, 1)).label('id'))\
        .filter(calibre_db.common_filters()).group_by(func.upper(func.substr(db.Books.sort, 1, 1))).all()

    elements = []
    if off == 0:
        elements.append({'id': "00", 'name':_("All")})
        shift = 1
    for entry in entries[
                 off + shift - 1:
                 int(off + int(config.config_books_per_page) - shift)]:
        elements.append({'id': entry.id, 'name': entry.id})

    pagination = Pagination((int(off) / (int(config.config_books_per_page)) + 1), config.config_books_per_page,
                            len(entries) + 1)
    return render_xml_template('feed.xml',
                               letterelements=elements,
                               folder='opds.feed_letter_books',
                               pagination=pagination)
        downloadBook = calibre_db.get_book(book.Downloads.book_id)
        if downloadBook:
    numBooks = entries.__len__()
                            config.config_books_per_page, numBooks)
    shift = 0
    off = int(request.args.get("offset") or 0)
    entries = calibre_db.session.query(func.upper(func.substr(db.Authors.sort, 1, 1)).label('id'))\
        .join(db.books_authors_link).join(db.Books).filter(calibre_db.common_filters())\
        .group_by(func.upper(func.substr(db.Authors.sort, 1, 1))).all()

    elements = []
    if off == 0:
        elements.append({'id': "00", 'name':_("All")})
        shift = 1
    for entry in entries[
                 off + shift - 1:
                 int(off + int(config.config_books_per_page) - shift)]:
        elements.append({'id': entry.id, 'name': entry.id})

    pagination = Pagination((int(off) / (int(config.config_books_per_page)) + 1), config.config_books_per_page,
                            len(entries) + 1)
    return render_xml_template('feed.xml',
                               letterelements=elements,
                               folder='opds.feed_letter_author',
                               pagination=pagination)
    off = request.args.get("offset") or 0
    entries, __, pagination = calibre_db.fill_indexpage((int(off) / (int(config.config_books_per_page)) + 1), 0,
                                                        db.Books,
                                                        db.Books.authors.any(db.Authors.id == book_id),
                                                        [db.Books.timestamp.desc()])
    return render_xml_template('feed.xml', entries=entries, pagination=pagination)
    off = request.args.get("offset") or 0
    entries, __, pagination = calibre_db.fill_indexpage((int(off) / (int(config.config_books_per_page)) + 1), 0,
                                                        db.Books,
                                                        db.Books.publishers.any(db.Publishers.id == book_id),
                                                        [db.Books.timestamp.desc()])
    return render_xml_template('feed.xml', entries=entries, pagination=pagination)
    shift = 0
    off = int(request.args.get("offset") or 0)
    entries = calibre_db.session.query(func.upper(func.substr(db.Tags.name, 1, 1)).label('id'))\
        .join(db.books_tags_link).join(db.Books).filter(calibre_db.common_filters())\
        .group_by(func.upper(func.substr(db.Tags.name, 1, 1))).all()
    elements = []
    if off == 0:
        elements.append({'id': "00", 'name':_("All")})
        shift = 1
    for entry in entries[
                 off + shift - 1:
                 int(off + int(config.config_books_per_page) - shift)]:
        elements.append({'id': entry.id, 'name': entry.id})
    pagination = Pagination((int(off) / (int(config.config_books_per_page)) + 1), config.config_books_per_page,
                            len(entries) + 1)
    return render_xml_template('feed.xml',
                               letterelements=elements,
                               folder='opds.feed_letter_category',
                               pagination=pagination)
    off = request.args.get("offset") or 0
    entries, __, pagination = calibre_db.fill_indexpage((int(off) / (int(config.config_books_per_page)) + 1), 0,
                                                        db.Books,
                                                        db.Books.tags.any(db.Tags.id == book_id),
                                                        [db.Books.timestamp.desc()])
    return render_xml_template('feed.xml', entries=entries, pagination=pagination)
    shift = 0
    off = int(request.args.get("offset") or 0)
    entries = calibre_db.session.query(func.upper(func.substr(db.Series.sort, 1, 1)).label('id'))\
        .join(db.books_series_link).join(db.Books).filter(calibre_db.common_filters())\
        .group_by(func.upper(func.substr(db.Series.sort, 1, 1))).all()
    elements = []
    if off == 0:
        elements.append({'id': "00", 'name':_("All")})
        shift = 1
    for entry in entries[
                 off + shift - 1:
                 int(off + int(config.config_books_per_page) - shift)]:
        elements.append({'id': entry.id, 'name': entry.id})
    pagination = Pagination((int(off) / (int(config.config_books_per_page)) + 1), config.config_books_per_page,
                            len(entries) + 1)
    return render_xml_template('feed.xml',
                               letterelements=elements,
                               folder='opds.feed_letter_series',
                               pagination=pagination)
                               (db.Ratings.rating / 2).label('name')) \
    off = request.args.get("offset") or 0
    entries, __, pagination = calibre_db.fill_indexpage((int(off) / (int(config.config_books_per_page)) + 1), 0,
                                                        db.Books,
                                                        db.Books.ratings.any(db.Ratings.id == book_id),
                                                        [db.Books.timestamp.desc()])
    return render_xml_template('feed.xml', entries=entries, pagination=pagination)
    # workaround, loading the user from the request and checking it's download rights here
        ip_Address = request.headers.get('X-Forwarded-For', request.remote_addr)
        log.warning('OPDS Login failed for user "%s" IP-address: %s', username.decode('utf-8'), ip_Address)
@opds.route("/opds/thumb_240_240/<book_id>")
@opds.route("/opds/cover_240_240/<book_id>")
@opds.route("/opds/cover_90_90/<book_id>")
@opds.route("/opds/cover/<book_id>")
@requires_basic_auth_if_no_ano
def feed_get_cover(book_id):
    return get_book_cover(book_id)


@opds.route("/opds/readbooks")
@requires_basic_auth_if_no_ano
def feed_read_books():
    result, pagination = render_read_books(int(off) / (int(config.config_books_per_page)) + 1, True, True)
    return render_xml_template('feed.xml', entries=result, pagination=pagination)
@opds.route("/opds/unreadbooks")
@requires_basic_auth_if_no_ano
def feed_unread_books():
    off = request.args.get("offset") or 0
    result, pagination = render_read_books(int(off) / (int(config.config_books_per_page)) + 1, False, True)
    return render_xml_template('feed.xml', entries=result, pagination=pagination)
    # right_edge: last right_edges count of all pages are shown as number, means, if 10 pages are paginated -> 9,10 shwn
    # left_edge: first left_edges count of all pages are shown as number                                    -> 1,2 shwn
    # left_current: left_current count below current page are shown as number, means if current page 5      -> 3,4 shwn
    # left_current: right_current count above current page are shown as number, means if current page 5     -> 6,7 shwn
try:
    from functools import wraps
except ImportError:
    pass  # We're not using Python 3
      <form class="padded-bottom" action="{{ url_for('editbook.convert_bookformat', book_id=book.id) }}" method="post" id="book_convert_frm">
<form role="form" action="{{ url_for('editbook.edit_book', book_id=book.id) }}" method="post" enctype="multipart/form-data" id="book_edit_frm">
        data-editable-url="{{ url_for('editbook.edit_list_book', param=parameter)}}"
            <th data-field="series_index" id="series_index" data-visible="{{visiblility.get('series_index')}}" data-edit-validate="{{ _('This Field is Required') }}" data-sortable="true" {% if g.user.role_edit() %} data-editable-type="number" data-editable-placeholder="1" data-editable-step="0.01" data-editable-min="0" data-editable-url="{{ url_for('editbook.edit_list_book', param='series_index')}}" data-edit="true" data-editable-title="{{_('Enter Title')}}"{% endif %}>{{_('Series Index')}}</th>
            <th data-field="comments" id="comments" data-escape="true" data-editable-mode="popup"  data-visible="{{visiblility.get('comments')}}" data-sortable="false" {% if g.user.role_edit() %} data-editable-type="wysihtml5" data-editable-url="{{ url_for('editbook.edit_list_book', param='comments')}}" data-edit="true" data-editable-title="{{_('Enter comments')}}"{% endif %}>{{_('Comments')}}</th>
                <th data-field="custom_column_{{ c.id|string }}" id="custom_column_{{ c.id|string }}" data-visible="{{visiblility.get('custom_column_'+ c.id|string)}}" data-sortable="false" {% if g.user.role_edit() %} data-editable-type="number" data-editable-placeholder="1" data-editable-step="1" data-editable-url="{{ url_for('editbook.edit_list_book', param='custom_column_'+ c.id|string)}}" data-edit="true" data-editable-title="{{_('Enter ') + c.name}}"{% endif %}>{{c.name}}</th>
                <th data-field="custom_column_{{ c.id|string }}" id="custom_column_{{ c.id|string }}" data-formatter="ratingFormatter" data-visible="{{visiblility.get('custom_column_'+ c.id|string)}}" data-sortable="false" {% if g.user.role_edit() %} data-editable-type="number" data-editable-placeholder="1" data-editable-step="0.5" data-editable-step="1" data-editable-min="1" data-editable-max="5" data-editable-url="{{ url_for('editbook.edit_list_book', param='custom_column_'+ c.id|string)}}" data-edit="true" data-editable-title="{{_('Enter ') + c.name}}"{% endif %}>{{c.name}}</th>
                <th data-field="custom_column_{{ c.id|string }}" id="custom_column_{{ c.id|string }}" data-visible="{{visiblility.get('custom_column_'+ c.id|string)}}" data-sortable="false" {% if g.user.role_edit() %} data-editable-type="number" data-editable-placeholder="1" data-editable-step="0.01" data-editable-url="{{ url_for('editbook.edit_list_book', param='custom_column_'+ c.id|string)}}" data-edit="true" data-editable-title="{{_('Enter ') + c.name}}"{% endif %}>{{c.name}}</th>
                <th data-field="custom_column_{{ c.id|string }}" id="custom_column_{{ c.id|string }}" data-visible="{{visiblility.get('custom_column_'+ c.id|string)}}" data-sortable="false" {% if g.user.role_edit() %} data-editable-type="select" data-editable-source={{ url_for('editbook.table_get_custom_enum', c_id=c.id)  }} data-editable-url="{{ url_for('editbook.edit_list_book', param='custom_column_'+ c.id|string)}}" data-edit="true" data-editable-title="{{_('Enter ') + c.name}}"{% endif %}>{{c.name}}</th>
                  <th data-field="custom_column_{{ c.id|string }}" id="custom_column_{{ c.id|string }}" data-escape="true" data-editable-mode="popup"  data-visible="{{visiblility.get('custom_column_'+ c.id|string)}}" data-sortable="false" {% if g.user.role_edit() %} data-editable-type="wysihtml5" data-editable-url="{{ url_for('editbook.edit_list_book', param='custom_column_'+ c.id|string)}}" data-edit="true" data-editable-title="{{_('Enter ') + c.name}}"{% endif %}>{{c.name}}</th>
          <a href="{{identifier}}" target="_blank" class="btn btn-xs btn-success" role="button">{{identifier.formatType()}}</a>
          <a href="{{ url_for('editbook.edit_book', book_id=entry.id) }}" class="btn btn-sm btn-primary" id="edit_book" role="button"><span class="glyphicon glyphicon-edit"></span> {{_('Edit Metadata')}}</a>
                    <form id="form-upload" class="navbar-form" action="{{ url_for('editbook.upload') }}" data-title="{{_('Uploading...')}}" data-footer="{{_('Close')}}" data-failed="{{_('Error')}}" data-message="{{_('Upload done, processing, please wait...')}}" method="post" enctype="multipart/form-data">
try:
    from functools import wraps
except ImportError:
    pass  # We're not using Python 3
from babel import Locale as LC

    resp.headers['Content-Security-Policy'] = "default-src 'self'" + ''.join([' '+host for host in config.config_trustedhosts.strip().split(',')]) + " 'unsafe-inline' 'unsafe-eval'; font-src 'self' data:; img-src 'self' data:"
    if request.endpoint == "editbook.edit_book" or config.config_use_google_drive:
    lbookmark = ub.Bookmark(user_id=current_user.id,
                            book_id=book_id,
                            format=book_format,
                            bookmark_key=bookmark_key)
    ub.session.merge(lbookmark)
    is_archived = change_archived_books(book_id, message="Book {} archivebit toggled".format(book_id))
        														False, 0,
        # Unary expression comparsion only working (for this expression) in sqlalchemy 1.4+
        #if not (order[0][0].compare(func.count(ub.Downloads.book_id).desc()) or
        #        order[0][0].compare(func.count(ub.Downloads.book_id).asc())):
        all_books = ub.session.query(ub.Downloads, func.count(ub.Downloads.book_id))\
            downloadBook = calibre_db.session.query(db.Books).filter(calibre_db.common_filters()).filter(
            if downloadBook:
                entries.append(downloadBook)
        numBooks = entries.__len__()
        pagination = Pagination(page, config.config_books_per_page, numBooks)
            if not calibre_db.session.query(db.Books).filter(calibre_db.common_filters()) \
                             .filter(db.Books.id == book.id).first():
                                     title=_(u"Downloaded books by %(user)s",user=user.name),
            return [] # ToDo: Handle error Case for opds
            pagename = "read"
            pagename = "unread"
                                     title=name, page=pagename, order=order[1])
    archived_books = (
        ub.session.query(ub.ArchivedBook)
        .filter(ub.ArchivedBook.user_id == int(current_user.id))
        .filter(ub.ArchivedBook.is_archived == True)
        .all()
    )
    pagename = "archived"
                                 title=name, page=pagename, order=sort_param[1])
    tags = calibre_db.session.query(db.Tags)\
        .join(db.books_tags_link)\
        .join(db.Books)\
        .group_by(text('books_tags_link.tag'))\
    series = calibre_db.session.query(db.Series)\
        .join(db.books_series_link)\
        .join(db.Books)\
        .group_by(text('books_series_link.series'))\
        .order_by(db.Series.name)\
    shelves = ub.session.query(ub.Shelf)\
        .filter(or_(ub.Shelf.is_public == 1, ub.Shelf.user_id == int(current_user.id)))\
    extensions = calibre_db.session.query(db.Data)\
        .join(db.Books)\
        .group_by(db.Data.format)\
                                 series=series,shelves=shelves, title=_(u"Advanced Search"), cc=cc, page="advsearch")
                                                                      False,
    search = request.args.get("search")
        join = db.books_authors_link, db.Books.id == db.books_authors_link.c.book, db.Authors, \
               db.books_series_link, db.Books.id == db.books_series_link.c.book, db.Series
    total_count = filtered_count = calibre_db.session.query(db.Books).filter(calibre_db.common_filters(allow_show_archived=True)).count()
        if search:
            books = calibre_db.search_query(search, config.config_read_column).all()
                                                          int(current_user.id) == ub.ArchivedBook.user_id))
    elif search:
        entries, filtered_count, __ = calibre_db.get_search_results(search,
                                                                    [order,''],
                                                                    True,
        for index in range(0, len(val.languages)):
            val.languages[index].language_name = isoLanguages.get_language_name(get_locale(), val.languages[
                index].lang_code)
        charlist = calibre_db.session.query(func.upper(func.substr(db.Authors.sort, 1, 1)).label('char')) \
            .join(db.books_authors_link).join(db.Books).filter(calibre_db.common_filters()) \
            .group_by(func.upper(func.substr(db.Authors.sort, 1, 1))).all()
        autor_copy = copy.deepcopy(entries)
        for entry in autor_copy:
        return render_title_template('list.html', entries=autor_copy, folder='web.books_list', charlist=charlist,
        entries = ub.session.query(ub.User, func.count(ub.Downloads.book_id).label('count'))\
        charlist = ub.session.query(func.upper(func.substr(ub.User.name, 1, 1)).label('char')) \
        return render_title_template('list.html', entries=entries, folder='web.books_list', charlist=charlist,
        charlist = calibre_db.session.query(func.upper(func.substr(db.Publishers.name, 1, 1)).label('char')) \
            .join(db.books_publishers_link).join(db.Books).filter(calibre_db.common_filters()) \
            .group_by(func.upper(func.substr(db.Publishers.name, 1, 1))).all()
        return render_title_template('list.html', entries=entries, folder='web.books_list', charlist=charlist,
            charlist = calibre_db.session.query(func.upper(func.substr(db.Series.sort, 1, 1)).label('char')) \
                .join(db.books_series_link).join(db.Books).filter(calibre_db.common_filters()) \
                .group_by(func.upper(func.substr(db.Series.sort, 1, 1))).all()
            return render_title_template('list.html', entries=entries, folder='web.books_list', charlist=charlist,
                .join(db.books_series_link).join(db.Series).filter(calibre_db.common_filters())\
            charlist = calibre_db.session.query(func.upper(func.substr(db.Series.sort, 1, 1)).label('char')) \
                .join(db.books_series_link).join(db.Books).filter(calibre_db.common_filters()) \
                .group_by(func.upper(func.substr(db.Series.sort, 1, 1))).all()

            return render_title_template('grid.html', entries=entries, folder='web.books_list', charlist=charlist,
                                   (db.Ratings.rating / 2).label('name')) \
        charlist = list()
            if upper_lang not in charlist:
                charlist.append(upper_lang)
                                     charlist=charlist, title=_(u"Languages"), page="langlist",
        charlist = calibre_db.session.query(func.upper(func.substr(db.Tags.name, 1, 1)).label('char')) \
            .join(db.books_tags_link).join(db.Books).filter(calibre_db.common_filters()) \
            .group_by(func.upper(func.substr(db.Tags.name, 1, 1))).all()
        return render_title_template('list.html', entries=entries, folder='web.books_list', charlist=charlist,
def adv_search_extension(q, include_extension_inputs, exclude_extension_inputs):
    return q
    q = q.outerjoin(ub.BookShelf, db.Books.id == ub.BookShelf.book_id)\
    elements = {'tag': db.Tags, 'serie':db.Series, 'shelf':ub.Shelf}
    q = query.outerjoin(db.books_series_link, db.Books.id == db.books_series_link.c.book)\
        .outerjoin(db.Series)\
    searchterm = []
                searchterm.extend([u"{} >= {}".format(c.name,
                                                      format_date(datetime.strptime(column_start, "%Y-%m-%d").date(),
                                                                      format='medium',
                                                                      locale=get_locale())
                                                      )])
                searchterm.extend([u"{} <= {}".format(c.name,
                                                      format_date(datetime.strptime(column_end, "%Y-%m-%d").date(),
                                                                      format='medium',
                                                                      locale=get_locale())
                                                      )])
            searchterm.extend([(u"{}: {}".format(c.name, term.get('custom_column_' + str(c.id))))])

    if any(tags.values()) or author_name or book_title or publisher or pub_start or pub_end or rating_low \
       or rating_high or description or cc_present or read_status:
        searchterm, pub_start, pub_end = extend_search_term(searchterm,
                                                            author_name,
                                                            book_title,
                                                            publisher,
                                                            pub_start,
                                                            pub_end,
                                                            tags,
                                                            rating_high,
                                                            rating_low,
                                                            read_status)
        q = adv_search_tag(q, tags['include_tag'], tags['exclude_tag'])
        q = adv_search_serie(q, tags['include_serie'], tags['exclude_serie'])
        q = adv_search_extension(q, tags['include_extension'], tags['exclude_extension'])
        q = adv_search_language(q, tags['include_language'], tags['exclude_language'])
        q = adv_search_ratings(q, rating_high, rating_low)
                                 adv_searchterm=searchterm,
                and user.name != "Guest":
                ip_Address = request.headers.get('X-Forwarded-For', request.remote_addr)
                log.warning('LDAP Login failed for user "%s" IP-address: %s', form['username'], ip_Address)
            ip_Address = request.headers.get('X-Forwarded-For', request.remote_addr)
                        log.info('Password reset for user "%s" IP-address: %s', form['username'], ip_Address)
                    log.warning('Username missing for password reset IP-address: %s', ip_Address)
                    log.warning('Login failed for user "%s" IP-address: %s', form['username'], ip_Address)
        ub.delete_user_session(current_user.id, flask_session.get('_id',""))
                # Query User name, if not existing, change
    translations = babel.list_translations() + [LC('en')]
        flash(_(u"Oops! Selected book title is unavailable. File does not exist or is not accessible"), category="error")
        flash(_(u"Oops! Selected book title is unavailable. File does not exist or is not accessible"), category="error")
        for index in range(0, len(entry.languages)):
            entry.languages[index].language_name = isoLanguages.get_language_name(get_locale(), entry.languages[
                index].lang_code)
        book_in_shelfs = []
            book_in_shelfs.append(sh.shelf)
                                     is_xhr=request.headers.get('X-Requested-With')=='XMLHttpRequest',
                                     books_shelfs=book_in_shelfs,
google-api-python-client>=1.7.11,<2.37.0
google-auth-oauthlib>=0.4.3,<0.5.0
google-api-python-client>=1.7.11,<2.37.0
scholarly>=1.2.0,<1.6
lxml>=3.8.0,<4.8.0

Fixed code:
from cps.editbooks import EditBook
    app.register_blueprint(EditBook)
    config.store_calibre_uuid(calibre_db, db.LibraryId)
from functools import wraps
from babel import Locale
    all_user = ub.session.query(ub.User).all()
    return render_title_template("admin.html", allUser=all_user, email=email_settings, config=config, commit=commit,
    read_column = calibre_db.session.query(db.CustomColumns)\
        .filter(and_(db.CustomColumns.datatype == 'bool', db.CustomColumns.mark_for_delete == 0)).all()
    restrict_columns = calibre_db.session.query(db.CustomColumns)\
        .filter(and_(db.CustomColumns.datatype == 'text', db.CustomColumns.mark_for_delete == 0)).all()
    translations = [Locale('en')] + babel.list_translations()
    translations = babel.list_translations() + [Locale('en')]
    all_user = ub.session.query(ub.User)
        all_user = all_user.filter(ub.User.role.op('&')(constants.ROLE_ANONYMOUS) != constants.ROLE_ANONYMOUS)
                                 users=all_user.all(),
            user.default = Locale.parse(user.default_language).get_language_name(get_locale())
    locale = babel.list_translations() + [Locale('en')]
                            [constants.ROLE_ADMIN, constants.ROLE_PASSWD, constants.ROLE_EDIT_SHELFS]:
        if not calibre_db.session.query(db.CustomColumns).filter(db.CustomColumns.id == column) \
              .filter(and_(db.CustomColumns.datatype == 'bool', db.CustomColumns.mark_for_delete == 0)).all():
        if not calibre_db.session.query(db.CustomColumns).filter(db.CustomColumns.id == column) \
              .filter(and_(db.CustomColumns.datatype == 'text', db.CustomColumns.mark_for_delete == 0)).all():
                or to_save["config_" + str(element['id']) + "_oauth_client_secret"] != element['oauth_client_secret']:
                and to_save["config_" + str(element['id']) + "_oauth_client_secret"]:
            or not config.config_ldap_user_object:
    db_valid = False
            or config.config_calibre_dir != to_save["config_calibre_dir"]:
        config.store_calibre_uuid(calibre_db, db.LibraryId)
                            and config.config_login_type == constants.LOGIN_LDAP)
    translations = [Locale('en')] + babel.list_translations()
    translations = babel.list_translations() + [Locale('en')]
import unidecode
class LibraryId(Base):
    def format_type(self):
    def __init__(self, comment, book):
        self.text = comment
        return self.timestamp.strftime('%Y-%m-%dT%H:%M:%S+00:00') or ''
class CustomColumns(Base):
class CalibreDB:
            self.init_session(expire_on_commit)
    def init_session(self, expire_on_commit=True):
    def setup_db_cc_classes(cls, cc):
                                         echo=False,
                                         isolation_level="SERIALIZABLE",
                                         connect_args={'check_same_thread': False},
                                         poolclass=StaticPool)
                database_uuid = local_session().query(LibraryId).one_or_none()
            inst.init_session()
            archived_books = (ub.session.query(ub.ArchivedBook)
                              .filter(ub.ArchivedBook.user_id == int(current_user.id))
                              .filter(ub.ArchivedBook.is_archived == True)
                              .all())
                    any(cc_classes[self.config.config_restricted_column].value.in_(pos_cc_list))
                    any(cc_classes[self.config.config_restricted_column].value.in_(neg_cc_list))
                                    and_(ub.ReadBook.user_id == int(current_user.id), ub.ReadBook.book_id == Books.id)))
                    query = self.session.query(database, None, ub.ArchivedBook.is_archived)
    def get_search_results(self, term, offset=None, order=None, limit=None,
        _log = logger.create()
        _log.error_or_exception(ex)
    clean_html = None
EditBook = Blueprint('edit-book', __name__)



        # check if an element with that name exists
# Modifies different Database objects, first check if elements have to be deleted,
    db_dict = dict([(identifier.type.lower(), identifier) for identifier in db_identifiers])

@EditBook.route("/ajax/delete/<int:book_id>", methods=["POST"])
@EditBook.route("/delete/<int:book_id>", defaults={'book_format': ""}, methods=["POST"])
@EditBook.route("/delete/<int:book_id>/<string:book_format>", methods=["POST"])
    cc = calibre_db.session.query(db.CustomColumns). \
        filter(db.CustomColumns.datatype.notin_(db.cc_exceptions)).all()
def render_delete_book_result(book_format, json_response, warning, book_id):
        if json_response:
            return json.dumps([warning, {"location": url_for("edit-book.edit_book", book_id=book_id),
            return redirect(url_for('edit-book.edit_book', book_id=book_id))
        if json_response:
def delete_book_from_table(book_id, book_format, json_response):
                    if json_response:
                        return json.dumps([{"location": url_for("edit-book.edit_book", book_id=book_id),
                                            "type": "danger",
                                            "format": "",
                                            "message": error}])
                        return redirect(url_for('edit-book.edit_book', book_id=book_id))
                    if json_response:
                        warning = {"location": url_for("edit-book.edit_book", book_id=book_id),
                                   "type": "warning",
                                   "format": "",
                                   "message": error}
                if json_response:
                    return json.dumps([{"location": url_for("edit-book.edit_book", book_id=book_id),
                    return redirect(url_for('edit-book.edit_book', book_id=book_id))
        return render_delete_book_result(book_format, json_response, warning, book_id)
    if json_response:
        return json.dumps({"location": url_for("edit-book.edit_book", book_id=book_id),
        return redirect(url_for('edit-book.edit_book', book_id=book_id))
    cc = calibre_db.session.query(db.CustomColumns).filter(db.CustomColumns.datatype.notin_(db.cc_exceptions)).all()
        flash(_(u"Oops! Selected book title is unavailable. File does not exist or is not accessible"),
              category="error")
    valid_source_formats = list()
    kepub_possible = None

    modify_date = False
        modify_date = True
    return modify_date
    modify_date = False
            modify_date = True
            book.comments.append(db.Comments(comment=comments, book=book.id))
            modify_date = True
    return modify_date
def edit_book_languages(languages, book, upload_mode=False, invalid=None):
    if not upload_mode:
    for lang in unknown_languages:
        log.error("'%s' is not a valid language", lang)
            invalid.append(lang)
            raise ValueError(_(u"'%(langname)s' is not a valid language", langname=lang))
    if upload_mode and len(input_l) == 1:

    cc = (calibre_db.session.query(db.CustomColumns)
          .filter(db.CustomColumns.datatype.notin_(db.cc_exceptions))
          .filter(db.CustomColumns.id == column_id)

    cc = calibre_db.session.query(db.CustomColumns).filter(db.CustomColumns.datatype.notin_(db.cc_exceptions)).all()


def upload_single_file(file_request, book, book_id):
    if 'btn-upload-format' in file_request.files:
        requested_file = file_request.files['btn-upload-format']
            upload_text = _(u"File format %(ext)s added to %(book)s", ext=file_ext.upper(), book=link)
            WorkerThread.add(current_user.name, TaskUpload(upload_text, escape(book.title)))
def upload_cover(cover_request, book):
    if 'btn-upload-cover' in cover_request.files:
        requested_file = cover_request.files['btn-upload-cover']
    input_authors, renamed = prepare_authors(author_name)
    '''input_authors = author_name.split('&')
                one_book.author_sort = one_book.author_sort.replace(sorted_renamed_author, sorted_old_author)'''
@EditBook.route("/admin/book/<int:book_id>", methods=['GET', 'POST'])
    modify_date = False
        flash(_(u"Oops! Selected book title is unavailable. File does not exist or is not accessible"),
              category="error")
        modify_date = True
            modify_date = True
        error = ""
                                                renamed_author=renamed)
                            modify_date = True
            modify_date |= edit_book_series_index(to_save["series_index"], book)
            modify_date |= edit_book_comments(Markup(to_save['description']).unescape(), book)
            modify_date |= modification
            modify_date |= edit_book_tags(to_save['tags'], book)
            modify_date |= edit_book_series(to_save["series"], book)
            modify_date |= edit_book_publisher(to_save['publisher'], book)
            modify_date |= edit_book_languages(to_save['languages'], book)
            modify_date |= edit_book_ratings(to_save, book)
            modify_date |= edit_all_cc_data(book_id, book, to_save)
            if modify_date:
def prepare_authors(authr):
    return input_authors, renamed


def prepare_authors_on_upload(title, authr):
    if title != _(u'Unknown') and authr != _(u'Unknown'):
        entry = calibre_db.check_exists_book(authr, title)
        if entry:
            log.info("Uploaded book probably exists in library")
            flash(_(u"Uploaded book probably exists in the library, consider to change before upload new: ")
                  + Markup(render_title_template('book_exists_flash.html', entry=entry)), category="warning")

    input_authors, renamed = prepare_authors(authr)
def create_book_on_upload(modify_date, meta):
    # combine path and normalize path from Windows systems
    modify_date |= modify_database_object(input_authors, db_book.authors, db.Authors, calibre_db.session,
                                          'author')
    modify_date |= edit_book_series_index(meta.series_id, db_book)
    invalid = []
    modify_date |= edit_book_languages(meta.languages, db_book, upload_mode=True, invalid=invalid)
        for lang in invalid:
            flash(_(u"'%(langname)s' is not a valid language", langname=lang), category="warning")
    modify_date |= edit_book_tags(meta.tags, db_book)
    modify_date |= edit_book_publisher(meta.publisher, db_book)
    modify_date |= edit_book_series(meta.series, db_book)

@EditBook.route("/upload", methods=["POST"])
                modify_date = False
                db_book, input_authors, title_dir, renamed_authors = create_book_on_upload(modify_date, meta)
                # Comments need book id therefore only possible after flush
                modify_date |= edit_book_comments(Markup(meta.description).unescape(), db_book)
                upload_text = _(u"File %(file)s uploaded", file=link)
                WorkerThread.add(current_user.name, TaskUpload(upload_text, escape(title)))
                        resp = {"location": url_for('edit-book.edit_book', book_id=book_id)}
@EditBook.route("/admin/book/convert/<int:book_id>", methods=['POST'])
        return redirect(url_for('edit-book.edit_book', book_id=book_id))
                book_format=book_format_to),
              category="success")
    return redirect(url_for('edit-book.edit_book', book_id=book_id))

@EditBook.route("/ajax/getcustomenum/<int:c_id>")
    cc = (calibre_db.session.query(db.CustomColumns)
          .filter(db.CustomColumns.id == c_id)
          .filter(db.CustomColumns.datatype.notin_(db.cc_exceptions)).one_or_none())
@EditBook.route("/ajax/editbooks/<param>", methods=['POST'])
    sort_param = ""
                                       'newValue': ', '.join([publisher.name for publisher in book.publishers])}),
                               mimetype='application/json')
            sort_param = book.sort
            ret = Response(json.dumps({
                'success': True,
                'newValue':  ' & '.join([author.replace('|', ',') for author in input_authors])}),
                mimetype='application/json')
                                                message="Book {} archive bit set to: {}".format(book.id, vals['value']))
            book.sort = sort_param
@EditBook.route("/ajax/sort_value/<field>/<int:bookid>")
@EditBook.route("/ajax/simulatemerge", methods=['POST'])
@EditBook.route("/ajax/mergebooks", methods=['POST'])
            to_name = helper.get_valid_filename(to_book.title,
                                                chars=96) + ' - ' + helper.get_valid_filename(to_book.authors[0].name,
                                                                                              chars=96)
                    delete_book_from_table(from_book.id, "", True)
@EditBook.route("/ajax/xchange", methods=['POST'])
    edited_books_id = False
            modify_date = False
                modify_date = True
                                            renamed_author=renamed)
            if modify_date:
    cf_name = tree.xpath('n:rootfiles/n:rootfile/@full-path', namespaces=ns)[0]
    cf = epub_zip.read(cf_name)
    cover_path = os.path.dirname(cf_name)
    cover_file = parse_epub_cover(ns, tree, epub_zip, cover_path, tmp_file_path)
    # if len(cover_section) > 0:
    for cs in cover_section:
        cover_file = _extract_cover(epub_zip, cs, cover_path, tmp_file_path)
        if cover_file:
            break
    if not cover_file:
            if cover_file: break
import unidecode

    import advocate
    from advocate.exceptions import UnacceptableAddressException
    use_advocate = True
    use_advocate = False
    advocate = requests
    UnacceptableAddressException = MissingSchema = BaseException
from . import logger, config, get_locale, db, ub
    if 'AZW3' in formats and 'MOBI' not in formats:
    extensions_reader = {'TXT', 'PDF', 'EPUB', 'CBZ', 'CBT', 'CBR', 'DJVU'}
            if ele.format.upper() in extensions_reader:
            email_text = _(u"%(book)s send to Kindle", book=link)
                             email_text, _(u'This e-mail has been sent via Calibre-Web.')))
    if config.config_unicode_filename:
        value = (unidecode.unidecode(value))
    value2 = None

            read_book = ub.ReadBook(user_id=current_user.id, book_id=book_id)
            read_book.read_status = ub.ReadBook.STATUS_FINISHED
            book = read_book
        except (OperationalError, InvalidRequestError) as ex:
            log.error(u"Read status could not set: {}".format(ex))
            return _("Read status could not set: {}".format(ex.orig))

                except (IOError, OSError) as ex:
                    log.error("Deleting book %s failed: %s", book.id, ex)
                    return False, _("Deleting book %(id)s failed: %(message)s", id=book.id, message=ex)
                    except (IOError, OSError) as ex:
                        log.error("Deleting authorpath for book %s failed: %s", book.id, ex)
                    + get_valid_filename(new_author.name, chars=42)
                                    os.path.normcase(os.path.join(all_new_path,
                                                                  all_new_name + '.' + file_format.format.lower())))
                        g_file = gd.getFileFromEbooksFolder(all_new_path,
                                                            file_format.name + '.' + file_format.format.lower())
                        if g_file:
                            gd.moveGdriveFileRemote(g_file, all_new_name + u'.' + file_format.format.lower())
                            gd.updateDatabaseOnEdit(g_file['id'], all_new_name + u'.' + file_format.format.lower())
                g_file = gd.getFileFromEbooksFolder(None, old_author_dir)
                if g_file:
                    gd.moveGdriveFolderRemote(g_file, new_author_rename_dir)
                    except OSError as ex:

        get_valid_filename(first_author, chars=42) + filename_ext
        g_file = gd.getFileFromEbooksFolder(os.path.dirname(book.path), titledir)
        if g_file:
            gd.moveGdriveFileRemote(g_file, new_titledir)
            gd.updateDatabaseOnEdit(g_file['id'], book.path)     # only child folder affected
        g_file = gd.getFileFromEbooksFolder(os.path.dirname(book.path), new_titledir)
        if g_file:
            gd.moveGdriveFolderRemote(g_file, new_authordir)
            gd.updateDatabaseOnEdit(g_file['id'], book.path)
            else:  # path is valid copy only files to new location (merge)
                                    os.path.normcase(os.path.join(new_path + dir_name[len(path):], file)))
        localbook.path = os.path.join(new_authordir, new_titledir).replace('\\', '/')
        g_file = gd.getFileFromEbooksFolder(book.path, name)
        g_file = gd.getFileFromEbooksFolder(os.path.dirname(book.path), book.path.split('/')[1])
    if g_file:
        gd.deleteDatabaseEntry(g_file['id'])
        g_file.Trash()
    inpt = [" ".join(inp.split()) for inp in inpt]

        raise Exception(_(u"This username is already taken"))
        if cli.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
        else:
            log.error("python modul advocate is not installed but is needed")
            return False, _("Python modul 'advocate' is not installed but is needed for cover downloads")
    except UnacceptableAddressException:
        log.error("Localhost was accessed for cover upload")
        return False, _("You are not allowed to access localhost for cover uploads")
            gd.uploadFileToEbooksFolder(os.path.join(book_path, 'cover.jpg').replace("\\", "/"),
        # startTime = time.time()
        # log.debug('%s', time.time() - startTime)
def check_unrar(unrar_location):
    if not unrar_location:
    if not os.path.exists(unrar_location):
        unrar_location = [unrar_location]
        value = process_wait(unrar_location, pattern='UNRAR (.*) freeware')
    ret_val = ""
        ret_val = format_unit(runtime.days, 'duration-day', length="long", locale=get_locale()) + ', '
        ret_val += '{:d}:{:02d}:{:02d}s'.format(hours, minutes, seconds)
        ret_val += '{:2d}:{:02d}s'.format(minutes, seconds)
        ret_val += '{:2d}s'.format(seconds)
    return ret_val
    tmpcc = calibre_db.session.query(db.CustomColumns)\
        .filter(db.CustomColumns.datatype.notin_(db.cc_exceptions)).all()
    data1= ""
    return render_element_index(db.Books.sort, None, 'opds.feed_letter_books')
        download_book = calibre_db.get_book(book.Downloads.book_id)
        if download_book:
    num_books = entries.__len__()
                            config.config_books_per_page, num_books)
    return render_element_index(db.Authors.sort, db.books_authors_link, 'opds.feed_letter_author')
    return render_xml_dataset(db.Authors, book_id)
    return render_xml_dataset(db.Publishers, book_id)
    return render_element_index(db.Tags.name, db.books_tags_link, 'opds.feed_letter_category')
    return render_xml_dataset(db.Tags, book_id)
    return render_element_index(db.Series.sort, db.books_series_link, 'opds.feed_letter_series')

                                       (db.Ratings.rating / 2).label('name')) \
    return render_xml_dataset(db.Tags, book_id)
    # workaround, loading the user from the request and checking its download rights here
@opds.route("/opds/thumb_240_240/<book_id>")
@opds.route("/opds/cover_240_240/<book_id>")
@opds.route("/opds/cover_90_90/<book_id>")
@opds.route("/opds/cover/<book_id>")
@requires_basic_auth_if_no_ano
def feed_get_cover(book_id):
    return get_book_cover(book_id)


@opds.route("/opds/readbooks")
@requires_basic_auth_if_no_ano
def feed_read_books():
    off = request.args.get("offset") or 0
    result, pagination = render_read_books(int(off) / (int(config.config_books_per_page)) + 1, True, True)
    return render_xml_template('feed.xml', entries=result, pagination=pagination)


@opds.route("/opds/unreadbooks")
@requires_basic_auth_if_no_ano
def feed_unread_books():
    off = request.args.get("offset") or 0
    result, pagination = render_read_books(int(off) / (int(config.config_books_per_page)) + 1, False, True)
    return render_xml_template('feed.xml', entries=result, pagination=pagination)


        ip_address = request.headers.get('X-Forwarded-For', request.remote_addr)
        log.warning('OPDS Login failed for user "%s" IP-address: %s', username.decode('utf-8'), ip_address)
def render_xml_dataset(data_table, book_id):
    entries, __, pagination = calibre_db.fill_indexpage((int(off) / (int(config.config_books_per_page)) + 1), 0,
                                                        db.Books,
                                                        data_table.any(data_table.id == book_id),
                                                        [db.Books.timestamp.desc()])
    return render_xml_template('feed.xml', entries=entries, pagination=pagination)
def render_element_index(database_column, linked_table, folder):
    shift = 0
    off = int(request.args.get("offset") or 0)
    entries = calibre_db.session.query(func.upper(func.substr(database_column, 1, 1)).label('id'))
    if linked_table:
        entries = entries.join(linked_table).join(db.Books)
    entries = entries.filter(calibre_db.common_filters()).group_by(func.upper(func.substr(database_column, 1, 1))).all()
    elements = []
    if off == 0:
        elements.append({'id': "00", 'name': _("All")})
        shift = 1
    for entry in entries[
                 off + shift - 1:
                 int(off + int(config.config_books_per_page) - shift)]:
        elements.append({'id': entry.id, 'name': entry.id})
    pagination = Pagination((int(off) / (int(config.config_books_per_page)) + 1), config.config_books_per_page,
                            len(entries) + 1)
    return render_xml_template('feed.xml',
                               letterelements=elements,
                               folder=folder,
                               pagination=pagination)
    # right_edge: last right_edges count of all pages are shown as number, means, if 10 pages are paginated -> 9,10 shown
    # left_edge: first left_edges count of all pages are shown as number                                    -> 1,2 shown
    # left_current: left_current count below current page are shown as number, means if current page 5      -> 3,4 shown
    # left_current: right_current count above current page are shown as number, means if current page 5     -> 6,7 shown
from functools import wraps
      <form class="padded-bottom" action="{{ url_for('edit-book.convert_bookformat', book_id=book.id) }}" method="post" id="book_convert_frm">
<form role="form" action="{{ url_for('edit-book.edit_book', book_id=book.id) }}" method="post" enctype="multipart/form-data" id="book_edit_frm">
        data-editable-url="{{ url_for('edit-book.edit_list_book', param=parameter)}}"
            <th data-field="series_index" id="series_index" data-visible="{{visiblility.get('series_index')}}" data-edit-validate="{{ _('This Field is Required') }}" data-sortable="true" {% if g.user.role_edit() %} data-editable-type="number" data-editable-placeholder="1" data-editable-step="0.01" data-editable-min="0" data-editable-url="{{ url_for('edit-book.edit_list_book', param='series_index')}}" data-edit="true" data-editable-title="{{_('Enter Title')}}"{% endif %}>{{_('Series Index')}}</th>
            <th data-field="comments" id="comments" data-escape="true" data-editable-mode="popup"  data-visible="{{visiblility.get('comments')}}" data-sortable="false" {% if g.user.role_edit() %} data-editable-type="wysihtml5" data-editable-url="{{ url_for('edit-book.edit_list_book', param='comments')}}" data-edit="true" data-editable-title="{{_('Enter comments')}}"{% endif %}>{{_('Comments')}}</th>
                <th data-field="custom_column_{{ c.id|string }}" id="custom_column_{{ c.id|string }}" data-visible="{{visiblility.get('custom_column_'+ c.id|string)}}" data-sortable="false" {% if g.user.role_edit() %} data-editable-type="number" data-editable-placeholder="1" data-editable-step="1" data-editable-url="{{ url_for('edit-book.edit_list_book', param='custom_column_'+ c.id|string)}}" data-edit="true" data-editable-title="{{_('Enter ') + c.name}}"{% endif %}>{{c.name}}</th>
                <th data-field="custom_column_{{ c.id|string }}" id="custom_column_{{ c.id|string }}" data-formatter="ratingFormatter" data-visible="{{visiblility.get('custom_column_'+ c.id|string)}}" data-sortable="false" {% if g.user.role_edit() %} data-editable-type="number" data-editable-placeholder="1" data-editable-step="0.5" data-editable-step="1" data-editable-min="1" data-editable-max="5" data-editable-url="{{ url_for('edit-book.edit_list_book', param='custom_column_'+ c.id|string)}}" data-edit="true" data-editable-title="{{_('Enter ') + c.name}}"{% endif %}>{{c.name}}</th>
                <th data-field="custom_column_{{ c.id|string }}" id="custom_column_{{ c.id|string }}" data-visible="{{visiblility.get('custom_column_'+ c.id|string)}}" data-sortable="false" {% if g.user.role_edit() %} data-editable-type="number" data-editable-placeholder="1" data-editable-step="0.01" data-editable-url="{{ url_for('edit-book.edit_list_book', param='custom_column_'+ c.id|string)}}" data-edit="true" data-editable-title="{{_('Enter ') + c.name}}"{% endif %}>{{c.name}}</th>
                <th data-field="custom_column_{{ c.id|string }}" id="custom_column_{{ c.id|string }}" data-visible="{{visiblility.get('custom_column_'+ c.id|string)}}" data-sortable="false" {% if g.user.role_edit() %} data-editable-type="select" data-editable-source={{ url_for('edit-book.table_get_custom_enum', c_id=c.id)  }} data-editable-url="{{ url_for('edit-book.edit_list_book', param='custom_column_'+ c.id|string)}}" data-edit="true" data-editable-title="{{_('Enter ') + c.name}}"{% endif %}>{{c.name}}</th>
                  <th data-field="custom_column_{{ c.id|string }}" id="custom_column_{{ c.id|string }}" data-escape="true" data-editable-mode="popup"  data-visible="{{visiblility.get('custom_column_'+ c.id|string)}}" data-sortable="false" {% if g.user.role_edit() %} data-editable-type="wysihtml5" data-editable-url="{{ url_for('edit-book.edit_list_book', param='custom_column_'+ c.id|string)}}" data-edit="true" data-editable-title="{{_('Enter ') + c.name}}"{% endif %}>{{c.name}}</th>
          <a href="{{identifier}}" target="_blank" class="btn btn-xs btn-success" role="button">{{identifier.format_type()}}</a>
          <a href="{{ url_for('edit-book.edit_book', book_id=entry.id) }}" class="btn btn-sm btn-primary" id="edit_book" role="button"><span class="glyphicon glyphicon-edit"></span> {{_('Edit Metadata')}}</a>
                    <form id="form-upload" class="navbar-form" action="{{ url_for('edit-book.upload') }}" data-title="{{_('Uploading...')}}" data-footer="{{_('Close')}}" data-failed="{{_('Error')}}" data-message="{{_('Upload done, processing, please wait...')}}" method="post" enctype="multipart/form-data">
from functools import wraps
from babel import Locale

    register_user_with_oauth = logout_oauth_user = get_oauth_status = None
    csp = "default-src 'self'"
    csp += ''.join([' ' + host for host in config.config_trustedhosts.strip().split(',')])
    csp += " 'unsafe-inline' 'unsafe-eval'; font-src 'self' data:; img-src 'self' data:"
    resp.headers['Content-Security-Policy'] = csp
    if request.endpoint == "edit-book.edit_book" or config.config_use_google_drive:


    l_bookmark = ub.Bookmark(user_id=current_user.id,
                             book_id=book_id,
                             format=book_format,
                             bookmark_key=bookmark_key)
    ub.session.merge(l_bookmark)
    is_archived = change_archived_books(book_id, message="Book {} archive bit toggled".format(book_id))

def generate_char_list(data_colum, db_link):
    return (calibre_db.session.query(func.upper(func.substr(data_colum, 1, 1)).label('char'))
            .join(db_link).join(db.Books).filter(calibre_db.common_filters())
            .group_by(func.upper(func.substr(data_colum, 1, 1))).all())


                                                                False, 0,

            # Unary expression comparsion only working (for this expression) in sqlalchemy 1.4+
            # if not (order[0][0].compare(func.count(ub.Downloads.book_id).desc()) or
            #        order[0][0].compare(func.count(ub.Downloads.book_id).asc())):
        all_books = ub.session.query(ub.Downloads, func.count(ub.Downloads.book_id)) \
            download_book = calibre_db.session.query(db.Books).filter(calibre_db.common_filters()).filter(
            if download_book:
                entries.append(download_book)
        num_books = entries.__len__()
        pagination = Pagination(page, config.config_books_per_page, num_books)
            if not calibre_db.session.query(db.Books).\
                                            filter(calibre_db.common_filters()).filter(db.Books.id == book.id).first():
                                     title=_(u"Downloaded books by %(user)s", user=user.name),
            return []  # ToDo: Handle error Case for opds
            page_name = "read"
            page_name = "unread"
                                     title=name, page=page_name, order=order[1])
    archived_books = (ub.session.query(ub.ArchivedBook)
                      .filter(ub.ArchivedBook.user_id == int(current_user.id))
                      .filter(ub.ArchivedBook.is_archived == True)
                      .all())
    page_name = "archived"
                                 title=name, page=page_name, order=sort_param[1])
    tags = calibre_db.session.query(db.Tags) \
        .join(db.books_tags_link) \
        .join(db.Books) \
        .group_by(text('books_tags_link.tag')) \
    series = calibre_db.session.query(db.Series) \
        .join(db.books_series_link) \
        .join(db.Books) \
        .group_by(text('books_series_link.series')) \
        .order_by(db.Series.name) \
    shelves = ub.session.query(ub.Shelf) \
        .filter(or_(ub.Shelf.is_public == 1, ub.Shelf.user_id == int(current_user.id))) \
    extensions = calibre_db.session.query(db.Data) \
        .join(db.Books) \
        .group_by(db.Data.format) \
                                 series=series, shelves=shelves, title=_(u"Advanced Search"), cc=cc, page="advsearch")

    search_param = request.args.get("search")
        join = db.books_authors_link, db.Books.id == db.books_authors_link.c.book, db.Authors, db.books_series_link, \
            db.Books.id == db.books_series_link.c.book, db.Series
    total_count = filtered_count = calibre_db.session.query(db.Books).filter(
        calibre_db.common_filters(allow_show_archived=True)).count()
        if search_param:
            books = calibre_db.search_query(search_param, config.config_read_column).all()
                                                           int(current_user.id) == ub.ArchivedBook.user_id))
    elif search_param:
        entries, filtered_count, __ = calibre_db.get_search_results(search_param,
                                                                    [order, ''],
        for lang_index in range(0, len(val.languages)):
            val.languages[lang_index].language_name = isoLanguages.get_language_name(get_locale(), val.languages[
                lang_index].lang_code)

        char_list = generate_char_list(db.Authors.sort, db.books_authors_link)
        author_copy = copy.deepcopy(entries)
        for entry in author_copy:
        return render_title_template('list.html', entries=author_copy, folder='web.books_list', charlist=char_list,

        entries = ub.session.query(ub.User, func.count(ub.Downloads.book_id).label('count')) \
        char_list = ub.session.query(func.upper(func.substr(ub.User.name, 1, 1)).label('char')) \
        return render_title_template('list.html', entries=entries, folder='web.books_list', charlist=char_list,
        char_list = generate_char_list(db.Publishers.name, db.books_publishers_link)
        return render_title_template('list.html', entries=entries, folder='web.books_list', charlist=char_list,
        char_list = generate_char_list(db.Series.sort, db.books_series_link)
            return render_title_template('list.html', entries=entries, folder='web.books_list', charlist=char_list,
                .join(db.books_series_link).join(db.Series).filter(calibre_db.common_filters()) \
            return render_title_template('grid.html', entries=entries, folder='web.books_list', charlist=char_list,
                                           (db.Ratings.rating / 2).label('name')) \
        char_list = list()
            if upper_lang not in char_list:
                char_list.append(upper_lang)
                                     charlist=char_list, title=_(u"Languages"), page="langlist",
        char_list = generate_char_list(db.Tags.name, db.books_tags_link)
        return render_title_template('list.html', entries=entries, folder='web.books_list', charlist=char_list,
def adv_search_text(q, include_inputs, exclude_inputs, data_value):
    for inp in include_inputs:
        q = q.filter(db.Books.data.any(data_value == inp))
    for excl in exclude_inputs:
        q = q.filter(not_(db.Books.data.any(data_value == excl)))
    return q


'''def adv_search_extension(q, include_extension_inputs, exclude_extension_inputs):
    return q'''

    q = q.outerjoin(ub.BookShelf, db.Books.id == ub.BookShelf.book_id) \

    elements = {'tag': db.Tags, 'serie': db.Series, 'shelf': ub.Shelf}
    q = query.outerjoin(db.books_series_link, db.Books.id == db.books_series_link.c.book) \
        .outerjoin(db.Series) \
    search_term = []
                search_term.extend([u"{} >= {}".format(c.name,
                                                       format_date(datetime.strptime(column_start, "%Y-%m-%d").date(),
                                                                   format='medium',
                                                                   locale=get_locale())
                                                       )])
                search_term.extend([u"{} <= {}".format(c.name,
                                                       format_date(datetime.strptime(column_end, "%Y-%m-%d").date(),
                                                                   format='medium',
                                                                   locale=get_locale())
                                                       )])
            search_term.extend([(u"{}: {}".format(c.name, term.get('custom_column_' + str(c.id))))])
    if any(tags.values()) or author_name or book_title or \
        publisher or pub_start or pub_end or rating_low or rating_high \
            or description or cc_present or read_status:
        search_term, pub_start, pub_end = extend_search_term(search_term,
                                                             author_name,
                                                             book_title,
                                                             publisher,
                                                             pub_start,
                                                             pub_end,
                                                             tags,
                                                             rating_high,
                                                             rating_low,
                                                             read_status)
        q = adv_search_text(q, tags['include_tag'], tags['exclude_tag'], db.Tags.id)
        q = adv_search_text(q, tags['include_serie'], tags['exclude_serie'], db.Series.id)
        q = adv_search_text(q, tags['include_extension'], tags['exclude_extension'], db.Data.format)
        q = adv_search_language(q, tags['include_language'], tags['exclude_language'], )
        q = adv_search_ratings(q, rating_high, rating_low, )
                                 adv_searchterm=search_term,


                    and user.name != "Guest":
                ip_address = request.headers.get('X-Forwarded-For', request.remote_addr)
                log.warning('LDAP Login failed for user "%s" IP-address: %s', form['username'], ip_address)
            ip_address = request.headers.get('X-Forwarded-For', request.remote_addr)
                        log.info('Password reset for user "%s" IP-address: %s', form['username'], ip_address)
                    log.warning('Username missing for password reset IP-address: %s', ip_address)
                    log.warning('Login failed for user "%s" IP-address: %s', form['username'], ip_address)
        ub.delete_user_session(current_user.id, flask_session.get('_id', ""))
                # Query username, if not existing, change
    translations = babel.list_translations() + [Locale('en')]
        flash(_(u"Oops! Selected book title is unavailable. File does not exist or is not accessible"),
              category="error")
        flash(_(u"Oops! Selected book title is unavailable. File does not exist or is not accessible"),
              category="error")
        for lang_index in range(0, len(entry.languages)):
            entry.languages[lang_index].language_name = isoLanguages.get_language_name(get_locale(), entry.languages[
                lang_index].lang_code)
        book_in_shelves = []
            book_in_shelves.append(sh.shelf)
                                     is_xhr=request.headers.get('X-Requested-With') == 'XMLHttpRequest',
                                     books_shelfs=book_in_shelves,
google-api-python-client>=1.7.11,<2.41.0
google-auth-oauthlib>=0.4.3,<0.6.0
google-api-python-client>=1.7.11,<2.41.0
scholarly>=1.2.0,<1.7
lxml>=3.8.0,<4.9.0
advocate>=1.0.0,<1.1.0