first commit

This commit is contained in:
huangzhenpc
2025-02-26 18:29:10 +08:00
parent 5d21c9468c
commit a8d1b41381
38 changed files with 2878 additions and 0 deletions

15
app/__init__.py Normal file
View File

@@ -0,0 +1,15 @@
from flask import Flask
from .config import Config
from .models import db
def create_app():
app = Flask(__name__)
app.config.from_object(Config)
# 初始化数据库
db.init_app(app)
with app.app_context():
db.create_all()
return app

10
app/config.py Normal file
View File

@@ -0,0 +1,10 @@
class Config:
# 使用 1Panel MySQL 内部连接
SQLALCHEMY_DATABASE_URI = 'mysql://gitea_HN5jYh:mysql_KbBZTN@1Panel-mysql-vjz9:3306/gitea_2l82ep'
# 如果内部连接不通,可以尝试使用外部连接
# SQLALCHEMY_DATABASE_URI = 'mysql://gitea_HN5jYh:mysql_KbBZTN@rnpanel.586vip.cn:3306/gitea_2l82ep'
SQLALCHEMY_TRACK_MODIFICATIONS = False
# 使用 1Panel Redis 内部连接
REDIS_URL = "redis://1Panel-redis-r3Pz:6379/0"
# 如果内部连接不通,可以尝试使用外部连接
# REDIS_URL = "redis://rnpanel.586vip.cn:6379/0"

11
app/models.py Normal file
View File

@@ -0,0 +1,11 @@
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
class Email(db.Model):
id = db.Column(db.Integer, primary_key=True)
subject = db.Column(db.String(255))
sender = db.Column(db.String(255))
recipient = db.Column(db.String(255))
body = db.Column(db.Text)
timestamp = db.Column(db.DateTime, server_default=db.func.now())

194
app/utils.py Normal file
View File

@@ -0,0 +1,194 @@
import smtplib
import os
import json
import logging
from email.parser import BytesParser
from email.policy import default
from datetime import datetime
from .models import db, Email
import redis
import smtpd
import asyncore
import base64
from .config import Config
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('smtp_server')
# 初始化 Redis 客户端
redis_client = redis.from_url(Config.REDIS_URL)
def receive_email():
# 这里实现邮件接收逻辑
# 假设我们从某个 SMTP 服务器接收邮件
# 解析邮件并存储到 Redis
# 示例:
raw_email = b'...' # 这里应该是接收到的原始邮件内容
email = BytesParser(policy=default).parsebytes(raw_email)
email_data = {
'subject': email['subject'],
'sender': email['from'],
'recipient': email['to'],
'body': email.get_body(preferencelist=('plain')).get_content()
}
# 将邮件信息存储到 Redis
redis_client.hmset(f'email:{email_data['subject']}', email_data)
class CustomSMTPServer(smtpd.SMTPServer):
def process_message(self, peer, mailfrom, rcpttos, data):
try:
# 记录接收到的邮件基本信息
logger.info(f'Received mail from {mailfrom} to {rcpttos}')
# 验证收件人域名
for rcpt in rcpttos:
if not rcpt.endswith('@nosqli.com'):
logger.warning(f'Rejected mail to {rcpt}: invalid domain')
continue
# 解析邮件
email = BytesParser(policy=default).parsebytes(data)
# 获取邮件正文
body = self._get_email_body(email)
# 处理附件
attachments = self._process_attachments(email)
# 构建邮件数据
timestamp = datetime.now().isoformat()
message_id = email.get('Message-ID', f'<{timestamp}@nosqli.com>')
email_data = {
'message_id': message_id,
'subject': email.get('subject', ''),
'sender': mailfrom,
'recipients': json.dumps(rcpttos),
'body': body,
'timestamp': timestamp,
'attachments': json.dumps(attachments),
'headers': json.dumps(dict(email.items())),
'peer': json.dumps(peer)
}
# 存储邮件
self._store_email(email_data)
logger.info(f'Successfully processed mail: {message_id}')
except Exception as e:
logger.error(f'Error processing email: {str(e)}', exc_info=True)
raise
def _get_email_body(self, email):
"""提取邮件正文"""
try:
if email.is_multipart():
for part in email.walk():
if part.get_content_type() == "text/plain":
return part.get_payload(decode=True).decode()
else:
return email.get_payload(decode=True).decode()
return ""
except Exception as e:
logger.error(f'Error extracting email body: {str(e)}')
return ""
def _process_attachments(self, email):
"""处理邮件附件"""
attachments = []
try:
if email.is_multipart():
for part in email.walk():
if part.get_content_maintype() == 'multipart':
continue
if part.get('Content-Disposition') is None:
continue
filename = part.get_filename()
if filename:
attachment_data = part.get_payload(decode=True)
attachments.append({
'filename': filename,
'content': base64.b64encode(attachment_data).decode(),
'content_type': part.get_content_type(),
'size': len(attachment_data)
})
except Exception as e:
logger.error(f'Error processing attachments: {str(e)}')
return attachments
def _store_email(self, email_data):
"""存储邮件到 Redis"""
try:
# 使用 message_id 作为主键
email_key = f'email:{email_data["message_id"]}'
redis_client.hmset(email_key, email_data)
# 为每个收件人创建索引
recipients = json.loads(email_data['recipients'])
for recipient in recipients:
recipient_key = f'recipient:{recipient}'
redis_client.lpush(recipient_key, email_key)
# 创建时间索引
time_key = f'time:{email_data["timestamp"]}'
redis_client.set(time_key, email_key)
# 设置过期时间可选这里设置为30天
redis_client.expire(email_key, 30 * 24 * 60 * 60)
except Exception as e:
logger.error(f'Error storing email: {str(e)}')
raise
def start_smtp_server(host='0.0.0.0', port=25):
"""启动 SMTP 服务器"""
try:
logger.info(f'Starting SMTP server on {host}:{port}')
server = CustomSMTPServer((host, port), None)
asyncore.loop()
except Exception as e:
logger.error(f'Error starting SMTP server: {str(e)}')
raise
def get_emails_by_recipient(recipient, limit=10):
"""获取指定收件人的最新邮件"""
try:
recipient_key = f'recipient:{recipient}'
email_keys = redis_client.lrange(recipient_key, 0, limit - 1)
emails = []
for key in email_keys:
email_data = redis_client.hgetall(key.decode())
if email_data:
# 转换数据为字符串
email_data = {k.decode(): v.decode() for k, v in email_data.items()}
emails.append(email_data)
return emails
except Exception as e:
print(f'Error fetching emails: {e}')
return []
def get_attachment(email_key, attachment_index):
"""获取指定邮件的附件"""
try:
email_data = redis_client.hgetall(email_key)
if email_data:
attachments = json.loads(email_data[b'attachments'].decode())
if 0 <= attachment_index < len(attachments):
return attachments[attachment_index]
return None
except Exception as e:
print(f'Error fetching attachment: {e}')
return None

74
old/app/__init__.py Normal file
View File

@@ -0,0 +1,74 @@
import os
import logging
from flask import Flask
from flask_cors import CORS
# 修改相对导入为绝对导入
import sys
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from config import active_config
def setup_logging(app):
"""设置日志"""
log_level = getattr(logging, active_config.LOG_LEVEL.upper(), logging.INFO)
# 确保日志目录存在
log_dir = os.path.dirname(active_config.LOG_FILE)
if not os.path.exists(log_dir):
os.makedirs(log_dir)
# 配置日志
logging.basicConfig(
level=log_level,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(active_config.LOG_FILE),
logging.StreamHandler()
]
)
app.logger.setLevel(log_level)
return app
def create_app(config=None):
"""创建并配置Flask应用"""
app = Flask(__name__)
# 加载配置
app.config.from_object(active_config)
# 如果提供了自定义配置,加载它
if config:
app.config.from_object(config)
# 允许跨域请求
CORS(app)
# 设置日志
app = setup_logging(app)
# 确保存储邮件的目录存在
os.makedirs(active_config.MAIL_STORAGE_PATH, exist_ok=True)
# 初始化数据库
from .models import init_db
init_db()
# 注册蓝图
from .api import api_bp
app.register_blueprint(api_bp)
# 首页路由
@app.route("/")
def index():
return {
"name": "Email System",
"version": "1.0.0",
"status": "running"
}
app.logger.info('应用初始化完成')
return app

Binary file not shown.

25
old/app/api/__init__.py Normal file
View File

@@ -0,0 +1,25 @@
# API模块初始化文件
from flask import Blueprint
import logging
# 创建API蓝图
api_bp = Blueprint('api', __name__, url_prefix='/api')
# 注册默认路由
@api_bp.route('/')
def index():
return {
'name': 'Email System API',
'version': '1.0.0',
'status': 'running'
}
# 导入并合并所有API路由
# 为避免可能的文件读取问题改为从routes.py模块中导入所有路由定义
try:
from .routes import *
# 导入解码邮件路由模块
from .decoded_email_routes import *
except Exception as e:
logging.error(f"导入API路由时出错: {str(e)}")
raise

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

View File

