如何编写批量删除重复邮件脚本

wen 实用脚本 2026-07-27 82

本文目录导读：

如何编写批量删除重复邮件脚本

IMAP协议脚本（推荐方案）
POP3协议脚本
Outlook/Exchange脚本（Windows）
高级功能：邮件去重工具
使用说明

IMAP协议脚本（推荐方案）

使用Python + imaplib

#!/usr/bin/env python3
"""
批量删除IMAP邮箱中的重复邮件
用法：python3 dedup_imap.py
"""
import imaplib
import email
from email.header import decode_header
import hashlib
from collections import defaultdict
# 邮箱配置
IMAP_SERVER = "imap.gmail.com"
EMAIL = "your_email@gmail.com"
PASSWORD = "your_app_password"  # Gmail需使用应用专用密码
def decode_email_header(header):
    """解码邮件头信息"""
    decoded_parts = []
    for part, encoding in decode_header(header):
        if isinstance(part, bytes):
            try:
                decoded_parts.append(part.decode(encoding or 'utf-8'))
            except:
                decoded_parts.append(part.decode('utf-8', errors='replace'))
        else:
            decoded_parts.append(part)
    return ''.join(decoded_parts)
def get_email_hash(msg):
    """生成邮件唯一标识（基于主题、发件人、日期和正文前100字符）"""
    subject = decode_email_header(msg.get('Subject', ''))
    sender = decode_email_header(msg.get('From', ''))
    date = msg.get('Date', '')
    # 提取正文前100个字符
    body = ""
    if msg.is_multipart():
        for part in msg.walk():
            if part.get_content_type() == "text/plain":
                body = part.get_payload(decode=True).decode('utf-8', errors='replace')[:100]
                break
    else:
        body = msg.get_payload(decode=True).decode('utf-8', errors='replace')[:100]
    # 创建哈希
    content = f"{subject}|{sender}|{date}|{body}"
    return hashlib.md5(content.encode('utf-8')).hexdigest()
def main():
    # 连接邮箱
    mail = imaplib.IMAP4_SSL(IMAP_SERVER)
    mail.login(EMAIL, PASSWORD)
    # 选择收件箱
    mail.select("INBOX")
    # 搜索所有邮件
    status, messages = mail.search(None, "ALL")
    email_ids = messages[0].split() if status == "OK" else []
    print(f"找到 {len(email_ids)} 封邮件")
    # 按哈希分组
    hash_map = defaultdict(list)
    for num in email_ids:
        status, data = mail.fetch(num, "(RFC822)")
        if status != "OK":
            continue
        msg = email.message_from_bytes(data[0][1])
        email_hash = get_email_hash(msg)
        hash_map[email_hash].append(num)
    # 找出重复邮件（保留第一个，删除其余）
    to_delete = []
    for email_hash, ids in hash_map.items():
        if len(ids) > 1:
            # 保留第一封，标记其余为删除
            to_delete.extend(ids[1:])
    if not to_delete:
        print("没有发现重复邮件")
        return
    print(f"发现 {len(to_delete)} 封重复邮件需要删除")
    # 确认操作
    confirm = input("确认删除？(yes/no): ")
    if confirm.lower() != "yes":
        print("操作已取消")
        return
    # 移动到垃圾箱（安全删除）
    for email_id in to_delete:
        mail.copy(email_id, "[Gmail]/Trash")  # Gmail专用
        mail.store(email_id, "+FLAGS", "\\Deleted")
    # 永久删除
    mail.expunge()
    print(f"已删除 {len(to_delete)} 封重复邮件")
    mail.logout()
if __name__ == "__main__":
    main()

POP3协议脚本

#!/usr/bin/env python3
"""
批量删除POP3邮箱中的重复邮件
"""
import poplib
import email
from email.header import decode_header
import hashlib
from collections import defaultdict
POP3_SERVER = "pop.gmail.com"
EMAIL = "your_email@gmail.com"
PASSWORD = "your_app_password"
def get_email_hash(msg):
    """与IMAP版本相同的哈希函数"""
    subject = decode_header(msg.get('Subject', ''))
    sender = decode_header(msg.get('From', ''))
    date = msg.get('Date', '')
    body = ""
    if msg.is_multipart():
        for part in msg.walk():
            if part.get_content_type() == "text/plain":
                body = part.get_payload(decode=True).decode('utf-8', errors='replace')[:100]
                break
    else:
        body = msg.get_payload(decode=True).decode('utf-8', errors='replace')[:100]
    content = f"{subject}|{sender}|{date}|{body}"
    return hashlib.md5(content.encode('utf-8')).hexdigest()
