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
import logging
from enum import Enum
from boltons.iterutils import unique
from click_extra.colorize import default_theme as theme
from .mail_box import create_box
TYPE_CHECKING = False
if TYPE_CHECKING:
from .deduplicate import Deduplicate
[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"]:
# Assert to please the type checker.
assert dedup.conf["export"]
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"]:
# Assert to please the type checker.
assert dedup.conf["export"]
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)
[docs]
class Action(Enum):
"""Define all available action IDs."""
COPY_SELECTED = "copy-selected"
COPY_DISCARDED = "copy-discarded"
MOVE_SELECTED = "move-selected"
MOVE_DISCARDED = "move-discarded"
DELETE_SELECTED = "delete-selected"
DELETE_DISCARDED = "delete-discarded"
def __str__(self) -> str:
return self.value
[docs]
def action_function(self) -> callable:
"""Return the action function associated with this action."""
func_name = self.name.lower()
return globals()[func_name]