operation_management_center/src/app/api/message.py

import datetime
import json

from flask import request, jsonify, current_app
from flask_restx import Resource, Namespace, reqparse
from sqlalchemy import select, insert, update
from sqlalchemy.orm import Session

from app.defines import StatesCode, Module, OperationType
from app.modle.message import Message
from app.modle.users import User
from app.utils.jwt_util import login_required
from app.utils.save_log import save_log
from app.utils.message_mq import add_message
from app.utils.util import to_dict
from config import Config

ns = Namespace('message', description='消息管理接口')

config = Config()


@ns.route('/message_list')
class MessageTypeApi(Resource):
    method_decorators = [login_required]

    @ns.doc(id='message_list', description='消息列表')
    def get(self):
        """获取消息列表"""
        data = {}
        for message in config.common.MESSAGR_TYPE:
            with Session(current_app.engine) as session:
                stmt = select(Message.id, Message.name).where(Message.type == message)

                for row in session.execute(stmt):
                    if data.get(message):
                        data[message].append({"id": row.id, "name": row.name})
                    else:
                        data[message] = [{"id": row.id, "name": row.name}]

        save_log(request, Module.MESSAGE, OperationType.INQUIRE, StatesCode.SUCCESS)

        return jsonify(code=StatesCode.SUCCESS, message='成功', data=data)


message_details = reqparse.RequestParser(bundle_errors=True)
message_details.add_argument(name='message_id', type=int, location='args', required=False, help='消息id')

message = reqparse.RequestParser(bundle_errors=True)

message.add_argument(name='message_id', type=str, location='form', required=False, help='消息id')
message.add_argument(name='name', type=str, location='form', required=False, help='消息名称')
message.add_argument(name='pipeline', type=str, location='form', required=False, help='通道配置')
message.add_argument(name='staff', type=str, location='form', required=False, help='人员配置: all role:'' company:'' ')
message.add_argument(name='send_time', type=str, location='form', required=False, help='发送时间 immediately 2020-10-20 10:10:10')
message.add_argument(name='title', type=str, location='form', required=False, help='标题')
message.add_argument(name='content', type=str, location='form', required=False, help='内容')
message.add_argument(name='style', type=str, location='form', required=False, help='样式')
message.add_argument(name='type', type=str, location='form', required=False, help='消息类型')