@@ -0,0 +1,708 @@
import base64
import re
import os
import email
from email import policy
from datetime import datetime, timedelta
import psutil
import time
from flask import jsonify, request, current_app
from sqlalchemy import or_, func, desc
from . import api_bp
from ..models import Email, Domain, Mailbox, get_session
from ..services import get_mail_store
# 调试接口 - 检查邮件接收状态
@api_bp.route('/debug_email', methods=['GET'])
def debug_email():
"""
调试接口:检查某个邮箱的邮件状态并提供详细信息
查询参数:
- email: 邮箱地址 (例如: newsadd1test@nosqli.com)
"""
try:
# 获取查询参数
email_address = request.args.get('email')
# 验证邮箱地址是否有效
if not email_address or '@' not in email_address:
return jsonify({
'success': False,
'error': '无效的邮箱地址',
'message': '请提供有效的邮箱地址格式为user@domain.com'
}), 400
# 解析邮箱地址
username, domain_name = email_address.split('@', 1)
result = {
'email_address': email_address,
'username': username,
'domain': domain_name,
'system_info': {},
'logs': [],
'files': []
}
# 查询数据库
db = get_session()
try:
# 查找域名
domain = db.query(Domain).filter_by(name=domain_name).first()
if domain:
result['system_info']['domain'] = {
'id': domain.id,
'name': domain.name,
'active': domain.active,
'created_at': str(domain.created_at) if hasattr(domain, 'created_at') else None
}
# 查找邮箱
mailbox = db.query(Mailbox).filter_by(
domain_id=domain.id,
address=username
).first()
if mailbox:
result['system_info']['mailbox'] = {
'id': mailbox.id,
'address': mailbox.address,
'full_address': f"{mailbox.address}@{domain_name}",
'created_at': str(mailbox.created_at) if hasattr(mailbox, 'created_at') else None
}
# 获取邮件
emails = db.query(Email).filter_by(mailbox_id=mailbox.id).all()
result['system_info']['emails_count'] = len(emails)
result['system_info']['emails'] = []
for email_obj in emails:
email_info = {
'id': email_obj.id,
'subject': email_obj.subject,
'sender': email_obj.sender,
'received_at': str(email_obj.received_at),
'verification_code': email_obj.verification_code if hasattr(email_obj, 'verification_code') else None
}
result['system_info']['emails'].append(email_info)
else:
result['system_info']['mailbox'] = "未找到邮箱记录"
else:
result['system_info']['domain'] = "未找到域名记录"
# 查找文件
email_data_dir = current_app.config.get('MAIL_STORAGE_PATH', 'email_data')
emails_dir = os.path.join(email_data_dir, 'emails')
if os.path.exists(emails_dir):
for file_name in os.listdir(emails_dir):
if file_name.endswith('.eml'):
file_path = os.path.join(emails_dir, file_name)
file_info = {
'name': file_name,
'path': file_path,
'size': os.path.getsize(file_path),
'modified': datetime.fromtimestamp(os.path.getmtime(file_path)).isoformat()
}
# 尝试读取文件内容并检查是否包含收件人地址
try:
with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
content = f.read(10000) # 只读取前10000个字符用于检查
if email_address.lower() in content.lower():
file_info['contains_address'] = True
result['files'].append(file_info)
except Exception as e:
file_info['error'] = str(e)
result['files'].append(file_info)
# 检查日志
log_file = current_app.config.get('LOG_FILE', os.path.join('logs', 'email_system.log'))
if os.path.exists(log_file):
try:
with open(log_file, 'r', encoding='utf-8', errors='replace') as f:
# 从日志尾部读取最后200行
lines = f.readlines()[-200:]
for line in lines:
if email_address.lower() in line.lower():
result['logs'].append(line.strip())
except Exception as e:
result['logs'] = [f"读取日志出错: {str(e)}"]
return jsonify({
'success': True,
'debug_info': result
}), 200
finally:
db.close()
except Exception as e:
current_app.logger.error(f"调试邮件出错: {str(e)}")
return jsonify({
'success': False,
'error': '服务器错误',
'message': str(e)
}), 500
# 创建邮箱接口
@api_bp.route('/add_mailbox', methods=['POST', 'GET'])
def add_mailbox():
"""
创建新邮箱,如果域名不存在则自动创建
查询参数GET方式或表单参数POST方式:
- email: 邮箱地址 (例如: testaa@nosqli.com)
- description: 邮箱描述 (可选)
"""
try:
# 获取参数
if request.method == 'POST':
data = request.json or {}
email_address = data.get('email')
description = data.get('description', '')
else: # GET方式
email_address = request.args.get('email')
description = request.args.get('description', '')
# 验证邮箱地址
if not email_address or '@' not in email_address:
return jsonify({
'success': False,
'error': '无效的邮箱地址',
'message': '请提供有效的邮箱地址格式为user@domain.com'
}), 400
# 解析邮箱地址
username, domain_name = email_address.split('@', 1)
# 创建或查找域名和邮箱
db = get_session()
try:
# 查找域名
domain = db.query(Domain).filter_by(name=domain_name).first()
# 如果域名不存在,创建域名
if not domain:
current_app.logger.info(f"域名 {domain_name} 不存在,开始创建")
domain = Domain(
name=domain_name,
description=f"自动创建的域名 {domain_name}",
active=True
)
db.add(domain)
db.commit()
current_app.logger.info(f"域名 {domain_name} 创建成功ID: {domain.id}")
# 查询邮箱是否已存在
mailbox = db.query(Mailbox).filter_by(
domain_id=domain.id,
address=username
).first()
# 如果邮箱已存在,返回已存在信息
if mailbox:
return jsonify({
'success': True,
'message': f'邮箱 {email_address} 已存在',
'mailbox': {
'id': mailbox.id,
'address': mailbox.address,
'domain_id': mailbox.domain_id,
'full_address': f"{mailbox.address}@{domain_name}",
'description': mailbox.description
}
}), 200
# 创建邮箱
mailbox = Mailbox(
domain_id=domain.id,
address=username,
description=description or f"自动创建的邮箱 {email_address}"
)
db.add(mailbox)
db.commit()
# 返回成功信息
return jsonify({
'success': True,
'message': f'邮箱 {email_address} 创建成功',
'mailbox': {
'id': mailbox.id,
'address': mailbox.address,
'domain_id': mailbox.domain_id,
'full_address': f"{mailbox.address}@{domain_name}",
'description': mailbox.description
}
}), 201
finally:
db.close()
except Exception as e:
current_app.logger.error(f"创建邮箱出错: {str(e)}")
return jsonify({
'success': False,
'error': '服务器错误',
'message': str(e)
}), 500
# 简化的URL路径直接通过邮箱地址获取邮件
@api_bp.route('/email', methods=['GET'])
def get_email_by_address():
"""
通过邮箱地址获取邮件的简化URL
等同于 /decoded_emails?email={email_address}&latest=0
查询参数:
- email: 邮箱地址 (必填)
- latest: 是否只返回最新的邮件 (1表示是0表示否默认0)
"""
# 重用已有的解码邮件接口
return get_decoded_emails()
@api_bp.route('/decoded_emails', methods=['GET'])
def get_decoded_emails():
"""
获取指定邮箱地址的所有邮件,并返回解码后的内容
查询参数:
- email: 邮箱地址 (例如: testaa@nosqli.com)
- latest: 是否只返回最新的邮件 (1表示是0表示否默认0)
- limit: 返回邮件数量 (默认10)
- offset: 查询起始位置 (默认0)
"""
try:
# 获取查询参数
email_address = request.args.get('email')
latest = request.args.get('latest', '0') == '1'
limit = int(request.args.get('limit', 10))
offset = int(request.args.get('offset', 0))
# 验证邮箱地址是否有效
if not email_address or '@' not in email_address:
return jsonify({
'success': False,
'error': '无效的邮箱地址',
'message': '请提供有效的邮箱地址格式为user@domain.com'
}), 400
# 解析邮箱地址
username, domain_name = email_address.split('@', 1)
# 查询数据库
db = get_session()
try:
# 查找域名
domain = db.query(Domain).filter_by(name=domain_name).first()
if not domain:
return jsonify({
'success': False,
'error': '域名不存在',
'message': f'域名 {domain_name} 不存在'
}), 404
# 查找邮箱
mailbox = db.query(Mailbox).filter_by(
domain_id=domain.id,
address=username
).first()
if not mailbox:
return jsonify({
'success': False,
'error': '邮箱不存在',
'message': f'邮箱 {email_address} 不存在'
}), 404
# 获取邮件
query = db.query(Email).filter_by(mailbox_id=mailbox.id)
# 按接收时间排序,最新的在前
query = query.order_by(Email.received_at.desc())
# 如果只要最新的一封
if latest:
emails = query.limit(1).all()
else:
emails = query.limit(limit).offset(offset).all()
# 处理结果
result_emails = []
for email_obj in emails:
# 获取原始邮件文件路径
email_file_path = os.path.join(
current_app.config.get('MAIL_STORAGE_PATH', 'email_data'),
'emails',
f'email_{email_obj.id}.eml'
)
# 解码邮件内容
decoded_email = decode_email(email_obj, email_file_path)
result_emails.append(decoded_email)
# 返回结果
return jsonify({
'success': True,
'email_address': email_address,
'total_emails': query.count(),
'emails': result_emails
}), 200
finally:
db.close()
except Exception as e:
current_app.logger.error(f"获取解码邮件出错: {str(e)}")
return jsonify({
'success': False,
'error': '服务器错误',
'message': str(e)
}), 500
@api_bp.route('/decoded_email/<int:email_id>', methods=['GET'])
def get_decoded_email_by_id(email_id):
"""获取指定ID的解码邮件内容"""
try:
db = get_session()
try:
# 获取邮件对象
email_obj = db.query(Email).filter_by(id=email_id).first()
if not email_obj:
return jsonify({
'success': False,
'error': '邮件不存在',
'message': f'ID为{email_id}的邮件不存在'
}), 404
# 获取原始邮件文件路径
email_file_path = os.path.join(
current_app.config.get('MAIL_STORAGE_PATH', 'email_data'),
'emails',
f'email_{email_obj.id}.eml'
)
# 解码邮件内容
decoded_email = decode_email(email_obj, email_file_path)
# 返回结果
return jsonify({
'success': True,
'email': decoded_email
}), 200
finally:
db.close()
except Exception as e:
current_app.logger.error(f"获取解码邮件出错: {str(e)}")
return jsonify({
'success': False,
'error': '服务器错误',
'message': str(e)
}), 500
def decode_email(email_obj, email_file_path):
"""解析并解码邮件内容"""
# 创建基本邮件信息
result = {
'id': email_obj.id,
'subject': email_obj.subject,
'sender': email_obj.sender,
'recipients': email_obj.recipients,
'received_at': email_obj.received_at.isoformat() if email_obj.received_at else None,
'read': email_obj.read,
'has_attachments': len(email_obj.attachments) > 0 if hasattr(email_obj, 'attachments') else False
}
# 从数据库中直接获取验证码
if hasattr(email_obj, 'verification_code') and email_obj.verification_code:
result['verification_code'] = email_obj.verification_code
if hasattr(email_obj, 'verification_link') and email_obj.verification_link:
result['verification_link'] = email_obj.verification_link
# 如果邮件对象有文本内容或HTML内容直接使用
if hasattr(email_obj, 'body_text') and email_obj.body_text:
result['body_text'] = email_obj.body_text
if hasattr(email_obj, 'body_html') and email_obj.body_html:
result['body_html'] = email_obj.body_html
# 如果有原始邮件文件,尝试解析
if os.path.exists(email_file_path):
try:
# 解析.eml文件
with open(email_file_path, 'r', encoding='utf-8', errors='replace') as f:
msg = email.message_from_file(f, policy=policy.default)
# 如果没有从数据库获取到内容,尝试从文件解析
if 'body_text' not in result or 'body_html' not in result:
body_text = ""
body_html = ""
# 处理多部分邮件
if msg.is_multipart():
for part in msg.iter_parts():
content_type = part.get_content_type()
if content_type == "text/plain":
try:
body_text = part.get_content()
except Exception:
payload = part.get_payload(decode=True)
if payload:
charset = part.get_content_charset() or 'utf-8'
try:
body_text = payload.decode(charset, errors='replace')
except:
body_text = payload.decode('utf-8', errors='replace')
elif content_type == "text/html":
try:
body_html = part.get_content()
except Exception:
payload = part.get_payload(decode=True)
if payload:
charset = part.get_content_charset() or 'utf-8'
try:
body_html = payload.decode(charset, errors='replace')
except:
body_html = payload.decode('utf-8', errors='replace')
else:
# 处理单部分邮件
content_type = msg.get_content_type()
try:
if content_type == "text/plain":
body_text = msg.get_content()
elif content_type == "text/html":
body_html = msg.get_content()
except Exception:
payload = msg.get_payload(decode=True)
if payload:
charset = msg.get_content_charset() or 'utf-8'
try:
decoded = payload.decode(charset, errors='replace')
if content_type == "text/plain":
body_text = decoded
elif content_type == "text/html":
body_html = decoded
except:
pass
# 如果找到了内容,添加到结果中
if body_text and 'body_text' not in result:
result['body_text'] = body_text
if body_html and 'body_html' not in result:
result['body_html'] = body_html
# 如果仍然没有提取到验证码,尝试从内容中提取
if 'verification_code' not in result:
verification_code = extract_verification_code(result.get('body_text', ''), result.get('body_html', ''))
if verification_code:
result['verification_code'] = verification_code
except Exception as e:
current_app.logger.error(f"解析邮件文件出错: {str(e)}")
return result
def extract_verification_code(body_text, body_html):
"""从邮件内容中提取验证码"""
# 首先尝试从HTML中提取
if body_html:
# 常用的验证码模式
patterns = [
r'letter-spacing:\s*\d+px[^>]*>([^<]+)<', # 特殊样式的验证码
r'<div[^>]*>(\d{6})</div>', # 6位数字验证码在div中
r'验证码[:]\s*([A-Z0-9]{4,8})', # 中文标记的验证码
r'code[^\d]+(\d{4,8})', # 英文标记的验证码
r'\b([A-Z0-9]{6})\b' # 6位大写字母或数字
]
for pattern in patterns:
matches = re.findall(pattern, body_html)
if matches:
return matches[0].strip()
# 如果HTML中没找到尝试从纯文本中提取
if body_text:
patterns = [
r'验证码[:]\s*([A-Z0-9]{4,8})', # 中文格式
r'code[^\d]+(\d{4,8})', # 英文格式
r'\b(\d{6})\b' # 6位数字
]
for pattern in patterns:
matches = re.findall(pattern, body_text)
if matches:
return matches[0].strip()
return None
# 系统诊断接口
@api_bp.route('/system_check', methods=['GET'])
def system_check():
"""
系统诊断接口:检查邮件系统各组件状态
"""
try:
result = {
'timestamp': datetime.now().isoformat(),
'system_status': 'normal',
'components': {},
'recent_activity': {},
'mailboxes': [],
'storage': {}
}
# 检查系统资源
try:
result['components']['system'] = {
'cpu_percent': psutil.cpu_percent(),
'memory_percent': psutil.virtual_memory().percent,
'disk_usage': psutil.disk_usage('/').percent
}
except Exception as e:
result['components']['system'] = {'error': str(e)}
# 检查数据库状态
db = get_session()
try:
# 获取域名数量
domain_count = db.query(Domain).count()
# 获取邮箱数量
mailbox_count = db.query(Mailbox).count()
# 获取邮件数量
email_count = db.query(Email).count()
# 获取最新邮件
latest_emails = db.query(Email).order_by(Email.received_at.desc()).limit(5).all()
result['components']['database'] = {
'status': 'connected',
'domain_count': domain_count,
'mailbox_count': mailbox_count,
'email_count': email_count
}
# 最近活动
result['recent_activity']['latest_emails'] = [
{
'id': email.id,
'subject': email.subject,
'sender': email.sender,
'received_at': email.received_at.isoformat() if email.received_at else None
} for email in latest_emails
]
# 获取所有活跃邮箱
active_mailboxes = db.query(Mailbox).order_by(Mailbox.id).limit(10).all()
result['mailboxes'] = [
{
'id': mb.id,
'address': mb.address,
'domain_id': mb.domain_id,
'full_address': f"{mb.address}@{mb.domain.name}" if hasattr(mb, 'domain') and mb.domain else f"{mb.address}@unknown",
'email_count': db.query(Email).filter_by(mailbox_id=mb.id).count()
} for mb in active_mailboxes
]
except Exception as e:
result['components']['database'] = {'status': 'error', 'error': str(e)}
finally:
db.close()
# 检查存储状态
email_data_dir = current_app.config.get('MAIL_STORAGE_PATH', 'email_data')
try:
emails_dir = os.path.join(email_data_dir, 'emails')
attachments_dir = os.path.join(email_data_dir, 'attachments')
# 检查目录是否存在
emails_dir_exists = os.path.exists(emails_dir)
attachments_dir_exists = os.path.exists(attachments_dir)
# 计算文件数量和大小
email_files_count = 0
email_files_size = 0
if emails_dir_exists:
for file_name in os.listdir(emails_dir):
if file_name.endswith('.eml'):
email_files_count += 1
email_files_size += os.path.getsize(os.path.join(emails_dir, file_name))
attachment_files_count = 0
attachment_files_size = 0
if attachments_dir_exists:
for file_name in os.listdir(attachments_dir):
attachment_files_count += 1
attachment_files_size += os.path.getsize(os.path.join(attachments_dir, file_name))
result['storage'] = {
'emails_dir': {
'exists': emails_dir_exists,
'path': emails_dir,
'file_count': email_files_count,
'size_bytes': email_files_size,
'size_mb': round(email_files_size / (1024 * 1024), 2) if email_files_size > 0 else 0
},
'attachments_dir': {
'exists': attachments_dir_exists,
'path': attachments_dir,
'file_count': attachment_files_count,
'size_bytes': attachment_files_size,
'size_mb': round(attachment_files_size / (1024 * 1024), 2) if attachment_files_size > 0 else 0
}
}
# 检查最近的邮件文件
if emails_dir_exists and email_files_count > 0:
files = [(os.path.getmtime(os.path.join(emails_dir, f)), f)
for f in os.listdir(emails_dir) if f.endswith('.eml')]
files.sort(reverse=True)
result['recent_activity']['latest_files'] = [
{
'filename': f,
'modified': datetime.fromtimestamp(t).isoformat(),
'age_seconds': int(time.time() - t)
} for t, f in files[:5]
]
except Exception as e:
result['storage'] = {'error': str(e)}
# 整体状态评估
if ('database' in result['components'] and result['components']['database'].get('status') != 'connected'):
result['system_status'] = 'warning'
if not emails_dir_exists or not attachments_dir_exists:
result['system_status'] = 'warning'
return jsonify({
'success': True,
'status': result['system_status'],
'diagnostics': result
}), 200
except Exception as e:
current_app.logger.error(f"系统诊断出错: {str(e)}")
return jsonify({
'success': False,
'error': '系统诊断失败',
'message': str(e)
}), 500

