首先介绍一下具体的功能设计。
整个系统由服务端、客户端组成。客户端包含前端(浏览器端、pyqt桌面端)和后台(用于添加要推送的消息)。
因为浏览器端不容易注意到实时提醒，所以我将实时提醒功能放到pyqt桌面端中实现(如托盘区闪烁等醒目的方式)。
浏览器端中只对消息进行拉取操作(刷新网页时)，而不采取实时推的方式。
pyqt桌面端主要接收新消息提示即可，通知用户到网页端查看最新消息。(登录时主动拉取一次消息判断有无新消息，或者服务器主动推送所有新消息)
实时推送消息和主动拉取消息两个功能实际上是完全分离的，可以独立行使各自的职能。
不过下面我就不写浏览器端的代码了， 直接把相关的逻辑也放到pyqt里实现。
然后再来讲一下拉取服务端消息的逻辑。
每个用户拥有自己的消息库。当消息库为空时，一次性拉取服务端所有消息入用户库；当消息库内有消息时，每次拉取只拉取用户库内最新的消息发布时间之后的消息(将未入用户库的消息入库后再从用户库拉取)。
代码共5部分
1.鉴权服务器：用于服务和服务之间对用户提供的token做权限检查
2.消息数据服务器：用于读取和处理用户消息在数据库中的状态。
3.消息在线推送服务器：将管理员发布的消息在线实时推送给所有在线用户(若使用发布器的在线推送功能)
4.管理员消息发布器：分为在线推送和离线入库两个功能，可独立或组合使用。在线推送消息不会入消息库。
5.用户消息监听器：登陆后可接收管理员推送的在线消息，也可主动拉取用户对应消息库中的所有已读、未读消息，并标记消息的已读、未读状态。
下面是代码部分(为了简化代码逻辑，后面代码中的token即用户名。)
1.鉴权服务器
from flask import Flask, request, jsonify
app = Flask(__name__)
class Lib:
@staticmethod
def auth_admin_token(token):
return True
@staticmethod
def auth_user_token(token):
return True
@app.route('/admin', methods = ['POST'])
def admin_auth_token():
data = request.json
if Lib.auth_admin_token(data['token']):
return jsonify({'code': 200, 'msg': '鉴权成功'})
return jsonify({'code': 400, 'msg': '鉴权失败'})
@app.route('/user', methods = ['POST'])
def user_auth_token():
data = request.json
if Lib.auth_user_token(data['token']):
return jsonify({'code': 200, 'msg': '鉴权成功'})
return jsonify({'code':400, 'msg': '鉴权失败'})
if __name__ == '__main__':
app.run(host = '0.0.0.0', port = 5009)
2.消息数据服务器
# coding : utf-8
# author : ['Wang Suyin', ]
# data : 2020/2/21 14:14
# software : PyCharm
# python_version : '3.5.3 64bit'
# file : 2.消息数据服务器.py
"""
说明文档：
"""
import datetime
import requests
import json
from flask import Flask,request,jsonify
app=Flask(__name__)
# 虚拟数据
db = {
'message_table': [
{'id': 1, 'message': '消息1', 'publish_time': (2020, 1, 20, 3, 4, 5)},
{'id': 2, 'message': '消息2', 'publish_time': (2020, 1, 20, 3, 4, 6)},
{'id': 3, 'message': '消息3', 'publish_time': (2020, 1, 20, 3, 4, 7)},
],
'user_to_messages_table': [
{'username': 'tester', 'message_id': 1, 'read': True},
{'username': 'tester', 'message_id': 2, 'read': False},
],
}
class Utils:
@staticmethod
def auth_admin_token(token):
data = {'token': token}
r = requests.post(url = 'http://127.0.0.1:5009/admin', json = data)
r = json.loads(r.te
