生产环境不使用OneSignal，如何通过Python向Chrome推送通知？

核心架构 #

采用Python后端+Chrome扩展的自研架构，无第三方推送服务依赖：

• Python后端负责生产告警消息、提供消息查询/推送能力
• Chrome扩展后台维持和后端的通信，拿到动态消息后自动触发系统通知，绑定点击跳转逻辑

1. manifest.json 配置修改 #

新增必要权限，适配通信和跳转能力：

{
  "name": "Check Dashboard",
  "version": "1.0",
  "content_scripts": [
    {
      "matches": ["<你实际需要注入的域名规则>"], // 收敛权限，不要用通配符
      "js": ["js/app.js"]
    }
  ],
  "background" : {
    "scripts" : ["js/background.js"],
    "persistent": true // 用WebSocket长连接时设为true，用定时拉取可设为false
  },
  "permissions": [
    "notifications",
    "tabs", // 用于打开仪表盘标签页
    "alarms", // 用于定时拉取消息
    "storage", // 用于存储已弹出的通知ID去重
    "https://<你的后端域名>/*" // 允许请求后端接口
  ],
  "browser_action": {
    "default_icon": "icon.png"
  },
  "manifest_version": 2
}

生产环境建议升级到Manifest V3，Manifest V2将于2024年停止支持，V3将background调整为service_worker配置即可，核心逻辑不变。

2. background.js 逻辑改造 #

新增消息拉取/监听、自动弹通知、点击跳转能力：

// 配置项
const CONFIG = {
  FETCH_INTERVAL: 30, // 定时拉取间隔，单位秒
  DASHBOARD_URL: "https://<你的仪表盘地址>",
  ALERT_API_URL: "https://<你的后端域名>/api/latest-alert"
}

// 定时拉取后端告警
chrome.alarms.create("fetchAlert", { periodInMinutes: CONFIG.FETCH_INTERVAL / 60 });
chrome.alarms.onAlarm.addListener((alarm) => {
  if (alarm.name === "fetchAlert") fetchAndShowAlert();
});

// 也可替换为WebSocket/SSE监听后端主动推送，适合低延迟场景
// const ws = new WebSocket("wss://<你的后端域名>/ws");
// ws.onmessage = (event) => showNotification(JSON.parse(event.data));

// 拉取告警并展示
async function fetchAndShowAlert() {
  try {
    const res = await fetch(CONFIG.ALERT_API_URL);
    const alertData = await res.json();
    // 有未读告警且未弹出过则展示
    if (alertData.hasNew && !await isAlertExist(alertData.id)) {
      showNotification(alertData);
      markAlertAsShown(alertData.id);
    }
  } catch (e) {
    // 网络异常可加重试逻辑
  }
}

// 弹出通知
function showNotification(options) {
  chrome.notifications.create(alertData.id, {
    title: options.title,
    message: options.message,
    iconUrl: '/icon.png',
    type: 'basic'
  });
}

// 监听通知点击跳转仪表盘
chrome.notifications.onClicked.addListener((notificationId) => {
  chrome.tabs.create({ url: CONFIG.DASHBOARD_URL });
  chrome.notifications.clear(notificationId);
});

// 通知去重相关逻辑
async function isAlertExist(alertId) {
  const res = await chrome.storage.local.get("shownAlerts");
  return (res.shownAlerts || []).includes(alertId);
}

async function markAlertAsShown(alertId) {
  const res = await chrome.storage.local.get("shownAlerts");
  const shownAlerts = res.shownAlerts || [];
  shownAlerts.push(alertId);
  // 只保留最近100条通知ID，避免存储超限
  if (shownAlerts.length > 100) shownAlerts.shift();
  await chrome.storage.local.set({ shownAlerts });
}

3. app.js 调整 #

原有手动触发通知的逻辑可按需保留或删除，不影响自动通知能力。

可根据现有技术栈选型，核心功能仅需2块：

1. 接收业务逻辑产生的告警消息，存储到数据库/缓存，给每条消息生成唯一ID
2. 提供供Chrome扩展调用的查询接口，或者WebSocket/SSE主动推送能力

FastAPI 极简示例：

from fastapi import FastAPI
from pydantic import BaseModel
import uuid

app = FastAPI()
# 生产环境替换为Redis/MySQL等持久化存储
latest_alert = None

class Alert(BaseModel):
    title: str
    message: str

# 供你的Python业务逻辑调用，上报新告警
@app.post("/api/report-alert")
async def report_alert(alert: Alert):
    global latest_alert
    latest_alert = {
        "id": str(uuid.uuid4()),
        "title": alert.title,
        "message": alert.message,
        "hasNew": True
    }
    return {"status": "ok"}

# 供Chrome扩展拉取最新告警
@app.get("/api/latest-alert")
async def get_latest_alert():
    global latest_alert
    if not latest_alert:
        return {"hasNew": False}
    res = latest_alert.copy()
    # 拉取后标记为已读，避免重复返回
    latest_alert["hasNew"] = False
    return res

• 后端服务必须配置HTTPS证书，Chrome扩展禁止向HTTP地址发起请求（本地开发除外）
• 扩展可打包为crx文件，企业内部使用可通过组策略批量推送安装，无需上架Chrome应用商店
• 可根据告警的实时性要求选择通信方式：秒级延迟用WebSocket/SSE，分钟级延迟用定时拉取即可
• 增加告警降级逻辑：扩展侧可配置每日最大通知数量，避免告警风暴打扰用户
• 权限最小化：content_scripts的匹配规则、后端接口权限都按需配置，不要使用通配符降低安全风险

内容的提问来源于stack exchange，提问作者Nishant Kumar