Binary file not shown.

376
old/app/api/email_routes.py Normal file
View File

@@ -0,0 +1,376 @@
from flask import request, jsonify, current_app, send_file
from io import BytesIO
import time
import os
import logging
from . import api_bp
from ..models import get_session, Email, Mailbox, Domain
from ..utils import email_parser
from ..config import config
# 获取邮箱的所有邮件
@api_bp.route('/mailboxes/<int:mailbox_id>/emails', methods=['GET'])
def get_mailbox_emails(mailbox_id):
"""获取指定邮箱的所有邮件"""
try:
page = int(request.args.get('page', 1))
limit = int(request.args.get('limit', 50))
unread_only = request.args.get('unread_only', 'false').lower() == 'true'
offset = (page - 1) * limit
db = get_session()
try:
# 检查邮箱是否存在
mailbox = db.query(Mailbox).filter_by(id=mailbox_id).first()
if not mailbox:
return jsonify({'error': '邮箱不存在'}), 404
# 查询邮件
query = db.query(Email).filter(Email.mailbox_id == mailbox_id)
if unread_only:
query = query.filter(Email.read == False)
# 获取总数
total = query.count()
# 分页获取邮件
emails = query.order_by(Email.received_at.desc()) \
.limit(limit) \
.offset(offset) \
.all()
# 返回结果
result = {
'success': True,
'total': total,
'page': page,
'limit': limit,
'emails': [email.to_dict() for email in emails]
}
return jsonify(result), 200
finally:
db.close()
except Exception as e:
current_app.logger.error(f"获取邮件列表出错: {str(e)}")
return jsonify({'success': False, 'error': '获取邮件列表失败', 'details': str(e)}), 500
# 获取特定邮件详情
@api_bp.route('/emails/<int:email_id>', methods=['GET'])
def get_email(email_id):
"""
获取单个邮件的详细信息
"""
try:
email_id = int(email_id)
session = get_session()
email = session.query(Email).filter(Email.id == email_id).first()
if not email:
return jsonify({
'success': False,
'error': f'未找到ID为{email_id}的邮件'
}), 404
# 获取邮件正文内容
body_text = None
body_html = None
try:
# 尝试从文件中读取邮件内容
if email.id:
email_path = os.path.join(config.DATA_DIR, 'emails', f'email_{email.id}.eml')
if os.path.exists(email_path):
logging.info(f"从文件读取邮件内容: {email_path}")
with open(email_path, 'r', encoding='utf-8', errors='ignore') as f:
try:
raw_email = f.read()
msg = email_parser.parsestr(raw_email)
if msg.is_multipart():
# 处理多部分邮件
for part in msg.walk():
content_type = part.get_content_type()
content_disposition = str(part.get("Content-Disposition"))
# 跳过附件
if "attachment" in content_disposition:
continue
# 处理文本内容
if content_type == "text/plain":
payload = part.get_payload(decode=True)
charset = part.get_content_charset() or 'utf-8'
try:
body_text = payload.decode(charset, errors='replace')
except Exception as e:
logging.error(f"解码纯文本内容失败: {e}")
body_text = payload.decode('utf-8', errors='replace')
# 处理HTML内容
elif content_type == "text/html":
payload = part.get_payload(decode=True)
charset = part.get_content_charset() or 'utf-8'
try:
body_html = payload.decode(charset, errors='replace')
except Exception as e:
logging.error(f"解码HTML内容失败: {e}")
body_html = payload.decode('utf-8', errors='replace')
else:
# 处理单部分邮件
content_type = msg.get_content_type()
payload = msg.get_payload(decode=True)
charset = msg.get_content_charset() or 'utf-8'
try:
decoded_content = payload.decode(charset, errors='replace')
except Exception as e:
logging.error(f"解码内容失败: {e}")
decoded_content = payload.decode('utf-8', errors='replace')
if content_type == "text/plain":
body_text = decoded_content
elif content_type == "text/html":
body_html = decoded_content
except Exception as e:
logging.error(f"解析邮件文件失败: {e}")
except Exception as e:
logging.error(f"读取邮件内容时出错: {e}")
# 如果文件读取失败,使用数据库中的内容
if body_text is None:
body_text = email.body_text
if body_html is None:
body_html = email.body_html
logging.info(f"邮件ID={email_id} 正文长度: text={len(body_text or '')}字节, html={len(body_html or '')}字节")
# 返回邮件信息,包括正文内容
return jsonify({
'success': True,
'email': {
'id': email.id,
'subject': email.subject,
'sender': email.sender,
'recipients': email.recipients,
'received_at': email.received_at.isoformat(),
'verification_code': email.verification_code,
'verification_link': email.verification_link,
'body_text': body_text,
'body_html': body_html
}
})
except Exception as e:
logging.error(f"获取邮件时出错: {str(e)}")
return jsonify({
'success': False,
'error': f'获取邮件时发生错误: {str(e)}'
}), 500
finally:
session.close()
# 删除邮件
@api_bp.route('/emails/<int:email_id>', methods=['DELETE'])
def delete_email(email_id):
"""删除特定邮件"""
try:
db = get_session()
try:
email = db.query(Email).filter_by(id=email_id).first()
if not email:
return jsonify({'error': '邮件不存在'}), 404
db.delete(email)
db.commit()
return jsonify({'message': '邮件已删除'}), 200
except Exception as e:
db.rollback()
raise
finally:
db.close()
except Exception as e:
current_app.logger.error(f"删除邮件出错: {str(e)}")
return jsonify({'error': '删除邮件失败', 'details': str(e)}), 500
# 下载附件
@api_bp.route('/attachments/<int:attachment_id>', methods=['GET'])
def download_attachment(attachment_id):
"""下载特定的附件"""
try:
from ..models import Attachment
db = get_session()
try:
attachment = db.query(Attachment).filter_by(id=attachment_id).first()
if not attachment:
return jsonify({'error': '附件不存在'}), 404
# 获取附件内容
content = attachment.get_content()
if not content:
return jsonify({'error': '附件内容不可用'}), 404
# 创建内存文件对象
file_obj = BytesIO(content)
# 返回文件下载响应
return send_file(
file_obj,
mimetype=attachment.content_type,
as_attachment=True,
download_name=attachment.filename
)
finally:
db.close()
except Exception as e:
current_app.logger.error(f"下载附件出错: {str(e)}")
return jsonify({'error': '下载附件失败', 'details': str(e)}), 500
# 获取最新邮件 (轮询API)
@api_bp.route('/mailboxes/<int:mailbox_id>/poll', methods=['GET'])
def poll_new_emails(mailbox_id):
"""轮询指定邮箱的新邮件"""
try:
# 获取上次检查时间
last_check = request.args.get('last_check')
if last_check:
try:
last_check_time = float(last_check)
except ValueError:
return jsonify({'error': '无效的last_check参数'}), 400
else:
last_check_time = time.time() - 300 # 默认检查最近5分钟
db = get_session()
try:
# 检查邮箱是否存在
mailbox = db.query(Mailbox).filter_by(id=mailbox_id).first()
if not mailbox:
return jsonify({'error': '邮箱不存在'}), 404
# 查询新邮件
new_emails = db.query(Email).filter(
Email.mailbox_id == mailbox_id,
Email.received_at >= time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(last_check_time))
).order_by(Email.received_at.desc()).all()
# 返回结果
result = {
'mailbox_id': mailbox_id,
'count': len(new_emails),
'emails': [email.to_dict() for email in new_emails],
'timestamp': time.time()
}
return jsonify(result), 200
finally:
db.close()
except Exception as e:
current_app.logger.error(f"轮询新邮件出错: {str(e)}")
return jsonify({'error': '轮询新邮件失败', 'details': str(e)}), 500
# 通过邮箱地址获取最新邮件
@api_bp.route('/emails/by-address', methods=['GET'])
def get_emails_by_address():
"""
通过邮箱地址获取最新邮件
参数:
email_address: 完整邮箱地址 (例如: user@example.com)
limit: 返回的邮件数量 (默认: 10)
since: 从指定时间戳后获取邮件 (可选)
unread_only: 是否只返回未读邮件 (默认: false)
返回:
最新的邮件列表
"""
try:
email_address = request.args.get('email_address')
if not email_address or '@' not in email_address:
return jsonify({'success': False, 'error': '无效的邮箱地址'}), 400
limit = int(request.args.get('limit', 10))
unread_only = request.args.get('unread_only', 'false').lower() == 'true'
since = request.args.get('since')
# 解析邮箱地址
try:
username, domain_name = email_address.split('@', 1)
except ValueError:
return jsonify({
'success': False,
'error': '邮箱地址格式无效'
}), 400
db = get_session()
try:
# 查找域名
domain = db.query(Domain).filter_by(name=domain_name, active=True).first()
if not domain:
return jsonify({
'success': False,
'error': f'域名 {domain_name} 不存在或未激活'
}), 404
# 查找邮箱
mailbox = db.query(Mailbox).filter_by(address=username, domain_id=domain.id).first()
if not mailbox:
# 自动创建邮箱 - 批量注册场景
mailbox = Mailbox(
address=username,
domain_id=domain.id,
description=f"自动创建 ({email_address})",
active=True
)
db.add(mailbox)
db.flush() # 获取新创建邮箱的ID
logging.info(f"自动创建邮箱: {email_address}, ID={mailbox.id}")
# 查询邮件
query = db.query(Email).filter(Email.mailbox_id == mailbox.id)
if unread_only:
query = query.filter(Email.read == False)
if since:
try:
since_time = float(since)
query = query.filter(
Email.received_at >= time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(since_time))
)
except ValueError:
logging.warning(f"无效的since参数: {since}")
# 获取最新的邮件
emails = query.order_by(Email.received_at.desc()).limit(limit).all()
# 获取总数
total = query.count()
# 提交数据库变更
db.commit()
# 返回结果
return jsonify({
'success': True,
'email_address': email_address,
'mailbox_id': mailbox.id,
'total': total,
'count': len(emails),
'emails': [email.to_dict() for email in emails],
'timestamp': time.time()
}), 200
except Exception as e:
db.rollback()
raise
finally:
db.close()
except Exception as e:
current_app.logger.error(f"获取邮件时出错: {str(e)}")
return jsonify({'success': False, 'error': '获取邮件失败'}), 500

