Python后台如何为Web App（H5+安卓iOS外壳）实现消息推送？

核心方案：使用第三方推送服务（如极光、个推）或自建WebSocket长连接。

对于H5+原生外壳的混合App，消息推送需要区分平台处理。这里提供两种主流方案的代码示例：

方案一：使用第三方推送服务（推荐）

以极光推送为例，后端集成其Python SDK：

# 安装：pip install jpush
import jpush as jpush
from jpush import common

def send_push_notification(alias, title, content, extras=None):
    """向指定用户发送推送"""
    app_key = '你的app_key'
    master_secret = '你的master_secret'
    
    _jpush = jpush.JPush(app_key, master_secret)
    push = _jpush.create_push()
    
    # 设置推送平台
    push.audience = jpush.audience(
        jpush.alias(alias)  # 按用户别名推送
    )
    
    push.notification = jpush.notification(
        android=jpush.android(
            alert=content,
            title=title,
            extras=extras or {}
        ),
        ios=jpush.ios(
            alert=content,
            extras=extras or {}
        )
    )
    
    push.platform = jpush.all_  # 全平台
    
    try:
        response = push.send()
        return response.status_code == 200
    except common.Unauthorized:
        print("认证失败，检查app_key和master_secret")
    except common.APIConnectionException:
        print("连接失败")
    except common.JPushFailure as e:
        print(f"JPush错误: {e}")
    return False

# 使用示例
send_push_notification(
    alias="user123",
    title="新消息",
    content="您有一条未读消息",
    extras={"type": "order_update", "order_id": "1001"}
)

前端需要：

1. 安卓/iOS壳中集成对应SDK
2. H5页面通过JS Bridge接收推送并展示
3. 用户登录时绑定alias到设备

方案二：自建WebSocket服务

使用FastAPI+WebSocket实现实时推送：

# 安装：pip install fastapi uvicorn websockets
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Dict
import json

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: Dict[str, WebSocket] = {}
    
    async def connect(self, websocket: WebSocket, user_id: str):
        await websocket.accept()
        self.active_connections[user_id] = websocket
    
    def disconnect(self, user_id: str):
        self.active_connections.pop(user_id, None)
    
    async def send_personal_message(self, message: dict, user_id: str):
        if user_id in self.active_connections:
            await self.active_connections[user_id].send_json(message)

manager = ConnectionManager()

@app.websocket("/ws/{user_id}")
async def websocket_endpoint(websocket: WebSocket, user_id: str):
    await manager.connect(websocket, user_id)
    try:
        while True:
            # 保持连接
            await websocket.receive_text()
    except WebSocketDisconnect:
        manager.disconnect(user_id)

# 推送接口
@app.post("/push/{user_id}")
async def push_message(user_id: str, message: dict):
    await manager.send_personal_message(message, user_id)
    return {"status": "sent"}

# 运行：uvicorn main:app --reload

前端连接示例：

// H5中使用WebSocket
const ws = new WebSocket(`ws://your-domain.com/ws/${userId}`);
ws.onmessage = (event) => {
    const msg = JSON.parse(event.data);
    // 处理推送消息
    showNotification(msg.title, msg.content);
};

总结：用第三方服务省心，自建WebSocket更可控。