50 lines
2.2 KiB
Python
50 lines
2.2 KiB
Python
import json
|
||
from datetime import datetime
|
||
|
||
from sqlalchemy import BigInteger, String, Text, Integer, TIMESTAMP, UniqueConstraint, Index
|
||
from sqlalchemy.orm import mapped_column
|
||
|
||
from app.models.mysql.archive_messages import Base
|
||
|
||
|
||
class ArchiveMediaFiles(Base):
|
||
__tablename__ = 'archive_media_files'
|
||
__table_args__ = {
|
||
'comment': '企微会话存档媒体文件表',
|
||
'mysql_charset': 'utf8mb4',
|
||
'mysql_collate': 'utf8mb4_unicode_ci'
|
||
}
|
||
|
||
id = mapped_column(BigInteger, primary_key=True, autoincrement=True, comment='自增主键')
|
||
msg_id = mapped_column(String(255), nullable=False, comment='企微消息 msgid')
|
||
archive_message_id = mapped_column(BigInteger, nullable=False, comment='archive_messages.id')
|
||
sdkfileid = mapped_column(Text, nullable=True, comment='企微媒体文件 sdkfileid')
|
||
cos_url = mapped_column(String(1024), nullable=False, comment='COS 文件访问 URL')
|
||
file_size = mapped_column(BigInteger, default=0, comment='文件大小（字节）')
|
||
file_type = mapped_column(String(32), nullable=True, comment='文件类型：image/voice/video/file')
|
||
status = mapped_column(Integer, default=1, comment='状态：1-有效 0-无效')
|
||
created_at = mapped_column(TIMESTAMP, nullable=False, default=datetime.now, comment='创建时间')
|
||
updated_at = mapped_column(TIMESTAMP, nullable=False, default=datetime.now, onupdate=datetime.now, comment='更新时间')
|
||
|
||
__table_args__ = (
|
||
UniqueConstraint('archive_message_id', name='uk_archive_msg_id'),
|
||
UniqueConstraint('msg_id', name='uk_msg_id'),
|
||
Index('idx_status', 'status'),
|
||
)
|
||
|
||
def to_dict(self):
|
||
return {
|
||
'id': self.id,
|
||
'msg_id': self.msg_id,
|
||
'archive_message_id': self.archive_message_id,
|
||
'sdkfileid': self.sdkfileid,
|
||
'cos_url': self.cos_url,
|
||
'file_size': self.file_size,
|
||
'file_type': self.file_type,
|
||
'status': self.status,
|
||
'created_at': self.created_at.isoformat() if self.created_at else None,
|
||
'updated_at': self.updated_at.isoformat() if self.updated_at else None
|
||
}
|
||
|
||
def __repr__(self):
|
||
return json.dumps(self.to_dict()) |