基于机器学习的垃圾邮件过滤系统

import os
import tarfile
import urllib.request
# 下载数据集
def download_data(url, target_folder):
    if not os.path.exists(target_folder):
        os.makedirs(target_folder)
    file_name = url.split("/")[-1]
    target_path = os.path.join(target_folder, file_name)
    if not os.path.exists(target_path):
        urllib.request.urlretrieve(url, target_path)
        print(f"Downloaded {file_name}")
    return target_path
# 解压数据集
def extract_data(file_path, target_folder):
    with tarfile.open(file_path, "r:gz") as tar:
        tar.extractall(target_folder)
        print(f"Extracted {os.path.basename(file_path)} to {target_folder}")
# 下载并解压数据集
url_list = [
    "https://spamassassin.apache.org/old/publiccorpus/20021010_easy_ham.tar.bz2",
    "https://spamassassin.apache.org/old/publiccorpus/20021010_hard_ham.tar.bz2",
    "https://spamassassin.apache.org/old/publiccorpus/20021010_spam.tar.bz2"
]
target_folder = "data"
for url in url_list:
    file_path = download_data(url, target_folder)
    extract_data(file_path, target_folder)

import os
import email
import email.policy
def read_email(file_path):
    with open(file_path, "rb") as f:
        return email.parser.BytesParser(policy=email.policy.default).parse(f)
ham_folder = "data/easy_ham"
spam_folder = "data/spam"
ham_files = [os.path.join(ham_folder, f) for f in os.listdir(ham_folder)]
spam_files = [os.path.join(spam_folder, f) for f in os.listdir(spam_folder)]
ham_emails = [read_email(f) for f in ham_files]
spam_emails = [read_email(f) for f in spam_files]

import re
import nltk
from nltk.corpus import stopwords
from nltk.stem import SnowballStemmer
from sklearn.feature_extraction.text import TfidfVectorizer
nltk.download('stopwords')
stemmer = SnowballStemmer("english")
stop_words = set(stopwords.words("english"))
# 清洗文本
def clean_text(text):
    text = re.sub(r'\W+', ' ', text)  # 移除非字母数字字符
    text = text.lower()  # 转换为小写
    text = re.sub(r'\d+', '', text)  # 移除数字
text = ' '.join([stemmer.stem(word) for word in text.split() if word not in stop_words])  # 移除停用词并进行词干提取
return text
# 提取邮件正文
def get_email_text(email_obj):
parts = []
for part in email_obj.walk():
if part.get_content_type() == 'text/plain':
parts.append(part.get_payload())
return ''.join(parts)
# 对所有邮件进行预处理
ham_texts = [clean_text(get_email_text(email)) for email in ham_emails]
spam_texts = [clean_text(get_email_text(email)) for email in spam_emails]
# 合并数据和标签
texts = ham_texts + spam_texts
labels = [0] * len(ham_texts) + [1] * len(spam_texts)
# 向量化文本
vectorizer = TfidfVectorizer()
X = vectorizer.fit_transform(texts)
y = labels

from sklearn.model_selection import train_test_split
from sklearn.naive_bayes import MultinomialNB
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
# 划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# 创建朴素贝叶斯模型并训练
model = MultinomialNB()
model.fit(X_train, y_train)
# 在测试集上进行预测
y_pred = model.predict(X_test)
# 计算评估指标
accuracy = accuracy_score(y_test, y_pred)
precision = precision_score(y_test, y_pred)
recall = recall_score(y_test, y_pred)
f1 = f1_score(y_test, y_pred)
print(f"Accuracy: {accuracy:.4f}")
print(f"Precision: {precision:.4f}")
print(f"Recall: {recall:.4f}")
print(f"F1 Score: {f1:.4f}")

