1
0
mirror of https://github.com/janeczku/calibre-web synced 2024-11-14 05:44:53 +00:00

Fix cover handling on upload formats

This commit is contained in:
Ozzie Isaacs 2024-08-18 10:12:04 +02:00
parent 1eb77c9cd4
commit 0f9b5d08e9
5 changed files with 67 additions and 57 deletions

View File

@ -26,7 +26,7 @@ from cps.constants import BookMeta
log = logger.create() log = logger.create()
def get_audio_file_info(tmp_file_path, original_file_extension, original_file_name): def get_audio_file_info(tmp_file_path, original_file_extension, original_file_name, no_cover_processing):
tmp_cover_name = None tmp_cover_name = None
audio_file = mutagen.File(tmp_file_path) audio_file = mutagen.File(tmp_file_path)
comments = None comments = None
@ -50,7 +50,7 @@ def get_audio_file_info(tmp_file_path, original_file_extension, original_file_na
pubdate = str(audio_file.tags.get('TDRC').text[0]) if "TDRC" in audio_file.tags else None pubdate = str(audio_file.tags.get('TDRC').text[0]) if "TDRC" in audio_file.tags else None
if not pubdate: if not pubdate:
pubdate = str(audio_file.tags.get('TDOR').text[0]) if "TDOR" in audio_file.tags else None pubdate = str(audio_file.tags.get('TDOR').text[0]) if "TDOR" in audio_file.tags else None
if cover_data: if cover_data and not no_cover_processing:
tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg') tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg')
cover_info = cover_data[0] cover_info = cover_data[0]
for dat in cover_data: for dat in cover_data:
@ -68,18 +68,19 @@ def get_audio_file_info(tmp_file_path, original_file_extension, original_file_na
publisher = audio_file.tags.get('LABEL')[0] if "LABEL" in audio_file else None publisher = audio_file.tags.get('LABEL')[0] if "LABEL" in audio_file else None
pubdate = audio_file.tags.get('DATE')[0] if "DATE" in audio_file else None pubdate = audio_file.tags.get('DATE')[0] if "DATE" in audio_file else None
cover_data = audio_file.tags.get('METADATA_BLOCK_PICTURE') cover_data = audio_file.tags.get('METADATA_BLOCK_PICTURE')
if cover_data: if not no_cover_processing:
tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg') if cover_data:
cover_info = mutagen.flac.Picture(base64.b64decode(cover_data[0])) tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg')
cover.cover_processing(tmp_file_path, cover_info.data, "." + cover_info.mime[-3:]) cover_info = mutagen.flac.Picture(base64.b64decode(cover_data[0]))
if hasattr(audio_file, "pictures"): cover.cover_processing(tmp_file_path, cover_info.data, "." + cover_info.mime[-3:])
cover_info = audio_file.pictures[0] if hasattr(audio_file, "pictures"):
for dat in audio_file.pictures: cover_info = audio_file.pictures[0]
if dat.type == mutagen.id3.PictureType.COVER_FRONT: for dat in audio_file.pictures:
cover_info = dat if dat.type == mutagen.id3.PictureType.COVER_FRONT:
break cover_info = dat
tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg') break
cover.cover_processing(tmp_file_path, cover_info.data, "." + cover_info.mime[-3:]) tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg')
cover.cover_processing(tmp_file_path, cover_info.data, "." + cover_info.mime[-3:])
elif original_file_extension in [".aac"]: elif original_file_extension in [".aac"]:
title = audio_file.tags.get('Title').value if "Title" in audio_file else None title = audio_file.tags.get('Title').value if "Title" in audio_file else None
author = audio_file.tags.get('Artist').value if "Artist" in audio_file else None author = audio_file.tags.get('Artist').value if "Artist" in audio_file else None
@ -90,7 +91,7 @@ def get_audio_file_info(tmp_file_path, original_file_extension, original_file_na
publisher = audio_file.tags.get('Label').value if "Label" in audio_file else None publisher = audio_file.tags.get('Label').value if "Label" in audio_file else None
pubdate = audio_file.tags.get('Year').value if "Year" in audio_file else None pubdate = audio_file.tags.get('Year').value if "Year" in audio_file else None
cover_data = audio_file.tags['Cover Art (Front)'] cover_data = audio_file.tags['Cover Art (Front)']
if cover_data: if cover_data and not no_cover_processing:
tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg') tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg')
with open(tmp_cover_name, "wb") as cover_file: with open(tmp_cover_name, "wb") as cover_file:
cover_file.write(cover_data.value.split(b"\x00",1)[1]) cover_file.write(cover_data.value.split(b"\x00",1)[1])
@ -104,7 +105,7 @@ def get_audio_file_info(tmp_file_path, original_file_extension, original_file_na
publisher = audio_file.tags.get('Label')[0].value if "Label" in audio_file else None publisher = audio_file.tags.get('Label')[0].value if "Label" in audio_file else None
pubdate = audio_file.tags.get('Year')[0].value if "Year" in audio_file else None pubdate = audio_file.tags.get('Year')[0].value if "Year" in audio_file else None
cover_data = audio_file.tags.get('WM/Picture', None) cover_data = audio_file.tags.get('WM/Picture', None)
if cover_data: if cover_data and not no_cover_processing:
tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg') tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg')
with open(tmp_cover_name, "wb") as cover_file: with open(tmp_cover_name, "wb") as cover_file:
cover_file.write(cover_data[0].value) cover_file.write(cover_data[0].value)
@ -118,7 +119,7 @@ def get_audio_file_info(tmp_file_path, original_file_extension, original_file_na
publisher = "" publisher = ""
pubdate = audio_file.tags.get('©day')[0] if "©day" in audio_file.tags else None pubdate = audio_file.tags.get('©day')[0] if "©day" in audio_file.tags else None
cover_data = audio_file.tags.get('covr', None) cover_data = audio_file.tags.get('covr', None)
if cover_data: if cover_data and not no_cover_processing:
tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg') tmp_cover_name = os.path.join(os.path.dirname(tmp_file_path), 'cover.jpg')
cover_type = None cover_type = None
for c in cover_data: for c in cover_data:

View File

@ -130,7 +130,7 @@ def _extract_cover(tmp_file_name, original_file_extension, rar_executable):
return cover.cover_processing(tmp_file_name, cover_data, extension) return cover.cover_processing(tmp_file_name, cover_data, extension)
def get_comic_info(tmp_file_path, original_file_name, original_file_extension, rar_executable): def get_comic_info(tmp_file_path, original_file_name, original_file_extension, rar_executable, no_cover_processing):
if use_comic_meta: if use_comic_meta:
try: try:
archive = ComicArchive(tmp_file_path, rar_exe_path=rar_executable) archive = ComicArchive(tmp_file_path, rar_exe_path=rar_executable)
@ -155,14 +155,17 @@ def get_comic_info(tmp_file_path, original_file_name, original_file_extension, r
lang = loaded_metadata.language or "" lang = loaded_metadata.language or ""
loaded_metadata.language = isoLanguages.get_lang3(lang) loaded_metadata.language = isoLanguages.get_lang3(lang)
if not no_cover_processing:
cover_file = _extract_cover(tmp_file_path, original_file_extension, rar_executable)
else:
cover_file = None
return BookMeta( return BookMeta(
file_path=tmp_file_path, file_path=tmp_file_path,
extension=original_file_extension, extension=original_file_extension,
title=loaded_metadata.title or original_file_name, title=loaded_metadata.title or original_file_name,
author=" & ".join([credit["person"] author=" & ".join([credit["person"]
for credit in loaded_metadata.credits if credit["role"] == "Writer"]) or 'Unknown', for credit in loaded_metadata.credits if credit["role"] == "Writer"]) or 'Unknown',
cover=_extract_cover(tmp_file_path, original_file_extension, rar_executable), cover=cover_file,
description=loaded_metadata.comments or "", description=loaded_metadata.comments or "",
tags="", tags="",
series=loaded_metadata.series or "", series=loaded_metadata.series or "",
@ -171,13 +174,17 @@ def get_comic_info(tmp_file_path, original_file_name, original_file_extension, r
publisher="", publisher="",
pubdate="", pubdate="",
identifiers=[]) identifiers=[])
if not no_cover_processing:
cover_file = _extract_cover(tmp_file_path, original_file_extension, rar_executable)
else:
cover_file = None
return BookMeta( return BookMeta(
file_path=tmp_file_path, file_path=tmp_file_path,
extension=original_file_extension, extension=original_file_extension,
title=original_file_name, title=original_file_name,
author='Unknown', author='Unknown',
cover=_extract_cover(tmp_file_path, original_file_extension, rar_executable), cover=cover_file,
description="", description="",
tags="", tags="",
series="", series="",

View File

@ -140,9 +140,7 @@ def upload():
input_authors[0], input_authors[0],
meta.file_path, meta.file_path,
title_dir + meta.extension.lower()) title_dir + meta.extension.lower())
move_coverfile(meta, db_book) move_coverfile(meta, db_book)
if modify_date: if modify_date:
calibre_db.set_metadata_dirty(book_id) calibre_db.set_metadata_dirty(book_id)
# save data to database, reread data # save data to database, reread data
@ -480,13 +478,10 @@ def do_edit_book(book_id, upload_formats=None):
meta = True meta = True
else: else:
# handle upload other formats from local disk # handle upload other formats from local disk
meta = upload_single_file(upload_formats, book, book_id) to_save, upload_format = upload_book_formats(upload_formats, book, book_id, book.has_cover)
# only merge metadata if file was uploaded and no error occurred (meta equals not false or none)
if meta:
upload_format = merge_metadata(to_save, meta)
# handle upload covers from local disk # handle upload covers from local disk
cover_upload_success = upload_cover(request, book) cover_upload_success = upload_cover(request, book)
if cover_upload_success: if cover_upload_success or to_save.get("format_cover"):
book.has_cover = 1 book.has_cover = 1
modify_date = True modify_date = True
@ -494,7 +489,7 @@ def do_edit_book(book_id, upload_formats=None):
if config.config_use_google_drive: if config.config_use_google_drive:
gdriveutils.updateGdriveCalibreFromLocal() gdriveutils.updateGdriveCalibreFromLocal()
if to_save.get("cover_url", None): if to_save.get("cover_url",):
if not current_user.role_upload(): if not current_user.role_upload():
edit_error = True edit_error = True
flash(_("User has no rights to upload cover"), category="error") flash(_("User has no rights to upload cover"), category="error")
@ -554,13 +549,15 @@ def do_edit_book(book_id, upload_formats=None):
calibre_db.session.commit() calibre_db.session.commit()
if config.config_use_google_drive: if config.config_use_google_drive:
gdriveutils.updateGdriveCalibreFromLocal() gdriveutils.updateGdriveCalibreFromLocal()
# if format is upladed and something goes wrong to_save is set to empty dictonary
if (len(to_save)
and edit_error is not True and title_author_error is not True and cover_upload_success is not False):
flash(_("Metadata successfully updated"), category="success")
if upload_formats: if upload_formats:
resp = {"location": url_for('edit-book.show_edit_book', book_id=book_id)} resp = {"location": url_for('edit-book.show_edit_book', book_id=book_id)}
return Response(json.dumps(resp), mimetype='application/json') return Response(json.dumps(resp), mimetype='application/json')
# if meta is not False \
if edit_error is not True and title_author_error is not True and cover_upload_success is not False:
flash(_("Metadata successfully updated"), category="success")
if "detail_view" in to_save: if "detail_view" in to_save:
return redirect(url_for('web.show_book', book_id=book.id)) return redirect(url_for('web.show_book', book_id=book.id))
else: else:
@ -590,7 +587,7 @@ def merge_metadata(to_save, meta):
for s_field, m_field in [ for s_field, m_field in [
('tags', 'tags'), ('author_name', 'author'), ('series', 'series'), ('tags', 'tags'), ('author_name', 'author'), ('series', 'series'),
('series_index', 'series_id'), ('languages', 'languages'), ('series_index', 'series_id'), ('languages', 'languages'),
('book_title', 'title'), ('description', 'description'),]: ('book_title', 'title'), ('description', 'description'),('format_cover', 'cover')]:
val = getattr(meta, m_field, '') val = getattr(meta, m_field, '')
if val: if val:
to_save[s_field] = val to_save[s_field] = val
@ -1205,26 +1202,26 @@ def edit_cc_data(book_id, book, to_save, cc):
# returns None if no file is uploaded # returns None if no file is uploaded
# returns False if an error occurs, in all other cases the ebook metadata is returned # returns False if an error occurs, in all other cases the ebook metadata is returned
def upload_single_file(requested_files, book, book_id): def upload_book_formats(requested_files, book, book_id, no_cover=True):
# Check and handle Uploaded file # Check and handle Uploaded file
# requested_file = file_request.files.get('btn-upload-format', None) to_save = dict()
# ToDo: Handle multiple files upload_format = False
meta = {}
allowed_extensions = config.config_upload_formats.split(',') allowed_extensions = config.config_upload_formats.split(',')
for requested_file in requested_files: for requested_file in requested_files:
current_filename = requested_file.filename
if config.config_check_extensions and allowed_extensions != ['']: if config.config_check_extensions and allowed_extensions != ['']:
if not validate_mime_type(requested_file, allowed_extensions): if not validate_mime_type(requested_file, allowed_extensions):
flash(_("File type isn't allowed to be uploaded to this server"), category="error") flash(_("File type isn't allowed to be uploaded to this server"), category="error")
continue continue
# return False # return False
# check for empty request # check for empty request
if requested_file.filename != '': if current_filename != '':
if not current_user.role_upload(): if not current_user.role_upload():
flash(_("User has no rights to upload additional file formats"), category="error") flash(_("User has no rights to upload additional file formats"), category="error")
continue continue
# return False # return False
if '.' in requested_file.filename: if '.' in current_filename:
file_ext = requested_file.filename.rsplit('.', 1)[-1].lower() file_ext = current_filename.rsplit('.', 1)[-1].lower()
if file_ext not in allowed_extensions and '' not in allowed_extensions: if file_ext not in allowed_extensions and '' not in allowed_extensions:
flash(_("File extension '%(ext)s' is not allowed to be uploaded to this server", ext=file_ext), flash(_("File extension '%(ext)s' is not allowed to be uploaded to this server", ext=file_ext),
category="error") category="error")
@ -1256,10 +1253,9 @@ def upload_single_file(requested_files, book, book_id):
# return False # return False
file_size = os.path.getsize(saved_filename) file_size = os.path.getsize(saved_filename)
is_format = calibre_db.get_book_format(book_id, file_ext.upper())
# Format entry already exists, no need to update the database # Format entry already exists, no need to update the database
if is_format: if calibre_db.get_book_format(book_id, file_ext.upper()):
log.warning('Book format %s already existing', file_ext.upper()) log.warning('Book format %s already existing', file_ext.upper())
else: else:
try: try:
@ -1279,11 +1275,13 @@ def upload_single_file(requested_files, book, book_id):
link = '<a href="{}">{}</a>'.format(url_for('web.show_book', book_id=book.id), escape(book.title)) link = '<a href="{}">{}</a>'.format(url_for('web.show_book', book_id=book.id), escape(book.title))
upload_text = N_("File format %(ext)s added to %(book)s", ext=file_ext.upper(), book=link) upload_text = N_("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))) WorkerThread.add(current_user.name, TaskUpload(upload_text, escape(book.title)))
meta = uploader.process(
return uploader.process( saved_filename,
saved_filename, *os.path.splitext(requested_file.filename), *os.path.splitext(current_filename),
rar_executable=config.config_rarfile_location) rar_executable=config.config_rarfile_location,
return meta if len(meta) else False no_cover=no_cover)
upload_format |= merge_metadata(to_save, meta)
return to_save if len(to_save) else False, upload_format
def upload_cover(cover_request, book): def upload_cover(cover_request, book):

View File

@ -66,7 +66,7 @@ def get_epub_layout(book, book_data):
return layout[0] return layout[0]
def get_epub_info(tmp_file_path, original_file_name, original_file_extension): def get_epub_info(tmp_file_path, original_file_name, original_file_extension, no_cover_processing):
ns = { ns = {
'n': 'urn:oasis:names:tc:opendocument:xmlns:container', 'n': 'urn:oasis:names:tc:opendocument:xmlns:container',
'pkg': 'http://www.idpf.org/2007/opf', 'pkg': 'http://www.idpf.org/2007/opf',
@ -117,7 +117,10 @@ def get_epub_info(tmp_file_path, original_file_name, original_file_extension):
epub_metadata = parse_epub_series(ns, tree, epub_metadata) epub_metadata = parse_epub_series(ns, tree, epub_metadata)
epub_zip = zipfile.ZipFile(tmp_file_path) epub_zip = zipfile.ZipFile(tmp_file_path)
cover_file = parse_epub_cover(ns, tree, epub_zip, cover_path, tmp_file_path) if not no_cover_processing:
cover_file = parse_epub_cover(ns, tree, epub_zip, cover_path, tmp_file_path)
else:
cover_file = None
identifiers = [] identifiers = []
for node in p.xpath('dc:identifier', namespaces=ns): for node in p.xpath('dc:identifier', namespaces=ns):

View File

@ -77,24 +77,25 @@ except ImportError as e:
use_audio_meta = False use_audio_meta = False
def process(tmp_file_path, original_file_name, original_file_extension, rar_executable): def process(tmp_file_path, original_file_name, original_file_extension, rar_executable, no_cover=False):
meta = default_meta(tmp_file_path, original_file_name, original_file_extension) meta = default_meta(tmp_file_path, original_file_name, original_file_extension)
extension_upper = original_file_extension.upper() extension_upper = original_file_extension.upper()
try: try:
if ".PDF" == extension_upper: if ".PDF" == extension_upper:
meta = pdf_meta(tmp_file_path, original_file_name, original_file_extension) meta = pdf_meta(tmp_file_path, original_file_name, original_file_extension, no_cover)
elif extension_upper in [".KEPUB", ".EPUB"] and use_epub_meta is True: elif extension_upper in [".KEPUB", ".EPUB"] and use_epub_meta is True:
meta = epub.get_epub_info(tmp_file_path, original_file_name, original_file_extension) meta = epub.get_epub_info(tmp_file_path, original_file_name, original_file_extension, no_cover)
elif ".FB2" == extension_upper and use_fb2_meta is True: elif ".FB2" == extension_upper and use_fb2_meta is True:
meta = fb2.get_fb2_info(tmp_file_path, original_file_extension) meta = fb2.get_fb2_info(tmp_file_path, original_file_extension)
elif extension_upper in ['.CBZ', '.CBT', '.CBR', ".CB7"]: elif extension_upper in ['.CBZ', '.CBT', '.CBR', ".CB7"]:
meta = comic.get_comic_info(tmp_file_path, meta = comic.get_comic_info(tmp_file_path,
original_file_name, original_file_name,
original_file_extension, original_file_extension,
rar_executable) rar_executable,
no_cover)
elif extension_upper in [".MP3", ".OGG", ".FLAC", ".WAV", ".AAC", ".AIFF", ".ASF", ".MP4", elif extension_upper in [".MP3", ".OGG", ".FLAC", ".WAV", ".AAC", ".AIFF", ".ASF", ".MP4",
".M4A", ".M4B", ".OGV", ".OPUS"] and use_audio_meta: ".M4A", ".M4B", ".OGV", ".OPUS"] and use_audio_meta:
meta = audio.get_audio_file_info(tmp_file_path, original_file_extension, original_file_name) meta = audio.get_audio_file_info(tmp_file_path, original_file_extension, original_file_name, no_cover)
except Exception as ex: except Exception as ex:
log.warning('cannot parse metadata, using default: %s', ex) log.warning('cannot parse metadata, using default: %s', ex)
@ -168,7 +169,7 @@ def parse_xmp(pdf_file):
} }
def pdf_meta(tmp_file_path, original_file_name, original_file_extension): def pdf_meta(tmp_file_path, original_file_name, original_file_extension, no_cover_processing):
doc_info = None doc_info = None
xmp_info = None xmp_info = None
@ -216,7 +217,7 @@ def pdf_meta(tmp_file_path, original_file_name, original_file_extension):
extension=original_file_extension, extension=original_file_extension,
title=title, title=title,
author=author, author=author,
cover=pdf_preview(tmp_file_path, original_file_name), cover=pdf_preview(tmp_file_path, original_file_name) if not no_cover_processing else None,
description=subject, description=subject,
tags=tags, tags=tags,
series="", series="",
@ -231,7 +232,7 @@ def pdf_preview(tmp_file_path, tmp_dir):
if use_generic_pdf_cover: if use_generic_pdf_cover:
return None return None
try: try:
cover_file_name = os.path.splitext(tmp_file_path)[0] + ".cover.jpg" cover_file_name = os.path.join(os.path.dirname(tmp_file_path), "cover.jpg")
with Image() as img: with Image() as img:
img.options["pdf:use-cropbox"] = "true" img.options["pdf:use-cropbox"] = "true"
img.read(filename=tmp_file_path + '[0]', resolution=150) img.read(filename=tmp_file_path + '[0]', resolution=150)