109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
"""
|
|
通知管理 API
|
|
"""
|
|
import uuid
|
|
import json
|
|
from datetime import datetime
|
|
from flask import Blueprint, request, jsonify
|
|
from config import DATA_DIR
|
|
from utils import load_data, save_data
|
|
|
|
notifications_bp = Blueprint('api_notifications', __name__)
|
|
|
|
NOTIFICATIONS_FILE = DATA_DIR / 'notifications.json'
|
|
|
|
|
|
@notifications_bp.route('/api/notifications')
|
|
def api_notifications():
|
|
"""获取通知列表"""
|
|
notifications = load_data(NOTIFICATIONS_FILE)
|
|
|
|
# 筛选参数
|
|
unread_only = request.args.get('unread', '0') == '1'
|
|
limit = int(request.args.get('limit', 50))
|
|
|
|
if unread_only:
|
|
notifications = [n for n in notifications if not n.get('read', False)]
|
|
|
|
# 按时间倒序
|
|
notifications = sorted(notifications, key=lambda x: x.get('created_at', ''), reverse=True)
|
|
|
|
return jsonify(notifications[:limit])
|
|
|
|
|
|
@notifications_bp.route('/api/notifications/unread-count')
|
|
def api_unread_count():
|
|
"""获取未读通知数量"""
|
|
notifications = load_data(NOTIFICATIONS_FILE)
|
|
count = len([n for n in notifications if not n.get('read', False)])
|
|
return jsonify({'count': count})
|
|
|
|
|
|
@notifications_bp.route('/api/notifications/<notification_id>/read', methods=['POST'])
|
|
def api_mark_read(notification_id):
|
|
"""标记通知为已读"""
|
|
notifications = load_data(NOTIFICATIONS_FILE)
|
|
notification = next((n for n in notifications if n['id'] == notification_id), None)
|
|
if not notification:
|
|
return jsonify({'error': 'Notification not found'}), 404
|
|
notification['read'] = True
|
|
notification['read_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
save_data(NOTIFICATIONS_FILE, notifications)
|
|
return jsonify({'success': True})
|
|
|
|
|
|
@notifications_bp.route('/api/notifications/read-all', methods=['POST'])
|
|
def api_mark_all_read():
|
|
"""标记所有通知为已读"""
|
|
notifications = load_data(NOTIFICATIONS_FILE)
|
|
for n in notifications:
|
|
if not n.get('read', False):
|
|
n['read'] = True
|
|
n['read_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
save_data(NOTIFICATIONS_FILE, notifications)
|
|
return jsonify({'success': True})
|
|
|
|
|
|
@notifications_bp.route('/api/notifications/<notification_id>', methods=['DELETE'])
|
|
def api_delete_notification(notification_id):
|
|
"""删除通知"""
|
|
notifications = load_data(NOTIFICATIONS_FILE)
|
|
notifications = [n for n in notifications if n['id'] != notification_id]
|
|
save_data(NOTIFICATIONS_FILE, notifications)
|
|
return jsonify({'success': True})
|
|
|
|
|
|
# ─── 内部函数：创建通知 ───────────────────────────────────────────────────────
|
|
|
|
def create_notification(title, message, level='info', category='system', data=None):
|
|
"""
|
|
创建一条通知
|
|
|
|
参数:
|
|
title: 通知标题
|
|
message: 通知内容
|
|
level: 级别 info/warning/error/success
|
|
category: 分类 system/api/review/statistics
|
|
data: 附加数据(dict)
|
|
"""
|
|
notifications = load_data(NOTIFICATIONS_FILE)
|
|
|
|
notification = {
|
|
'id': uuid.uuid4().hex[:12],
|
|
'title': title,
|
|
'message': message,
|
|
'level': level, # info, warning, error, success
|
|
'category': category, # system, api, review, statistics
|
|
'data': data or {},
|
|
'read': False,
|
|
'created_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
}
|
|
|
|
notifications.append(notification)
|
|
|
|
# 只保留最近500条通知
|
|
if len(notifications) > 500:
|
|
notifications = sorted(notifications, key=lambda x: x.get('created_at', ''), reverse=True)[:500]
|
|
|
|
save_data(NOTIFICATIONS_FILE, notifications)
|
|
return notification |