@ns.route('/message')
class MessageApi(Resource):
    method_decorators = [login_required]

    @ns.doc(id='message_details', description='获取消息详情')
    @ns.expect(message_details)
    def get(self):
        """获取消息详情"""

        message_id = request.args.get('message_id')

        if message_id is None:
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message='消息id不能为空')

        with Session(current_app.engine) as session:
            stmt = select(Message).where(Message.id == message_id)
            results = session.execute(stmt).scalars().all()

        save_log(request, Module.MESSAGE, OperationType.INQUIRE, StatesCode.SUCCESS)

        return jsonify(code=StatesCode.SUCCESS, message='成功', data=to_dict(results))

    @ns.doc(id='add_message', description='新增消息')
    @ns.expect(message)
    def post(self):
        """新增消息"""

        # 添加rabbit mq ，定时发送创建定时任务发送，立即发送直接添加rabbit mq
        name = request.form.get('name')
        pipeline = request.form.get('pipeline')
        staff = request.form.get('staff')
        send_time = request.form.get('send_time')
        title = request.form.get('title')
        content = request.form.get('content')
        style = request.form.get('style')
        message_type = request.form.get('type')

        if pipeline != 'instation':
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message='仅支持站内通知 instation')

        staff_list = []

        with Session(User) as session:
            if staff == 'all':
                stmt = select(User.user_name)
                staff_list.extend(session.execute(stmt).scalars().all())

            elif staff.split(':')[0] == 'company':
                stmt = select(User.user_name).where(User.company == staff.split(':')[1])
                staff_list.extend(session.execute(stmt).scalars().all())

            elif staff.split(':')[0] == 'role':
                stmt = select(User.user_name).where(User.role == staff.split(':')[1])
                staff_list.extend(session.execute(stmt).scalars().all())

            else:
                return jsonify(code=StatesCode.UNKNOWN_ERROR, message='无效的人员配置')

        try:
            if send_time != 'immediately':
                datetime.datetime.strptime(send_time, '%Y-%m-%d %H:%M:%S')
        except Exception:
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message='时间格式错误。例：2023-10-10 12:12:12')

        # 添加数据库
        with Session(current_app.engine) as session:
            stmt = insert(Message).values(
                name=name,
                pipeline=pipeline,
                staff=staff,
                send_time=send_time,
                title=title,
                content=content,
                style=style,
                type=message_type
            )
            session.execute(stmt)
            session.commit()

        message_data = json.dumps(
            {
                "staff": staff_list,
                "name": name,
                "title": title,
                "content": content,
                "style": style
            }
        )

        # 判断发送时间，立即发送，或添加添加定时任务
        if send_time == 'immediately':
            add_message(message_data)

        else:
            current_app.scheduler.add_job(id=name,
                                          func=add_message,
                                          trigger='date',
                                          run_date=datetime.datetime.strptime(send_time, '%Y-%m-%d %H:%M:%S'),
                                          args=[message_data]
                                          )

        save_log(request, Module.MESSAGE, OperationType.ADD, StatesCode.SUCCESS)

        return jsonify(code=StatesCode.SUCCESS, message='成功')

    @ns.doc(id='update_message', description='修改消息')
    @ns.expect(message)
    def put(self):
        """修改消息"""
        # 修改数据库，重新添加

        message_id = request.form.get('message_id')
        name = request.form.get('name')
        pipeline = request.form.get('pipeline')
        staff = request.form.get('staff')
        send_time = request.form.get('send_time')
        title = request.form.get('title')
        content = request.form.get('content')
        style = request.form.get('style')
        message_type = request.form.get('type')

        if pipeline != 'instation':
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message='仅支持站内通知 instation')

        staff_list = []

        with Session(User) as session:
            if staff == 'all':
                stmt = select(User.user_name)
                staff_list.extend(session.execute(stmt).scalars().all())

            elif staff.split(':')[0] == 'company':
                stmt = select(User.user_name).where(User.company == staff.split(':')[1])
                staff_list.extend(session.execute(stmt).scalars().all())

            elif staff.split(':')[0] == 'role':
                stmt = select(User.user_name).where(User.role == staff.split(':')[1])
                staff_list.extend(session.execute(stmt).scalars().all())

            else:
                return jsonify(code=StatesCode.UNKNOWN_ERROR, message='无效的人员配置')

        try:
            if send_time != 'immediately':
                datetime.datetime.strptime(send_time, '%Y-%m-%d %H:%M:%S')
        except Exception:
            return jsonify(code=StatesCode.UNKNOWN_ERROR, message='时间格式错误。例：2023-10-10 12:12:12')

        # 添加数据库
        with Session(current_app.engine) as session:
            stmt = update(Message).where(Message.id == message_id).values(
                name=name,
                pipeline=pipeline,
                staff=staff,
                send_time=send_time,
                title=title,
                content=content,
                style=style,
                type=message_type
            )
            session.execute(stmt)
            session.commit()

        message_data = json.dumps(
            {
                "staff": staff_list,
                "name": name,
                "title": title,
                "content": content,
                "style": style
            }
        )

        # 判断发送时间，立即发送，或添加添加定时任务
        if send_time == 'immediately':
            add_message(message_data)

        else:
            current_app.scheduler.add_job(id=name,
                                          func=add_message,
                                          trigger='date',
                                          run_date=datetime.datetime.strptime(send_time, '%Y-%m-%d %H:%M:%S'),
                                          args=[message_data]
                                          )

        save_log(request, Module.MESSAGE, OperationType.UPDATE, StatesCode.SUCCESS)

        return jsonify(code=StatesCode.SUCCESS, message='成功')