vf_react/apns_push.py

"""
APNs 推送通知模块
用于向 iOS 设备发送推送通知

使用前需要安装:
pip install PyAPNs2

配置:
1. 将 AuthKey_HSF578B626.p8 文件放到服务器安全目录
2. 设置环境变量或修改下面的配置
"""

import os
from datetime import datetime
from typing import List, Optional
from apns2.client import APNsClient, NotificationPriority
from apns2.payload import Payload, PayloadAlert
from apns2.credentials import TokenCredentials

# ==================== APNs 配置 ====================

# APNs 认证配置
APNS_KEY_ID = 'HSF578B626'
APNS_TEAM_ID = '6XML2LHR2J'
APNS_BUNDLE_ID = 'com.valuefrontier.meagent'

# Auth Key 文件路径（基于当前文件位置）
_BASE_DIR = os.path.dirname(os.path.abspath(__file__))
APNS_AUTH_KEY_PATH = os.environ.get(
    'APNS_AUTH_KEY_PATH',
    os.path.join(_BASE_DIR, 'cert', 'AuthKey_HSF578B626.p8')
)

# 是否使用沙盒环境（开发测试用 True，生产用 False）
APNS_USE_SANDBOX = os.environ.get('APNS_USE_SANDBOX', 'false').lower() == 'true'

# ==================== APNs 客户端初始化 ====================

_apns_client = None


def get_apns_client():
    """获取 APNs 客户端（单例）"""
    global _apns_client

    if _apns_client is None:
        if not os.path.exists(APNS_AUTH_KEY_PATH):
            print(f"[APNs] 警告: Auth Key 文件不存在: {APNS_AUTH_KEY_PATH}")
            return None

        try:
            credentials = TokenCredentials(
                auth_key_path=APNS_AUTH_KEY_PATH,
                auth_key_id=APNS_KEY_ID,
                team_id=APNS_TEAM_ID
            )
            _apns_client = APNsClient(
                credentials=credentials,
                use_sandbox=APNS_USE_SANDBOX
            )
            print(f"[APNs] 客户端初始化成功 (sandbox={APNS_USE_SANDBOX})")
        except Exception as e:
            print(f"[APNs] 客户端初始化失败: {e}")
            return None

    return _apns_client


# ==================== 推送函数 ====================

def send_push_notification(
    device_tokens: List[str],
    title: str,
    body: str,
    data: Optional[dict] = None,
    badge: int = 1,
    sound: str = "default",
    priority: int = 10
) -> dict:
    """
    发送推送通知到多个设备

    Args:
        device_tokens: 设备 Token 列表
        title: 通知标题
        body: 通知内容
        data: 自定义数据（会传递给 App）
        badge: 角标数字
        sound: 提示音
        priority: 优先级 (10=立即, 5=省电)

    Returns:
        dict: {success: int, failed: int, errors: list}
    """
    client = get_apns_client()
    if not client:
        return {'success': 0, 'failed': len(device_tokens), 'errors': ['APNs 客户端未初始化']}

    # 构建通知内容
    alert = PayloadAlert(title=title, body=body)
    payload = Payload(
        alert=alert,
        sound=sound,
        badge=badge,
        custom=data or {}
    )

    # 设置优先级
    notification_priority = NotificationPriority.Immediate if priority == 10 else NotificationPriority.PowerConsideration

    results = {'success': 0, 'failed': 0, 'errors': []}

    for token in device_tokens:
        try:
            client.send_notification(
                token_hex=token,
                notification=payload,
                topic=APNS_BUNDLE_ID,
                priority=notification_priority
            )
            results['success'] += 1
            print(f"[APNs] 推送成功: {token[:20]}...")
        except Exception as e:
            results['failed'] += 1
            results['errors'].append(f"{token[:20]}...: {str(e)}")
            print(f"[APNs] 推送失败 {token[:20]}...: {e}")

    return results


def send_event_notification(event, device_tokens: List[str]) -> dict:
    """
    发送事件推送通知

    Args:
        event: Event 模型实例
        device_tokens: 设备 Token 列表

    Returns:
        dict: 推送结果
    """
    if not device_tokens:
        return {'success': 0, 'failed': 0, 'errors': ['没有设备 Token']}

    # 根据重要性设置标题
    importance_labels = {
        'S': '重大事件',
        'A': '重要事件',
        'B': '一般事件',
        'C': '普通事件'
    }
    title = f"【{importance_labels.get(event.importance, '新事件')}】"

    # 通知内容（截断过长的标题）
    body = event.title[:100] + ('...' if len(event.title) > 100 else '')

    # 自定义数据
    data = {
        'event_id': event.id,
        'importance': event.importance,
        'title': event.title,
        'event_type': event.event_type,
        'created_at': event.created_at.isoformat() if event.created_at else None
    }

    return send_push_notification(
        device_tokens=device_tokens,
        title=title,
        body=body,
        data=data,
        badge=1,
        priority=10 if event.importance in ['S', 'A'] else 5
    )


# ==================== 数据库操作（需要在 app.py 中实现） ====================

def get_all_device_tokens(db_session) -> List[str]:
    """
    获取所有活跃的设备 Token
    需要在 app.py 中实现实际的数据库查询
    """
    # 示例 SQL:
    # SELECT device_token FROM user_device_tokens WHERE is_active = 1
    pass


def get_subscribed_device_tokens(db_session, importance_filter: List[str] = None) -> List[str]:
    """
    获取订阅了推送的设备 Token

    Args:
        importance_filter: 重要性过滤，如 ['S', 'A'] 表示只获取订阅了重要事件的用户
    """
    # 示例 SQL:
    # SELECT device_token FROM user_device_tokens
    # WHERE is_active = 1
    # AND (subscribe_all = 1 OR subscribe_importance IN ('S', 'A'))
    pass


# ==================== 在 app.py 中调用示例 ====================
"""
在 app.py 的 broadcast_new_event() 函数中添加:

from apns_push import send_event_notification

def broadcast_new_event(event):
    # ... 现有的 WebSocket 推送代码 ...

    # 添加 APNs 推送（只推送重要事件）
    if event.importance in ['S', 'A']:
        try:
            # 获取所有设备 token
            device_tokens = db.session.query(UserDeviceToken.device_token).filter(
                UserDeviceToken.is_active == True
            ).all()
            tokens = [t[0] for t in device_tokens]

            if tokens:
                result = send_event_notification(event, tokens)
                print(f"[APNs] 事件推送结果: {result}")
        except Exception as e:
            print(f"[APNs] 推送失败: {e}")
"""