264 lines
11 KiB
Python
264 lines
11 KiB
Python
"""Orquestador principal de la migración/respaldo IMAP.
|
|
|
|
Estrategia en dos fases por ejecución:
|
|
|
|
1. **Análisis**: para cada carpeta del servidor A se reproduce su
|
|
estructura en B, se determina desde qué UID hay que buscar (usando el
|
|
``migration-state.json`` de una ejecución anterior) y se leen solo los
|
|
encabezados (``Message-ID``, tamaño, fecha, flags) de los mensajes
|
|
nuevos. Con esto se calcula qué mensajes realmente faltan por copiar,
|
|
sin tener que descargar ni comparar el cuerpo completo de todos los
|
|
mensajes en cada ejecución.
|
|
2. **Copia**: se transfieren uno a uno, mensaje por mensaje, solo los
|
|
mensajes pendientes detectados en la fase anterior.
|
|
|
|
El estado (``last_scanned_a_uid`` por carpeta) solo avanza hasta el punto
|
|
donde todos los mensajes de esa carpeta quedaron resueltos (copiados o ya
|
|
existentes). Si un mensaje falla o la ejecución se interrumpe, el estado no
|
|
avanza más allá de ese punto, así que la siguiente ejecución lo vuelve a
|
|
intentar sin perder mensajes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
|
|
from fingerprint import compute_fingerprint
|
|
from imap_client import ImapClient
|
|
from progress import ProgressTracker, format_bytes
|
|
from state import MigrationState
|
|
|
|
|
|
def translate_folder_path(name: str, src_delim: str, dst_delim: str) -> str:
|
|
if not src_delim or src_delim == dst_delim:
|
|
return name
|
|
parts = [p for p in name.split(src_delim) if p != ""]
|
|
return dst_delim.join(parts)
|
|
|
|
|
|
def _is_noselect(flags: str) -> bool:
|
|
lowered = flags.lower()
|
|
return "\\noselect" in lowered or "\\nonexistent" in lowered
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class PendingItem:
|
|
folder: str
|
|
dst_folder: str
|
|
uid: int
|
|
size: int
|
|
fingerprint: str
|
|
|
|
|
|
class Migrator:
|
|
def __init__(self, config, logger, progress: ProgressTracker | None = None):
|
|
self.config = config
|
|
self.logger = logger
|
|
self.client_a = ImapClient(config.server_a, config.timeout, logger, config.retries)
|
|
self.client_b = ImapClient(config.server_b, config.timeout, logger, config.retries)
|
|
self.state = MigrationState(config.state_file, config.server_a.user)
|
|
self.progress = progress or ProgressTracker()
|
|
self._stop_requested = False
|
|
|
|
def request_stop(self) -> None:
|
|
self._stop_requested = True
|
|
|
|
def run(self) -> int:
|
|
try:
|
|
print("Conectando al servidor origen...")
|
|
self.logger.info("Conectando al servidor origen...")
|
|
self.client_a.connect()
|
|
print("OK\n")
|
|
|
|
print("Conectando al servidor destino...")
|
|
self.logger.info("Conectando al servidor destino...")
|
|
self.client_b.connect()
|
|
print("OK\n")
|
|
except Exception as exc:
|
|
self.logger.error(f"Error de conexión: {exc}")
|
|
print(f"Error de conexión: {exc}")
|
|
return 3
|
|
|
|
exit_code = 0
|
|
try:
|
|
print("Analizando carpetas...")
|
|
self.logger.info("Analizando carpetas...")
|
|
folders_a = self.client_a.list_folders()
|
|
selectable = [f for f in folders_a if not _is_noselect(f.flags)]
|
|
|
|
b_delim = self._mirror_folder_structure(selectable)
|
|
|
|
pending, folder_scan_info, total_exists, total_skipped = self._scan_folders(selectable, b_delim)
|
|
|
|
total_new_messages = len(pending)
|
|
total_new_bytes = sum(item.size for item in pending)
|
|
|
|
print(f"\nCarpetas encontradas: {len(selectable)}")
|
|
print(f"Mensajes en servidor origen: {total_exists}")
|
|
print(f"Mensajes nuevos a copiar: {total_new_messages}")
|
|
print(f"Tamaño estimado a transferir: {format_bytes(total_new_bytes)}\n")
|
|
self.logger.info(f"Carpetas encontradas: {len(selectable)}")
|
|
self.logger.info(f"Mensajes en servidor origen: {total_exists}")
|
|
self.logger.info(f"Mensajes nuevos a copiar: {total_new_messages}, existentes: {total_skipped}")
|
|
|
|
self.progress.set_total_folders(len(selectable))
|
|
self.progress.set_total_messages(total_new_messages)
|
|
self.progress.set_total_bytes(total_new_bytes)
|
|
self.progress.messages_skipped = total_skipped
|
|
self.progress.folders_done = len(selectable)
|
|
self.progress.start()
|
|
|
|
if not self._stop_requested:
|
|
print("Migrando...")
|
|
self._copy_pending(pending, folder_scan_info)
|
|
|
|
self.progress.stop()
|
|
exit_code = 0 if self.progress.messages_errors == 0 else 1
|
|
except Exception as exc:
|
|
self.logger.exception(f"Error inesperado durante la migración: {exc}")
|
|
exit_code = 1
|
|
finally:
|
|
self.state.save()
|
|
self.client_a.close()
|
|
self.client_b.close()
|
|
self.progress.print_summary(self.config.server_a.host, self.config.server_b.host, self.logger)
|
|
|
|
return exit_code
|
|
|
|
# -- fase de estructura de carpetas -----------------------------------
|
|
|
|
def _mirror_folder_structure(self, selectable) -> str:
|
|
folders_b = self.client_b.list_folders()
|
|
b_delim = folders_b[0].delimiter if folders_b else "/"
|
|
created: set[str] = set()
|
|
for folder in selectable:
|
|
dst_path = translate_folder_path(folder.name, folder.delimiter, b_delim)
|
|
self._ensure_folder_path(dst_path, b_delim, created)
|
|
return b_delim
|
|
|
|
def _ensure_folder_path(self, dst_path: str, b_delim: str, created: set[str]) -> None:
|
|
parts = [p for p in dst_path.split(b_delim) if p != ""]
|
|
prefix_parts: list[str] = []
|
|
for part in parts:
|
|
prefix_parts.append(part)
|
|
path = b_delim.join(prefix_parts)
|
|
if path in created:
|
|
continue
|
|
try:
|
|
self.client_b.create_folder(path)
|
|
self.logger.info(f"Carpeta asegurada en destino: {path}")
|
|
except Exception as exc:
|
|
self.logger.warning(f"No se pudo crear/verificar la carpeta {path!r} en destino: {exc}")
|
|
created.add(path)
|
|
|
|
# -- fase de análisis --------------------------------------------------
|
|
|
|
def _scan_folders(self, selectable, b_delim: str):
|
|
pending: list[PendingItem] = []
|
|
folder_scan_info: dict[str, dict] = {}
|
|
total_exists = 0
|
|
total_skipped = 0
|
|
|
|
for folder in selectable:
|
|
if self._stop_requested:
|
|
break
|
|
dst_folder = translate_folder_path(folder.name, folder.delimiter, b_delim)
|
|
try:
|
|
exists = self.client_a.select(folder.name, readonly=True)
|
|
total_exists += exists
|
|
uidvalidity = self.client_a.get_uidvalidity(folder.name)
|
|
except Exception as exc:
|
|
self.logger.error(f"No se pudo analizar la carpeta {folder.name!r}: {exc}")
|
|
continue
|
|
|
|
prev_uidvalidity = self.state.get_uidvalidity(folder.name)
|
|
if prev_uidvalidity is not None and prev_uidvalidity != uidvalidity:
|
|
self.logger.warning(
|
|
f"UIDVALIDITY cambió en {folder.name!r} ({prev_uidvalidity} -> {uidvalidity}); "
|
|
"se reescaneará la carpeta completa (los mensajes ya migrados se detectan por Message-ID)."
|
|
)
|
|
self.state.set_uidvalidity(folder.name, uidvalidity)
|
|
|
|
last_scanned = self.state.get_last_scanned_uid(folder.name)
|
|
since_uid = last_scanned + 1 if last_scanned else None
|
|
|
|
try:
|
|
uids = self.client_a.search_uids(since_uid=since_uid)
|
|
except Exception as exc:
|
|
self.logger.error(f"No se pudo buscar mensajes en {folder.name!r}: {exc}")
|
|
continue
|
|
|
|
max_uid = last_scanned
|
|
if uids:
|
|
try:
|
|
headers = self.client_a.fetch_headers(uids, self.config.header_batch_size)
|
|
except Exception as exc:
|
|
self.logger.error(f"No se pudieron leer encabezados de {folder.name!r}: {exc}")
|
|
folder_scan_info[folder.name] = {"max_uid": max_uid}
|
|
continue
|
|
|
|
for uid in sorted(headers.keys()):
|
|
info = headers[uid]
|
|
max_uid = max(max_uid, uid)
|
|
fp = compute_fingerprint(info.get("message_id"), info.get("size", 0), info.get("internaldate_raw", ""))
|
|
if self.state.has_fingerprint(folder.name, fp):
|
|
total_skipped += 1
|
|
continue
|
|
pending.append(PendingItem(folder.name, dst_folder, uid, info.get("size", 0), fp))
|
|
|
|
folder_scan_info[folder.name] = {"max_uid": max_uid}
|
|
|
|
return pending, folder_scan_info, total_exists, total_skipped
|
|
|
|
# -- fase de copia -------------------------------------------------------
|
|
|
|
def _copy_pending(self, pending: list[PendingItem], folder_scan_info: dict[str, dict]) -> None:
|
|
succeeded: dict[str, set[int]] = {folder: set() for folder in folder_scan_info}
|
|
processed_since_save = 0
|
|
current_folder = None
|
|
|
|
for item in pending:
|
|
if self._stop_requested:
|
|
self.logger.warning("Detención solicitada; guardando estado y saliendo de forma segura.")
|
|
break
|
|
|
|
if item.folder != current_folder:
|
|
current_folder = item.folder
|
|
self.progress.set_current_folder(item.folder)
|
|
|
|
try:
|
|
self.client_a.select(item.folder, readonly=True)
|
|
content, flags_raw, internaldate_raw = self.client_a.fetch_full(item.uid)
|
|
flags = " ".join(f for f in flags_raw.split() if f.lower() != "\\recent")
|
|
b_uid = self.client_b.append(item.dst_folder, content, flags, internaldate_raw)
|
|
self.state.mark_migrated(item.folder, item.fingerprint, item.uid, b_uid, item.size)
|
|
succeeded.setdefault(item.folder, set()).add(item.uid)
|
|
self.progress.message_copied(item.size)
|
|
self.logger.debug(f"Copiado UID {item.uid} ({item.folder} -> {item.dst_folder})")
|
|
except Exception as exc:
|
|
self.progress.message_error()
|
|
self.logger.error(f"Error copiando UID {item.uid} en {item.folder!r}: {exc}")
|
|
|
|
processed_since_save += 1
|
|
if processed_since_save >= 50:
|
|
self._finalize_scan_state(pending, folder_scan_info, succeeded)
|
|
self.state.save()
|
|
processed_since_save = 0
|
|
|
|
self._finalize_scan_state(pending, folder_scan_info, succeeded)
|
|
self.state.save()
|
|
|
|
def _finalize_scan_state(self, pending: list[PendingItem], folder_scan_info: dict[str, dict], succeeded: dict[str, set[int]]) -> None:
|
|
pending_uids_by_folder: dict[str, list[int]] = {}
|
|
for item in pending:
|
|
pending_uids_by_folder.setdefault(item.folder, []).append(item.uid)
|
|
|
|
for folder, info in folder_scan_info.items():
|
|
all_pending = pending_uids_by_folder.get(folder, [])
|
|
ok_uids = succeeded.get(folder, set())
|
|
unresolved = [u for u in all_pending if u not in ok_uids]
|
|
final_uid = (min(unresolved) - 1) if unresolved else info["max_uid"]
|
|
prev = self.state.get_last_scanned_uid(folder)
|
|
if final_uid > prev:
|
|
self.state.set_last_scanned_uid(folder, final_uid)
|