306 lines
12 KiB
Python
306 lines
12 KiB
Python
import logging
|
|
import os
|
|
import email
|
|
from email.policy import default
|
|
from sqlalchemy.orm import Session
|
|
from datetime import datetime
|
|
import re
|
|
|
|
from ..models.domain import Domain
|
|
from ..models.mailbox import Mailbox
|
|
from ..models.email import Email
|
|
from ..models.attachment import Attachment
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
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}")
|
|
try:
|
|
# 解析邮件内容
|
|
email_subject = None
|
|
body_text = ""
|
|
body_html = ""
|
|
attachments = []
|
|
|
|
# 提取邮件主题
|
|
if hasattr(message, 'subject') and message.subject:
|
|
email_subject = message.subject
|
|
|
|
# 提取邮件内容
|
|
if hasattr(message, 'get_body'):
|
|
try:
|
|
for part in message.walk():
|
|
content_type = part.get_content_type()
|
|
|
|
if content_type == "text/plain":
|
|
body_text = part.get_payload(decode=True).decode(part.get_content_charset() or 'utf-8', errors='replace')
|
|
|
|
elif content_type == "text/html":
|
|
body_html = part.get_payload(decode=True).decode(part.get_content_charset() or 'utf-8', errors='replace')
|
|
|
|
# 处理附件
|
|
if part.get_filename():
|
|
file_name = part.get_filename()
|
|
content = part.get_payload(decode=True)
|
|
attachments.append((file_name, content))
|
|
except Exception as e:
|
|
logging.error(f"获取邮件内容时出错: {str(e)}")
|
|
|
|
# 如果仍然没有内容，尝试从原始数据中提取
|
|
if not body_text and not body_html and raw_data:
|
|
logging.info("从原始数据中提取内容")
|
|
try:
|
|
# 简单提取，可能不适用于所有情况
|
|
if '<html>' in raw_data.lower():
|
|
body_html = raw_data
|
|
else:
|
|
body_text = raw_data
|
|
except Exception as e:
|
|
logging.error(f"从原始数据提取内容失败: {str(e)}")
|
|
|
|
logging.info(f"提取完成: 纯文本={len(body_text)}字节, HTML={len(body_html)}字节, 附件数={len(attachments)}")
|
|
|
|
# 保存到数据库
|
|
session = self.db_session_factory()
|
|
try:
|
|
# 查找收件人对应的邮箱
|
|
mailbox_id = None
|
|
recipients_list = recipients if isinstance(recipients, list) else [recipients]
|
|
|
|
# 确保收件人列表不为空
|
|
if not recipients_list:
|
|
logging.error("收件人列表为空，无法确定邮箱")
|
|
return False, "收件人列表为空"
|
|
|
|
# 尝试找到或创建收件人邮箱
|
|
for recipient in recipients_list:
|
|
# 跳过空收件人
|
|
if not recipient or '@' not in recipient:
|
|
continue
|
|
|
|
# 提取域名和用户名
|
|
username, domain = recipient.lower().split('@', 1)
|
|
logging.info(f"处理收件人: 用户名={username}, 域名={domain}")
|
|
|
|
# 1. 先查找域名
|
|
domain_obj = session.query(Domain).filter(Domain.name.ilike(domain)).first()
|
|
|
|
# 如果域名不存在，创建它
|
|
if not domain_obj:
|
|
logging.info(f"创建新域名: {domain}")
|
|
domain_obj = Domain(
|
|
name=domain.lower(),
|
|
description=f"系统自动创建的域名 ({domain})",
|
|
active=True
|
|
)
|
|
session.add(domain_obj)
|
|
session.flush()
|
|
|
|
# 2. 查找或创建邮箱
|
|
mailbox = session.query(Mailbox).filter(
|
|
Mailbox.domain_id == domain_obj.id,
|
|
Mailbox.address.ilike(username)
|
|
).first()
|
|
|
|
# 如果邮箱不存在，创建它
|
|
if not mailbox:
|
|
logging.info(f"创建新邮箱: {username}@{domain}")
|
|
mailbox = Mailbox(
|
|
domain_id=domain_obj.id,
|
|
address=username.lower(),
|
|
description=f"系统自动创建的邮箱 ({username}@{domain})"
|
|
)
|
|
session.add(mailbox)
|
|
session.flush()
|
|
|
|
# 设置邮箱ID并结束循环
|
|
mailbox_id = mailbox.id
|
|
logging.info(f"使用邮箱ID: {mailbox_id} ({username}@{domain})")
|
|
break
|
|
|
|
# 最终检查是否获取到了邮箱ID
|
|
if mailbox_id is None:
|
|
error_msg = f"无法确定有效的收件邮箱，无法保存邮件。收件人: {recipients}"
|
|
logging.error(error_msg)
|
|
return False, error_msg
|
|
|
|
# 创建邮件记录
|
|
email_obj = Email(
|
|
mailbox_id=mailbox_id, # 确保始终有邮箱ID
|
|
sender=sender,
|
|
recipients=str(recipients),
|
|
subject=email_subject,
|
|
body_text=body_text,
|
|
body_html=body_html,
|
|
)
|
|
|
|
# 提取验证码和验证链接
|
|
email_obj.extract_verification_data()
|
|
|
|
session.add(email_obj)
|
|
session.flush() # 获取新创建邮件的ID
|
|
|
|
# 保存附件
|
|
for file_name, content in attachments:
|
|
attachment = Attachment(
|
|
email_id=email_obj.id,
|
|
filename=file_name,
|
|
content_type="application/octet-stream",
|
|
size=len(content)
|
|
)
|
|
session.add(attachment)
|
|
|
|
# 保存附件内容到文件
|
|
attachment_path = os.path.join(self.storage_path, 'attachments', f"attachment_{attachment.id}")
|
|
os.makedirs(os.path.dirname(attachment_path), exist_ok=True)
|
|
with open(attachment_path, 'wb') as f:
|
|
f.write(content)
|
|
|
|
# 保存原始邮件数据
|
|
raw_path = os.path.join(self.storage_path, 'emails', f"email_{email_obj.id}.eml")
|
|
os.makedirs(os.path.dirname(raw_path), exist_ok=True)
|
|
|
|
# 写入原始邮件
|
|
with open(raw_path, 'w', encoding='utf-8', errors='replace') as f:
|
|
if isinstance(message, str):
|
|
f.write(message)
|
|
else:
|
|
# 如果是邮件对象，尝试获取原始文本
|
|
try:
|
|
f.write(message.as_string())
|
|
except Exception:
|
|
# 如果失败，使用提供的原始数据
|
|
if raw_data:
|
|
f.write(raw_data)
|
|
|
|
session.commit()
|
|
logging.info(f"邮件保存成功: ID={email_obj.id}")
|
|
return True, f"邮件已保存: ID={email_obj.id}"
|
|
|
|
except Exception as e:
|
|
session.rollback()
|
|
logging.error(f"数据库操作失败: {str(e)}")
|
|
raise
|
|
finally:
|
|
session.close()
|
|
|
|
except Exception as e:
|
|
logging.error(f"邮件保存失败: {str(e)}")
|
|
return False, f"保存邮件失败: {str(e)}"
|
|
|
|
def get_emails_for_mailbox(self, mailbox_id, limit=50, offset=0, unread_only=False):
|
|
"""获取指定邮箱的邮件列表"""
|
|
db = self.db_session_factory()
|
|
try:
|
|
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()
|
|
|
|
return {
|
|
'total': total,
|
|
'items': [email.to_dict() for email in emails]
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"获取邮件列表时出错: {str(e)}")
|
|
return {'total': 0, 'items': []}
|
|
finally:
|
|
db.close()
|
|
|
|
def get_email_by_id(self, email_id, mark_as_read=True):
|
|
"""获取指定ID的邮件详情"""
|
|
db = self.db_session_factory()
|
|
try:
|
|
email = db.query(Email).filter(Email.id == email_id).first()
|
|
|
|
if not email:
|
|
return None
|
|
|
|
if mark_as_read and not email.read:
|
|
email.read = True
|
|
email.last_read = datetime.now()
|
|
db.commit()
|
|
|
|
# 获取附件信息
|
|
attachments = [attachment.to_dict() for attachment in email.attachments]
|
|
|
|
# 构建完整响应
|
|
result = email.to_dict()
|
|
result['body_text'] = email.body_text
|
|
result['body_html'] = email.body_html
|
|
result['attachments'] = attachments
|
|
|
|
return result
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"获取邮件详情时出错: {str(e)}")
|
|
return None
|
|
finally:
|
|
db.close()
|
|
|
|
def delete_email(self, email_id):
|
|
"""删除指定ID的邮件"""
|
|
db = self.db_session_factory()
|
|
try:
|
|
email = db.query(Email).filter(Email.id == email_id).first()
|
|
|
|
if not email:
|
|
return False
|
|
|
|
db.delete(email)
|
|
db.commit()
|
|
return True
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"删除邮件时出错: {str(e)}")
|
|
return False
|
|
finally:
|
|
db.close()
|
|
|
|
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() |