最终版:轻量级邮件接收与API系统搭建完整教程
本教程将指导您在一台全新的 Debian 12 服务器上,搭建一个私有的、仅用于接收邮件的系统。该系统能够将收到的邮件存入数据库,并通过两种方式访问:
一个供自动化脚本调用的 API 接口,可根据收件人和主题筛选邮件并返回其内容。
一个简单的网页界面,用于人工浏览和查看所有收到的邮件。
一、前期准备 (Prerequisites)
一台服务器:拥有一个固定的公网 IP 地址。本教程使用 Debian 12 系统和 IP
89.138.1.152
。一个域名:您拥有该域名的 DNS 管理权限。本教程使用
mail.sijuly.nyc.com
作为邮件域名。SSH 访问权限:能够以
root
用户身份登录服务器。
二、DNS 配置 (DNS Configuration)
登录您的域名 sijuly.nyc.mn
的 DNS 管理后台,添加以下两条记录,将邮件路由指向您的服务器。
A 记录:
类型:
A
主机/名称:
mail
值/指向:
89.158.16.152
TTL: 自动
MX 记录:
类型:
MX
主机/名称:
mail
值/服务器:
mail.sijuly.nyc.com
优先级:
10
TTL: 自动
三、服务器环境配置 (Server Environment Configuration)
⚠️ 重要提示:本教程所有命令均以
root
用户身份执行,全程无需使用sudo
命令。
更新系统:
Bash
apt-get update && apt-get upgrade -y
安装必要的软件包:
Bash
apt-get install python3-pip python3-venv ufw sqlite3 -y
配置防火墙:
Bash
# 允许 SSH (保证您的连接不会中断) ufw allow ssh # 允许 SMTP (用于接收邮件) ufw allow 25/tcp # 允许我们的 API 和 Web 界面 ufw allow 2099/tcp # 启用防火墙 ufw enable # 出现提示时,输入 y 并按回车
四、部署应用程序 (Deploying the Application)
1. 创建项目结构和虚拟环境
Bash
# 创建并进入项目目录
mkdir -p /opt/mail_api
cd /opt/mail_api
# 创建 Python 虚拟环境
python3 -m venv venv
# 激活虚拟环境
source venv/bin/activate
2. 安装 Python 依赖库
确保您已激活虚拟环境(命令提示符前有
(venv)
字样)。
Bash
pip install aiosmtpd flask gunicorn
3. 创建 app.py
(API、数据库和Web界面)
使用 nano /opt/mail_api/app.py
命令创建文件,并将以下全部代码粘贴进去。
Python
import sqlite3
import re
from flask import Flask, request, Response
from email import message_from_bytes
from email.header import decode_header
from markupsafe import escape
# --- 配置 ---
DB_FILE = 'emails.db'
# !!! 强烈建议:为了安全,请将下面的 Token 修改为一个长而复杂的随机字符串
YOUR_API_TOKEN = "2088"
# --- 数据库操作 ---
def get_db_conn():
conn = sqlite3.connect(DB_FILE, check_same_thread=False)
conn.row_factory = sqlite3.Row
return conn
def init_db():
conn = get_db_conn()
c = conn.cursor()
# 最终的表结构
c.execute('''
CREATE TABLE IF NOT EXISTS received_emails (
id INTEGER PRIMARY KEY AUTOINCREMENT, recipient TEXT, sender TEXT,
subject TEXT, body TEXT, body_type TEXT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
''')
conn.commit()
conn.close()
# --- 核心邮件处理逻辑 ---
def process_email_data(to_address, raw_email_data):
msg = message_from_bytes(raw_email_data)
subject_raw, encoding = decode_header(msg['Subject'])[0]
if isinstance(subject_raw, bytes):
subject = subject_raw.decode(encoding or 'utf-8', errors='ignore')
else:
subject = str(subject_raw)
sender = msg.get('From')
body, body_type = "", "text/plain"
if msg.is_multipart():
html_part, text_part = None, None
for part in msg.walk():
if "text/html" in part.get_content_type(): html_part = part
elif "text/plain" in part.get_content_type(): text_part = part
if html_part:
body = html_part.get_payload(decode=True).decode(html_part.get_content_charset() or 'utf-8', errors='ignore')
body_type = "text/html"
elif text_part:
body = text_part.get_payload(decode=True).decode(text_part.get_content_charset() or 'utf-8', errors='ignore')
body_type = "text/plain"
else:
body = msg.get_payload(decode=True).decode(msg.get_content_charset() or 'utf-8', errors='ignore')
body_type = msg.get_content_type()
try:
conn = get_db_conn()
cursor = conn.cursor()
cursor.execute("INSERT INTO received_emails (recipient, sender, subject, body, body_type) VALUES (?, ?, ?, ?, ?)",
(to_address, sender, subject, body, body_type))
conn.commit()
print(f"邮件已存入数据库: To='{to_address}', Subject='{subject}', Type='{body_type}'")
finally:
if conn: conn.close()
# --- Flask 应用 ---
app = Flask(__name__)
# 接口1:给脚本调用的API(已设置为不删除邮件)
@app.route('/Mail', methods=['GET'])
def get_mail_content():
token = request.args.get('token')
mail_address_to_find = request.args.get('mail')
if not token or token != YOUR_API_TOKEN:
return Response("❌ 无效的 token!", status=401, mimetype="text/plain; charset=utf-8")
if not mail_address_to_find:
return Response("❌ 参数错误:请提供 mail 地址。", status=400, mimetype="text/plain; charset=utf-8")
subject_expected_1, subject_expected_2 = "Verify your email address", "验证您的电子邮件地址"
try:
conn = get_db_conn()
cursor = conn.cursor()
cursor.execute("SELECT id, subject, body, body_type FROM received_emails WHERE recipient = ? ORDER BY timestamp DESC LIMIT 50", (mail_address_to_find,))
messages = cursor.fetchall()
for msg in messages:
if subject_expected_1.lower() in msg['subject'].lower() or subject_expected_2 in msg['subject']:
if re.search(r"\b(\d{6})\b", msg['body']):
# cursor.execute("DELETE FROM received_emails WHERE id = ?", (msg['id'],)) # 已注释,不执行删除
# conn.commit()
print(f"成功匹配并返回邮件 (ID: {msg['id']}) for {mail_address_to_find}")
return Response(msg['body'], mimetype=f"{msg['body_type']}; charset=utf-8")
conn.close()
return Response(f"❌ 未找到 <{mail_address_to_find}> 符合条件的邮件。", status=404, mimetype="text/plain; charset=utf-8")
finally:
if 'conn' in locals() and conn: conn.close()
# 接口2:【新增】邮件列表查看页面
@app.route('/view_emails')
def view_emails():
html = """
<!DOCTYPE html><html><head><title>收件箱</title>
<style>body{font-family: sans-serif; margin: 2em;} table{border-collapse: collapse; width: 100%;}
th, td{border: 1px solid #ddd; padding: 8px; text-align: left;}
tr:nth-child(even){background-color: #f2f2f2;} th{background-color: #4CAF50; color: white;}</style>
</head><body><h2>收件箱 (最新 100 封)</h2>
<table><tr><th>时间</th><th>发件人</th><th>收件人</th><th>主题</th><th>操作</th></tr>
"""
conn = get_db_conn()
cursor = conn.cursor()
cursor.execute("SELECT id, timestamp, sender, recipient, subject FROM received_emails ORDER BY timestamp DESC LIMIT 100")
emails = cursor.fetchall()
conn.close()
for email in emails:
html += f"<tr><td>{escape(email['timestamp'])}</td><td>{escape(email['sender'])}</td><td>{escape(email['recipient'])}</td><td>{escape(email['subject'])}</td>"
html += f'<td><a href="/view_email/{email["id"]}" target="_blank">查看</a></td></tr>'
html += "</table></body></html>"
return Response(html, mimetype="text/html; charset=utf-8")
# 接口3:【新增】单封邮件详情页面
@app.route('/view_email/<int:email_id>')
def view_email_detail(email_id):
conn = get_db_conn()
cursor = conn.cursor()
cursor.execute("SELECT body, body_type FROM received_emails WHERE id = ?", (email_id,))
email = cursor.fetchone()
conn.close()
if email:
return Response(email['body'], mimetype=f"{email['body_type']}; charset=utf-8")
return "邮件未找到", 404
init_db()
4. 创建 smtp_server.py
(邮件接收服务)
使用 nano /opt/mail_api/smtp_server.py
命令创建文件,并将以下全部代码粘贴进去。
Python
import asyncio
from aiosmtpd.controller import Controller
from app import process_email_data
class CustomSMTPHandler:
async def handle_DATA(self, server, session, envelope):
print(f'收到邮件 from <{envelope.mail_from}> to <{envelope.rcpt_tos}>')
for recipient in envelope.rcpt_tos:
# 关键修复:增加了 await
await asyncio.to_thread(process_email_data, recipient, envelope.content)
return '250 OK'
if __name__ == '__main__':
controller = Controller(CustomSMTPHandler(), hostname='0.0.0.0', port=25)
print("SMTP 服务器正在启动,监听 0.0.0.0:25...")
controller.start()
print("SMTP 服务器已启动。按 Ctrl+C 关闭。")
try:
asyncio.get_event_loop().run_forever()
except KeyboardInterrupt:
print("正在关闭 SMTP 服务器...")
finally:
controller.stop()
五、设置为系统服务 (持久化运行)
为了让程序在后台稳定运行并开机自启,我们创建 systemd
服务。
1. 创建 API 服务 (mail-api.service
)
使用 nano /etc/systemd/system/mail-api.service
创建文件,并粘贴以下内容:
Ini, TOML
[Unit]
Description=Gunicorn instance to serve the Mail API
After=network.target
[Service]
User=root
Group=www-data
WorkingDirectory=/opt/mail_api
Environment="PATH=/opt/mail_api/venv/bin"
ExecStart=/opt/mail_api/venv/bin/gunicorn --workers 3 --bind 0.0.0.0:2099 app:app
Restart=always
[Install]
WantedBy=multi-user.target
2. 创建 SMTP 服务 (mail-smtp.service
)
使用 nano /etc/systemd/system/mail-smtp.service
创建文件,并粘贴以下内容:
Ini, TOML
[Unit]
Description=Custom Python SMTP Server for Mail Receiving
After=network.target
[Service]
User=root
WorkingDirectory=/opt/mail_api
ExecStart=/opt/mail_api/venv/bin/python3 smtp_server.py
AmbientCapabilities=CAP_NET_BIND_SERVICE
Restart=always
[Install]
WantedBy=multi-user.target
3. 启动并验证服务
Bash
# 重新加载配置
systemctl daemon-reload
# 同时重启并设置开机自启
systemctl restart mail-api.service mail-smtp.service
systemctl enable mail-api.service mail-smtp.service
# 分别检查两个服务的状态,确保都是绿色的 active (running)
systemctl status mail-api.service
systemctl status mail-smtp.service
六、使用与测试 (Usage and Testing)
1. 发送测试邮件
从您的个人邮箱向 test@mail.sijuly.nyc.com
发送一封邮件。
主题:
Verify your email address
正文:
This is a test, my code is 123456
2. 通过网页界面查看邮件
在您的浏览器中打开: http://89.13.126.152:2099/view_emails
您应该能看到刚刚收到的邮件列表。
3. 通过 API 接口提取邮件
在您本地电脑的终端运行:
Bash
curl "http://89.138.116.122:2099/Mail?token=2088&mail=test@mail.sijuly.nyc.com"
您应该能看到返回的邮件正文(HTML 或纯文本)。
七、常见问题与维护 (FAQ and Maintenance)
如何查看实时日志?
API 服务日志:
journalctl -u mail-api.service -f
SMTP 服务日志:
journalctl -u mail-smtp.service -f
如何更新代码?
用
nano /opt/mail_api/app.py
(或smtp_server.py
) 编辑文件并保存。用
systemctl restart mail-api.service
(或mail-smtp.service
) 重启对应的服务。
数据库文件在哪里?
位于
/opt/mail_api/emails.db
。您可以用sqlite3 /opt/mail_api/emails.db
命令进入并进行SQL
查询。