def main():
    # 连接POP3服务器
    mail = poplib.POP3_SSL(POP3_SERVER)
    mail.user(EMAIL)
    mail.pass_(PASSWORD)
    num_messages = len(mail.list()[1])
    print(f"找到 {num_messages} 封邮件")
    # 分析所有邮件
    hash_map = defaultdict(list)
    for i in range(1, num_messages + 1):
        response, lines, octets = mail.retr(i)
        msg_content = b'\n'.join(lines)
        msg = email.message_from_bytes(msg_content)
        email_hash = get_email_hash(msg)
        hash_map[email_hash].append(i)
    # 找出重复邮件（保留最后一个收到的）
    to_delete = []
    for email_hash, ids in hash_map.items():
        if len(ids) > 1:
            # POP3按顺序编号，保留编号最大的（最新）
            ids.sort()
            to_delete.extend(ids[:-1])
    if not to_delete:
        print("没有发现重复邮件")
        mail.quit()
        return
    print(f"发现 {len(to_delete)} 封重复邮件")
    # POP3只能通过UIDL删除特定邮件
    confirm = input("确认删除？(yes/no): ")
    if confirm.lower() != "yes":
        print("操作已取消")
        mail.quit()
        return
    # 获取UID列表
    uids = mail.uidl()[1]
    # 删除重复邮件（从后往前删除避免索引错乱）
    for msg_num in sorted(to_delete, reverse=True):
        mail.dele(msg_num)
    print(f"已标记删除 {len(to_delete)} 封重复邮件")
    mail.quit()
if __name__ == "__main__":
    main()

Outlook/Exchange脚本（Windows）

# PowerShell脚本 - 删除Outlook重复邮件
Add-Type -Assembly "Microsoft.Office.Interop.Outlook"
$outlook = New-Object -ComObject Outlook.Application
$namespace = $outlook.GetNamespace("MAPI")
# 选择收件箱
$inbox = $namespace.GetDefaultFolder([Microsoft.Office.Interop.Outlook.OlDefaultFolders]::olFolderInbox)
$items = $inbox.Items
# 按主题分组
$groups = @{}
foreach ($item in $items) {
    if ($item.Class -eq 43) {  # 邮件对象
        $subject = $item.Subject
        if (-not $groups.ContainsKey($subject)) {
            $groups[$subject] = @()
        }
        $groups[$subject] += $item
    }
}
$deleted = 0
foreach ($subject in $groups.Keys) {
    $group = $groups[$subject]
    if ($group.Count -gt 1) {
        # 保留最新的，删除其余的
        $sorted = $group | Sort-Object ReceivedTime -Descending
        for ($i = 1; $i -lt $sorted.Count; $i++) {
            $sorted[$i].Delete()
            $deleted++
        }
    }
}
Write-Host "已删除 $deleted 封重复邮件"

高级功能：邮件去重工具

