user01/calibre-web
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update mass edit

Browse Source 
Refactored delete User function
Updated testresults
This commit is contained in:
Ozzie Isaacs
2024-12-09 17:08:58 +01:00
parent 89e9958222
commit 08527ae3ce
10 changed files with 473 additions and 372 deletions
Download Patch File Download Diff File Expand all files Collapse all files

421
cps/editbooks.py
View File

@@ -73,17 +73,18 @@ def edit_required(f):
return inner
@editbook.route("/ajax/delete/<int:book_id>", methods=["POST"])
@editbook.route("/ajax/deletebook", methods=["POST"])
@user_login_required
def delete_book_from_details(book_id):
return delete_book_from_table(book_id, "", True) # , mimetype='application/json')
def delete_books_ajax():
book_ids = request.get_json().get("bookid")
return check_delete_book(book_ids, "", True)
@editbook.route("/delete/<int:book_id>", defaults={'book_format': ""}, methods=["POST"])
@editbook.route("/delete/<int:book_id>/<string:book_format>", methods=["POST"])
@user_login_required
def delete_book_ajax(book_id, book_format):
return delete_book_from_table(book_id, book_format, False, request.form.to_dict().get('location', ""))
def delete_book(book_id, book_format):
return check_delete_book(book_id, book_format, False, request.form.to_dict().get('location', ""))
@editbook.route("/admin/book/<int:book_id>", methods=['GET'])
@@ -213,7 +214,7 @@ def table_get_custom_enum(c_id):
@login_required_if_no_ano
@edit_required
def edit_list_book(param):
vals = request.form.to_dict()
vals = request.get_json() # form.to_dict(flat=False)
return edit_book_param(param, vals)
@editbook.route("/ajax/editselectedbooks", methods=['POST'])
@@ -233,42 +234,39 @@ def edit_selected_books():
comments = d.get('comments')
checkA = d.get('checkA')
if len(selections) != 0:
for book_id in selections:
vals = {
"pk": book_id,
"value": None,
"checkA": checkA,
}
if title:
vals['value'] = title
edit_book_param('title', vals)
if title_sort:
vals['value'] = title_sort
edit_book_param('sort', vals)
if author_sort:
vals['value'] = author_sort
edit_book_param('author_sort', vals)
if authors:
vals['value'] = authors
edit_book_param('authors', vals)
if categories:
vals['value'] = categories
edit_book_param('tags', vals)
if series:
vals['value'] = series
edit_book_param('series', vals)
if languages:
vals['value'] = languages
edit_book_param('languages', vals)
if publishers:
vals['value'] = publishers
edit_book_param('publishers', vals)
if comments:
vals['value'] = comments
edit_book_param('comments', vals)
return json.dumps({'success': True})
return ""
vals = {
"pk": selections,
"value": None,
"checkA": checkA,
}
if title:
vals['value'] = title
edit_book_param('title', vals)
if title_sort:
vals['value'] = title_sort
edit_book_param('sort', vals)
if author_sort:
vals['value'] = author_sort
edit_book_param('author_sort', vals)
if authors:
vals['value'] = authors
edit_book_param('authors', vals)
if categories:
vals['value'] = categories
edit_book_param('tags', vals)
if series:
vals['value'] = series
edit_book_param('series', vals)
if languages:
vals['value'] = languages
edit_book_param('languages', vals)
if publishers:
vals['value'] = publishers
edit_book_param('publishers', vals)
if comments:
vals['value'] = comments
edit_book_param('comments', vals)
return json.dumps({'success': True})
# Separated from /editbooks so that /editselectedbooks can also use this
#
@@ -284,94 +282,96 @@ def edit_selected_books():
@login_required_if_no_ano
@edit_required
def edit_book_param(param, vals):
book = calibre_db.get_book(vals['pk'])
calibre_db.create_functions(config)
sort_param = ""
ret = ""
try:
if param == 'series_index':
edit_book_series_index(vals['value'], book)
ret = make_response(jsonify(success=True, newValue=book.series_index))
elif param == 'tags':
edit_book_tags(vals['value'], book)
ret = make_response(jsonify(success=True, newValue=', '.join([tag.name for tag in book.tags])))
elif param == 'series':
edit_book_series(vals['value'], book)
ret = make_response(jsonify(success=True, newValue=', '.join([serie.name for serie in book.series])))
elif param == 'publishers':
edit_book_publisher(vals['value'], book)
ret = make_response(jsonify(success=True,
newValue=', '.join([publisher.name for publisher in book.publishers])))
elif param == 'languages':
invalid = list()
edit_book_languages(vals['value'], book, invalid=invalid)
if invalid:
ret = make_response(jsonify(success=False,
msg='Invalid languages in request: {}'.format(','.join(invalid))))
else:
lang_names = list()
for lang in book.languages:
lang_names.append(isoLanguages.get_language_name(get_locale(), lang.lang_code))
ret = make_response(jsonify(success=True, newValue=', '.join(lang_names)))
elif param == 'author_sort':
book.author_sort = vals['value']
ret = make_response(jsonify(success=True, newValue=book.author_sort))
elif param == 'title':
sort_param = book.sort
if handle_title_on_edit(book, vals.get('value', "")):
rename_error = helper.update_dir_structure(book.id, config.get_book_path())
if not rename_error:
ret = make_response(jsonify(success=True, newValue=book.title))
elements = vals.get('pk',[])
ret = {}
for elem in elements:
book = calibre_db.get_book(elem)
calibre_db.create_functions(config)
sort_param = ""
try:
if param == 'series_index':
edit_book_series_index(vals['value'], book)
ret = jsonify(success=True, newValue=book.series_index)
elif param == 'tags':
edit_book_tags(vals['value'], book)
ret = jsonify(success=True, newValue=', '.join([tag.name for tag in book.tags]))
elif param == 'series':
edit_book_series(vals['value'], book)
ret = jsonify(success=True, newValue=', '.join([serie.name for serie in book.series]))
elif param == 'publishers':
edit_book_publisher(vals['value'], book)
ret = jsonify(success=True,
newValue=', '.join([publisher.name for publisher in book.publishers]))
elif param == 'languages':
invalid = list()
edit_book_languages(vals['value'], book, invalid=invalid)
if invalid:
ret = jsonify(success=False, msg='Invalid languages in request: {}'.format(','.join(invalid)))
else:
ret = make_response(jsonify(success=False, msg=rename_error))
elif param == 'sort':
book.sort = vals['value']
ret = make_response(jsonify(success=True,newValue=book.sort))
elif param == 'comments':
edit_book_comments(vals['value'], book)
ret = make_response(jsonify(success=True, newValue=book.comments[0].text))
elif param == 'authors':
input_authors, __ = handle_author_on_edit(book, vals['value'], vals.get('checkA', None) == "true")
rename_error = helper.update_dir_structure(book.id, config.get_book_path(), input_authors[0])
if not rename_error:
ret = make_response(jsonify(
success=True,
newValue=' & '.join([author.replace('|', ',') for author in input_authors])))
lang_names = list()
for lang in book.languages:
lang_names.append(isoLanguages.get_language_name(get_locale(), lang.lang_code))
ret = jsonify(success=True, newValue=', '.join(lang_names))
elif param == 'author_sort':
book.author_sort = vals['value']
ret = jsonify(success=True, newValue=book.author_sort)
elif param == 'title':
sort_param = book.sort
if handle_title_on_edit(book, vals.get('value', "")):
rename_error = helper.update_dir_structure(book.id, config.get_book_path())
if not rename_error:
ret = jsonify(success=True, newValue=book.title)
else:
ret = jsonify(success=False, msg=rename_error)
elif param == 'sort':
book.sort = vals['value']
ret = jsonify(success=True,newValue=book.sort)
elif param == 'comments':
edit_book_comments(vals['value'], book)
ret = jsonify(success=True, newValue=book.comments[0].text)
elif param == 'authors':
input_authors, __ = handle_author_on_edit(book, vals['value'], vals.get('checkA', None) == True)
rename_error = helper.update_dir_structure(book.id, config.get_book_path(), input_authors[0])
if not rename_error:
ret = jsonify(
success=True,
newValue=' & '.join([author.replace('|', ',') for author in input_authors]))
else:
ret = jsonify(success=False, msg=rename_error)
elif param == 'is_archived':
is_archived = change_archived_books(book.id, vals['value'] == "True",
message="Book {} archive bit set to: {}".format(book.id,
vals['value']))
if is_archived:
kobo_sync_status.remove_synced_book(book.id)
continue
elif param == 'read_status':
error = helper.edit_book_read_status(book.id, vals['value'] == "True")
if error:
return error, 400
continue
elif param.startswith("custom_column_"):
new_val = dict()
new_val[param] = vals['value']
edit_single_cc_data(book.id, book, param[14:], new_val)
# ToDo: Very hacky find better solution
if vals['value'] in ["True", "False"]:
ret = {}
else:
ret = jsonify(success=True, newValue=vals['value'])
else:
ret = make_response(jsonify(success=False, msg=rename_error))
elif param == 'is_archived':
is_archived = change_archived_books(book.id, vals['value'] == "True",
message="Book {} archive bit set to: {}".format(book.id, vals['value']))
if is_archived:
kobo_sync_status.remove_synced_book(book.id)
return ""
elif param == 'read_status':
ret = helper.edit_book_read_status(book.id, vals['value'] == "True")
if ret:
return ret, 400
elif param.startswith("custom_column_"):
new_val = dict()
new_val[param] = vals['value']
edit_single_cc_data(book.id, book, param[14:], new_val)
# ToDo: Very hacky find better solution
if vals['value'] in ["True", "False"]:
ret = ""
else:
ret = make_response(jsonify(success=True, newValue=vals['value']))
else:
return _("Parameter not found"), 400
book.last_modified = datetime.now(timezone.utc)
return _("Parameter not found"), 400
book.last_modified = datetime.now(timezone.utc)
calibre_db.session.commit()
# revert change for sort if automatic fields link is deactivated
if param == 'title' and vals.get('checkT') == "false":
book.sort = sort_param
calibre_db.session.commit()
except (OperationalError, IntegrityError, StaleDataError) as e:
calibre_db.session.rollback()
log.error_or_exception("Database error: {}".format(e))
ret = make_response(jsonify(success=False,
msg='Database error: {}'.format(e.orig if hasattr(e, "orig") else e)))
# revert change for sort if automatic fields link is deactivated
if param == 'title' and vals.get('checkT') == False:
book.sort = sort_param
calibre_db.session.commit()
except (OperationalError, IntegrityError, StaleDataError, AttributeError) as e:
calibre_db.session.rollback()
log.error_or_exception("Database error: {}".format(e))
ret = jsonify(success=False, msg='Database error: {}'.format(e.orig if hasattr(e, "orig") else e))
return ret
@@ -433,16 +433,6 @@ def archive_selected_books():
return json.dumps({'success': True})
return ""
@editbook.route("/ajax/deleteselectedbooks", methods=['POST'])
@user_login_required
@edit_required
def delete_selected_books():
vals = request.get_json().get('selections')
if vals:
for book_id in vals:
delete_book_from_table(book_id, "", True)
return json.dumps({'success': True})
return ""
@editbook.route("/ajax/readselectedbooks", methods=['POST'])
@user_login_required
@@ -498,7 +488,7 @@ def merge_list_book():
element.format,
element.uncompressed_size,
to_name))
delete_book_from_table(from_book.id, "", True)
check_delete_book([from_book.id], "", True)
return make_response(jsonify(success=True))
return ""
@@ -968,86 +958,109 @@ def delete_whole_book(book_id, book):
calibre_db.session.query(db.Books).filter(db.Books.id == book_id).delete()
def render_delete_book_result(book_format, json_response, warning, book_id, location=""):
def render_delete_book_result(book_format, book_id, location=""):
if book_format:
if json_response:
return jsonify([warning, {"location": url_for("edit-book.show_edit_book", book_id=book_id),
"type": "success",
"format": book_format,
"message": _('Book Format Successfully Deleted')}])
else:
flash(_('Book Format Successfully Deleted'), category="success")
return redirect(url_for('edit-book.show_edit_book', book_id=book_id))
flash(_('Book Format Successfully Deleted'), category="success")
return redirect(url_for('edit-book.show_edit_book', book_id=book_id))
else:
if json_response:
return jsonify([warning, {"location": get_redirect_location(location, "web.index"),
"type": "success",
"format": book_format,
"message": _('Book Successfully Deleted')}])
else:
flash(_('Book Successfully Deleted'), category="success")
return redirect(get_redirect_location(location, "web.index"))
flash(_('Book Successfully Deleted'), category="success")
return redirect(get_redirect_location(location, "web.index"))
def delete_book_from_table(book_id, book_format, json_response, location=""):
warning = {}
def check_delete_book(book_id, book_format, json_response, location=""):
if current_user.role_delete_books():
book = calibre_db.get_book(book_id)
if book:
try:
result, error = helper.delete_book(book, config.get_book_path(), book_format=book_format.upper())
if not result:
if json_response:
return jsonify([{"location": url_for("edit-book.show_edit_book", book_id=book_id),
"type": "danger",
"format": "",
"message": error}])
else:
flash(error, category="error")
return redirect(url_for('edit-book.show_edit_book', book_id=book_id))
if error:
if json_response:
warning = {"location": url_for("edit-book.show_edit_book", book_id=book_id),
"type": "warning",
"format": "",
"message": error}
else:
flash(error, category="warning")
if not book_format:
delete_whole_book(book_id, book)
else:
calibre_db.session.query(db.Data).filter(db.Data.book == book.id).\
filter(db.Data.format == book_format).delete()
if book_format.upper() in ['KEPUB', 'EPUB', 'EPUB3']:
kobo_sync_status.remove_synced_book(book.id, True)
calibre_db.session.commit()
except Exception as ex:
log.error_or_exception(ex)
calibre_db.session.rollback()
if json_response:
return jsonify([{"location": url_for("edit-book.show_edit_book", book_id=book_id),
"type": "danger",
"format": "",
"message": ex}])
else:
flash(str(ex), category="error")
return redirect(url_for('edit-book.show_edit_book', book_id=book_id))
if json_response:
# if json response is set, it's possible to delete more than one book, but never a format is deleted
res = list()
for b in book_id:
ret = delete_book_from_table(b)
if ret:
res.extend([ret])
if len(res) == 0:
return [{"location": get_redirect_location(location, "web.index"),
"type": "success",
"format": "",
"message": _('Book Successfully Deleted')}]
return jsonify(res)
else:
# book not found
log.error('Book with id "%s" could not be deleted: not found', book_id)
return render_delete_book_result(book_format, json_response, warning, book_id, location)
return delete_book_from_UI(book_id, book_format, location)
message = _("You are missing permissions to delete books")
if json_response:
return jsonify({"location": url_for("edit-book.show_edit_book", book_id=book_id),
"type": "danger",
"format": "",
"message": message})
try:
return jsonify({"location": url_for("edit-book.show_edit_book", book_id=int(book_id)),
"type": "danger",
"format": "",
"message": message})
except TypeError as e:
return jsonify({"location": url_for("web.index"), "type": "danger", "format": "",
"message": str(e)})
else:
flash(message, category="error")
return redirect(url_for('edit-book.show_edit_book', book_id=book_id))
def delete_book_from_UI(book_id, book_format, location=""):
book = calibre_db.get_book(book_id)
if book:
try:
result, error = helper.delete_book(book, config.get_book_path(), book_format=book_format.upper())
if not result:
flash(error, category="error")
return redirect(url_for('edit-book.show_edit_book', book_id=book_id))
if error:
flash(error, category="warning")
if not book_format:
delete_whole_book(book_id, book)
else:
calibre_db.session.query(db.Data).filter(db.Data.book == book.id). \
filter(db.Data.format == book_format).delete()
if book_format.upper() in ['KEPUB', 'EPUB', 'EPUB3']:
kobo_sync_status.remove_synced_book(book.id, True)
calibre_db.session.commit()
except Exception as ex:
log.error_or_exception(ex)
calibre_db.session.rollback()
flash(str(ex), category="error")
return redirect(url_for('edit-book.show_edit_book', book_id=book_id))
else:
# book not found
log.error('Book with id "%s" could not be deleted: not found', book_id)
return render_delete_book_result(book_format, book_id, location)
def delete_book_from_table(book_id):
book = calibre_db.get_book(book_id)
if book:
try:
result, error = helper.delete_book(book, config.get_book_path(), book_format="")
if not result:
return {"location": url_for("edit-book.show_edit_book", book_id=book_id),
"type": "danger",
"format": "",
"message": error}
delete_whole_book(book_id, book)
calibre_db.session.commit()
if error:
return {"location": url_for("edit-book.show_edit_book", book_id=book_id),
"type": "warning",
"format": "",
"message": error}
except Exception as ex:
log.error_or_exception(ex)
calibre_db.session.rollback()
return {"location": url_for("edit-book.show_edit_book", book_id=book_id),
"type": "danger",
"format": "",
"message": ex}
else:
# book not found
log.error('Book with id "%s" could not be deleted: not found', book_id)
return {"location": url_for("edit-book.show_edit_book", book_id=book_id),
"type": "danger",
"format": "",
"message": _('Book with id "{}" could not be deleted: not found'.format(book_id))}
def render_edit_book(book_id):
cc = calibre_db.session.query(db.CustomColumns).filter(db.CustomColumns.datatype.notin_(db.cc_exceptions)).all()
book = calibre_db.get_filtered_book(book_id, allow_show_archived=True)