Source code for mail_deduplicate.action

# Copyright Kevin Deldycke <kevin@deldycke.com> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
from __future__ import annotations

from typing import TYPE_CHECKING
import logging

from boltons.dictutils import FrozenDict
from boltons.iterutils import unique
from click_extra.colorize import default_theme as theme

from mail_deduplicate.mail_box import create_box

if TYPE_CHECKING:
    from .deduplicate import Deduplicate

COPY_SELECTED = "copy-selected"
COPY_DISCARDED = "copy-discarded"
MOVE_SELECTED = "move-selected"
MOVE_DISCARDED = "move-discarded"
DELETE_SELECTED = "delete-selected"
DELETE_DISCARDED = "delete-discarded"
"""Define all available action IDs."""


[docs] def copy_mails(dedup: Deduplicate, mails) -> None: """Copy provided ``mails`` to a brand new box or an existing one.""" if not dedup.conf.dry_run: box = create_box( dedup.conf.export, dedup.conf.export_format, dedup.conf.export_append, ) for mail in mails: logging.debug(f"Copying {mail!r} to {dedup.conf.export}...") dedup.stats["mail_copied"] += 1 if dedup.conf.dry_run: logging.warning("DRY RUN: Skip action.") else: box.add(mail) logging.info(f"{mail!r} copied.") logging.debug(f"Close {dedup.conf.export}") if not dedup.conf.dry_run: box.close()
[docs] def move_mails(dedup: Deduplicate, mails) -> None: """Move provided ``mails`` to a brand new box or an existing one.""" if not dedup.conf.dry_run: box = create_box( dedup.conf.export, dedup.conf.export_format, dedup.conf.export_append, ) for mail in mails: logging.debug( f"Move {mail!r} form {mail.source_path} to {dedup.conf.export}..." ) dedup.stats["mail_moved"] += 1 if dedup.conf.dry_run: logging.warning("DRY RUN: Skip action.") else: box.add(mail) dedup.sources[mail.source_path].remove(mail.mail_id) logging.info(f"{mail!r} copied.") logging.debug(f"Close {dedup.conf.export}") if not dedup.conf.dry_run: box.close()
[docs] def delete_mails(dedup: Deduplicate, mails) -> None: """Remove provided ``mails`` in-place, from their original boxes.""" for mail in mails: logging.debug(f"Deleting {mail!r} in-place...") dedup.stats["mail_deleted"] += 1 if dedup.conf.dry_run: logging.warning("DRY RUN: Skip action.") else: dedup.sources[mail.source_path].remove(mail.mail_id) logging.info(f"{mail!r} deleted.")
[docs] def copy_selected(dedup: Deduplicate) -> None: """Copy all selected mails to a brand new box.""" copy_mails(dedup, dedup.selection)
[docs] def copy_discarded(dedup: Deduplicate) -> None: """Copy all discarded mails to a brand new box.""" copy_mails(dedup, dedup.discard)
[docs] def move_selected(dedup: Deduplicate) -> None: """Move all selected mails to a brand new box.""" move_mails(dedup, dedup.selection)
[docs] def move_discarded(dedup: Deduplicate) -> None: """Move all discarded mails to a brand new box.""" move_mails(dedup, dedup.discard)
[docs] def delete_selected(dedup: Deduplicate) -> None: """Remove in-place all selected mails, from their original boxes.""" delete_mails(dedup, dedup.selection)
[docs] def delete_discarded(dedup: Deduplicate) -> None: """Remove in-place all discarded mails, from their original boxes.""" delete_mails(dedup, dedup.discard)
ACTIONS = FrozenDict( { COPY_SELECTED: copy_selected, COPY_DISCARDED: copy_discarded, MOVE_SELECTED: move_selected, MOVE_DISCARDED: move_discarded, DELETE_SELECTED: delete_selected, DELETE_DISCARDED: delete_discarded, }, ) """Map action ID's to their implementation."""
[docs] def perform_action(dedup: Deduplicate) -> None: """Performs the action on selected mail candidates.""" logging.info(f"Perform {theme.choice(dedup.conf.action)} action...") selection_count = len(dedup.selection) if selection_count == 0: logging.warning("No mail selected to perform action on.") return logging.info(f"{selection_count} mails selected for action.") # Check our indexing and selection methods are not flagging candidates # several times. assert len(unique(dedup.selection)) == len(dedup.selection) assert len(dedup.selection) == dedup.stats["mail_selected"] # Hunt down for action implementation. method = ACTIONS.get(dedup.conf.action) if not method: msg = f"{dedup.conf.action} action not implemented yet." raise NotImplementedError(msg) method(dedup)