基于Flask、Flutter和PostgreSQL，实现用户收新物品时的通知方案咨询

一、后端（Flask+PostgreSQL）核心实现 #

1. 数据库设计 #

先建个notifications表存储通知数据，关联用户和物品：

CREATE TABLE notifications (
    id SERIAL PRIMARY KEY,
    recipient_id INT REFERENCES users(id) NOT NULL,
    sender_id INT REFERENCES users(id) NOT NULL,
    item_id INT REFERENCES items(id) NOT NULL,
    content TEXT NOT NULL,
    is_read BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

2. 触发通知写入 #

在物品发送的Flask接口里，完成物品逻辑后，同步插入通知记录：

@app.route('/send-item', methods=['POST'])
def send_item():
    # 先处理物品发送的核心逻辑（权限校验、物品关联等）
    recipient_id = request.json.get('recipient_id')
    sender_id = current_user.id
    item_id = # 新创建的物品ID
    content = f"用户{sender_id}给你发送了新物品"
    
    # 插入通知到数据库
    db.session.execute(
        "INSERT INTO notifications (recipient_id, sender_id, item_id, content) VALUES (:rid, :sid, :iid, :content)",
        {"rid": recipient_id, "sid": sender_id, "iid": item_id, "content": content}
    )
    db.session.commit()
    
    # 触发推送逻辑（两种方案二选一或结合使用）
    push_notification(recipient_id, content)
    return jsonify({"status": "success"})

3. 后端推送机制 #

方案A：WebSocket实时推送（适合APP前台在线场景）

用Flask-SocketIO实现长连接：

• 安装依赖：pip install flask-socketio
• 核心代码：

from flask_socketio import SocketIO, emit
from datetime import datetime

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'
socketio = SocketIO(app, cors_allowed_origins="*")

# 生产环境建议用Redis存用户-连接映射，避免重启丢失
user_socket_map = {}

@socketio.on('connect')
def handle_connect():
    user_id = request.args.get('user_id')
    if user_id:
        user_socket_map[user_id] = request.sid

@socketio.on('disconnect')
def handle_disconnect():
    for user_id, sid in list(user_socket_map.items()):
        if sid == request.sid:
            del user_socket_map[user_id]
            break

def push_notification(user_id, content):
    sid = user_socket_map.get(user_id)
    if sid:
        emit('new_notification', {
            "content": content, 
            "timestamp": datetime.utcnow().isoformat()
        }, room=sid)

方案B：FCM推送（适合APP后台离线场景）

调用Firebase Cloud Messaging API发送系统级通知：

import requests

FCM_SERVER_KEY = 'your-fcm-server-key'

def push_notification(user_id, content):
    # 从users表获取用户的FCM令牌（Flutter端注册后上传）
    user = User.query.get(user_id)
    if not user.fcm_token:
        return
    
    headers = {
        'Authorization': f'key={FCM_SERVER_KEY}',
        'Content-Type': 'application/json'
    }
    payload = {
        'to': user.fcm_token,
        'notification': {
            'title': '新物品通知',
            'body': content
        },
        'data': {
            'type': 'item',
            'item_id': '对应物品ID'
        }
    }
    requests.post('https://fcm.googleapis.com/fcm/send', headers=headers, json=payload)

二、Flutter端处理 #

1. WebSocket连接（对应方案A） #

用socket_io_client包实现长连接：

import 'package:socket_io_client/socket_io_client.dart' as IO;

IO.Socket? socket;

void initSocket(String userId) {
  socket = IO.io('http://your-flask-server:5000', <String, dynamic>{
    'transports': ['websocket'],
    'query': {'user_id': userId},
  });

  socket?.onConnect((_) => print('WebSocket连接成功'));

  socket?.on('new_notification', (data) {
    // 收到通知后，调用本地通知组件展示
    print('新通知：${data['content']}');
    showNotification('新物品通知', data['content']);
  });

  socket?.onDisconnect((_) => print('WebSocket连接断开'));
}

2. FCM集成（对应方案B） #

用firebase_messaging包注册令牌并处理推送：

import 'package:firebase_messaging/firebase_messaging.dart';

Future<void> initFCM() async {
  FirebaseMessaging messaging = FirebaseMessaging.instance;
  String? token = await messaging.getToken();
  // 将令牌上传到Flask后端，存到users表的fcm_token字段
  if (token != null) {
    await uploadFCMToken(token);
  }

  // 处理前台收到的推送
  FirebaseMessaging.onMessage.listen((RemoteMessage message) {
    showNotification(
      message.notification?.title ?? '新通知',
      message.notification?.body ?? ''
    );
  });

  // 处理点击通知打开APP的场景
  FirebaseMessaging.onMessageOpenedApp.listen((RemoteMessage message) {
    // 跳转到物品详情页
    navigateToItemDetail(message.data['item_id']);
  });
}

3. 本地通知展示 #

用flutter_local_notifications包实现系统通知：

import 'package:flutter_local_notifications/flutter_local_notifications.dart';

FlutterLocalNotificationsPlugin flutterLocalNotificationsPlugin = FlutterLocalNotificationsPlugin();

Future<void> initLocalNotifications() async {
  const AndroidInitializationSettings androidSettings = AndroidInitializationSettings('@mipmap/ic_launcher');
  const InitializationSettings initSettings = InitializationSettings(android: androidSettings);
  await flutterLocalNotificationsPlugin.initialize(initSettings);
}

void showNotification(String title, String body) async {
  const AndroidNotificationDetails androidDetails = AndroidNotificationDetails(
    'item_notification_channel',
    '新物品通知',
    importance: Importance.max,
    priority: Priority.high,
  );
  const NotificationDetails notificationDetails = NotificationDetails(android: androidDetails);
  await flutterLocalNotificationsPlugin.show(
    0,
    title,
    body,
    notificationDetails,
  );
}

三、生产环境补充 #

• WebSocket的用户连接映射建议用Redis存储，避免服务器重启后丢失；
• FCM需要在Firebase控制台配置项目，获取服务器密钥和对应平台的配置文件；
• 可以在Flutter端启动时调用/api/notifications/unread接口拉取未读通知，作为推送的兜底方案。

内容的提问来源于stack exchange，提问作者whitebat 199