58 lines
1.6 KiB
Python
58 lines
1.6 KiB
Python
import time
|
||
from typing import Optional, Union
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.db import DbOper
|
||
from app.db.models.message import Message
|
||
from app.schemas import MessageChannel, NotificationType
|
||
|
||
|
||
class MessageOper(DbOper):
|
||
"""
|
||
消息数据管理
|
||
"""
|
||
|
||
def __init__(self, db: Session = None):
|
||
super().__init__(db)
|
||
|
||
def add(self,
|
||
channel: MessageChannel = None,
|
||
mtype: NotificationType = None,
|
||
title: str = None,
|
||
text: str = None,
|
||
image: str = None,
|
||
link: str = None,
|
||
userid: str = None,
|
||
action: int = 1,
|
||
**kwargs):
|
||
"""
|
||
新增媒体服务器数据
|
||
:param channel: 消息渠道
|
||
:param mtype: 消息类型
|
||
:param title: 标题
|
||
:param text: 文本内容
|
||
:param image: 图片
|
||
:param link: 链接
|
||
:param userid: 用户ID
|
||
:param action: 消息方向：0-接收息，1-发送消息
|
||
"""
|
||
kwargs.update({
|
||
"channel": channel.value if channel else '',
|
||
"mtype": mtype.value if mtype else '',
|
||
"title": title,
|
||
"text": text,
|
||
"image": image,
|
||
"link": link,
|
||
"userid": userid,
|
||
"action": action,
|
||
"reg_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
|
||
})
|
||
Message(**kwargs).create(self._db)
|
||
|
||
def list_by_page(self, page: int = 1, count: int = 30) -> Optional[str]:
|
||
"""
|
||
获取媒体服务器数据ID
|
||
"""
|
||
return Message.list_by_page(self._db, page, count)
|