ios app

apns_push.py

@@ -0,0 +1,225 @@
+"""
+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}")
+"""