View File

@@ -0,0 +1,206 @@
from flask import request, jsonify, current_app
from sqlalchemy.exc import IntegrityError
import random
import string
from . import api_bp
from ..models import get_session, Domain, Mailbox
# 获取所有邮箱
@api_bp.route('/mailboxes', methods=['GET'])
def get_mailboxes():
"""获取所有邮箱列表"""
try:
page = int(request.args.get('page', 1))
limit = int(request.args.get('limit', 50))
offset = (page - 1) * limit
db = get_session()
try:
# 查询总数
total = db.query(Mailbox).count()
# 获取分页数据
mailboxes = db.query(Mailbox).order_by(Mailbox.created_at.desc()) \
.limit(limit) \
.offset(offset) \
.all()
# 转换为字典列表
result = {
'total': total,
'page': page,
'limit': limit,
'mailboxes': [mailbox.to_dict() for mailbox in mailboxes]
}
return jsonify(result), 200
finally:
db.close()
except Exception as e:
current_app.logger.error(f"获取邮箱列表出错: {str(e)}")
return jsonify({'error': '获取邮箱列表失败', 'details': str(e)}), 500
# 创建邮箱
@api_bp.route('/mailboxes', methods=['POST'])
def create_mailbox():
"""创建新邮箱"""
try:
data = request.json
# 验证必要参数
if not data or 'domain_id' not in data:
return jsonify({'error': '缺少必要参数'}), 400
db = get_session()
try:
# 查询域名是否存在
domain = db.query(Domain).filter_by(id=data['domain_id'], active=True).first()
if not domain:
return jsonify({'error': '指定的域名不存在或未激活'}), 404
# 生成或使用给定地址
if 'address' not in data or not data['address']:
# 生成随机地址
address = ''.join(random.choices(string.ascii_lowercase + string.digits, k=10))
else:
address = data['address']
# 创建邮箱
mailbox = Mailbox(
address=address,
domain_id=domain.id,
description=data.get('description', ''),
active=True
)
db.add(mailbox)
db.commit()
return jsonify({
'message': '邮箱创建成功',
'mailbox': mailbox.to_dict()
}), 201
except IntegrityError:
db.rollback()
return jsonify({'error': '邮箱地址已存在'}), 409
except Exception as e:
db.rollback()
raise
finally:
db.close()
except Exception as e:
current_app.logger.error(f"创建邮箱出错: {str(e)}")
return jsonify({'error': '创建邮箱失败', 'details': str(e)}), 500
# 批量创建邮箱
@api_bp.route('/mailboxes/batch', methods=['POST'])
def batch_create_mailboxes():
"""批量创建邮箱"""
try:
data = request.json
# 验证必要参数
if not data or 'domain_id' not in data or 'count' not in data:
return jsonify({'error': '缺少必要参数'}), 400
domain_id = data['domain_id']
count = min(int(data['count']), 100) # 限制最大数量为100
prefix = data.get('prefix', '')
db = get_session()
try:
# 查询域名是否存在
domain = db.query(Domain).filter_by(id=domain_id, active=True).first()
if not domain:
return jsonify({'error': '指定的域名不存在或未激活'}), 404
created_mailboxes = []
# 批量创建
for _ in range(count):
# 生成随机地址
if prefix:
address = f"{prefix}{random.randint(1000, 9999)}"
else:
address = ''.join(random.choices(string.ascii_lowercase + string.digits, k=10))
# 尝试创建,如果地址已存在则重试
retries = 0
while retries < 3: # 最多尝试3次
try:
mailbox = Mailbox(
address=address,
domain_id=domain.id,
active=True
)
db.add(mailbox)
db.flush() # 验证但不提交
created_mailboxes.append(mailbox)
break
except IntegrityError:
db.rollback()
# 地址已存在,重新生成
address = ''.join(random.choices(string.ascii_lowercase + string.digits, k=10))
retries += 1
# 提交所有更改
db.commit()
return jsonify({
'message': f'成功创建 {len(created_mailboxes)} 个邮箱',
'mailboxes': [mailbox.to_dict() for mailbox in created_mailboxes]
}), 201
except Exception as e:
db.rollback()
raise
finally:
db.close()
except Exception as e:
current_app.logger.error(f"批量创建邮箱出错: {str(e)}")
return jsonify({'error': '批量创建邮箱失败', 'details': str(e)}), 500
# 获取特定邮箱
@api_bp.route('/mailboxes/<int:mailbox_id>', methods=['GET'])
def get_mailbox(mailbox_id):
"""获取指定ID的邮箱信息"""
try:
db = get_session()
try:
mailbox = db.query(Mailbox).filter_by(id=mailbox_id).first()
if not mailbox:
return jsonify({'error': '邮箱不存在'}), 404
return jsonify(mailbox.to_dict()), 200
finally:
db.close()
except Exception as e:
current_app.logger.error(f"获取邮箱详情出错: {str(e)}")
return jsonify({'error': '获取邮箱详情失败', 'details': str(e)}), 500
# 删除邮箱
@api_bp.route('/mailboxes/<int:mailbox_id>', methods=['DELETE'])
def delete_mailbox(mailbox_id):
"""删除指定ID的邮箱"""
try:
db = get_session()
try:
mailbox = db.query(Mailbox).filter_by(id=mailbox_id).first()
if not mailbox:
return jsonify({'error': '邮箱不存在'}), 404
db.delete(mailbox)
db.commit()
return jsonify({'message': '邮箱已删除'}), 200
except Exception as e:
db.rollback()
raise
finally:
db.close()
except Exception as e:
current_app.logger.error(f"删除邮箱出错: {str(e)}")
return jsonify({'error': '删除邮箱失败', 'details': str(e)}), 500

