71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
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
|
||
} |