如何编写自动化邮件归档脚本

wen 实用脚本 2026-08-01 200

本文目录导读：

如何编写自动化邮件归档脚本

方案一：Python + IMAP（通用性强）
方案二：Windows PowerShell脚本
方案三：Shell + Mutt（Linux/Mac）
配置文件示例
最佳实践和建议
使用方法

我来为您设计一个完整的自动化邮件归档脚本方案,包含多种实现方式和最佳实践。

Python + IMAP（通用性强）

#!/usr/bin/env python3
"""
自动化邮件归档脚本 - Python版本
支持IMAP协议，适用于大多数邮箱服务
"""
import imaplib
import email
from email.header import decode_header
import datetime
import os
import logging
import time
from pathlib import Path
import json
from typing import List, Dict, Optional
import hashlib
class EmailArchiver:
    """邮件自动化归档类"""
    def __init__(self, config_file: str = "config.json"):
        """
        初始化归档器
        Args:
            config_file: 配置文件路径
        """
        self.config = self._load_config(config_file)
        self.imap = None
        self.setup_logging()
    def _load_config(self, config_file: str) -> Dict:
        """加载配置文件"""
        with open(config_file, 'r', encoding='utf-8') as f:
            return json.load(f)
    def setup_logging(self):
        """设置日志记录"""
        log_config = self.config.get('logging', {})
        log_file = log_config.get('file', 'archive.log')
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(log_file, encoding='utf-8'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def connect(self) -> bool:
        """连接到邮件服务器"""
        try:
            imap_config = self.config['imap']
            self.imap = imaplib.IMAP4_SSL(
                imap_config['server'], 
                imap_config['port']
            )
            self.imap.login(
                imap_config['username'], 
                imap_config['password']
            )
            self.logger.info(f"成功连接到 {imap_config['server']}")
            return True
        except Exception as e:
            self.logger.error(f"连接失败: {str(e)}")
            return False
    def disconnect(self):
        """断开连接"""
        if self.imap:
            try:
                self.imap.logout()
                self.logger.info("已断开连接")
            except:
                pass
    def _decode_header(self, header_value: str) -> str:
        """解码邮件头"""
        if header_value is None:
            return ""
        decoded_parts = []
        for part, encoding in decode_header(header_value):
            if isinstance(part, bytes):
                try:
                    decoded = part.decode(encoding or 'utf-8')
                except:
                    decoded = part.decode('utf-8', errors='ignore')
            else:
                decoded = part
            decoded_parts.append(decoded)
        return ''.join(decoded_parts)
    def _get_email_date(self, msg) -> datetime.datetime:
        """获取邮件日期"""
        date_str = msg.get('Date', '')
        try:
            from email.utils import parsedate_to_datetime
            return parsedate_to_datetime(date_str)
        except:
            return datetime.datetime.now()
    def _extract_email_content(self, msg) -> str:
        """提取邮件内容"""
        content = ""
        if msg.is_multipart():
            for part in msg.walk():
                if part.get_content_type() == "text/plain":
                    content = part.get_payload(decode=True)
                    if content:
                        try:
                            content = content.decode('utf-8')
                        except:
                            content = content.decode('utf-8', errors='ignore')
                        break
        else:
            content = msg.get_payload(decode=True)
            if content:
                try:
                    content = content.decode('utf-8')
                except:
                    content = content.decode('utf-8', errors='ignore')
        return str(content)
    def get_inbox_emails(self, folder: str = "INBOX") -> List[Dict]:
        """获取收件箱邮件"""
        emails = []
        try:
            # 选择文件夹（只读模式）
            status, messages = self.imap.select(folder, readonly=True)
            if status != 'OK':
                self.logger.error(f"无法打开文件夹 {folder}")
                return emails
            # 获取邮件数量
            email_count = int(messages[0])
            self.logger.info(f"文件夹 {folder} 中有 {email_count} 封邮件")
            # 获取所有邮件ID
            status, msg_ids = self.imap.search(None, "ALL")
            if status != 'OK':
                return emails
            # 配置中的归档条件
            archive_config = self.config.get('archive_rules', {})
            days_to_archive = archive_config.get('days_old', 30)
            archive_folders = archive_config.get('folders', ['INBOX'])
            cutoff_date = datetime.datetime.now() - datetime.timedelta(
                days=days_to_archive
            )
            # 处理每封邮件
            for msg_id in msg_ids[0].split():
                try:
                    # 获取邮件详情
                    status, msg_data = self.imap.fetch(msg_id, '(RFC822)')
                    if status != 'OK':
                        continue
                    raw_email = msg_data[0][1]
                    msg = email.message_from_bytes(raw_email)
                    # 解析邮件信息
                    email_date = self._get_email_date(msg)
                    subject = self._decode_header(msg.get('Subject', ''))
                    sender = self._decode_header(msg.get('From', ''))
                    recipients = self._decode_header(msg.get('To', ''))
                    content = self._extract_email_content(msg)
                    # 检查是否满足归档条件
                    if email_date < cutoff_date:
                        email_info = {
                            'id': msg_id.decode(),
                            'date': email_date.strftime('%Y-%m-%d'),
                            'time': email_date.strftime('%H:%M:%S'),
                            'subject': subject,
                            'sender': sender,
                            'recipients': recipients,
                            'content_preview': content[:500],
                            'folder': folder,
                            'archived': False
                        }
                        # 应用过滤规则
                        if self._should_archive(email_info):
                            emails.append(email_info)
                except Exception as e:
                    self.logger.error(f"处理邮件 {msg_id} 时出错: {str(e)}")
                    continue
        except Exception as e:
            self.logger.error(f"获取邮件失败: {str(e)}")
        return emails
    def _should_archive(self, email_info: Dict) -> bool:
        """判断是否应该归档某封邮件"""
        rules = self.config.get('filter_rules', {})
        # 排除重要发件人
        exclude_senders = rules.get('exclude_senders', [])
        for sender in exclude_senders:
            if sender.lower() in email_info['sender'].lower():
                return False
        # 排除特定主题
        exclude_subjects = rules.get('exclude_subjects', [])
        for subject in exclude_subjects:
            if subject.lower() in email_info['subject'].lower():
                return False
        # 仅包含特定发件人
        include_senders = rules.get('include_senders', [])
        if include_senders:
            for sender in include_senders:
                if sender.lower() in email_info['sender'].lower():
                    return True
            return False
        return True
    def archive_emails(self, emails: List[Dict]) -> int:
        """归档邮件"""
        archived_count = 0
        for email_info in emails:
            try:
                if email_info.get('archived', False):
                    continue
                # 根据月份创建归档文件夹
                year_month = email_info['date'][:7]  # YYYY-MM
                archive_folder = f"Archive/{year_month}"
                # 创建归档文件夹（如果不存在）
                status, _ = self.imap.select(email_info['folder'])
                if status != 'OK':
                    continue
                # 复制邮件到归档文件夹
                msg_id = email_info['id'].encode()
                status, _ = self.imap.copy(msg_id, archive_folder)
                if status == 'OK':
                    # 如果配置为移动（而不是复制），则删除原邮件
                    if self.config.get('archive_rules', {}).get('move', True):
                        self.imap.store(msg_id, '+FLAGS', '\\Deleted')
                    email_info['archived'] = True
                    archived_count += 1
                    self.logger.info(
                        f"已归档邮件: {email_info['subject'][:50]}"
                    )
            except Exception as e:
                self.logger.error(
                    f"归档邮件 {email_info.get('id')} 失败: {str(e)}"
                )
                continue
        # 如果使用了删除标记，执行expunge
        if archived_count > 0 and self.config.get('archive_rules', {}).get('move', True):
            try:
                self.imap.expunge()
            except:
                pass
        return archived_count
    def save_metadata(self, emails: List[Dict], output_file: str = "archive_metadata.json"):
        """保存归档元数据"""
        metadata = {
            'timestamp': datetime.datetime.now().isoformat(),
            'archived_emails': emails
        }
        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(metadata, f, ensure_ascii=False, indent=2)
        self.logger.info(f"元数据已保存到 {output_file}")
    def run_archive_chunk(self, chunk_size: int = 100):
        """分块运行归档任务"""
        all_emails = []
        for folder in self.config.get('archive_rules', {}).get('folders', ['INBOX']):
            self.logger.info(f"处理文件夹: {folder}")
            emails = self.get_inbox_emails(folder)
            all_emails.extend(emails)
            # 分块处理
            for i in range(0, len(emails), chunk_size):
                chunk = emails[i:i+chunk_size]
                if chunk:
                    archived = self.archive_emails(chunk)
                    self.logger.info(f"分块 {i//chunk_size + 1}: 归档了 {archived} 封邮件")
                    # 延迟以避免服务器限制
                    time.sleep(1)
        # 保存元数据
        self.save_metadata(all_emails)
        return len(all_emails)
    def run(self):
        """主运行函数"""
        self.logger.info("=== 开始邮件归档任务 ===")
        if not self.connect():
            return
        try:
            archived_count = self.run_archive_chunk()
            self.logger.info(f"总共归档了 {archived_count} 封邮件")
        except Exception as e:
            self.logger.error(f"归档任务执行失败: {str(e)}")
        finally:
            self.disconnect()
            self.logger.info("=== 邮件归档任务结束 ===")
def main():
    """主函数"""
    # 创建配置示例
    config = {
        "imap": {
            "server": "imap.example.com",
            "port": 993,
            "username": "your_email@example.com",
            "password": "your_password"
        },
        "archive_rules": {
            "days_old": 30,
            "folders": ["INBOX"],
            "move": True
        },
        "filter_rules": {
            "exclude_senders": ["urgent@company.com"],
            "exclude_subjects": ["Important", "Urgent"],
            "include_senders": []  # 空列表表示归档所有邮件
        },
        "logging": {
            "file": "email_archive.log"
        }
    }
    # 保存配置
    with open("config.json", 'w', encoding='utf-8') as f:
        json.dump(config, f, indent=2, ensure_ascii=False)
    # 运行归档器
    archiver = EmailArchiver("config.json")
    archiver.run()
if __name__ == "__main__":
    main()

Windows PowerShell脚本

# EmailArchiver.ps1 - Windows版邮件归档脚本
param(
    [string]$Mailbox,
    [int]$DaysOld = 30,
    [string]$OutputPath = "C:\EmailArchive",
    [switch]$CreateScheduledTask
)
# 添加Microsoft Outlook库
Add-Type -AssemblyName "Microsoft.Office.Interop.Outlook"
class EmailArchiver {
    [string]$Mailbox
    [int]$DaysOld
    [string]$OutputPath
    [object]$Outlook
    EmailArchiver([string]$mailbox, [int]$daysOld, [string]$outputPath) {
        $this.Mailbox = $mailbox
        $this.DaysOld = $daysOld
        $this.OutputPath = $outputPath
        # 初始化Outlook
        $this.Outlook = New-Object -ComObject Outlook.Application
        # 创建输出目录
        if (-not (Test-Path $this.OutputPath)) {
            New-Item -Path $this.OutputPath -ItemType Directory -Force
        }
    }
    [void] StartArchive() {
        Write-Host "开始归档邮件..." -ForegroundColor Green
        try {
            # 获取命名空间
            $namespace = $this.Outlook.GetNamespace("MAPI")
            # 获取收件箱
            $inbox = $namespace.GetDefaultFolder([Microsoft.Office.Interop.Outlook]::OlDefaultFolders.olFolderInbox)
            # 计算截止日期
            $cutoffDate = (Get-Date).AddDays(-$this.DaysOld)
            # 遍历邮件
            $items = $inbox.Items
            $items.Sort("[ReceivedTime]", $true)  # 按接收时间排序
            $archivedCount = 0
            foreach ($item in $items) {
                if ($item.Class -eq [Microsoft.Office.Interop.Outlook]::OlObjectClass.olMail) {
                    $email = [Microsoft.Office.Interop.Outlook.MailItem]$item
                    # 检查邮件日期
                    if ($email.ReceivedTime -lt $cutoffDate) {
                        # 创建归档文件夹结构
                        $yearMonth = $email.ReceivedTime.ToString("yyyy-MM")
                        $archiveFolder = Join-Path $this.OutputPath $yearMonth
                        if (-not (Test-Path $archiveFolder)) {
                            New-Item -Path $archiveFolder -ItemType Directory -Force
                        }
                        # 生成文件名
                        $safeSubject = $email.Subject -replace '[^\w\-]', '_'
                        $fileName = "$($email.ReceivedTime.ToString('yyyyMMdd_HHmmss'))_$safeSubject.eml"
                        $filePath = Join-Path $archiveFolder $fileName
                        # 转换并保存为EML格式
                        $this.ConvertToEML($email, $filePath)
                        # 可选的：移动到Outlook归档文件夹
                        if ($this.Mailbox) {
                            $this.MoveToArchiveFolder($email)
                        }
                        $archivedCount++
                        Write-Host "已归档: $($email.Subject)" -ForegroundColor Gray
                    }
                }
            }
            Write-Host "归档完成，共归档 $archivedCount 封邮件" -ForegroundColor Green
        } catch {
            Write-Error "归档失败: $_"
        } finally {
            # 清理COM对象
            [System.Runtime.Interopservices.Marshal]::ReleaseComObject($this.Outlook) | Out-Null
            [GC]::Collect()
            [GC]::WaitForPendingFinalizers()
        }
    }
    [void] ConvertToEML($email, $outputPath) {
        try {
            # 创建和写出EML内容
            $emlContent = @"
From: $($email.SenderEmailAddress)
To: $($email.To)
CC: $($email.CC)
Subject: $($email.Subject)
Date: $($email.ReceivedTime.ToString("r"))
$($email.Body)
"@
            # 保存EML文件
            [System.IO.File]::WriteAllText($outputPath, $emlContent, [System.Text.Encoding]::UTF8)
        } catch {
            Write-Warning "转换邮件失败: $_"
        }
    }
    [void] MoveToArchiveFolder($email) {
        try {
            # 获取命名空间
            $namespace = $this.Outlook.GetNamespace("MAPI")
            # 创建归档文件夹
            $emailFolder = $namespace.Folders.Item($email.Parent.FolderPath)
            $archiveFolderName = "Archive_$($email.ReceivedTime.ToString('yyyy-MM'))"
            $archiveFolder = $null
            try {
                $archiveFolder = $emailFolder.Folders.Item($archiveFolderName)
            } catch {
                $archiveFolder = $emailFolder.Folders.Add($archiveFolderName)
            }
            # 移动邮件
            $email.Move($archiveFolder)
        } catch {
            Write-Warning "移动邮件失败: $_"
        }
    }
}
# 主程序
function Main {
    $archiver = [EmailArchiver]::new(
        $Mailbox,
        $DaysOld,
        $OutputPath
    )
    $archiver.StartArchive()
    # 创建计划任务
    if ($CreateScheduledTask) {
        CreateScheduledTask
    }
}
function CreateScheduledTask {
    Write-Host "创建计划任务..." -ForegroundColor Cyan
    $scriptPath = $MyInvocation.MyCommand.Path
    $action = New-ScheduledTaskAction -Execute "PowerShell.exe" -Argument "-File `"$scriptPath`""
    # 每周一上午9点运行
    $trigger = New-ScheduledTaskTrigger -Weekly -DaysOfWeek Monday -At "09:00"
    $settings = New-ScheduledTaskSettingsSet -StartWhenAvailable -DontStopOnIdleEnd
    Register-ScheduledTask -TaskName "Email Archive" -Action $action -Trigger $trigger -Settings $settings -Force
    Write-Host "计划任务已创建" -ForegroundColor Green
}
# 检查是否作为计划任务运行
if ($env:SCHEDULER_EXECUTING -eq $true -or $CreateScheduledTask) {
    Main
} else {
    # 交互式运行
    Main
}

Shell + Mutt（Linux/Mac）

#!/bin/bash
# email_archive.sh - Linux/Mac邮件归档脚本
# 配置变量
IMAP_SERVER="imap.example.com"
EMAIL_USER="your_email@example.com"
EMAIL_PASS="your_password"
ARCHIVE_DIR="$HOME/email_archive"
DAYS_OLD=30
LOG_FILE="$HOME/email_archive.log"
# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m' # No Color
# 日志函数
log() {
    echo "$(date '+%Y-%m-%d %H:%M:%S') - $1" | tee -a "$LOG_FILE"
}
# 错误日志
error_log() {
    echo -e "${RED}$(date '+%Y-%m-%d %H:%M:%S') - ERROR: $1${NC}" | tee -a "$LOG_FILE"
}
# 检查依赖
check_dependencies() {
    if ! command -v curl &>/dev/null; then
        error_log "curl未安装"
        exit 1
    fi
}
# 配置Mutt环境
setup_mutt() {
    cat > /tmp/archive_mutt.conf << EOF
set imap_user="$EMAIL_USER"
set imap_pass="$EMAIL_PASS"
set imap_check_subscribed
set use_imap
set folder="imap://$IMAP_SERVER"
set spoolfile="imap://$IMAP_SERVER/INBOX"
set record="+INBOX/Sent"
set editor="echo"
set askcc=no
set askbcc=no
set move=ask-yes
set save_name=yes
set delete=yes
set delete_untagged=yes
set sort=threads
set read_ratio=0.4
set timeout=120
set mail_check=150
EOF
}
# 创建归档目录结构
create_archive_structures() {
    mkdir -p "$ARCHIVE_DIR"/{metadata,logs,emails}
    log "创建归档目录结构"
}
# 归档邮件函数
archive_emails() {
    local folder=$1
    local cutoff_date=$(date -d "$DAYS_OLD days ago" +%Y-%m-%d)
    log "开始归档文件夹 $folder 中的邮件（$DAYS_OLD天前）"
    # 使用curl直接操作IMAP
    local temp_dir="$ARCHIVE_DIR/emails/$(date +%Y-%m)"
    mkdir -p "$temp_dir"
    # 获取邮件列表
    curl -s --imap "$IMAP_SERVER" \
         --user "$EMAIL_USER:$EMAIL_PASS" \
         --create "$folder" \
         --search "SINCE $cutoff_date" \
         2>/dev/null | while read email_id; do
            if [ ! -z "$email_id" ]; then
                log "处理邮件 ID: $email_id"
                # 下载邮件
                local filename="${temp_dir}/email_${email_id}_$(date +%Y%m%d%H%M%S).eml"
                curl -s --imap "$IMAP_SERVER" \
                     --user "$EMAIL_USER:$EMAIL_PASS" \
                     --fetch "$email_id" "$folder" \
                     2>/dev/null > "$filename"
                # 保存元数据
                echo "{\"id\":\"$email_id\",\"date\":\"$(date +%Y-%m-%d)\",\"folder\":\"$folder\"}" >> \
                    "$ARCHIVE_DIR/metadata/archive_$(date +%Y%m).json"
                log "已保存邮件 $email_id 到 $filename"
                # 可选：删除原邮件
                # curl -s --imap "$IMAP_SERVER" \
                #      --user "$EMAIL_USER:$EMAIL_PASS" \
                #      --delete "$email_id" "$folder" \
                #      2>/dev/null
            fi
        done
    log "文件夹 $folder 归档完成"
}
# 压缩并归档
compress_archive() {
    local archive_file="$ARCHIVE_DIR/email_archive_$(date +%Y%m).tar.gz"
    log "压缩归档文件..."
    tar -czf "$archive_file" -C "$ARCHIVE_DIR" emails metadata logs
    # 删除原始文件
    rm -rf "$ARCHIVE_DIR/emails/" "$ARCHIVE_DIR/metadata/"
    log "归档压缩完成: $archive_file"
    echo "$archive_file"
}
# 生成报告
generate_report() {
    local archive_file=$1
    local report_file="$ARCHIVE_DIR/archive_report_$(date +%Y%m).html"
    log "生成HTML报告..."
    # 从压缩文件读取元数据
    local email_count=$(tar -tzf "$archive_file" | grep -c "\.eml$")
    local archive_size=$(du -h "$archive_file" | cut -f1)
    cat > "$report_file" << EOF
<!DOCTYPE html>
<html>
<head>邮件归档报告 - $(date +%Y年%m月)</title>
    <style>
        body { font-family: Arial, sans-serif; margin: 20px; }
        .header { background-color: #4CAF50; color: white; padding: 20px; }
        .stats { margin: 20px 0; }
        .stat-item { margin: 10px 0; padding: 10px; background-color: #f9f9f9; }
    </style>
</head>
<body>
    <div class="header">
        <h1>邮件归档报告</h1>
        <p>归档时间：$(date)</p>
    </div>
    <div class="stats">
        <h2>归档统计</h2>
        <div class="stat-item">
            <strong>归档邮件数量：</strong> $email_count 封
        </div>
        <div class="stat-item">
            <strong>归档文件大小：</strong> $archive_size
        </div>
        <div class="stat-item">
            <strong>归档来源：</strong> IMAP邮件服务器
        </div>
    </div>
</body>
</html>
EOF
    log "报告已生成: $report_file"
}
# 主程序
main() {
    log "开始执行邮件归档任务"
    check_dependencies
    setup_mutt
    create_archive_structures
    # 归档收件箱
    archive_emails "INBOX"
    # 可选：归档其他文件夹
    # archive_emails "Work"
    # archive_emails "Projects/ClientA"
    # 压缩归档
    archive_file=$(compress_archive)
    # 生成报告
    generate_report "$archive_file"
    # 可选：上传到云存储
    # upload_to_s3 "$archive_file"
    # 清理临时文件
    rm -rf "$ARCHIVE_DIR/logs"/*
    log "邮件归档任务完成！"
    echo -e "${GREEN}归档完成，报告位置：$ARCHIVE_DIR/archive_report_$(date +%Y%m).html${NC}"
}
# 运行主程序
main

配置文件示例

创建 config.json 文件：

{
  "imap": {
    "server": "imap.example.com",
    "port": 993,
    "username": "your_email@example.com",
    "password": "your_email_password",
    "security": "ssl"
  },
  "archive_rules": {
    "days_old": 30,
    "folders": ["INBOX", "Work"],
    "move": true,
    "delete_after": false,
    "compress": true
  },
  "filter_rules": {
    "exclude_senders": ["no-reply@newsletter.com", "notification@updates.com"],
    "exclude_subjects": ["Support Ticket", "Invoice"]
  },
  "output": {
    "path": "/path/to/archive",
    "format": "eml",
    "metadata": true,
    "report": true
  },
  "logging": {
    "file": "/var/log/email_archive.log",
    "level": "INFO"
  },
  "schedule": {
    "enabled": true,
    "frequency": "weekly",
    "day": "Monday",
    "time": "09:00"
  },
  "notifications": {
    "email": "admin@example.com",
    "slack_webhook": "https://hooks.slack.com/services/xxx",
    "enabled": false
  }
}

最佳实践和建议

安全建议

保护好配置文件中的密码（使用环境变量或密钥管理服务）
使用OAuth2.0替代基本认证
设置文件夹权限只有管理员可访问

# 环境变量保护密码
export EMAIL_ARCHIVE_PASSWORD="your_secret_password"

自动化计划

Windows cron (任务计划程序)：

schtasks /create /tn "Email Archive" /tr "powershell.exe -File C:\scripts\EmailArchiver.ps1" /sc weekly /d MON /st 09:00 /ru SYSTEM

Linux cron：

# 每周一凌晨2点运行
0 2 * * 1 /path/to/email_archive.sh >> /var/log/email_archive_cron.log 2>&1
# 每天检查邮件
0 */6 * * * /path/to/email_archive.sh --incremental

性能优化

# 分批处理
for batch in chunks(email_list, 100):
    process_batch(batch)
    time.sleep(5)  # 避免服务器限制

错误处理

# 添加重试机制
def retry_on_failure(func, max_attempts=3):
    for attempt in range(max_attempts):
        try:
            return func()
        except Exception as e:
            if attempt == max_attempts - 1:
                raise
            time.sleep(2 ** attempt)  # 指数退避