gian/MailMigration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added attachment extraction and statistics features

Browse Source
This commit is contained in:
bwaide 2025-02-10 17:46:11 +01:00
parent fcba8b8062
commit de58238025
8 changed files with 845 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@ -1,3 +1,7 @@
.env
__pycache__/
.env_test
.DS_Store
migration.log
statistics.json
venv/

380
README.md
View File

@ -1,16 +1,368 @@
# Installation
pip install tqdm
Copy .env_template to .env and fill out your credentials for source and destination IMAP server
Check config.json for the right mapping.
# Run
The following command runs a simulation, logging into both servers, checking the number of emails and total data volume to be migrated. Highly recommended to run the script first in simulation mode.
python migrate.py --simulate
If everything looks fine, run:
python migrate.py
Be aware: for larger mailboxes this might run hours.
# IMAP Email Migration Script
A Python-based email migration tool designed to migrate emails from a Gmail inbox (or other IMAP sources) to another IMAP mailbox. The script is highly configurable and extensible, featuring folder mapping, attachment extraction (with external storage and link replacement), checkpointing for restartability, and detailed logging and statistics.
---
## Table of Contents
- [Overview](#overview)
- [Features](#features)
- [Prerequisites](#prerequisites)
- [Installation](#installation)
- [Configuration](#configuration)
- [.env File](#env-file)
- [config.json](#configjson)
- [Usage](#usage)
- [Simulation Mode](#simulation-mode)
- [Live Mode](#live-mode)
- [Folder Mapping & Label Translation](#folder-mapping--label-translation)
- [Attachment Extraction](#attachment-extraction)
- [Checkpointing and Resuming](#checkpointing-and-resuming)
- [Logging and Statistics](#logging-and-statistics)
- [Error Handling and Reconnection](#error-handling-and-reconnection)
- [Extensibility](#extensibility)
- [License](#license)
---
## Overview
This migration script uses IMAP to fetch emails from a source (typically Gmail) and appends them into a destination mailbox. It preserves metadata such as flags, internal dates, and folder structure. Additionally, the script includes features for:
- **Folder mapping:** Converting Gmail’s label-based folder structure to a traditional IMAP folder structure.
- **Attachment extraction:** Optionally extracting attachments that match certain criteria (file type and size), storing them externally, and replacing the attachments in the migrated email with file links.
- **Checkpointing:** Saving progress to resume migration if the process is interrupted.
- **Statistics collection:** Recording various metrics (e.g., number of emails per sender or folder).
- **Robust error handling and reconnection:** Automatically reconnecting if the IMAP connection is lost.
- **Detailed logging:** Logging output to both console (INFO level) and a file (DEBUG level).
---
## Features
**Folder Mapping:**
- Uses a mapping function to translate Gmail labels to destination folder names.
- Prepares missing destination folders automatically based on the source structure.
**Attachment Extraction:**
- Extracts attachments of specific file types and sizes.
- Saves extracted attachments in a structured folder hierarchy (e.g., organized by month based on email sent date).
- Replaces attachments in the migrated email with a file:// link.
**Checkpointing & Resume:**
- Automatically saves progress (including counters and last processed email index) to a JSON file.
- Resumes migration from the last checkpoint if the script is interrupted.
**Statistics Collection (Expandable):**
- Uses helper functions (e.g. `add_statistic()`) to record information such as file extensions, source folders, and senders.
- Stores statistics in a JSON file for later analysis.
**Robust Logging:**
- Logs are written to a file and INFO-level messages are also printed to the console.
- Logs include details about folder mapping, email processing, reconnections, and errors.
**Error Handling & Reconnection:**
- Automatically reconnects on connection failures using exponential backoff.
- Handles CTRL+C gracefully, performing cleanup before exit.
---
## Prerequisites
**Python 3.7+**
- The following Python modules (most are part of the standard library):
- `imaplib`
- `email`
- `json`
- `os`
- `time`
- `base64`
- `hashlib`
- `argparse`
- `shlex`
- `logging`
**Third-party modules:**
- `python-dotenv` (for loading the `.env` file)
- `tqdm` (for progress bars)
- `python-dateutil` (for date parsing)
Install third-party modules via pip:
```bash
pip install python-dotenv tqdm python-dateutil
```
Alternative
```bash
pip install -r requirements.txt
```
# Installation
**Clone the Repository:**
```bash
git clone https://github.com/bwaide/MailMigration.git
cd mailmigration
```
**Set Up a Virtual Environment (Recommended):**
```bash
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
```
**Install Dependencies:**
```bash
pip install -r requirements.txt
```
## Configuration
The migration script is configured via environment variables (loaded from a `.env` file) and a JSON configuration file (`config.json`).
### GMail Account
Google doesn't allow login via IMAP with your user password for security reasons. You therefore have to create an Application password to be used just for the migration. It is strongly advised to delete the app password after the migration is done.
You can create the Application password here:
https://myaccount.google.com/apppasswords
_Note:_ You have to have Two-Factor-Authentication (2FA) turned on to be able to create application passwords.
### .env File
Create a `.env` file (or rename the provided `.env_template` to `.env`) with the following variables:
```ini
SOURCE_IMAP_SERVER=imap.gmail.com
SOURCE_EMAIL=yourgmail@example.com
SOURCE_PASSWORD=yourpassword
DEST_IMAP_SERVER=imap.destination.com
DEST_EMAIL=yourdest@example.com
DEST_PASSWORD=yourdestpassword
```
### config.json
The configuration file contains global settings and module-specific settings. For example:
```json
`{
"global": {
"folder_mapping": {
"Inbox": "INBOX",
"Gesendet": "Gesendete Objekte",
"Sent Mail": "Gesendete Objekte",
"Sent": "Gesendete Objekte",
"Alle Nachrichten": "Archive",
"All Mail": "Archive",
"Important": "INBOX",
"Markiert": "INBOX",
"Starred": "INBOX",
"Wichtig": "INBOX",
"Papierkorb": null,
"Trash": null,
"Spam": null,
"Entwürfe": null,
"Drafts": null,
"Draft": null,
"Jobsuche": null,
"INBOX/Haus": "INBOX.Haus",
"INBOX/Haus/BS Energy": "INBOX.Haus.BS Energy",
"INBOX/Haus/Montana (Gas)": "INBOX.Haus.Montana (Gas)",
"INBOX/Haus/Vermietung": "INBOX.Haus.Vermietung",
"INBOX/Haus/eprimo (Strom)": "INBOX.Haus.eprimo (Strom)",
"INBOX/Hochzeit": "INBOX.Hochzeit",
"INBOX/NAK": "INBOX.NAK",
"INBOX/Shopping": "INBOX.Shopping"
},
"archive_folder": "Archive",
"folder_prefix": "INBOX.",
"log_file": "migration.log",
"root_folder": "[Gmail]/All Mail",
"debug_delay": 0.0,
"labels_as_flagged": [
"Important",
"[Gmail]/Important",
"Starred",
"[Gmail]/Starred"
]
},
"extract_attachments": {
"enabled": true,
"attachment_whitelist": [".pdf", ".zip", ".docx", ".xlsx"],
"max_attachment_size": 10240,
"external_storage_path": "/Users/bwaide/Documents/Development/GoogleMigration/downloads",
"storage_url": "file:///Users/bwaide/Documents/Development/GoogleMigration/downloads"
}
}`
```
_Note:_ You can expand this configuration with additional modules or plugins as needed.
----------
## Usage
### Command-Line Arguments
Run the script from the command line. It supports a simulation mode:
- **Simulation Mode:**
Runs the migration without actually transferring emails. Check for warnings and errors in `migration.log` afterwards. Use the output in `statistics.json` to adjust the configuration, especially the folder mapping and settings for attachment extraction.
```bash
python migrate.py --simulate`
```
- **Live Mode:**
Runs the migration for real. You will be prompted to confirm before actual migration begins.
```bash
python migrate.py
```
_Note:_ For larger mailboxes the migration can take hours. From my experience running this script on a laptop performs at roughly 2 - 4 mails/second. For mailboxes with 10,000 mails this means 1 hour.
----------
## Folder Mapping & Label Translation
The script converts Gmail’s label-based folder structure into a destination folder structure using a centralized mapping function.
- **Mapping Rules:**
1. If the input contains `[Gmail]/Inbox` (or cleans to "INBOX"), the destination is `"INBOX"`.
2. If a label exists in the global `folder_mapping` (from `config.json`), that value is used.
3. Otherwise, the script chooses the most specific (deepest) label and prepends the global `folder_prefix`.
4. If no label is available, it falls back to `archive_folder`.
- **Implementation:**
The function `map_labels_to_destination(gmail_labels)` is used throughout the script for both folder creation and email migration. This ensures consistency.
----------
## Attachment Extraction
If enabled in the configuration, the script will:
1. **Examine each email for attachments.**
2. **For attachments that meet criteria (file extension, size):**
- Save them to an external location.
- Organize them into month-based subfolders (e.g., `"2023_10"`).
- Replace the attachment in the email with a link starting with `file://` pointing to the saved file.
3. **For attachments that don’t meet the criteria:**
They are kept as part of the email.
----------
## Checkpointing and Resuming
- **Checkpoint File:**
The script saves its progress (last processed email index, total size, and skipped count) to a file (default: `migration_checkpoint.json`).
- **Resumption:**
On start, the script loads the checkpoint and resumes from the last processed email.
- **Automatic Cleanup:**
The checkpoint file is automatically deleted if the migration completes successfully.
----------
## Logging and Statistics
- **Logging:**
Logging is configured to write DEBUG-level messages to a file and INFO-level messages to the console.
Log messages include details about folder mapping, email processing, errors, and reconnections.
- **Statistics (Expandable):**
The script can be extended to collect statistics (for example, counts of emails per sender, file extensions, etc.) via helper functions like `add_statistic(category, key, amount)`.
----------
## Error Handling and Reconnection
- The script handles `KeyboardInterrupt` (CTRL+C) gracefully and performs cleanup.
- If an IMAP connection is lost (e.g., due to a system error), the script automatically attempts to reconnect using exponential backoff.
- In case of an unexpected crash, the checkpoint file allows the migration to resume without duplicating emails.
----------
## Extensibility
The configuration system is modular, with a nested JSON structure supporting multiple modules (global settings and module-specific settings such as for attachment extraction).
You can extend the system by adding new sections to the JSON config and writing corresponding helper functions.
----------
## License
MIT License
See https://github.com/bwaide/MailMigration/blob/main/LICENSE for details.
----------
## Contact
Björn Waide
Contact me at mailto:bjoern.waide@net-positive-ventures.com

182
attachments.py Normal file
View File

@ -0,0 +1,182 @@
from email import message_from_bytes
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import os
import hashlib
from email.utils import parsedate_to_datetime
from pathlib import Path
from datetime import datetime
from config import get_config
from stats import add_statistic
# Global variable: Define which attachment types and sizes should be extracted
EXTRACT_ATTACHMENTS = get_config("extract_attachments", "enabled", False)
ATTACHMENT_WHITELIST = get_config("extract_attachments", "attachment_whitelist", [".pdf", ".zip", ".docx", ".xlsx"]) # Allowed file types
MIN_ATTACHMENT_SIZE = get_config("extract_attachments", "min_attachment_size", 0 * 1024) # Min size for attachments to have to be downloaded
MAX_ATTACHMENT_SIZE = get_config("extract_attachments", "max_attachment_size", 100 * 1024 * 1024) # Max size per attachment in Bytes
download_folder = os.path.expanduser("~")+"/Downloads/"
EXTERNAL_STORAGE_PATH = get_config("extract_attachments", "external_storage_path", download_folder) # Path to store extracted files
def normalize_filename(filename):
"""
Normalizes the filename by removing any trailing query string or extra characters
from the file extension. For example:
"document.pdf?=" becomes "document.pdf"
"""
if '?' in filename:
filename = filename.split('?')[0]
return filename.strip()
def get_normalized_extension(filename):
"""
Returns the normalized file extension (in lower-case) for the given filename.
"""
filename = normalize_filename(filename)
return os.path.splitext(filename)[1].lower()
def should_extract_attachment(part):
"""Decide whether an attachment should be extracted based on type and size."""
filename = part.get_filename()
if not filename:
return False
# Normalize filename and extension.
filename = normalize_filename(filename)
file_ext = get_normalized_extension(filename)
file_size = len(part.get_payload(decode=True)) # File size in bytes
add_statistic("attachment_types", file_ext)
return file_ext in ATTACHMENT_WHITELIST and file_size < MAX_ATTACHMENT_SIZE and file_size > MIN_ATTACHMENT_SIZE
def save_attachment(part, email_date_str):
"""
Save an extracted attachment to disk in a month-based folder (e.g., "2023_10")
and return a file:// URL pointing to the stored file.
Args:
part: The email attachment part.
email_date_str: The email's sent date as a string
(e.g., "Mon, 09 Oct 2023 12:34:56 -0400" or '"10-Feb-2025 07:56:34 +0100"')
Returns:
A file:// URL to the stored attachment, or None if filename is missing.
"""
filename = part.get_filename()
if not filename:
return None
# Normalize the filename to remove trailing query strings etc.
filename = normalize_filename(filename)
# Strip extra quotes from the date string, if present.
date_str = email_date_str.strip('"')
# Parse the email date string into a datetime object.
try:
email_date = parsedate_to_datetime(date_str)
except Exception as e:
try:
# Fallback: try using datetime.strptime with the appropriate format.
email_date = datetime.strptime(date_str, "%d-%b-%Y %H:%M:%S %z")
except Exception as e2:
print(f"Date parsing failed for: {date_str}")
print(e2)
email_date = datetime.now()
# Create a folder name based on the email's date (e.g., "2023_10")
month_folder = email_date.strftime("%Y_%m")
storage_dir = os.path.join(EXTERNAL_STORAGE_PATH, month_folder)
os.makedirs(storage_dir, exist_ok=True)
# Generate a unique filename using a hash prefix to avoid collisions.
hash_prefix = hashlib.md5(filename.encode()).hexdigest()[:8]
storage_filename = f"{hash_prefix}_{filename}"
storage_path = os.path.join(storage_dir, storage_filename)
# Write the attachment data to disk.
file_data = part.get_payload(decode=True)
with open(storage_path, "wb") as f:
f.write(file_data)
# Create a file:// URL pointing to the stored file using pathlib.
file_url = Path(storage_path).as_uri()
return file_url
def extract_and_replace_attachments(raw_msg, email_date_str):
"""
Extracts attachments meeting the criteria and replaces them with a link.
Reassembles the email so that the body appears only once.
Args:
raw_msg (bytes): Raw email message in bytes.
email_date_str: Sent date of the message (used for organizing storage).
Returns:
bytes: The modified email message (as bytes) with attachments replaced by links.
"""
# Parse the raw message into an email object
email_msg = message_from_bytes(raw_msg)
# If the message isn't multipart, return it unchanged
if not email_msg.is_multipart():
return raw_msg
# Create a new email container (multipart/mixed)
new_email = MIMEMultipart("mixed")
# Copy key headers from the original message
for header in ["Subject", "From", "To", "Date"]:
if email_msg[header]:
new_email[header] = email_msg[header]
# Initialize lists to store parts
body_parts = [] # For text parts (the email body)
keep_parts = [] # For attachments that should not be extracted
attachment_links = [] # For links to extracted attachments
# Walk through all parts (leaf nodes) of the email
for part in email_msg.walk():
# Skip container parts
if part.is_multipart():
continue
# Check if the part has a filename (i.e. it's an attachment)
filename = part.get_filename()
if filename:
# This part is an attachment.
if should_extract_attachment(part):
link = save_attachment(part, email_date_str)
if link:
attachment_links.append(link)
else:
# Keep attachments that don't meet extraction criteria.
keep_parts.append(part)
else:
# If there is no filename, assume it's a body (text) part.
body_parts.append(part)
# Choose a single body part: Prefer "text/plain" if available.
selected_body = None
for part in body_parts:
if part.get_content_type() == "text/plain":
selected_body = part
break
if not selected_body and body_parts:
selected_body = body_parts[0]
# Assemble the new email.
if selected_body:
new_email.attach(selected_body)
# Attach any attachments that should be kept intact.
for part in keep_parts:
new_email.attach(part)
# If any attachments were extracted, add a summary text part with links.
if attachment_links:
links_text = "\n\n[Attachments extracted and stored separately:]\n" + "\n".join(attachment_links)
new_email.attach(MIMEText(links_text, "plain"))
return new_email.as_bytes()

49
config.json
View File

@ -1,4 +1,11 @@
{
"general": {
"log_file": "migration.log",
"debug_delay": 0.0,
"statistics_file": "statistics.json",
"reconnect_interval": 500
},
"mapping": {
"folder_mapping": {
"Inbox": "INBOX",
"Gesendet": "Gesendete Objekte",
@ -20,13 +27,49 @@
},
"archive_folder": "Archive",
"folder_prefix": "INBOX.",
"log_file": "migration.log",
"root_folder": "[Gmail]/Alle Nachrichten",
"debug_delay": 0.0,
"root_folder": "[Gmail]/All Mail",
"labels_as_flagged": [
"Important",
"[Gmail]/Important",
"Starred",
"[Gmail]/Starred"
]
},
"extract_attachments": {
"enabled": true,
"attachment_whitelist": [
".pdf",
".jpg",
".png",
".doc",
".jpeg",
".docx",
".zip",
".wav",
".xls",
".csv",
".txt",
".xlsx",
".pps",
".ppt",
".pptx",
".rtf",
".mp3",
".mov",
".m4a",
".mpg",
".wmv",
".tiff",
".tif",
".swf",
".mpeg",
".odt",
".mobi",
".docm",
".pages"
],
"min_attachment_size": 0,
"max_attachment_size": 104857600,
"external_storage_path": "/Users/bwaide/Documents/Email"
}
}

29
config.py Normal file
View File

@ -0,0 +1,29 @@
import json
import logging
def load_config(config_path="config.json"):
with open(config_path, "r") as file:
return json.load(file)
my_config = load_config()
def get_config(section, parameter, default):
return my_config.get(section, {}).get(parameter, default)
# -------------------------------
# Logging Setup
# -------------------------------
log_file = get_config("general", "log_file", "migration.log") # see below for config loading
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(logging.DEBUG)
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S")
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)

160
migrate.py
View File

@ -6,36 +6,16 @@ import base64
import time
import os
import json
import argparse
from dotenv import load_dotenv
from email import message_from_bytes
from tqdm import tqdm # Progress bar
from tqdm import tqdm
import shlex
import logging
def load_config(config_path="config.json"):
with open(config_path, "r") as file:
return json.load(file)
from config import logger
from config import get_config
config = load_config()
from stats import add_statistic, save_statistics_file, format_statistics, load_statistics_file
# -------------------------------
# Logging Setup (as before)
# -------------------------------
log_file = config.get("log_file", "migration.log") # see below for config loading
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(logging.DEBUG)
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S")
file_handler.setFormatter(formatter)
#console_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
import attachments
# -------------------------------
# Configuration Loading
@ -53,18 +33,20 @@ DEST_PASSWORD = os.getenv("DEST_PASSWORD")
CHECKPOINT_FILE = "migration_checkpoint.json"
print(SOURCE_EMAIL)
# Global settings (used across the script)
DEBUG_DELAY = get_config("general", "debug_delay", 0)
RECONNECT_INTERVAL = get_config("general", "reconnect_interval", 500)
STATISTICS_FILE = get_config("general", "statistics_file", "statistics.json")
# These keys come from the config file
FOLDER_MAPPING = config["folder_mapping"]
DEFAULT_ARCHIVE_FOLDER = config["archive_folder"]
FOLDER_PREFIX = config["folder_prefix"]
ROOT_FOLDER = config["root_folder"]
DEBUG_DELAY = config["debug_delay"]
FOLDER_MAPPING = get_config("mapping", "folder_mapping", {})
DEFAULT_ARCHIVE_FOLDER = get_config("mapping", "archive_folder", "Archive")
FOLDER_PREFIX = get_config("mapping", "folder_prefix", "INBOX.")
ROOT_FOLDER = get_config("mapping", "root_folder", "[Gmail]/All Mail")
# Global variable defining labels that should be treated as "\Flagged" in IMAP
LABELS_AS_FLAGGED = config["labels_as_flagged"]
LABELS_AS_FLAGGED = get_config("mapping", "labels_as_flagged", ["Important", "[Gmail]/Important", "Starred", "[Gmail]/Starred"])
def load_checkpoint():
"""Loads the last processed email index and counters from a checkpoint file."""
@ -211,13 +193,13 @@ def reconnect_imap(server, email, password, max_retries=5):
attempt = 0
while attempt < max_retries:
try:
logger.info(f"🔄 Attempting to reconnect to {server} (Attempt {attempt + 1}/{max_retries})...")
logger.debug(f"🔄 Attempting to reconnect to {server} (Attempt {attempt + 1}/{max_retries})...")
conn = imaplib.IMAP4_SSL(server)
conn.login(email, password)
logger.info(f"Reconnected successfully to {server}")
logger.debug(f"Reconnected successfully to {server}")
return conn
except imaplib.IMAP4.abort as e:
logger.warning(f"Reconnect attempt {attempt + 1} failed: {e}")
logger.debug(f"Reconnect attempt {attempt + 1} failed: {e}")
time.sleep(2 ** attempt) # Exponential backoff
attempt += 1
logger.error(f"Could not reconnect to {server} after {max_retries} attempts. Exiting.")
@ -283,7 +265,7 @@ def convertFlags(flags):
formatted_flags = " ".join(flags)
return formatted_flags
def extract_gmail_labels(msg_data):
def extract_gmail_labels_(msg_data):
labels = []
for response_part in msg_data:
if isinstance(response_part, tuple):
@ -295,14 +277,66 @@ def extract_gmail_labels(msg_data):
break
return labels
import re
import shlex
def extract_gmail_labels(msg_data):
"""
Extracts Gmail labels from the email's metadata (X-GM-LABELS).
Returns a list of correctly parsed labels while handling different formatting issues.
"""
labels = []
for response_part in msg_data:
if isinstance(response_part, tuple):
header = response_part[0].decode(errors="ignore")
# Updated regex to match labels inside X-GM-LABELS
match = re.search(r'X-GM-LABELS \((.+?)\)\sRFC822', header)
if match:
raw_labels = match.group(1).strip()
# Handle cases where the labels might be quoted or unquoted
try:
labels = shlex.split(raw_labels) # Properly split labels
except ValueError as e:
logger.error(f"Error parsing labels: {e}. Raw labels: {raw_labels}")
labels = [] # Fallback to empty list
break # Stop processing after the first match
return labels
def collect_sender_statistic(raw_msg):
from email import message_from_bytes
import email.utils
raw_sender = ""
sender_email = "Unkown"
try:
email_msg = message_from_bytes(raw_msg)
raw_sender = str(email_msg.get("From", "Unknown"))
_, sender_email = email.utils.parseaddr(raw_sender)
except Exception as e:
logger.debug(f"WARNING: Extracting sender information failed: '{raw_sender}'")
finally:
add_statistic("sender", sender_email, 1)
# -------------------------------
# Centralized Migration Function
# -------------------------------
def migrate_all(source_conn, dest_conn, simulation=False):
source_folder = f'"{ROOT_FOLDER}"'
success = False
logger.info(f"\n🔍 Scanning {source_folder}...")
# In case of an early interruption, rebuild from last known checkpoint
current_email, total_size_mb, skipped_emails = load_checkpoint()
if (current_email > 0):
# At the beginning of your main function:
load_statistics_file(STATISTICS_FILE)
try:
status, _ = source_conn.select(source_folder, readonly=True)
@ -322,7 +356,7 @@ def migrate_all(source_conn, dest_conn, simulation=False):
logger.info(f"🔄 Resuming from email {current_email + 1}...")
# Initialize tqdm progress bar
progress_bar = tqdm(total=total_emails, desc="Processing emails", unit="email")
progress_bar = tqdm(total=total_emails, desc="Processing emails", unit="email", initial=current_email)
while current_email < len(msg_nums):
num = msg_nums[current_email]
@ -366,6 +400,12 @@ def migrate_all(source_conn, dest_conn, simulation=False):
formatted_internal_date = convertDate(internal_date)
formatted_flags = convertFlags(flags)
collect_sender_statistic(raw_msg)
# Process attachments (if enabled)
if attachments.EXTRACT_ATTACHMENTS:
raw_msg = attachments.extract_and_replace_attachments(raw_msg, formatted_internal_date)
destination_folder = map_labels_to_destination(labels)
if (destination_folder is None):
@ -388,10 +428,16 @@ def migrate_all(source_conn, dest_conn, simulation=False):
progress_bar.update(1)
# 🛑 **Save checkpoint every 100 emails**
# **Save checkpoint every 100 emails**
if current_email % 100 == 0:
save_checkpoint(current_email, total_size_mb, skipped_emails)
# --- Force a reconnection every 500 emails ---
if current_email > 0 and current_email % RECONNECT_INTERVAL == 0:
logger.debug("Forcing periodic reconnection of IMAP connections...")
reconnect(source_folder)
# ---------------------------------------------------------
# **Reduce IMAP request rate** (prevent timeouts/rate limiting)
time.sleep(DEBUG_DELAY) # Adjust delay if needed
@ -399,25 +445,37 @@ def migrate_all(source_conn, dest_conn, simulation=False):
except imaplib.IMAP4.abort as e:
logger.error(f"IMAP connection lost during migration (mail #{num}). Error: {e}")
source_conn = reconnect_imap(SOURCE_IMAP_SERVER, SOURCE_EMAIL, SOURCE_PASSWORD)
dest_conn = reconnect_imap(DEST_IMAP_SERVER, DEST_EMAIL, DEST_PASSWORD)
# Re-select the folder after reconnecting
source_conn.select(source_folder, readonly=True)
reconnect(source_folder)
progress_bar.close()
# Delete checkpoint on successful completion
delete_checkpoint()
total_size_mb = round(total_size_mb / (1024 * 1024), 2)
success = True
except KeyboardInterrupt:
logger.warning("\nCTRL+C detected! Exiting gracefully...")
except Exception as e: # Catch any unexpected crash
logger.error(f"🚨 Unexpected error: {e}. Saving checkpoint before exiting.")
save_checkpoint(current_email, total_size_mb, skipped_emails)
raise # Re-raise the exception so it can be handled by the main script
finally:
save_statistics_file(STATISTICS_FILE)
save_checkpoint(current_email, total_size_mb, skipped_emails)
# Delete checkpoint on successful completion
if (success):
delete_checkpoint()
return total_emails, total_size_mb, skipped_emails
def reconnect(source_folder):
reconnect_imap(DEST_IMAP_SERVER, DEST_EMAIL, DEST_PASSWORD)
source_conn = reconnect_imap(SOURCE_IMAP_SERVER, SOURCE_EMAIL, SOURCE_PASSWORD)
# Re-select the source folder after reconnecting.
source_conn.select(source_folder, readonly=True)
# -------------------------------
# Now update your get_folder_mapping_info to use the consolidated function.
def get_folder_mapping_info(source_conn, dest_conn):
@ -437,6 +495,7 @@ def get_folder_mapping_info(source_conn, dest_conn):
mapping_info = {}
for src in source_folders:
add_statistic("source_folders", src)
# Get the destination folder using the consolidated function.
dest = map_labels_to_destination([src])
# Skip folders that explicitly map to null.
@ -487,6 +546,11 @@ def migrate(source_conn, dest_conn, simulation):
logger.info(f"Skipped emails: {skipped}\t(Check migration log for details)")
logger.info(f"Total size: {total_size_mb} MB")
# Too long to print it to the console
logger.debug("\n=== STATISTICS ===")
logger.debug(format_statistics())
def prepare_and_migrate(simulation):
logger.info("\n" + ("SIMULATION MODE: Scanning mailbox" if simulation else "LIVE MODE: Starting migration"))
@ -522,5 +586,7 @@ def main():
prepare_and_migrate(simulation)
save_statistics_file(STATISTICS_FILE)
if __name__ == '__main__':
main()

4
requirements.txt Normal file
View File

@ -0,0 +1,4 @@
python-dateutil==2.9.0.post0
python-dotenv==1.0.1
six==1.17.0
tqdm==4.67.1

75
stats.py Normal file
View File

@ -0,0 +1,75 @@
import os
import json
# Global dictionary to store statistics.
STATISTICS = {}
def add_statistic(category, key, amount=1):
"""
Increments the count for a given key under a given category by 'amount'.
If the category or key does not exist, they are created.
For example:
add_statistic("file_extensions", ".pdf", 1)
add_statistic("source_folder", "Archive", 1)
add_statistic("sender", "bjoern@waide.de", 1)
"""
if category not in STATISTICS:
STATISTICS[category] = {}
if key not in STATISTICS[category]:
STATISTICS[category][key] = 0
STATISTICS[category][key] += amount
def save_statistics_file(statistics_file="statistics.json"):
"""
Writes the STATISTICS dictionary to a JSON file with indentation.
"""
try:
with open(statistics_file, "w") as f:
json.dump(STATISTICS, f, indent=4)
print(f"Statistics saved to {statistics_file}")
except Exception as e:
print(f"ERROR: Could not save statistics to {statistics_file}: {e}")
def load_statistics_file(statistics_file="statistics.json"):
"""
Loads the statistics from the given JSON file and pre-fills the global STATISTICS variable.
If the file does not exist or an error occurs, STATISTICS remains an empty dict.
Returns:
None
"""
global STATISTICS
if os.path.exists(statistics_file):
try:
with open(statistics_file, "r") as f:
loaded_stats = json.load(f)
# You can choose to either replace STATISTICS or merge with the existing one.
# Here we replace it:
STATISTICS = loaded_stats
print(f"Statistics loaded from {statistics_file}")
except Exception as e:
print(f"ERROR: Could not load statistics from {statistics_file}: {e}")
else:
STATISTICS = {}
print(f"No statistics file found. Starting with an empty statistics dictionary.")
def format_statistics():
"""
Returns a formatted string representation of the STATISTICS dictionary.
For each category in STATISTICS, the keys are sorted in decreasing order
by their associated count.
"""
lines = []
# Optionally, sort the categories alphabetically.
for category, data in sorted(STATISTICS.items()):
lines.append(f"Category: {category}")
# Sort the items (key, count) in decreasing order by count.
sorted_items = sorted(data.items(), key=lambda item: item[1], reverse=True)
for key, count in sorted_items:
lines.append(f" {key}: {count}")
lines.append("") # Add an empty line between categories for readability
return "\n".join(lines)