本文目录导读：
我来为您设计一个完整的自动化邮件归档脚本方案,包含多种实现方式和最佳实践。
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) # 指数退避
监控和报告
- 记录归档运行状态
- 发送通知邮件或Slack消息
- 定期检查日志
使用方法
- 配置脚本参数（根据实际情况修改）
- 测试运行：先在小规模邮件上测试
- 设置权限：确保脚本有适当的文件权限
- 创建计划任务：使用cron或任务计划程序
- 监控运行：定期检查日志和输出
这个脚本提供了灵活的配置和多种归档方式，您可以根据实际需求调整，建议先在测试环境中运行,确保一切正常后再部署到生产环境。