341
old/app/api/routes.py Normal file
View File

@@ -0,0 +1,341 @@
from flask import Blueprint, request, jsonify, current_app
import json
from datetime import datetime, timedelta
import os
import time
import psutil
import sys
import platform
from sqlalchemy import func
from ..models import get_session, Domain, Mailbox, Email
from ..services import get_smtp_server, get_email_processor
api_bp = Blueprint('api', __name__, url_prefix='/api')
@api_bp.route('/domains', methods=['GET'])
def get_domains():
"""获取所有可用域名"""
db = get_session()
try:
domains = db.query(Domain).filter_by(active=True).all()
return jsonify({
'success': True,
'domains': [domain.to_dict() for domain in domains]
})
except Exception as e:
current_app.logger.exception(f"获取域名失败: {str(e)}")
return jsonify({'success': False, 'error': '获取域名失败'}), 500
finally:
db.close()
@api_bp.route('/domains', methods=['POST'])
def create_domain():
"""创建新域名"""
data = request.json
if not data or 'name' not in data:
return jsonify({'success': False, 'error': '缺少必要字段'}), 400
db = get_session()
try:
# 检查域名是否已存在
domain_exists = db.query(Domain).filter_by(name=data['name']).first()
if domain_exists:
return jsonify({'success': False, 'error': '域名已存在'}), 400
# 创建新域名
domain = Domain(
name=data['name'],
description=data.get('description', ''),
active=data.get('active', True)
)
db.add(domain)
db.commit()
return jsonify({
'success': True,
'message': '域名创建成功',
'domain': domain.to_dict()
})
except Exception as e:
db.rollback()
current_app.logger.exception(f"创建域名失败: {str(e)}")
return jsonify({'success': False, 'error': '创建域名失败'}), 500
finally:
db.close()
@api_bp.route('/mailboxes', methods=['GET'])
def get_mailboxes():
"""获取所有邮箱"""
db = get_session()
try:
mailboxes = db.query(Mailbox).all()
return jsonify({
'success': True,
'mailboxes': [mailbox.to_dict() for mailbox in mailboxes]
})
except Exception as e:
current_app.logger.exception(f"获取邮箱失败: {str(e)}")
return jsonify({'success': False, 'error': '获取邮箱失败'}), 500
finally:
db.close()
@api_bp.route('/mailboxes', methods=['POST'])
def create_mailbox():
"""创建新邮箱"""
data = request.json
if not data or 'address' not in data or 'domain_id' not in data:
return jsonify({'success': False, 'error': '缺少必要字段'}), 400
db = get_session()
try:
# 检查域名是否存在
domain = db.query(Domain).filter_by(id=data['domain_id'], active=True).first()
if not domain:
return jsonify({'success': False, 'error': '域名不存在或未激活'}), 400
# 检查邮箱是否已存在
mailbox_exists = db.query(Mailbox).filter_by(
address=data['address'], domain_id=data['domain_id']).first()
if mailbox_exists:
return jsonify({'success': False, 'error': '邮箱已存在'}), 400
# 创建新邮箱
mailbox = Mailbox(
address=data['address'],
domain_id=data['domain_id'],
description=data.get('description', ''),
active=data.get('active', True)
)
db.add(mailbox)
db.commit()
return jsonify({
'success': True,
'message': '邮箱创建成功',
'mailbox': mailbox.to_dict()
})
except Exception as e:
db.rollback()
current_app.logger.exception(f"创建邮箱失败: {str(e)}")
return jsonify({'success': False, 'error': '创建邮箱失败'}), 500
finally:
db.close()
@api_bp.route('/mailboxes/batch', methods=['POST'])
def batch_create_mailboxes():
"""批量创建邮箱"""
data = request.json
if not data or 'domain_id' not in data or 'usernames' not in data or not isinstance(data['usernames'], list):
return jsonify({'success': False, 'error': '缺少必要字段或格式不正确'}), 400
domain_id = data['domain_id']
usernames = data['usernames']
description = data.get('description', '')
db = get_session()
try:
# 检查域名是否存在
domain = db.query(Domain).filter_by(id=domain_id, active=True).first()
if not domain:
return jsonify({'success': False, 'error': '域名不存在或未激活'}), 400
created_mailboxes = []
existed_mailboxes = []
for username in usernames:
# 检查邮箱是否已存在
mailbox_exists = db.query(Mailbox).filter_by(
username=username, domain_id=domain_id).first()
if mailbox_exists:
existed_mailboxes.append(username)
continue
# 创建新邮箱
mailbox = Mailbox(
username=username,
domain_id=domain_id,
description=description,
active=True
)
db.add(mailbox)
created_mailboxes.append(username)
db.commit()
return jsonify({
'success': True,
'message': f'成功创建 {len(created_mailboxes)} 个邮箱,{len(existed_mailboxes)} 个已存在',
'created': created_mailboxes,
'existed': existed_mailboxes
})
except Exception as e:
db.rollback()
current_app.logger.exception(f"批量创建邮箱失败: {str(e)}")
return jsonify({'success': False, 'error': '批量创建邮箱失败'}), 500
finally:
db.close()
@api_bp.route('/mailboxes/<int:mailbox_id>', methods=['GET'])
def get_mailbox(mailbox_id):
"""获取特定邮箱的信息"""
db = get_session()
try:
mailbox = db.query(Mailbox).filter_by(id=mailbox_id).first()
if not mailbox:
return jsonify({'success': False, 'error': '邮箱不存在'}), 404
# 更新最后访问时间
mailbox.last_accessed = datetime.utcnow()
db.commit()
return jsonify({
'success': True,
'mailbox': mailbox.to_dict()
})
except Exception as e:
current_app.logger.exception(f"获取邮箱信息失败: {str(e)}")
return jsonify({'success': False, 'error': '获取邮箱信息失败'}), 500
finally:
db.close()
@api_bp.route('/mailboxes/<int:mailbox_id>/emails', methods=['GET'])
def get_emails(mailbox_id):
"""获取邮箱中的所有邮件"""
db = get_session()
try:
mailbox = db.query(Mailbox).filter_by(id=mailbox_id).first()
if not mailbox:
return jsonify({'success': False, 'error': '邮箱不存在'}), 404
# 更新最后访问时间
mailbox.last_accessed = datetime.utcnow()
db.commit()
emails = db.query(Email).filter_by(mailbox_id=mailbox_id).order_by(Email.received_at.desc()).all()
return jsonify({
'success': True,
'emails': [email.to_dict() for email in emails]
})
except Exception as e:
current_app.logger.exception(f"获取邮件失败: {str(e)}")
return jsonify({'success': False, 'error': '获取邮件失败'}), 500
finally:
db.close()
@api_bp.route('/emails/<int:email_id>', methods=['GET'])
def get_email(email_id):
"""获取特定邮件的详细内容"""
db = get_session()
try:
email = db.query(Email).filter_by(id=email_id).first()
if not email:
return jsonify({'success': False, 'error': '邮件不存在'}), 404
# 标记为已读
if not email.read:
email.read = True
db.commit()
return jsonify({
'success': True,
'email': email.to_dict()
})
except Exception as e:
current_app.logger.exception(f"获取邮件详情失败: {str(e)}")
return jsonify({'success': False, 'error': '获取邮件详情失败'}), 500
finally:
db.close()
@api_bp.route('/emails/<int:email_id>/verification', methods=['GET'])
def get_verification_info(email_id):
"""获取邮件中的验证信息(链接和验证码)"""
db = get_session()
try:
email = db.query(Email).filter_by(id=email_id).first()
if not email:
return jsonify({'success': False, 'error': '邮件不存在'}), 404
verification_links = json.loads(email.verification_links) if email.verification_links else []
verification_codes = json.loads(email.verification_codes) if email.verification_codes else []
return jsonify({
'success': True,
'email_id': email_id,
'verification_links': verification_links,
'verification_codes': verification_codes
})
except Exception as e:
current_app.logger.exception(f"获取验证信息失败: {str(e)}")
return jsonify({'success': False, 'error': '获取验证信息失败'}), 500
finally:
db.close()
@api_bp.route('/status', methods=['GET'])
def system_status():
"""获取系统状态"""
session = get_session()
# 获取基本统计信息
domain_count = session.query(func.count(Domain.id)).scalar()
mailbox_count = session.query(func.count(Mailbox.id)).scalar()
email_count = session.query(func.count(Email.id)).scalar()
# 获取最近24小时的邮件数量
recent_emails = session.query(func.count(Email.id)).filter(
Email.received_at > datetime.now() - timedelta(hours=24)
).scalar()
# 获取系统资源信息
cpu_percent = psutil.cpu_percent(interval=0.5)
memory = psutil.virtual_memory()
disk = psutil.disk_usage('/')
# 获取服务状态
smtp_server = get_smtp_server()
email_processor = get_email_processor()
smtp_status = "running" if smtp_server and smtp_server.controller else "stopped"
processor_status = "running" if email_processor and email_processor.is_running else "stopped"
# 构建响应
status = {
"system": {
"uptime": round(time.time() - psutil.boot_time()),
"time": datetime.now().isoformat(),
"platform": platform.platform(),
"python_version": sys.version
},
"resources": {
"cpu_percent": cpu_percent,
"memory_percent": memory.percent,
"memory_used": memory.used,
"memory_total": memory.total,
"disk_percent": disk.percent,
"disk_used": disk.used,
"disk_total": disk.total
},
"application": {
"domain_count": domain_count,
"mailbox_count": mailbox_count,
"email_count": email_count,
"recent_emails_24h": recent_emails,
"storage_path": os.path.abspath("email_data"),
"services": {
"smtp_server": smtp_status,
"email_processor": processor_status
}
}
}
return jsonify(status)