#!/usr/bin/env python3
"""
完整邮件去重工具 - 支持多个邮箱和邮件夹
"""
import imaplib
import email
import hashlib
import sqlite3
import json
from pathlib import Path
from datetime import datetime
class EmailDedup:
    def __init__(self, config_file='email_config.json'):
        self.load_config(config_file)
        self.setup_database()
    def load_config(self, config_file):
        with open(config_file, 'r') as f:
            self.config = json.load(f)
    def setup_database(self):
        """创建去重数据库"""
        db_path = Path.home() / '.email_dedup.db'
        self.conn = sqlite3.connect(str(db_path))
        self.cursor = self.conn.cursor()
        self.cursor.execute('''
            CREATE TABLE IF NOT EXISTS email_hashes (
                hash TEXT PRIMARY KEY,
                account TEXT,
                first_seen TIMESTAMP
            )
        ''')
        self.conn.commit()
    def connect_mailbox(self, account):
        """连接邮件服务器"""
        cfg = self.config['accounts'][account]
        mail = imaplib.IMAP4_SSL(cfg['server'])
        mail.login(cfg['email'], cfg['password'])
        return mail, cfg
    def scan_mailbox(self, account, folder='INBOX'):
        """扫描邮件夹"""
        mail, cfg = self.connect_mailbox(account)
        mail.select(folder)
        status, messages = mail.search(None, "ALL")
        email_ids = messages[0].split() if status == "OK" else []
        to_delete = []
        for num in email_ids:
            status, data = mail.fetch(num, "(RFC822)")
            if status != "OK":
                continue
            msg = email.message_from_bytes(data[0][1])
            email_hash = self.compute_hash(msg)
            # 检查是否已存在
            self.cursor.execute(
                "SELECT first_seen FROM email_hashes WHERE hash = ?",
                (email_hash,)
            )
            result = self.cursor.fetchone()
            if result:
                to_delete.append(num)
            else:
                self.cursor.execute(
                    "INSERT INTO email_hashes VALUES (?, ?, ?)",
                    (email_hash, account, datetime.now())
                )
        return mail, to_delete
    def compute_hash(self, msg):
        """计算邮件哈希"""
        subject = msg.get('Subject', '')
        sender = msg.get('From', '')
        date = msg.get('Date', '')
        # 获取正文
        body = ""
        if msg.is_multipart():
            for part in msg.walk():
                if part.get_content_type() == "text/plain":
                    body = part.get_payload(decode=True).decode('utf-8', errors='replace')
                    break
        else:
            body = msg.get_payload(decode=True).decode('utf-8', errors='replace')
        # 添加附件信息
        attachments = []
        for part in msg.walk():
            if part.get_content_disposition() == 'attachment':
                attachments.append(part.get_filename())
        content = f"{subject}|{sender}|{date}|{body[:200]}|{attachments}"
        return hashlib.sha256(content.encode('utf-8')).hexdigest()
    def delete_duplicates(self, mail, email_ids, confirm=True):
        """删除重复邮件"""
        if not email_ids:
            print("没有重复邮件")
            return
        print(f"发现 {len(email_ids)} 封重复邮件")
        if confirm:
            response = input("确认删除？(yes/no): ")
            if response.lower() != 'yes':
                return
        for num in email_ids:
            mail.store(num, "+FLAGS", "\\Deleted")
        mail.expunge()
        print(f"已删除 {len(email_ids)} 封重复邮件")
        self.conn.commit()
    def run(self):
        """执行去重操作"""
        print("开始扫描邮箱...")
        for account in self.config['accounts']:
            print(f"处理账号: {account}")
            mail, to_delete = self.scan_mailbox(account)
            self.delete_duplicates(mail, to_delete)
            mail.logout()
        self.conn.close()
# 配置文件示例：email_config.json
"""
{
    "accounts": {
        "work": {
            "server": "imap.gmail.com",
            "email": "work@company.com",
            "password": "app_password_1"
        },
        "personal": {
            "server": "imap.gmail.com",
            "email": "personal@gmail.com",
            "password": "app_password_2"
        }
    }
}
"""
if __name__ == "__main__":
    dedup = EmailDedup()
    dedup.run()

使用说明

环境准备

# 安装Python依赖
pip install imaplib email hashlib
# Gmail用户需要启用"允许不够安全的应用"或使用应用专用密码

安全建议

备份邮件：在删除前先备份重要邮件
测试运行：先在少量邮件上测试
使用应用密码：不要使用主密码
开启两步验证：提高账户安全性
日志记录：记录删除的邮件信息

注意事项

不同邮件提供商的IMAP路径可能不同
某些邮箱对批量操作有限制
建议先移动到垃圾箱而不是永久删除
考虑使用邮件客户端内置的去重功能

选择适合你邮箱提供商的脚本,并根据实际需求调整哈希算法和匹配规则。

本文地址： https://www.bmjup.com/post/31776.html