|
|
@@ -89,8 +89,12 @@ async def create_message(
|
|
|
|
|
|
# 2. 确定接收者 (Receiver Resolution)
|
|
|
final_receiver_id = None
|
|
|
+ is_broadcast = getattr(message_in, "is_broadcast", False)
|
|
|
|
|
|
- if message_in.receiver_id:
|
|
|
+ if is_broadcast:
|
|
|
+ if message_in.type != MessageType.NOTIFICATION:
|
|
|
+ raise HTTPException(status_code=403, detail="广播模式仅支持系统通知 (NOTIFICATION)")
|
|
|
+ elif message_in.receiver_id:
|
|
|
# 方式 A: 直接指定 User ID
|
|
|
final_receiver_id = message_in.receiver_id
|
|
|
user = db.query(User).filter(User.id == final_receiver_id).first()
|
|
|
@@ -122,7 +126,7 @@ async def create_message(
|
|
|
)
|
|
|
final_receiver_id = mapping.user_id
|
|
|
else:
|
|
|
- raise HTTPException(status_code=400, detail="必须指定 receiver_id 或 (app_id + app_user_id)")
|
|
|
+ raise HTTPException(status_code=400, detail="必须指定 receiver_id 或 (app_id + app_user_id)，或者设置 is_broadcast=True")
|
|
|
|
|
|
# 3. 处理 SSO 跳转链接 (Link Generation)
|
|
|
final_action_url = message_in.action_url
|
|
|
@@ -152,49 +156,110 @@ async def create_message(
|
|
|
pass # 提取失败则保持原样
|
|
|
|
|
|
# 4. 创建消息
|
|
|
- message = Message(
|
|
|
- sender_id=sender_id,
|
|
|
- receiver_id=final_receiver_id,
|
|
|
- app_id=app_id_int, # 使用 Integer ID 存储到数据库
|
|
|
- type=message_in.type,
|
|
|
- content_type=message_in.content_type,
|
|
|
- title=message_in.title,
|
|
|
- content=content_val,
|
|
|
- action_url=final_action_url,
|
|
|
- action_text=message_in.action_text
|
|
|
- )
|
|
|
- db.add(message)
|
|
|
- db.commit()
|
|
|
- db.refresh(message)
|
|
|
-
|
|
|
- # 5. 触发实时推送 (WebSocket)
|
|
|
- # 处理用于推送的消息内容 (签名)
|
|
|
- processed_msg = _process_message_content(message)
|
|
|
-
|
|
|
- push_payload = {
|
|
|
- "type": "NEW_MESSAGE",
|
|
|
- "data": {
|
|
|
- "id": processed_msg.id,
|
|
|
- "type": processed_msg.type,
|
|
|
- "content_type": processed_msg.content_type,
|
|
|
- "title": processed_msg.title,
|
|
|
- "content": processed_msg.content, # 使用签名后的 URL
|
|
|
- "action_url": processed_msg.action_url,
|
|
|
- "action_text": processed_msg.action_text,
|
|
|
- "sender_name": "系统通知" if not sender_id else "用户私信", # 简化处理
|
|
|
- "sender_id": sender_id, # Add sender_id for frontend to decide left/right
|
|
|
- "created_at": str(processed_msg.created_at),
|
|
|
- # 附加应用信息，便于前端按应用拆分系统通知会话
|
|
|
- "app_id": message.app_id,
|
|
|
- "app_name": message.app.app_name if message.app else None,
|
|
|
+ if is_broadcast:
|
|
|
+ # 查找所有活跃用户 (is_deleted=0)
|
|
|
+ active_users = db.query(User.id).filter(
|
|
|
+ User.status == "ACTIVE",
|
|
|
+ User.is_deleted == 0
|
|
|
+ ).all()
|
|
|
+
|
|
|
+ if not active_users:
|
|
|
+ raise HTTPException(status_code=400, detail="没有可发送的活跃用户")
|
|
|
+
|
|
|
+ messages_to_insert = []
|
|
|
+ for u in active_users:
|
|
|
+ messages_to_insert.append(
|
|
|
+ Message(
|
|
|
+ sender_id=sender_id,
|
|
|
+ receiver_id=u.id,
|
|
|
+ app_id=app_id_int,
|
|
|
+ type=message_in.type,
|
|
|
+ content_type=message_in.content_type,
|
|
|
+ title=message_in.title,
|
|
|
+ content=content_val,
|
|
|
+ action_url=final_action_url,
|
|
|
+ action_text=message_in.action_text
|
|
|
+ )
|
|
|
+ )
|
|
|
+
|
|
|
+ db.add_all(messages_to_insert)
|
|
|
+ db.commit()
|
|
|
+
|
|
|
+ # 使用第一条消息进行签名作为接口的返回值
|
|
|
+ db.refresh(messages_to_insert[0])
|
|
|
+ processed_msg = _process_message_content(messages_to_insert[0])
|
|
|
+
|
|
|
+ # 提取推送所需数据，避免在异步任务中触发延迟加载
|
|
|
+ push_tasks = [{"id": msg.id, "receiver_id": msg.receiver_id} for msg in messages_to_insert]
|
|
|
+
|
|
|
+ async def send_broadcast_ws():
|
|
|
+ for t in push_tasks:
|
|
|
+ push_payload = {
|
|
|
+ "type": "NEW_MESSAGE",
|
|
|
+ "data": {
|
|
|
+ "id": t["id"],
|
|
|
+ "type": processed_msg.type,
|
|
|
+ "content_type": processed_msg.content_type,
|
|
|
+ "title": processed_msg.title,
|
|
|
+ "content": processed_msg.content,
|
|
|
+ "action_url": processed_msg.action_url,
|
|
|
+ "action_text": processed_msg.action_text,
|
|
|
+ "sender_name": "系统通知" if not sender_id else "用户私信",
|
|
|
+ "sender_id": sender_id,
|
|
|
+ "created_at": str(processed_msg.created_at),
|
|
|
+ "app_id": processed_msg.app_id,
|
|
|
+ "app_name": processed_msg.app_name,
|
|
|
+ }
|
|
|
+ }
|
|
|
+ await manager.send_personal_message(push_payload, t["receiver_id"])
|
|
|
+
|
|
|
+ background_tasks.add_task(send_broadcast_ws)
|
|
|
+ return processed_msg
|
|
|
+
|
|
|
+ else:
|
|
|
+ message = Message(
|
|
|
+ sender_id=sender_id,
|
|
|
+ receiver_id=final_receiver_id,
|
|
|
+ app_id=app_id_int, # 使用 Integer ID 存储到数据库
|
|
|
+ type=message_in.type,
|
|
|
+ content_type=message_in.content_type,
|
|
|
+ title=message_in.title,
|
|
|
+ content=content_val,
|
|
|
+ action_url=final_action_url,
|
|
|
+ action_text=message_in.action_text
|
|
|
+ )
|
|
|
+ db.add(message)
|
|
|
+ db.commit()
|
|
|
+ db.refresh(message)
|
|
|
+
|
|
|
+ # 5. 触发实时推送 (WebSocket)
|
|
|
+ # 处理用于推送的消息内容 (签名)
|
|
|
+ processed_msg = _process_message_content(message)
|
|
|
+
|
|
|
+ push_payload = {
|
|
|
+ "type": "NEW_MESSAGE",
|
|
|
+ "data": {
|
|
|
+ "id": processed_msg.id,
|
|
|
+ "type": processed_msg.type,
|
|
|
+ "content_type": processed_msg.content_type,
|
|
|
+ "title": processed_msg.title,
|
|
|
+ "content": processed_msg.content, # 使用签名后的 URL
|
|
|
+ "action_url": processed_msg.action_url,
|
|
|
+ "action_text": processed_msg.action_text,
|
|
|
+ "sender_name": "系统通知" if not sender_id else "用户私信", # 简化处理
|
|
|
+ "sender_id": sender_id, # Add sender_id for frontend to decide left/right
|
|
|
+ "created_at": str(processed_msg.created_at),
|
|
|
+ # 附加应用信息，便于前端按应用拆分系统通知会话
|
|
|
+ "app_id": message.app_id,
|
|
|
+ "app_name": message.app.app_name if message.app else None,
|
|
|
+ }
|
|
|
}
|
|
|
- }
|
|
|
+
|
|
|
+ # 使用后台任务发送 WS 消息，避免阻塞 HTTP 响应
|
|
|
+ # 如果是发给自己，receiver_id == sender_id，ws 会收到一次
|
|
|
+ background_tasks.add_task(manager.send_personal_message, push_payload, final_receiver_id)
|
|
|
|
|
|
- # 使用后台任务发送 WS 消息，避免阻塞 HTTP 响应
|
|
|
- # 如果是发给自己，receiver_id == sender_id，ws 会收到一次
|
|
|
- background_tasks.add_task(manager.send_personal_message, push_payload, final_receiver_id)
|
|
|
-
|
|
|
- return processed_msg
|
|
|
+ return processed_msg
|
|
|
|
|
|
@router.get("/", response_model=List[MessageResponse])
|
|
|
def read_messages(
|
liuq