迁移至新Firebase Cloud Messaging API实现主题消息推送问题咨询

已禁用FCM旧版Legacy API，切换至V1 API后，原主题订阅代码迁移后无法正常运行，以下是正确的实现方案：

原有可运行代码（旧版API） #

...
@classmethod
def subscribe_to_topic(cls, device_id, topics: list):
    success = []
    for topic in topics:
        try:
            url = f'https://iid.googleapis.com/iid/v1/{device_id}/rel/topics/{topic}'
            headers = _get_header()
            result = requests.post(url, headers=headers)
            success.append(True) if result.status_code == 200 else success.append(False)
            logger.info(f'subscribe user <{device_id}> in topic <{topic}> with status code: {result.status_code}')
        except Exception as e:
            logger.error(f'{e}')
    return all(success)

迁移后尝试的代码（无法正常运行） #

...
@classmethod
def subscribe_to_topic(cls, device_id, topics: list):
    """Subscribes a device to a topic using FCM v1 API."""
    payload = {"registrationTokens": [device_id]}
    success = []
    for topic in topics:
        try:
            url = (f"https://fcm.googleapis.com/v1/"
                   f"projects/{environ.get('FCM_PROJECT_NAME')}/topics/{topic}:subscribe")
            headers = _get_header()
            result = requests.post(url, headers=headers, json=payload)
            success.append(True) if result.status_code == 200 else success.append(False)
            logger.info(f'subscribe user <{device_id}> in topic <{topic}> with status code: {result.status_code}')
        except Exception as e:
            logger.error(f'{e}')
    return all(success)

补充背景 #

• 原有代码支持单设备订阅多个主题及消息推送
• 需完全迁移至FCM V1 API

关键修正点与实现示例 #

1. 替换认证方式：使用OAuth2令牌而非Server Key #

FCM V1 API不再支持旧的Server Key认证，必须使用Google服务账号的OAuth2访问令牌。以下是获取令牌的方法：

import requests
from google.oauth2 import service_account
from google.auth.transport.requests import Request

def get_fcm_v1_headers():
    # 加载服务账号密钥文件（从Firebase控制台下载）
    credentials = service_account.Credentials.from_service_account_file(
        'path/to/service-account-key.json',
        scopes=['https://www.googleapis.com/auth/firebase.messaging']
    )
    # 如果令牌过期，刷新获取新令牌
    if not credentials.valid:
        credentials.refresh(Request())
    return {
        'Authorization': f'Bearer {credentials.token}',
        'Content-Type': 'application/json'
    }

2. 修正主题订阅代码 #

V1 API的主题订阅端点格式正确，但需注意响应处理（可能存在部分成功的情况），以下是修正后的订阅代码：

import os
import requests
import logging

logger = logging.getLogger(__name__)

class FCMService:
    @classmethod
    def subscribe_to_topic(cls, device_id, topics: list):
        """使用FCM V1 API订阅设备到多个主题"""
        project_id = os.environ.get('FCM_PROJECT_NAME')
        if not project_id:
            logger.error('FCM_PROJECT_NAME环境变量未设置')
            return False
        
        success = []
        headers = get_fcm_v1_headers()
        payload = {"registrationTokens": [device_id]}

        for topic in topics:
            try:
                url = f"https://fcm.googleapis.com/v1/projects/{project_id}/topics/{topic}:subscribe"
                response = requests.post(url, headers=headers, json=payload)
                
                # V1 API成功状态码为200，同时需检查响应体中的错误
                if response.status_code == 200:
                    response_data = response.json()
                    if response_data.get('failureCount', 0) == 0:
                        success.append(True)
                        logger.info(f'设备<{device_id}>订阅主题<{topic}>成功')
                    else:
                        success.append(False)
                        logger.error(f'设备<{device_id}>订阅主题<{topic}>失败: {response_data}')
                else:
                    success.append(False)
                    logger.error(f'订阅主题<{topic}>请求失败，状态码: {response.status_code}, 响应: {response.text}')
            except Exception as e:
                success.append(False)
                logger.error(f'订阅主题<{topic}>发生异常: {str(e)}')
        
        return all(success)

3. FCM V1 API消息推送示例（主题推送） #

以下是向主题推送消息的实现：

@classmethod
    def send_topic_message(cls, topic: str, title: str, body: str):
        """使用FCM V1 API向主题推送消息"""
        project_id = os.environ.get('FCM_PROJECT_NAME')
        if not project_id:
            logger.error('FCM_PROJECT_NAME环境变量未设置')
            return False
        
        url = f"https://fcm.googleapis.com/v1/projects/{project_id}/messages:send"
        headers = get_fcm_v1_headers()
        
        # 构建推送消息体（支持Android/iOS/Web多平台）
        payload = {
            "message": {
                "topic": topic,
                "notification": {
                    "title": title,
                    "body": body
                },
                "android": {
                    "notification": {
                        "sound": "default"
                    }
                },
                "apns": {
                    "payload": {
                        "aps": {
                            "sound": "default"
                        }
                    }
                }
            }
        }

        try:
            response = requests.post(url, headers=headers, json=payload)
            if response.status_code == 200:
                logger.info(f'向主题<{topic}>推送消息成功')
                return True
            else:
                logger.error(f'推送消息失败，状态码: {response.status_code}, 响应: {response.text}')
                return False
        except Exception as e:
            logger.error(f'推送消息发生异常: {str(e)}')
            return False

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