This is an action to merge duplicates produced by the Find Duplicates plugin. You can merge the books using one of two options:
- Merging the duplicates while the results of the Find Duplicates PI are still in display in calibre view.
- Alternatively, there is an option in the Find Duplicates plugin to export it results to a json file. You can later use this json file to merge the books, when the results of the Find Duplicate plugin have been cleared.
Note: Duplicates will be merged into the first book in each group. You can control the order of books in duplicate groups using the Find Duplicates advanced mode (group sorting option)
Warning: This action can destroy your library as it merges book irreversibly. You use it at your risk, without any guarantee. You must always backup your library before using the action, and you must examine the results carefully until you are satisfied. If you brick your library without having backed it up, and without having examined the result meticulously, you bear full responsibility. Don't come here to complain or ask for support. You have been warned.
Code:
import json
from functools import partial
from qt.core import (QApplication, Qt, QWidget, QVBoxLayout, QGroupBox,
QRadioButton, QCheckBox)
from calibre import prints
from calibre.constants import DEBUG
from calibre.gui2 import choose_files, error_dialog, question_dialog
from calibre_plugins.action_chains.common_utils import DoubleProgressDialog, truncate
from calibre_plugins.action_chains.actions.base import ChainAction
#from calibre.gui2.actions.edit_metadata import EditMetadataAction
class MergeConfigWidget(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)
groupbox = QGroupBox(_('Duplicate source'))
groupbox_layout = QVBoxLayout()
groupbox.setLayout(groupbox_layout)
l.addWidget(groupbox)
from_plugin_opt = self.from_plugin_opt = QRadioButton(_('From Find Duplicates Results'))
from_file_opt = self.from_file_opt = QRadioButton(_('From Find Duplicates Exported Json'))
groupbox_layout.addWidget(from_plugin_opt)
groupbox_layout.addWidget(from_file_opt)
from_plugin_opt.setChecked(True)
self.confirm_chk = QCheckBox(_('Ask for confirmation before merging'))
self.confirm_chk.setChecked(True)
l.addWidget(self.confirm_chk)
l.addStretch(1)
self.setMinimumSize(300,300)
def load_settings(self, settings):
if settings:
if settings['opt'] == 'from_plugin':
self.from_plugin_opt.setChecked(True)
elif settings['opt'] == 'from_file':
self.from_file_opt.setChecked(True)
self.confirm_chk.setChecked(settings.get('confirm', True))
def save_settings(self):
settings = {}
if self.from_plugin_opt.isChecked():
settings['opt'] = 'from_plugin'
elif self.from_file_opt:
settings['opt'] = 'from_file'
settings['confirm'] = self.confirm_chk.isChecked()
return settings
class MergeDuplicates(ChainAction):
name = 'Merge Duplicates'
def merge_books(self, dest_id, src_ids):
gui = self.plugin_action.gui
db = gui.current_db
em = gui.iactions['Edit Metadata']
em.add_formats(dest_id, em.formats_for_ids([dest_id]+src_ids))
em.merge_metadata(dest_id, src_ids)
#em.delete_books_after_merge(src_ids)
# delete books now and will notify db at the end
db.new_api.remove_books(src_ids)
self.deleted_ids.extend(src_ids)
def notify_db_deleted_books(self):
gui = self.plugin_action.gui
gui.library_view.model().ids_deleted(self.deleted_ids)
def pd_callback(self, db, duplicates, pbar):
entangled_groups = set()
for book_id, group_ids in duplicates['entangled_groups_for_book']:
for group_id in group_ids:
entangled_groups.add(group_id)
pbar.update_overall(len(duplicates['books_for_group']))
self.deleted_ids = []
for group_id, book_ids in duplicates['books_for_group'].items():
if group_id in entangled_groups:
msg = _('Group_id () in entangled groups. skipping'.format(group_id))
if DEBUG:
prints('Action Chains: '+msg)
else:
dest_id = book_ids[0]
src_ids = book_ids[1:]
title = db.new_api.field_for('title', dest_id)
title = truncate(title, 30)
msg = _('Group ({}): merging into: {}'.format(group_id, title))
self.merge_books(dest_id, src_ids)
pbar.update_progress(1, msg)
def run(self, gui, settings, chain):
if settings.get('confirm', True):
message = _('The following action will merge books in your library and data will be permenatly lost. '
'Are you sure you want to proceed?')
if not question_dialog(gui, _('Are you sure?'), message, show_copy_button=False):
return
if settings['opt'] == 'from_file':
filters=[(_('Settings'), ['json'])]
json_file = choose_files(gui, 'Choose duplicates json file',
_('Select duplicates json file'), filters=filters,
select_only_single_file=True)
if not json_file:
return
json_file = json_file[0]
with open(json_file) as f:
duplicates = json.load(f)
if duplicates['library_uuid'] != gui.current_db.library_id:
return error_dialog(gui,
'uuid error',
'Library uuid in duplicates json does not match current library uuid.\n'
'Quitting without merging',
show=True)
elif settings['opt'] == 'from_plugin':
duplicates = {}
find_duplicates_plugin = gui.iactions.get('Find Duplicates')
if not find_duplicates_plugin:
return
duplicate_finder = find_duplicates_plugin.duplicate_finder
# if not hasattr(duplicate_finder, '_groups_for_book_map'):
# return
if not duplicate_finder.has_results():
return
entangled_books = {}
for book_id, groups in duplicate_finder._groups_for_book_map.items():
if len(groups) > 1:
entangled_books[book_id] = list(groups)
duplicates['books_for_group'] = duplicate_finder._books_for_group_map
duplicates['entangled_groups_for_book'] = entangled_books
callback = partial(self.pd_callback, gui.current_db, duplicates)
pd = DoubleProgressDialog(1, callback, gui, window_title=_('Merging ...'))
gui.tags_view.blockSignals(True)
QApplication.setOverrideCursor(Qt.ArrowCursor)
try:
pd.exec_()
pd.thread = None
if pd.error is not None:
return error_dialog(gui, _('Failed'),
pd.error[0], det_msg=pd.error[1],
show=True)
finally:
self.notify_db_deleted_books()
QApplication.restoreOverrideCursor()
gui.tags_view.recount()
def config_widget(self):
return MergeConfigWidget
def default_settings(self):
return {'opt': 'from_plugin'}