Commit inicial
This commit is contained in:
lastdragon
@@ -0,0 +1,159 @@
|
||||
"""Parseo de argumentos de línea de comandos y configuración de la aplicación."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import getpass
|
||||
import os
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
|
||||
DEFAULT_PORT = 993
|
||||
DEFAULT_TIMEOUT = 30
|
||||
DEFAULT_RETRIES = 3
|
||||
DEFAULT_STATE_FILE = "migration-state.json"
|
||||
DEFAULT_LOG_DIR = "logs"
|
||||
DEFAULT_HEADER_BATCH_SIZE = 500
|
||||
|
||||
EXIT_PARAM_ERROR = 2
|
||||
|
||||
|
||||
@dataclass
|
||||
class ServerConfig:
|
||||
label: str
|
||||
host: str
|
||||
port: int
|
||||
user: str
|
||||
password: str
|
||||
domain: str | None = None
|
||||
ip: str | None = None
|
||||
verify_tls: bool = True
|
||||
use_ssl: bool = True
|
||||
|
||||
@property
|
||||
def connect_host(self) -> str:
|
||||
return self.ip or self.host
|
||||
|
||||
@property
|
||||
def tls_hostname(self) -> str:
|
||||
return self.host
|
||||
|
||||
|
||||
@dataclass
|
||||
class AppConfig:
|
||||
server_a: ServerConfig
|
||||
server_b: ServerConfig
|
||||
timeout: int
|
||||
retries: int
|
||||
state_file: str
|
||||
log_dir: str
|
||||
header_batch_size: int
|
||||
verbose: bool
|
||||
|
||||
|
||||
def _resolve_password(cli_value: str | None, env_var: str, prompt_label: str) -> str:
|
||||
if cli_value:
|
||||
return cli_value
|
||||
env_value = os.environ.get(env_var)
|
||||
if env_value:
|
||||
return env_value
|
||||
try:
|
||||
return getpass.getpass(f"{prompt_label}: ")
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
print("\nNo se proporcionó contraseña.", file=sys.stderr)
|
||||
sys.exit(EXIT_PARAM_ERROR)
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="imap_migrate.py",
|
||||
description="Migra o respalda el contenido completo de una cuenta IMAP entre dos servidores.",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="""
|
||||
Ejemplo:
|
||||
python3 imap_migrate.py \\
|
||||
--aserver imap.servidor-a.com --aport 993 \\
|
||||
--auser usuario@dominio.com --apassword 'passwordA' \\
|
||||
--bserver imap.servidor-b.com --bport 993 \\
|
||||
--buser usuario@dominio.com --bpassword 'passwordB'
|
||||
|
||||
Si no se proporcionan las contraseñas, se solicitarán de forma interactiva
|
||||
(o pueden definirse en las variables de entorno IMAP_A_PASSWORD / IMAP_B_PASSWORD).
|
||||
|
||||
Ejecutar el programa varias veces sobre el mismo --state-file continúa una
|
||||
migración incremental sin volver a copiar mensajes ya migrados.
|
||||
""",
|
||||
)
|
||||
|
||||
group_a = parser.add_argument_group("Servidor origen (A)")
|
||||
group_a.add_argument("--aserver", required=True, help="Host del servidor IMAP origen")
|
||||
group_a.add_argument("--aport", type=int, default=DEFAULT_PORT, help=f"Puerto del servidor origen (por defecto {DEFAULT_PORT})")
|
||||
group_a.add_argument("--auser", required=True, help="Usuario/cuenta en el servidor origen")
|
||||
group_a.add_argument("--apassword", default=None, help="Password del servidor origen (si se omite, se solicita de forma interactiva)")
|
||||
group_a.add_argument("--adomain", default=None, help="Dominio lógico de la cuenta origen (informativo / uso futuro multi-cuenta)")
|
||||
group_a.add_argument("--aip", default=None, help="IP a la que conectar en el servidor origen, usando --aserver como hostname TLS")
|
||||
group_a.add_argument("--a-no-verify-tls", action="store_true", help="Deshabilita verificación de certificado TLS en el servidor origen (inseguro)")
|
||||
group_a.add_argument("--a-no-ssl", action="store_true", help="Conecta sin SSL/TLS al servidor origen (inseguro, no recomendado)")
|
||||
|
||||
group_b = parser.add_argument_group("Servidor destino (B)")
|
||||
group_b.add_argument("--bserver", required=True, help="Host del servidor IMAP destino")
|
||||
group_b.add_argument("--bport", type=int, default=DEFAULT_PORT, help=f"Puerto del servidor destino (por defecto {DEFAULT_PORT})")
|
||||
group_b.add_argument("--buser", required=True, help="Usuario/cuenta en el servidor destino")
|
||||
group_b.add_argument("--bpassword", default=None, help="Password del servidor destino (si se omite, se solicita de forma interactiva)")
|
||||
group_b.add_argument("--bdomain", default=None, help="Dominio lógico de la cuenta destino (informativo / uso futuro multi-cuenta)")
|
||||
group_b.add_argument("--bip", default=None, help="IP a la que conectar en el servidor destino, usando --bserver como hostname TLS")
|
||||
group_b.add_argument("--b-no-verify-tls", action="store_true", help="Deshabilita verificación de certificado TLS en el servidor destino (inseguro)")
|
||||
group_b.add_argument("--b-no-ssl", action="store_true", help="Conecta sin SSL/TLS al servidor destino (inseguro, no recomendado)")
|
||||
|
||||
parser.add_argument("--no-verify-tls", action="store_true", help="Deshabilita verificación de certificado TLS en AMBOS servidores (inseguro)")
|
||||
|
||||
parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT, help=f"Timeout de conexión en segundos (por defecto {DEFAULT_TIMEOUT})")
|
||||
parser.add_argument("--retries", type=int, default=DEFAULT_RETRIES, help=f"Número de reintentos ante fallos de conexión (por defecto {DEFAULT_RETRIES})")
|
||||
parser.add_argument("--state-file", default=DEFAULT_STATE_FILE, help=f"Archivo de estado de la migración (por defecto {DEFAULT_STATE_FILE})")
|
||||
parser.add_argument("--log-dir", default=DEFAULT_LOG_DIR, help=f"Directorio de logs (por defecto {DEFAULT_LOG_DIR})")
|
||||
parser.add_argument("--header-batch-size", type=int, default=DEFAULT_HEADER_BATCH_SIZE, help="Cantidad de mensajes a analizar por lote (por defecto %(default)s)")
|
||||
parser.add_argument("--verbose", action="store_true", help="Muestra en consola los mismos detalles que se registran en el log")
|
||||
|
||||
return parser
|
||||
|
||||
|
||||
def parse_args(argv: list[str] | None = None) -> AppConfig:
|
||||
parser = build_parser()
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
a_password = _resolve_password(args.apassword, "IMAP_A_PASSWORD", "Password servidor A")
|
||||
b_password = _resolve_password(args.bpassword, "IMAP_B_PASSWORD", "Password servidor B")
|
||||
|
||||
server_a = ServerConfig(
|
||||
label="A",
|
||||
host=args.aserver,
|
||||
port=args.aport,
|
||||
user=args.auser,
|
||||
password=a_password,
|
||||
domain=args.adomain,
|
||||
ip=args.aip,
|
||||
verify_tls=not (args.no_verify_tls or args.a_no_verify_tls),
|
||||
use_ssl=not args.a_no_ssl,
|
||||
)
|
||||
server_b = ServerConfig(
|
||||
label="B",
|
||||
host=args.bserver,
|
||||
port=args.bport,
|
||||
user=args.buser,
|
||||
password=b_password,
|
||||
domain=args.bdomain,
|
||||
ip=args.bip,
|
||||
verify_tls=not (args.no_verify_tls or args.b_no_verify_tls),
|
||||
use_ssl=not args.b_no_ssl,
|
||||
)
|
||||
|
||||
return AppConfig(
|
||||
server_a=server_a,
|
||||
server_b=server_b,
|
||||
timeout=args.timeout,
|
||||
retries=args.retries,
|
||||
state_file=args.state_file,
|
||||
log_dir=args.log_dir,
|
||||
header_batch_size=args.header_batch_size,
|
||||
verbose=args.verbose,
|
||||
)
|
||||
Reference in New Issue
Block a user