View File

@@ -0,0 +1,46 @@
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
import os
import sys
# 修改相对导入为绝对导入
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
from config import active_config
# 创建数据库引擎
engine = create_engine(active_config.SQLALCHEMY_DATABASE_URI)
# 创建会话工厂
session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)
# 创建模型基类
Base = declarative_base()
# 获取数据库会话
def get_session():
"""获取数据库会话"""
return Session()
# 初始化数据库
def init_db():
"""初始化数据库,创建所有表"""
# 导入所有模型以确保它们被注册
from .domain import Domain
from .mailbox import Mailbox
from .email import Email
from .attachment import Attachment
# 创建表
Base.metadata.create_all(engine)
return get_session()
# 导出模型类
from .domain import Domain
from .mailbox import Mailbox
from .email import Email
from .attachment import Attachment
__all__ = ['Base', 'get_session', 'init_db', 'Domain', 'Mailbox', 'Email', 'Attachment']

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

View File

@@ -0,0 +1,71 @@
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, LargeBinary
from sqlalchemy.orm import relationship
from datetime import datetime
import os
from . import Base
class Attachment(Base):
"""附件模型"""
__tablename__ = 'attachments'
id = Column(Integer, primary_key=True)
email_id = Column(Integer, ForeignKey('emails.id'), nullable=False, index=True)
filename = Column(String(255), nullable=False)
content_type = Column(String(100), nullable=True)
size = Column(Integer, nullable=False, default=0)
storage_path = Column(String(500), nullable=True) # 用于文件系统存储
content = Column(LargeBinary, nullable=True) # 用于小型附件的直接存储
created_at = Column(DateTime, default=datetime.utcnow)
# 关联关系
email = relationship("Email", back_populates="attachments")
@property
def is_stored_in_fs(self):
"""判断附件是否存储在文件系统中"""
return bool(self.storage_path and not self.content)
def save_to_filesystem(self, content, base_path):
"""将附件保存到文件系统"""
# 确保目录存在
os.makedirs(base_path, exist_ok=True)
# 创建文件路径
file_path = os.path.join(
base_path,
f"{self.email_id}_{self.id}_{self.filename}"
)
# 写入文件
with open(file_path, 'wb') as f:
f.write(content)
# 更新对象属性
self.storage_path = file_path
self.size = len(content)
self.content = None # 清空内存中的内容
return file_path
def get_content(self, attachments_dir=None):
"""获取附件内容,无论是从数据库还是文件系统"""
if self.content:
return self.content
if self.storage_path and os.path.exists(self.storage_path):
with open(self.storage_path, 'rb') as f:
return f.read()
return None
def to_dict(self):
"""转换为字典用于API响应"""
return {
"id": self.id,
"email_id": self.email_id,
"filename": self.filename,
"content_type": self.content_type,
"size": self.size,
"created_at": self.created_at.isoformat() if self.created_at else None
}

35
old/app/models/domain.py Normal file
View File

@@ -0,0 +1,35 @@
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.orm import relationship
from datetime import datetime
from . import Base
class Domain(Base):
"""邮件域名模型"""
__tablename__ = 'domains'
id = Column(Integer, primary_key=True)
name = Column(String(255), unique=True, nullable=False, index=True)
description = Column(String(500), nullable=True)
active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# 关系
mailboxes = relationship("Mailbox", back_populates="domain", cascade="all, delete-orphan")
def __repr__(self):
return f"<Domain {self.name}>"
def to_dict(self):
"""转换为字典用于API响应"""
return {
"id": self.id,
"name": self.name,
"description": self.description,
"active": self.active,
"created_at": self.created_at.isoformat() if self.created_at else None,
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
"mailbox_count": len(self.mailboxes) if self.mailboxes else 0
}

148
old/app/models/email.py Normal file
View File

@@ -0,0 +1,148 @@
import os
import json
from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, Boolean, JSON
from sqlalchemy.orm import relationship
from datetime import datetime
import re
import sys
import logging
from . import Base
import config
active_config = config.active_config
class Email(Base):
"""电子邮件模型"""
__tablename__ = 'emails'
id = Column(Integer, primary_key=True)
mailbox_id = Column(Integer, ForeignKey('mailboxes.id'), nullable=False, index=True)
sender = Column(String(255), nullable=False)
recipients = Column(String(1000), nullable=False)
subject = Column(String(500), nullable=True)
body_text = Column(Text, nullable=True)
body_html = Column(Text, nullable=True)
received_at = Column(DateTime, default=datetime.utcnow)
read = Column(Boolean, default=False)
headers = Column(JSON, nullable=True)
# 提取的验证码和链接
verification_code = Column(String(100), nullable=True)
verification_link = Column(String(1000), nullable=True)
# 关联关系
mailbox = relationship("Mailbox", back_populates="emails")
attachments = relationship("Attachment", back_populates="email", cascade="all, delete-orphan")
def save_raw_email(self, raw_content):
"""保存原始邮件内容到文件"""
storage_path = active_config.MAIL_STORAGE_PATH
mailbox_dir = os.path.join(storage_path, str(self.mailbox_id))
os.makedirs(mailbox_dir, exist_ok=True)
# 保存原始邮件内容
file_path = os.path.join(mailbox_dir, f"{self.id}.eml")
with open(file_path, 'wb') as f:
f.write(raw_content)
def extract_verification_data(self):
"""
尝试从邮件内容中提取验证码和验证链接
这个方法会在邮件保存时自动调用
"""
logger = logging.getLogger(__name__)
# 合并文本和HTML内容用于搜索
content = f"{self.subject or ''} {self.body_text or ''} {self.body_html or ''}"
logger.info(f"开始提取邮件ID={self.id}的验证信息,内容长度={len(content)}")
# 首先检查是否是Cursor验证邮件
if "Verify your email" in self.subject and (
"cursor.sh" in self.sender.lower() or
"cursor" in self.sender.lower()
):
logger.info("检测到Cursor验证邮件")
# 从HTML中提取6位数字验证码
cursor_patterns = [
r'(\d{6})</div>', # 匹配Cursor邮件中的6位数字验证码格式
r'<div[^>]*>(\d{6})</div>', # 更宽松的匹配
r'>(\d{6})<', # 最简单的形式
r'(\d{6})' # 任何6位数字
]
for pattern in cursor_patterns:
matches = re.findall(pattern, content)
if matches:
self.verification_code = matches[0]
logger.info(f"从Cursor邮件中提取到验证码: {self.verification_code}")
break
return
# 提取可能的验证码4-8位数字或字母组合
code_patterns = [
r'\b([A-Z0-9]{4,8})\b', # 大写字母和数字
r'验证码[:]\s*([A-Z0-9]{4,8})', # 中文格式
r'验证码是[:]\s*([A-Z0-9]{4,8})', # 中文格式2
r'code[:]\s*([A-Z0-9]{4,8})', # 英文格式
r'code is[:]\s*([A-Z0-9]{4,8})', # 英文格式2
r'code[:]\s*<[^>]*>([A-Z0-9]{4,8})', # HTML格式
r'<div[^>]*>([0-9]{4,8})</div>', # HTML分隔的数字
r'<strong[^>]*>([A-Z0-9]{4,8})</strong>', # 粗体验证码
]
for pattern in code_patterns:
matches = re.findall(pattern, content, re.IGNORECASE)
if matches:
# 过滤掉明显不是验证码的结果
filtered_matches = [m for m in matches if len(m) >= 4 and not m.lower() in ['code', 'verify', 'http', 'https']]
if filtered_matches:
self.verification_code = filtered_matches[0]
logger.info(f"提取到验证码: {self.verification_code}")
break
# 提取验证链接
link_patterns = [
r'https?://\S+(?:verify|confirm|activate)\S+',
r'https?://\S+(?:token|auth|account)\S+',
]
for pattern in link_patterns:
matches = re.findall(pattern, content, re.IGNORECASE)
if matches:
self.verification_link = matches[0]
logger.info(f"提取到验证链接: {self.verification_link}")
break
# 如果没有找到验证码,但邮件主题暗示这是验证邮件
verify_subjects = ['verify', 'confirmation', 'activate', 'validation', '验证', '确认']
if not self.verification_code and any(subj in self.subject.lower() for subj in verify_subjects):
logger.info("根据主题判断这可能是验证邮件,但未能提取到验证码")
# 尝试从HTML中提取明显的数字序列
if self.body_html:
number_matches = re.findall(r'(\d{4,8})', self.body_html)
filtered_numbers = [n for n in number_matches if len(n) >= 4 and len(n) <= 8]
if filtered_numbers:
self.verification_code = filtered_numbers[0]
logger.info(f"从HTML中提取到可能的验证码: {self.verification_code}")
logger.info(f"验证信息提取完成: code={self.verification_code}, link={self.verification_link}")
def __repr__(self):
return f"<Email {self.id}: {self.subject}>"
def to_dict(self):
"""转换为字典用于API响应"""
return {
"id": self.id,
"mailbox_id": self.mailbox_id,
"sender": self.sender,
"recipients": self.recipients,
"subject": self.subject,
"received_at": self.received_at.isoformat() if self.received_at else None,
"read": self.read,
"verification_code": self.verification_code,
"verification_link": self.verification_link,
"has_attachments": len(self.attachments) > 0 if self.attachments else False
}

