Code:
#!/usr/bin/env python
# ~*~ coding: utf-8 ~*~
__license__ = 'GPL v3'
__copyright__ = '2022, Ahmed Zaki <azaki00.dev@gmail.com>'
__docformat__ = 'restructuredtext en'
from qt.core import (QApplication, Qt, QWidget, QVBoxLayout, QCheckBox,
QGroupBox, QRadioButton, QAction)
import copy
import types
from functools import partial
from calibre import prints
from calibre.constants import DEBUG
from calibre.gui2 import Dispatcher, error_dialog
from calibre.ptempfile import PersistentTemporaryFile
from calibre.gui2.metadata.bulk_download import Job, download
from calibre.gui2.actions.edit_metadata import EditMetadataAction
from polyglot.builtins import iteritems
from calibre_plugins.action_chains.actions.base import ChainAction
from calibre_plugins.action_chains.common_utils import responsive_wait, responsive_wait_until
def unfinished_job_ids(gui):
return set([job.id for job in gui.job_manager.unfinished_jobs()])
class ModifiedEditMetadataAction(EditMetadataAction):
name = 'Modified Edit Metadata'
action_spec = (_('Modified Edit metadata'), 'edit_input.png', _('Change the title/author/cover etc. of books'), _(''))
action_type = 'current'
action_add_menu = True
def __init__(self, parent, site_customization):
EditMetadataAction.__init__(self, parent, site_customization)
self.do_genesis()
@property
def unique_name(self):
bn = self.__class__.__name__
return 'Interface Action: %s (%s)'%(bn, self.name)
def genesis(self):
pass
def location_selected(self, loc):
pass
def library_changed(self):
pass
def shutting_down(self):
pass
def metadata_downloaded(self, job):
if job.failed:
self.gui.job_exception(job, dialog_title=_('Failed to download metadata'))
return
from calibre.gui2.metadata.bulk_download import get_job_details
(aborted, id_map, tdir, log_file, failed_ids, failed_covers, all_failed,
det_msg, lm_map) = get_job_details(job)
if aborted:
return self.cleanup_bulk_download(tdir)
if all_failed:
num = len(failed_ids | failed_covers)
self.cleanup_bulk_download(tdir)
return error_dialog(self.gui, _('Download failed'), ngettext(
'Failed to download metadata or cover for the selected book.',
'Failed to download metadata or covers for any of the {} books.', num
).format(num), det_msg=det_msg, show=True)
self.gui.status_bar.show_message(_('Metadata download completed'), 3000)
msg = '<p>' + ngettext(
'Finished downloading metadata for the selected book.',
'Finished downloading metadata for <b>{} books</b>.', len(id_map)).format(len(id_map)) + ' ' + \
_('Proceed with updating the metadata in your library?')
show_copy_button = False
checkbox_msg = None
if failed_ids or failed_covers:
show_copy_button = True
num = len(failed_ids.union(failed_covers))
msg += '<p>'+_('Could not download metadata and/or covers for %d of the books. Click'
' "Show details" to see which books.')%num
checkbox_msg = _('Show the &failed books in the main book list '
'after updating metadata')
if getattr(job, 'metadata_and_covers', None) == (False, True):
# Only covers, remove failed cover downloads from id_map
for book_id in failed_covers:
if hasattr(id_map, 'discard'):
id_map.discard(book_id)
payload = (id_map, tdir, log_file, lm_map,
failed_ids.union(failed_covers))
if self.do_review:
QApplication.setOverrideCursor(Qt.ArrowCursor)
try:
self.apply_downloaded_metadata(True, payload, self.restrict_to_failed)
finally:
QApplication.restoreOverrideCursor()
else:
self.apply_downloaded_metadata(False, payload, self.restrict_to_failed)
class ConfigWidget(QWidget):
def __init__(self, plugin_action):
QWidget.__init__(self)
self.plugin_action = plugin_action
self.gui = plugin_action.gui
self.db = self.gui.current_db
self._init_controls()
def _init_controls(self):
l = self.l = QVBoxLayout()
self.setLayout(l)
opt_gb = QGroupBox(_('Options'))
opt_gb_l = QVBoxLayout()
opt_gb.setLayout(opt_gb_l)
l.addWidget(opt_gb)
self.metadata_opt = QRadioButton(_('Download Metadata'))
self.covers_opt = QRadioButton(_('Download Covers'))
self.both_opt = QRadioButton(_('Download Both'))
self.both_opt.setChecked(True)
opt_gb_l.addWidget(self.metadata_opt)
opt_gb_l.addWidget(self.covers_opt)
opt_gb_l.addWidget(self.both_opt)
self.review_chk = QCheckBox(_('Review downloaded metadata before applying them'))
self.wait_chk = QCheckBox(_('Wait for metadata download jobs to finish'))
self.wait_chk.setToolTip(_('Check this if this action in not the last action in the chain.'))
l.addWidget(self.review_chk)
l.addWidget(self.wait_chk)
l.addStretch(1)
self.setMinimumSize(500,300)
def load_settings(self, settings):
if settings:
self.metadata_opt.setChecked(settings.get('download_metadata', False))
self.covers_opt.setChecked(settings.get('download_covers', False))
self.both_opt.setChecked(settings.get('download_both', True))
self.review_chk.setChecked(settings.get('review', True))
self.wait_chk.setChecked(settings.get('wait_jobs', False))
def save_settings(self):
settings = {}
settings['download_metadata'] = self.metadata_opt.isChecked()
settings['download_covers'] = self.covers_opt.isChecked()
settings['download_both'] = self.both_opt.isChecked()
settings['review'] = self.review_chk.isChecked()
settings['wait_jobs'] = self.wait_chk.isChecked()
return settings
class DownloadMetadata(ChainAction):
name = 'Download Metadata'
support_scopes = True
def run(self, gui, settings, chain):
identify = settings.get('download_metadata') or settings.get('download_both', True)
covers = settings.get('download_covers') or settings.get('download_both', True)
wait_jobs = settings.get('wait_jobs', False)
ensure_fields = None
edit_metadata = ModifiedEditMetadataAction(gui, '')
edit_metadata.do_review = settings.get('review', True)
edit_metadata.restrict_to_failed = settings.get('restrict_to_failed', True)
callback = Dispatcher(edit_metadata.metadata_downloaded)
ids = chain.scope().get_book_ids()
if len(ids) == 0:
return error_dialog(gui, _('Cannot download metadata'),
_('No books selected'), show=True)
jobs_before_ids = unfinished_job_ids(gui)
tf = PersistentTemporaryFile('_metadata_bulk.log')
tf.close()
job = Job('metadata bulk download',
ngettext(
'Download metadata for one book',
'Download metadata for {} books', len(ids)).format(len(ids)),
download, (ids, tf.name, gui.current_db, identify, covers,
ensure_fields), {}, callback)
job.metadata_and_covers = (identify, covers)
job.download_debug_log = tf.name
gui.job_manager.run_threaded_job(job)
gui.status_bar.show_message(_('Metadata download started'), 3000)
if wait_jobs:
# wait for jobs spawned by action to kick in
responsive_wait(1)
# save ids of jobs started after running the action
ids_jobs_by_action = unfinished_job_ids(gui).difference(jobs_before_ids)
# wait for jobs to finish
responsive_wait_until(lambda: ids_jobs_by_action.intersection(unfinished_job_ids(gui)) == set())
def validate(self, settings):
#if not settings:
#return (_('Settings Error'), _('You must configure this action before running it'))
return True
def config_widget(self):
return ConfigWidget