如何在Python脚本中监听Firebase数据库变更并实现用户下载通知？

1. 设计Firebase数据库结构 #

• 用户节点：按用户ID划分独立节点（/users/{user_id}），存储用户的FCM推送令牌（fcm_token）、允许下载的文件ID列表（allowed_files）
• 文件节点：集中存储可下载文件（/available_files/{file_id}），包含文件名、下载链接、权限标签、上传时间等字段，新增文件时触发监听

2. 配置Python环境与依赖 #

• 安装必要库：

  pip install firebase-admin python-dotenv
  

• 用python-dotenv管理敏感配置，避免硬编码。给每个用户准备.env文件，包含：

  FIREBASE_DB_URL=你的实时数据库URL
  USER_ID_TOKEN=用户专属的Firebase Auth ID令牌
  FIREBASE_SERVICE_ACCOUNT_PATH=受限权限的服务账号密钥路径
  

3. 初始化Firebase客户端 #

import firebase_admin
from firebase_admin import credentials, db, auth, messaging
from dotenv import load_dotenv
import os

load_dotenv()
# 初始化Firebase，使用受限权限的服务账号
cred = credentials.Certificate(os.getenv("FIREBASE_SERVICE_ACCOUNT_PATH"))
firebase_admin.initialize_app(cred, {
    'databaseURL': os.getenv("FIREBASE_DB_URL")
})

4. 实现数据库变更监听 #

监听available_files节点的新增事件，一旦有新文件添加就触发权限验证：

def handle_new_file(event):
    new_file = event.data
    if new_file:
        validate_permission_and_notify(new_file)

# 启动持续监听
file_ref = db.reference('available_files')
file_ref.listen(handle_new_file)

5. 用户权限验证逻辑 #

通过用户的ID令牌验证身份，并检查是否有权限下载该文件：

def validate_permission_and_notify(new_file):
    user_token = os.getenv("USER_ID_TOKEN")
    try:
        # 验证ID令牌有效性，获取用户ID
        decoded_token = auth.verify_id_token(user_token)
        user_id = decoded_token['uid']
        
        # 读取用户的权限配置
        user_ref = db.reference(f'users/{user_id}')
        user_data = user_ref.get()
        
        # 检查文件是否在用户允许列表中
        if new_file['file_id'] in user_data.get('allowed_files', []):
            send_download_notification(user_data['fcm_token'], new_file)
    except Exception as e:
        print(f"权限验证失败: {str(e)}")

6. 发送下载通知 #

利用FCM给用户推送新文件通知：

def send_download_notification(fcm_token, file_data):
    message = messaging.Message(
        notification=messaging.Notification(
            title="新文件可下载",
            body=f"文件「{file_data['name']}」已上线，点击查看下载链接"
        ),
        token=fcm_token
    )
    try:
        messaging.send(message)
        print("通知已发送")
    except Exception as e:
        print(f"通知发送失败: {str(e)}")

7. 脚本分发与用户操作 #

• 可选：用pyinstaller将脚本打包成可执行文件，降低用户使用门槛：

  pyinstaller --onefile your_script.py
  

• 给每个用户分发：
  • 可执行文件/脚本
  • 专属.env文件（包含用户的ID令牌和Firebase配置）
  • 简单说明：告知用户如何启动脚本，以及定期更新ID令牌的方法

8. 安全优化要点 #

• 限制服务账号权限：仅授予实时数据库只读权限，防止用户篡改数据
• 令牌有效期处理：ID令牌有效期为1小时，脚本可添加自动刷新逻辑，或提示用户定期更新
• 文件权限细化：在文件节点添加allowed_user_ids字段，直接指定允许下载的用户ID，提升验证效率

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