50
old/app/models/mailbox.py Normal file
View File

@@ -0,0 +1,50 @@
from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from datetime import datetime
import secrets
from . import Base
class Mailbox(Base):
"""邮箱模型"""
__tablename__ = 'mailboxes'
id = Column(Integer, primary_key=True)
address = Column(String(255), unique=True, nullable=False, index=True)
domain_id = Column(Integer, ForeignKey('domains.id'), nullable=False)
password_hash = Column(String(255), nullable=True)
description = Column(String(500), nullable=True)
active = Column(Boolean, default=True)
api_key = Column(String(64), unique=True, default=lambda: secrets.token_hex(16))
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
last_accessed = Column(DateTime, nullable=True)
# 关系
domain = relationship("Domain", back_populates="mailboxes")
emails = relationship("Email", back_populates="mailbox", cascade="all, delete-orphan")
@property
def full_address(self):
"""获取完整邮箱地址 (包含域名)"""
return f"{self.address}@{self.domain.name}"
def __repr__(self):
return f"<Mailbox {self.full_address}>"
def to_dict(self):
"""转换为字典用于API响应"""
return {
"id": self.id,
"address": self.address,
"domain_id": self.domain_id,
"domain_name": self.domain.name if self.domain else None,
"full_address": self.full_address,
"description": self.description,
"active": self.active,
"created_at": self.created_at.isoformat() if self.created_at else None,
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
"last_accessed": self.last_accessed.isoformat() if self.last_accessed else None,
"email_count": len(self.emails) if self.emails else 0
}

View File

@@ -0,0 +1,44 @@
# 服务层初始化文件
# 这里将导入所有服务模块以便于统一调用
from .smtp_server import SMTPServer
from .email_processor import EmailProcessor
from .mail_store import MailStore
# 全局服务实例
_smtp_server = None
_email_processor = None
_mail_store = None
def register_smtp_server(instance):
"""注册SMTP服务器实例"""
global _smtp_server
_smtp_server = instance
def register_email_processor(instance):
"""注册邮件处理器实例"""
global _email_processor
_email_processor = instance
def register_mail_store(instance):
"""注册邮件存储实例"""
global _mail_store
_mail_store = instance
def get_smtp_server():
"""获取SMTP服务器实例"""
return _smtp_server
def get_email_processor():
"""获取邮件处理器实例"""
return _email_processor
def get_mail_store():
"""获取邮件存储实例"""
return _mail_store
__all__ = [
'SMTPServer', 'EmailProcessor', 'MailStore',
'register_smtp_server', 'register_email_processor', 'register_mail_store',
'get_smtp_server', 'get_email_processor', 'get_mail_store'
]

Binary file not shown.

View File

@@ -0,0 +1,123 @@
import logging
import re
import threading
import time
from queue import Queue
logger = logging.getLogger(__name__)
class EmailProcessor:
"""邮件处理器,负责处理邮件并提取验证信息"""
def __init__(self, mail_store):
"""
初始化邮件处理器
参数:
mail_store: 邮件存储服务实例
"""
self.mail_store = mail_store
self.processing_queue = Queue()
self.is_running = False
self.worker_thread = None
def start(self):
"""启动邮件处理器"""
if self.is_running:
logger.warning("邮件处理器已在运行")
return False
self.is_running = True
self.worker_thread = threading.Thread(
target=self._processing_worker,
daemon=True
)
self.worker_thread.start()
logger.info("邮件处理器已启动")
return True
def stop(self):
"""停止邮件处理器"""
if not self.is_running:
logger.warning("邮件处理器未在运行")
return False
self.is_running = False
if self.worker_thread:
self.worker_thread.join(timeout=5.0)
self.worker_thread = None
logger.info("邮件处理器已停止")
return True
def queue_email_for_processing(self, email_id):
"""将邮件添加到处理队列"""
self.processing_queue.put(email_id)
return True
def _processing_worker(self):
"""处理队列中的邮件的工作线程"""
while self.is_running:
try:
# 获取队列中的邮件最多等待1秒
try:
email_id = self.processing_queue.get(timeout=1.0)
except:
continue
# 处理邮件
self._process_email(email_id)
# 标记任务完成
self.processing_queue.task_done()
except Exception as e:
logger.error(f"处理邮件时出错: {str(e)}")
def _process_email(self, email_id):
"""处理单个邮件,提取验证码和链接"""
# 从邮件存储获取邮件
email_data = self.mail_store.get_email_by_id(email_id, mark_as_read=False)
if not email_data:
logger.warning(f"找不到ID为 {email_id} 的邮件")
return False
# 提取验证码和链接已经在Email模型的extract_verification_data方法中实现
# 这里可以添加更复杂的提取逻辑或后处理
logger.info(f"邮件 {email_id} 处理完成")
return True
@staticmethod
def extract_verification_code(content):
"""从内容中提取验证码"""
code_patterns = [
r'\b[A-Z0-9]{4,8}\b', # 基本验证码格式
r'验证码[:]\s*([A-Z0-9]{4,8})',
r'验证码是[:]\s*([A-Z0-9]{4,8})',
r'code[:]\s*([A-Z0-9]{4,8})',
r'码[:]\s*(\d{4,8})' # 纯数字验证码
]
for pattern in code_patterns:
matches = re.findall(pattern, content, re.IGNORECASE)
if matches:
return matches[0]
return None
@staticmethod
def extract_verification_link(content):
"""从内容中提取验证链接"""
link_patterns = [
r'(https?://\S+(?:verify|confirm|activate)\S+)',
r'(https?://\S+(?:token|auth|account)\S+)',
r'href\s*=\s*["\']([^"\']+(?:verify|confirm|activate)[^"\']*)["\']'
]
for pattern in link_patterns:
matches = re.findall(pattern, content, re.IGNORECASE)
if matches:
return matches[0]
return None

View File