from sklearn.model_selection import train_test_split
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
# 在测试集上进行预测
y_pred = model.predict(X_test)
# 计算评估指标
accuracy = accuracy_score(y_test, y_pred)
precision = precision_score(y_test, y_pred)
recall = recall_score(y_test, y_pred)
f1 = f1_score(y_test, y_pred)
print(f"Accuracy: {accuracy:.4f}")
print(f"Precision: {precision:.4f}")
print(f"Recall: {recall:.4f}")
print(f"F1 Score: {f1:.4f}")

from sklearn.model_selection import cross_val_score
# 使用交叉验证计算评估指标
cv_accuracy = cross_val_score(model, X, y, cv=5, scoring='accuracy').mean()
cv_precision = cross_val_score(model, X, y, cv=5, scoring='precision').mean()
cv_recall = cross_val_score(model, X, y, cv=5, scoring='recall').mean()
cv_f1 = cross_val_score(model, X, y, cv=5, scoring='f1').mean()
print(f"Cross-Validation Accuracy: {cv_accuracy:.4f}")
print(f"Cross-Validation Precision: {cv_precision:.4f}")
print(f"Cross-Validation Recall: {cv_recall:.4f}")
print(f"Cross-Validation F1 Score: {cv_f1:.4f}")

import pickle
# 保存模型
with open('spam_classifier_model.pkl', 'wb') as file:
    pickle.dump(model, file)
# 保存向量化器
with open('vectorizer.pkl', 'wb') as file:
    pickle.dump(vectorizer, file)

import email
from email.message import EmailMessage
# 加载模型和向量化器
with open('spam_classifier_model.pkl', 'rb') as file:
    model = pickle.load(file)
with open('vectorizer.pkl', 'rb') as file:
    vectorizer = pickle.load(file)
# 邮件处理函数
def process_email(raw_email):
    # 解析邮件并提取正文
    email_obj = email.message_from_string(raw_email)
    email_text = get_email_text(email_obj)
    # 文本预处理
    cleaned_text = clean_text(email_text)
    vectorized_text = vectorizer.transform([cleaned_text])
    # 使用模型预测
    prediction = model.predict(vectorized_text)
    return "Spam" if prediction[0] == 1 else "Ham"
# 示例：读取邮件文本并使用处理函数判断是否为垃圾邮件
with open('example_email.txt', 'r') as file:
    raw_email = file.read()
result = process_email(raw_email)
print(f"Result: {result}")

Option Explicit
Sub ProcessNewEmail(Item As Outlook.MailItem)
    ' 调用Python脚本处理邮件，并获得预测结果
    Dim result As String
    result = RunPythonScript("process_email.py", Item.Body)
    ' 根据预测结果处理邮件
    If result = "Spam" Then
        ' 将邮件移动到垃圾邮件文件夹
        Dim spamFolder As Outlook.MAPIFolder
        Set spamFolder = Application.Session.GetDefaultFolder(olFolderJunk)
        Item.Move spamFolder
    End If
End Sub

import sys
def process_email(email_text):
    # ...
    # 文本预处理和模型预测的代码
    # ...
    return "Spam" if prediction[0] == 1 else "Ham"
if __name__ == "__main__":
    email_text = sys.argv[1]
    result = process_email(email_text)
    print(result)

Function RunPythonScript(scriptPath As String, emailBody As String) As String
    Dim shell As Object
    Dim command As String
    Dim tempFile As String
    Dim fso As Object
    Dim file As Object
    Dim result As String
    ' 创建一个临时文件用于存储Python脚本的输出
    Set fso = CreateObject("Scripting.FileSystemObject")
    tempFile = fso.GetSpecialFolder(2) & "\" & fso.GetTempName
    ' 构建命令行
command = "python " & scriptPath & " """ & emailBody & """" & " > " & tempFile
' 执行命令行
Set shell = CreateObject("WScript.Shell")
shell.Run command, 0, True
' 读取临时文件中的输出结果
Set file = fso.OpenTextFile(tempFile, 1)
result = file.ReadAll
file.Close
' 删除临时文件
fso.DeleteFile tempFile
' 返回结果
RunPythonScript = Trim(result)