@@ -0,0 +1,140 @@
import logging
import os
import email
from email.policy import default
from sqlalchemy.orm import Session
from datetime import datetime
import re
import redis
from ..models.domain import Domain
from ..models.mailbox import Mailbox
from ..models.email import Email
from ..models.attachment import Attachment
logging.basicConfig(
level=logging.DEBUG, # 设置日志级别为DEBUG
format='%(asctime)s - %(levelname)s - %(message)s', # 日志格式
filename='app.log', # 日志文件名
filemode='a' # 追加模式
)
logger = logging.getLogger(__name__)
# 连接到 Redis
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)
class MailStore:
"""邮件存储服务,负责保存和检索邮件"""
def __init__(self, db_session_factory, storage_path=None):
"""
初始化邮件存储服务
参数:
db_session_factory: 数据库会话工厂函数
storage_path: 附件存储路径
"""
self.db_session_factory = db_session_factory
self.storage_path = storage_path or os.path.join(os.getcwd(), 'email_data')
# 确保存储目录存在
if not os.path.exists(self.storage_path):
os.makedirs(self.storage_path)
async def save_email(self, message, sender, recipients, raw_data=None):
logging.info(f"开始保存邮件: 发件人={sender}, 收件人={recipients}")
# 处理收件人列表
if recipients is None:
logging.error("收件人列表为None无法保存邮件")
return False, "收件人列表为None"
elif isinstance(recipients, list):
recipients_list = recipients # 如果是列表,直接使用
else:
recipients_list = recipients.split(",") # 假设是以逗号分隔的字符串
# 确保收件人列表不为空
if not recipients_list:
logging.error("收件人列表为空,无法保存邮件")
return False, "收件人列表为空"
# 解析邮件内容
email_subject = message.subject if message.subject else "无主题"
body_text = message.get_body(preferencelist=('plain')).get_content()
received_at = datetime.now().isoformat()
# 存储邮件到 Redis
for recipient in recipients_list:
email_id = f"email:{recipient}:{received_at}"
redis_client.hset(email_id, mapping={
"subject": email_subject,
"sender": sender,
"recipients": recipients,
"body": body_text,
"received_at": received_at
})
logging.info(f"邮件已保存到 Redis: {email_id}")
return True, "邮件已保存到 Redis"
def get_emails_for_mailbox(self, mailbox_id, limit=50, offset=0, unread_only=False):
"""获取指定邮箱的邮件列表"""
try:
# 从 Redis 获取邮件
keys = redis_client.keys(f"email:*")
emails = []
for key in keys:
email_data = redis_client.hgetall(key)
emails.append(email_data)
total = len(emails)
return {
'total': total,
'items': emails[offset:offset + limit]
}
except Exception as e:
logger.error(f"获取邮件列表时出错: {str(e)}")
return {'total': 0, 'items': []}
def get_email_by_id(self, email_id):
"""获取指定ID的邮件详情"""
try:
email_data = redis_client.hgetall(f"email:{email_id}")
if not email_data:
return None
return email_data
except Exception as e:
logger.error(f"获取邮件详情时出错: {str(e)}")
return None
def delete_email(self, email_id):
"""删除指定ID的邮件"""
try:
redis_client.delete(f"email:{email_id}")
return True
except Exception as e:
logger.error(f"删除邮件时出错: {str(e)}")
return False
def get_attachment_content(self, attachment_id):
"""获取附件内容"""
db = self.db_session_factory()
try:
attachment = db.query(Attachment).filter(Attachment.id == attachment_id).first()
if not attachment:
return None
content = attachment.get_content()
return {
'content': content,
'filename': attachment.filename,
'content_type': attachment.content_type
}
except Exception as e:
logger.error(f"获取附件内容时出错: {str(e)}")
return None
finally:
db.close()

View File

@@ -0,0 +1,230 @@
import asyncio
import logging
import email
import platform
from email.policy import default
from aiosmtpd.controller import Controller
from aiosmtpd.smtp import SMTP as SMTPProtocol
from aiosmtpd.handlers import Message
import os
import sys
import threading
import traceback
from ..models.domain import Domain
from ..models.mailbox import Mailbox
from ..utils import email_parser
from ..models import Email
from aiosmtpd.smtp import SMTP, Session, Envelope
logger = logging.getLogger(__name__)
# 检测是否Windows环境
IS_WINDOWS = platform.system().lower() == 'windows'
class EmailHandler(Message):
"""处理接收的电子邮件"""
def __init__(self, mail_store):
super().__init__()
self.mail_store = mail_store
def handle_message(self, message):
"""处理邮件消息这是Message类的抽象方法必须实现"""
# 这个方法在异步DATA处理完成后被调用但我们的邮件处理逻辑已经在handle_DATA中实现
# 所以这里只是一个空实现
return
async def handle_DATA(self, server, session, envelope):
"""处理邮件数据"""
try:
logging.info(f"收到邮件: 发件人={envelope.mail_from}, 收件人={envelope.rcpt_tos}")
logging.debug(f"邮件内容: {envelope.content.decode('utf-8', errors='replace')}")
# 检查收件人列表是否有效
if not envelope.rcpt_tos:
logging.error("收件人列表无效,无法保存邮件")
return '550 收件人列表无效'
elif not isinstance(envelope.rcpt_tos, (list, str)):
logging.error("收件人列表格式不正确,无法保存邮件")
return '550 收件人列表格式不正确'
# 保存原始邮件数据
raw_data = envelope.content.decode('utf-8', errors='replace')
# 解析邮件数据
message = email_parser.Parser(policy=default).parsestr(raw_data)
subject = message.get('Subject', '')
logging.info(f"邮件主题: {subject}")
# 记录邮件结构和内容
logging.debug(f"邮件结构: is_multipart={message.is_multipart()}")
if message.is_multipart():
logging.debug(f"多部分邮件: 部分数量={len(list(message.walk()))}")
for i, part in enumerate(message.walk()):
content_type = part.get_content_type()
logging.debug(f"部分 {i+1}: 内容类型={content_type}")
# 使用邮件存储服务保存邮件
success, error_msg = await self.mail_store.save_email(
message,
envelope.mail_from,
envelope.rcpt_tos,
raw_data=raw_data
)
if success:
logging.info(f"邮件保存成功: 来自 {envelope.mail_from} 发送给 {envelope.rcpt_tos}")
return '250 消息接收完成'
else:
logging.error(f"邮件保存失败#server: {error_msg}")
# 即使保存失败,也返回成功状态码,避免邮件服务器重试
return '250 消息已收到'
except Exception as e:
logging.error(f"处理邮件时出错: {str(e)}")
traceback.print_exc()
return '451 处理邮件时出现错误,请稍后重试'
# 为Windows环境自定义SMTP控制器
if IS_WINDOWS:
class WindowsSafeController(Controller):
"""Windows环境安全的Controller跳过连接测试"""
def _trigger_server(self):
"""Windows环境下跳过SMTP服务器自检连接测试"""
# 在Windows环境下我们跳过自检连接测试
logger.info("Windows环境: 跳过SMTP服务器连接自检")
return
class SMTPServer:
"""SMTP服务器实现"""
def __init__(self, host='0.0.0.0', port=25, mail_store=None):
self.host = host
self.port = port
self.mail_store = mail_store
self.controller = None
self.server_thread = None
def start(self):
"""启动SMTP服务器"""
if self.controller:
logger.warning("SMTP服务器已经在运行")
return
try:
handler = EmailHandler(self.mail_store)
# 根据环境选择适当的Controller
if IS_WINDOWS:
# Windows环境使用自定义Controller
logger.info(f"Windows环境: 使用自定义Controller启动SMTP服务器 {self.host}:{self.port}")
self.controller = WindowsSafeController(
handler,
hostname=self.host,
port=self.port
)
else:
# 非Windows环境使用标准Controller
self.controller = Controller(
handler,
hostname=self.host,
port=self.port
)
# 在单独的线程中启动服务器
self.server_thread = threading.Thread(
target=self.controller.start,
daemon=True
)
self.server_thread.start()
logger.info(f"SMTP服务器已启动在 {self.host}:{self.port}")
return True
except Exception as e:
logger.error(f"启动SMTP服务器失败: {str(e)}")
return False
def stop(self):
"""停止SMTP服务器"""
if not self.controller:
logger.warning("SMTP服务器没有运行")
return
try:
self.controller.stop()
self.controller = None
self.server_thread = None
logger.info("SMTP服务器已停止")
return True
except Exception as e:
logger.error(f"停止SMTP服务器失败: {str(e)}")
return False
class MailHandler:
"""邮件处理器用于处理接收的SMTP邮件"""
def __init__(self, mail_store):
self.mail_store = mail_store
async def handle_EHLO(self, server, session, envelope, hostname):
session.host_name = hostname
return '250-AUTH PLAIN\n250-SIZE 52428800\n250 SMTPUTF8'
async def handle_MAIL(self, server, session, envelope, address, mail_options=None):
if not mail_options:
mail_options = []
envelope.mail_from = address
envelope.mail_options.extend(mail_options)
return '250 OK'
async def handle_RCPT(self, server, session, envelope, address, rcpt_options=None):
if not rcpt_options:
rcpt_options = []
envelope.rcpt_tos.append(address)
envelope.rcpt_options.extend(rcpt_options)
return '250 OK'
async def handle_DATA(self, server, session, envelope):
"""处理接收到的邮件数据"""
try:
logging.info(f"收到邮件: 发件人={envelope.mail_from}, 收件人={envelope.rcpt_tos}")
# 保存原始邮件数据
raw_data = envelope.content.decode('utf-8', errors='replace')
# 解析邮件数据
message = email_parser.parsestr(raw_data)
subject = message.get('Subject', '')
logging.info(f"邮件主题: {subject}")
# 记录邮件结构和内容
logging.debug(f"邮件结构: is_multipart={message.is_multipart()}")
if message.is_multipart():
logging.debug(f"多部分邮件: 部分数量={len(list(message.walk()))}")
for i, part in enumerate(message.walk()):
content_type = part.get_content_type()
logging.debug(f"部分 {i+1}: 内容类型={content_type}")
# 使用邮件存储服务保存邮件
success, error_msg = await self.mail_store.save_email(
message,
envelope.mail_from,
envelope.rcpt_tos,
raw_data=raw_data
)
if success:
logging.info(f"邮件保存成功: 来自 {envelope.mail_from} 发送给 {envelope.rcpt_tos}")
return '250 消息接收完成'
else:
logging.error(f"邮件保存失败: {error_msg}")
return '451 处理邮件时出现错误,请稍后重试'
except Exception as e:
logging.error(f"处理邮件时出错: {str(e)}")
traceback.print_exc()
return '451 处理邮件时出现错误,请稍后重试'

View File

View File

@@ -0,0 +1,9 @@
"""
工具类和辅助函数
"""
import email.parser as email_parser
import email.policy
# 创建邮件解析器实例,用于解析邮件
parser = email_parser.Parser()
parsestr = parser.parsestr

5
requirements.txt Normal file
View File

@@ -0,0 +1,5 @@
Flask==2.0.1
Flask-SQLAlchemy==2.5.1
Flask-Mail==0.9.1
redis==3.5.3
mysqlclient==2.0.3

17
run.py Normal file
View File

@@ -0,0 +1,17 @@
import threading
from app import create_app
from app.utils import start_smtp_server
app = create_app()
def run_smtp_server():
start_smtp_server(host='0.0.0.0', port=25)
if __name__ == '__main__':
# 在单独的线程中启动 SMTP 服务器
smtp_thread = threading.Thread(target=run_smtp_server)
smtp_thread.daemon = True
smtp_thread.start()
# 启动 Flask 应用
app.run(host='0.0.0.0', port=5000, debug=True)