diff --git a/banban-mini/src/app.config.ts b/banban-mini/src/app.config.ts index 4947029..7309acc 100644 --- a/banban-mini/src/app.config.ts +++ b/banban-mini/src/app.config.ts @@ -3,6 +3,7 @@ export default defineAppConfig({ "pages/login/index", "pages/bind/index", "pages/device/index", + "pages/notification/index", "pages/chat/index", "pages/chat/detail/index", "pages/location/index", diff --git a/banban-mini/src/pages/chat/detail/index.tsx b/banban-mini/src/pages/chat/detail/index.tsx index fbb68be..4a14443 100644 --- a/banban-mini/src/pages/chat/detail/index.tsx +++ b/banban-mini/src/pages/chat/detail/index.tsx @@ -109,6 +109,7 @@ const RECORD_LONG_PRESS_DELAY_MS = 350 type LoadChatOptions = { showLoading?: boolean + scrollToBottom?: 'jump' | 'animate' | false } function compareVersion(left: string, right: string): number { @@ -232,6 +233,7 @@ export default function ChatDetail() { const [playingMessageId, setPlayingMessageId] = useState(null) const [recordHint, setRecordHint] = useState('长按开始留言') const [scrollTarget, setScrollTarget] = useState('') + const [scrollWithAnimation, setScrollWithAnimation] = useState(false) const audioContextRef = useRef(null) const recorderManagerRef = useRef(null) const activeConversationIdRef = useRef(initialConversationId) @@ -251,12 +253,14 @@ export default function ChatDetail() { const recordStartPendingRef = useRef(false) const recordingActiveRef = useRef(false) - const scrollToBottom = () => { + const scrollToBottom = (animated: boolean) => { if (scrollTimerRef.current) { clearTimeout(scrollTimerRef.current) } + setScrollWithAnimation(animated) setScrollTarget('') scrollTimerRef.current = setTimeout(() => { + setScrollWithAnimation(animated) setScrollTarget(CHAT_BOTTOM_ANCHOR_ID) scrollTimerRef.current = null }, 80) @@ -373,8 +377,9 @@ export default function ChatDetail() { } const msgs = await getMessages(conversationId, conversationSource, decodedConversationTypeName) setMessages(msgs) - if (msgs.length > 0) { - scrollToBottom() + const scrollMode = options.scrollToBottom ?? (!hasLoadedChatRef.current ? 'jump' : false) + if (msgs.length > 0 && scrollMode) { + scrollToBottom(scrollMode === 'animate') } hasLoadedChatRef.current = true } catch (error: any) { @@ -480,9 +485,15 @@ export default function ChatDetail() { if (response.conversation_id && response.conversation_id !== activeConversationIdRef.current) { skipNextAutoLoadRef.current = true setActiveConversationId(response.conversation_id) - await latestLoadChatDataRef.current(response.conversation_id, { showLoading: false }) + await latestLoadChatDataRef.current(response.conversation_id, { + showLoading: false, + scrollToBottom: 'animate', + }) } else { - await latestLoadChatDataRef.current(activeConversationIdRef.current, { showLoading: false }) + await latestLoadChatDataRef.current(activeConversationIdRef.current, { + showLoading: false, + scrollToBottom: 'animate', + }) } Taro.showToast({ title: '留言已发送', icon: 'success' }) } catch (error: any) { @@ -752,7 +763,12 @@ export default function ChatDetail() { - + {loading ? ( 加载中... diff --git a/banban-mini/src/pages/notification/index.scss b/banban-mini/src/pages/notification/index.scss new file mode 100644 index 0000000..da5ceb7 --- /dev/null +++ b/banban-mini/src/pages/notification/index.scss @@ -0,0 +1,37 @@ +.notification-page { + min-height: 100vh; + background: #f5f7fa; + display: flex; + align-items: center; + justify-content: center; + padding: 48px; + box-sizing: border-box; +} + +.notification-panel { + width: 100%; + max-width: 520px; + background: #ffffff; + border-radius: 8px; + padding: 40px 32px; + box-sizing: border-box; + box-shadow: 0 10px 24px rgba(31, 41, 55, 0.08); + display: flex; + flex-direction: column; + align-items: center; + gap: 16px; +} + +.notification-title { + color: #1f2937; + font-size: 34px; + font-weight: 700; + line-height: 1.3; +} + +.notification-message { + color: #6b7280; + font-size: 28px; + line-height: 1.5; + text-align: center; +} diff --git a/banban-mini/src/pages/notification/index.tsx b/banban-mini/src/pages/notification/index.tsx new file mode 100644 index 0000000..3b0f392 --- /dev/null +++ b/banban-mini/src/pages/notification/index.tsx @@ -0,0 +1,59 @@ +import { View, Text } from '@tarojs/components' +import { useEffect, useState } from 'react' +import Taro, { useRouter } from '@tarojs/taro' +import { getToken } from '@/services/auth' +import { resolveWechatMpNotificationLink } from '@/services/notification' +import './index.scss' + +function buildQuery(params: Record): string { + return Object.entries(params) + .filter(([, value]) => value !== undefined && value !== null && String(value) !== '') + .map(([key, value]) => `${encodeURIComponent(key)}=${encodeURIComponent(String(value))}`) + .join('&') +} + +function buildCurrentPageUrl(token: string): string { + return `/pages/notification/index?token=${encodeURIComponent(token)}` +} + +export default function NotificationEntry() { + const router = useRouter() + const token = String(router.params.token || '').trim() + const [message, setMessage] = useState('正在打开留言...') + + useEffect(() => { + const openNotification = async () => { + if (!token) { + setMessage('通知链接无效') + return + } + + if (!getToken()) { + Taro.setStorageSync('postLoginRedirect', buildCurrentPageUrl(token)) + Taro.reLaunch({ url: '/pages/login/index' }) + return + } + + try { + const result = await resolveWechatMpNotificationLink(token) + const query = buildQuery(result.params || {}) + const route = result.route.startsWith('/') ? result.route : `/${result.route}` + Taro.redirectTo({ url: query ? `${route}?${query}` : route }) + } catch (error: any) { + console.error('[notification] resolve failed:', error) + setMessage(error?.message || '通知链接已失效') + } + } + + void openNotification() + }, [token]) + + return ( + + + 伴伴 + {message} + + + ) +} diff --git a/banban-mini/src/services/chat.ts b/banban-mini/src/services/chat.ts index 7ad9237..4031117 100644 --- a/banban-mini/src/services/chat.ts +++ b/banban-mini/src/services/chat.ts @@ -122,6 +122,8 @@ interface ChildConversationItem { last_message_preview?: string | null last_message_at?: string | null message_count: number + created_at: string + updated_at: string } interface ChildConversationListResponse { @@ -307,9 +309,10 @@ function canSendParentConversation(parentUserId: number | null, context: Binding return Boolean(parentUserId && parentUserId === context.currentUserId) } -function getConversationRank(conversation: ChatConversation): number { - if (isParentChildConversation(conversation.conversationTypeName)) return 0 - return 1 +function compareConversationsByNewest(left: ChatConversation, right: ChatConversation): number { + const timeDiff = parseTime(right.sortAt) - parseTime(left.sortAt) + if (timeDiff !== 0) return timeDiff + return right.key.localeCompare(left.key) } async function getBindingContext(): Promise { @@ -385,6 +388,7 @@ function toImConversation(item: ChildConversationItem, context: BindingContext): const parentUserId = isParentConversation ? parseNumericId(item.peer_id) : null const canSend = isParentConversation ? canSendParentConversation(parentUserId, context) : false const parentConversationName = getParentConversationName(parentUserId, item.peer_id, item.peer_name, context) + const sortAt = item.last_message_at || item.updated_at || item.created_at return { id: item.conversation_id, @@ -400,8 +404,8 @@ function toImConversation(item: ChildConversationItem, context: BindingContext): : '其他家长和孩子的留言记录' : meta.description, lastMessage: item.last_message_preview || (isParentConversation ? (canSend ? '还没有消息，点进去发送第一条' : '暂无留言记录') : '暂无消息'), - time: formatListTime(item.last_message_at), - sortAt: item.last_message_at || '', + time: formatListTime(sortAt), + sortAt, conversationTypeName: item.conversation_type_name, peerId: item.peer_id, childId: context.childId, @@ -496,11 +500,7 @@ export async function getConversations(): Promise { } } - return conversations.sort((left, right) => { - const rankDiff = getConversationRank(left) - getConversationRank(right) - if (rankDiff !== 0) return rankDiff - return parseTime(right.sortAt) - parseTime(left.sortAt) - }) + return conversations.sort(compareConversationsByNewest) } export async function getMessages( diff --git a/banban-mini/src/services/notification.ts b/banban-mini/src/services/notification.ts new file mode 100644 index 0000000..fa9d489 --- /dev/null +++ b/banban-mini/src/services/notification.ts @@ -0,0 +1,10 @@ +import { request } from './api' + +export interface WechatMpNotificationLink { + route: string + params: Record +} + +export function resolveWechatMpNotificationLink(token: string): Promise { + return request(`/banban/wechat-mp/notification-links/${encodeURIComponent(token)}`) +} diff --git a/talkingq-url/banban/routers/device_im.py b/talkingq-url/banban/routers/device_im.py index f7071a5..03de5a5 100644 --- a/talkingq-url/banban/routers/device_im.py +++ b/talkingq-url/banban/routers/device_im.py @@ -103,10 +103,11 @@ async def list_device_conversations( last_message_preview, last_message_at, message_count, - created_at + created_at, + updated_at FROM im_conversations WHERE {where} - ORDER BY COALESCE(last_message_at, created_at) DESC, id DESC + ORDER BY COALESCE(last_message_at, updated_at, created_at) DESC, id DESC LIMIT :fetch_limit """ ), diff --git a/talkingq-url/banban/routers/im.py b/talkingq-url/banban/routers/im.py index 4c8abff..7b4a85e 100644 --- a/talkingq-url/banban/routers/im.py +++ b/talkingq-url/banban/routers/im.py @@ -247,6 +247,8 @@ def _row_to_conversation_item( last_message_preview=row["last_message_preview"], last_message_at=row["last_message_at"], message_count=int(row["message_count"]), + created_at=row["created_at"], + updated_at=row["updated_at"], ) @@ -331,10 +333,11 @@ async def list_child_conversations( last_message_preview, last_message_at, message_count, - created_at + created_at, + updated_at FROM im_conversations WHERE {where} - ORDER BY COALESCE(last_message_at, created_at) DESC, id DESC + ORDER BY COALESCE(last_message_at, updated_at, created_at) DESC, id DESC LIMIT :fetch_limit """ ), diff --git a/talkingq-url/banban/routers/wechat_mp.py b/talkingq-url/banban/routers/wechat_mp.py index e5e3b6f..e8b0e4c 100644 --- a/talkingq-url/banban/routers/wechat_mp.py +++ b/talkingq-url/banban/routers/wechat_mp.py @@ -24,6 +24,11 @@ class WechatMpBindStatusResponse(BaseModel): updated_at: str | None = None +class WechatMpNotificationLinkResponse(BaseModel): + route: str + params: dict[str, str | int | None] + + @router.post("/bind/start", response_model=WechatMpBindStartResponse) async def start_wechat_mp_bind( current_user_id: int = Depends(get_current_user_id), @@ -48,6 +53,21 @@ async def get_wechat_mp_bind_status( ) +@router.get("/notification-links/{token}", response_model=WechatMpNotificationLinkResponse) +async def resolve_wechat_mp_notification_link( + token: str, + current_user_id: int = Depends(get_current_user_id), +) -> WechatMpNotificationLinkResponse: + result = await wechat_mp_notification_service.resolve_notification_link( + token=token, + user_id=current_user_id, + ) + return WechatMpNotificationLinkResponse( + route=str(result["route"]), + params=dict(result["params"]), + ) + + @router.get("/oauth/callback", response_class=HTMLResponse) async def wechat_mp_oauth_callback( code: str = Query(default=""), diff --git a/talkingq-url/banban/schemas/im.py b/talkingq-url/banban/schemas/im.py index cdab21d..7301470 100644 --- a/talkingq-url/banban/schemas/im.py +++ b/talkingq-url/banban/schemas/im.py @@ -14,6 +14,8 @@ class ChildConversationItem(BaseModel): last_message_preview: str | None = None last_message_at: datetime | None = None message_count: int + created_at: datetime + updated_at: datetime class ChildConversationListResponse(BaseModel): diff --git a/talkingq-url/banban/service/im.py b/talkingq-url/banban/service/im.py index c1330d8..de54d17 100644 --- a/talkingq-url/banban/service/im.py +++ b/talkingq-url/banban/service/im.py @@ -47,6 +47,16 @@ PARTICIPANT_TYPE_NAMES = { CHILD_PARTICIPANT_TYPE: "child", } + +@dataclass(frozen=True) +class DeviceParentLeaveMessageTarget: + parent_user_id: int + conversation_id: int + message_id: int + child_id: int + child_name: str | None + device_id: str + _AUDIO_CONTENT_TYPE_TO_EXT = { "audio/mpeg": "mp3", "audio/mp3": "mp3", @@ -644,7 +654,7 @@ class ImService(DatabaseServiceBase): client_msg_id: str | None = None, ext_json: dict[str, Any] | None = None, audio_content: bytes | None = None, - ) -> tuple[DeviceIdentity, ConversationMessageCreateResult]: + ) -> tuple[DeviceIdentity, ConversationMessageCreateResult, list[DeviceParentLeaveMessageTarget]]: normalized_media_file_key = str(media_file_key or "").strip() if not normalized_media_file_key: raise HTTPException(status_code=400, detail="media_file_key is required") @@ -656,6 +666,7 @@ class ImService(DatabaseServiceBase): if not family_identities: raise HTTPException(status_code=404, detail="device family has no members") first_result: ConversationMessageCreateResult | None = None + notification_targets: list[DeviceParentLeaveMessageTarget] = [] for owner_identity in family_identities: member_client_msg_id = (client_msg_id or "").strip() if len(family_identities) > 1 or not member_client_msg_id: @@ -676,6 +687,16 @@ class ImService(DatabaseServiceBase): client_msg_id=member_client_msg_id, ext_json=ext_json, ) + notification_targets.append( + DeviceParentLeaveMessageTarget( + parent_user_id=owner_identity.owner_user_id, + conversation_id=member_result.conversation_id, + message_id=member_result.message.id, + child_id=owner_identity.child_id, + child_name=owner_identity.child_name, + device_id=owner_identity.device_id, + ) + ) if audio_content: await self.schedule_message_media_duration_parse( message_id=member_result.message.id, @@ -695,6 +716,7 @@ class ImService(DatabaseServiceBase): child_name=family_identities[0].child_name, ), first_result, + notification_targets, ) except Exception: await db_session.rollback() diff --git a/talkingq-url/banban/service/wechat_mp_notification.py b/talkingq-url/banban/service/wechat_mp_notification.py index 7c05af1..a38ab60 100644 --- a/talkingq-url/banban/service/wechat_mp_notification.py +++ b/talkingq-url/banban/service/wechat_mp_notification.py @@ -1,20 +1,34 @@ import asyncio +import base64 +import hashlib +import hmac +import json import secrets import time from collections.abc import Mapping +from dataclasses import dataclass from datetime import datetime, timedelta from typing import Any from urllib.parse import urlencode import httpx +from fastapi import HTTPException from sqlalchemy import text +from banban.service.im import DeviceParentLeaveMessageTarget from config import settings from services.database_service_base import DatabaseServiceBase from utils.logger import session_logger WECHAT_MP_ACCOUNT_TYPE = "service_account" +WECHAT_MP_NOTIFICATION_TOKEN_VERSION = 1 + + +@dataclass(frozen=True) +class WechatMpRecipient: + user_id: int + openid: str class WechatMpNotificationService(DatabaseServiceBase): @@ -99,6 +113,113 @@ class WechatMpNotificationService(DatabaseServiceBase): finally: await db_session.close() + async def resolve_notification_link(self, *, token: str, user_id: int) -> Mapping[str, Any]: + payload = self._decode_notification_token(token) + if payload.get("type") != "leave_message": + raise HTTPException(status_code=400, detail="unsupported notification type") + if int(payload.get("user_id") or 0) != user_id: + raise HTTPException(status_code=403, detail="notification does not belong to current user") + + conversation_id = int(payload.get("conversation_id") or 0) + message_id = int(payload.get("message_id") or 0) + child_id = int(payload.get("child_id") or 0) + device_id = str(payload.get("device_id") or "").strip() + if conversation_id <= 0 or message_id <= 0 or child_id <= 0: + raise HTTPException(status_code=400, detail="invalid notification token") + + db_session = await self.get_session() + try: + user_id_str = str(user_id) + child_id_str = str(child_id) + result = await db_session.execute( + text( + """ + SELECT + c.id AS conversation_id, + ch.child_id, + ch.child_name, + db.device_id + FROM im_conversations AS c + JOIN im_messages AS m + ON m.id = :message_id + AND m.conversation_id = c.id + AND m.deleted_at IS NULL + JOIN children AS ch + ON ch.child_id = :child_id + AND ch.status = 1 + LEFT JOIN parent_child_relations AS pcr + ON pcr.child_id = ch.child_id + AND pcr.user_id = :user_id + AND pcr.status = 1 + LEFT JOIN device_bindings AS db + ON db.child_id = ch.child_id + AND db.status = 1 + LEFT JOIN device_family_members AS dfm + ON dfm.device_id = db.device_id + AND dfm.user_id = :user_id + AND dfm.status = 1 + WHERE c.id = :conversation_id + AND c.status = 1 + AND c.conversation_type = 2 + AND ( + ( + c.participant_a_type = 2 + AND c.participant_a_id = :child_id_str + AND c.participant_b_type = 1 + AND c.participant_b_id = :user_id_str + ) + OR ( + c.participant_b_type = 2 + AND c.participant_b_id = :child_id_str + AND c.participant_a_type = 1 + AND c.participant_a_id = :user_id_str + ) + ) + AND ( + pcr.id IS NOT NULL + OR db.owner_user_id = :user_id + OR dfm.id IS NOT NULL + ) + LIMIT 1 + """ + ), + { + "message_id": message_id, + "conversation_id": conversation_id, + "child_id": child_id, + "child_id_str": child_id_str, + "user_id": user_id, + "user_id_str": user_id_str, + }, + ) + row = result.mappings().first() + if not row: + raise HTTPException(status_code=404, detail="notification target not found") + + resolved_child_name = str(row.get("child_name") or payload.get("child_name") or "").strip() + resolved_device_id = str(row.get("device_id") or device_id).strip() + return { + "route": "pages/chat/detail/index", + "params": { + "id": conversation_id, + "source": "im", + "name": "家长沟通", + "peerKind": "parent", + "roleKey": "", + "conversationTypeName": "parent_child", + "peerId": user_id_str, + "childId": child_id, + "childName": resolved_child_name, + "parentUserId": user_id, + "deviceId": resolved_device_id, + "channelLabel": "微信小程序", + "canSend": "1", + "messageId": message_id, + }, + } + finally: + await db_session.close() + def schedule_leave_message_notification(self, **kwargs: Any) -> None: self._schedule_best_effort( self.notify_leave_message_best_effort(**kwargs), @@ -120,22 +241,35 @@ class WechatMpNotificationService(DatabaseServiceBase): *, device_id: str, child_name: str | None = None, + targets: list[DeviceParentLeaveMessageTarget | Mapping[str, Any]] | None = None, ) -> None: if not self.is_configured(): return - recipients = await self._list_device_family_mp_openids(device_id=device_id) + recipients = await self._list_device_family_mp_recipients(device_id=device_id) if not recipients: session_logger.info(device_id, "wechat_mp", "留言公众号推送跳过：没有绑定服务号openid的家长") return - await self._send_to_recipients( - device_id=device_id, - notification_type="leave_message", - recipients=recipients, - title=child_name or device_id, - condition="收到新的设备留言", - page=settings.wechat_mp_chat_page, - dedup_seconds=max(0, int(settings.wechat_mp_leave_message_dedup_seconds or 0)), - ) + target_by_user_id = { + self._target_int(target, "parent_user_id"): target + for target in targets or [] + if self._target_int(target, "parent_user_id") > 0 + } + dedup_seconds = max(0, int(settings.wechat_mp_leave_message_dedup_seconds or 0)) + + for recipient in recipients: + page = self._build_leave_message_page( + recipient_user_id=recipient.user_id, + target=target_by_user_id.get(recipient.user_id), + ) + await self._send_to_recipient( + device_id=device_id, + notification_type="leave_message", + openid=recipient.openid, + title=child_name or device_id, + condition="收到新的设备留言", + page=page, + dedup_seconds=dedup_seconds, + ) def schedule_low_battery_notification(self, *, device_id: str, power: Any, child_name: str | None = None) -> None: self._schedule_best_effort( @@ -240,35 +374,62 @@ class WechatMpNotificationService(DatabaseServiceBase): return for openid in recipients: - if self._is_deduped( + await self._send_to_recipient( device_id=device_id, notification_type=notification_type, openid=openid, - dedup_seconds=dedup_seconds, - ): - session_logger.info(device_id, "wechat_mp", f"公众号推送防抖跳过: type={notification_type}") - continue - - result = await self._send_template_message( - openid=openid, - template_id=template_id, title=title, condition=condition, page=page, + dedup_seconds=dedup_seconds, + template_id=template_id, + ) + + async def _send_to_recipient( + self, + *, + device_id: str, + notification_type: str, + openid: str, + title: str, + condition: str, + page: str, + dedup_seconds: int, + template_id: str | None = None, + ) -> None: + resolved_template_id = template_id or self._template_id(notification_type) + if not resolved_template_id: + return + + if self._is_deduped( + device_id=device_id, + notification_type=notification_type, + openid=openid, + dedup_seconds=dedup_seconds, + ): + session_logger.info(device_id, "wechat_mp", f"公众号推送防抖跳过: type={notification_type}") + return + + result = await self._send_template_message( + openid=openid, + template_id=resolved_template_id, + title=title, + condition=condition, + page=page, + ) + if result.get("ok"): + self._mark_sent(device_id=device_id, notification_type=notification_type, openid=openid) + session_logger.info( + device_id, + "wechat_mp", + f"公众号推送成功: type={notification_type}, msgid={result.get('msgid')}", + ) + else: + session_logger.warning( + device_id, + "wechat_mp", + f"公众号推送失败: type={notification_type}, errcode={result.get('errcode')}, errmsg={result.get('errmsg')}", ) - if result.get("ok"): - self._mark_sent(device_id=device_id, notification_type=notification_type, openid=openid) - session_logger.info( - device_id, - "wechat_mp", - f"公众号推送成功: type={notification_type}, msgid={result.get('msgid')}", - ) - else: - session_logger.warning( - device_id, - "wechat_mp", - f"公众号推送失败: type={notification_type}, errcode={result.get('errcode')}, errmsg={result.get('errmsg')}", - ) async def _send_template_message( self, @@ -306,12 +467,16 @@ class WechatMpNotificationService(DatabaseServiceBase): } async def _list_device_family_mp_openids(self, *, device_id: str) -> list[str]: + recipients = await self._list_device_family_mp_recipients(device_id=device_id) + return [recipient.openid for recipient in recipients] + + async def _list_device_family_mp_recipients(self, *, device_id: str) -> list[WechatMpRecipient]: db_session = await self.get_session() try: result = await db_session.execute( text( """ - SELECT pwa.openid + SELECT recipients.user_id, pwa.openid FROM ( SELECT db.owner_user_id AS user_id, @@ -351,15 +516,20 @@ class WechatMpNotificationService(DatabaseServiceBase): "account_type": WECHAT_MP_ACCOUNT_TYPE, }, ) - openids: list[str] = [] + recipients: list[WechatMpRecipient] = [] seen: set[str] = set() for row in result.mappings().all(): openid = str(row["openid"]).strip() if not openid or openid in seen: continue seen.add(openid) - openids.append(openid) - return openids + recipients.append( + WechatMpRecipient( + user_id=int(row["user_id"]), + openid=openid, + ) + ) + return recipients finally: await db_session.close() @@ -560,6 +730,83 @@ class WechatMpNotificationService(DatabaseServiceBase): raise RuntimeError(f"wechat mp user info failed: errcode={data.get('errcode')} errmsg={data.get('errmsg')}") return data + def _build_leave_message_page( + self, + *, + recipient_user_id: int, + target: DeviceParentLeaveMessageTarget | Mapping[str, Any] | None, + ) -> str: + if target is None: + return settings.wechat_mp_chat_page + + child_id = self._target_int(target, "child_id") + conversation_id = self._target_int(target, "conversation_id") + message_id = self._target_int(target, "message_id") + if child_id <= 0 or conversation_id <= 0 or message_id <= 0: + return settings.wechat_mp_chat_page + + token = self._create_notification_token( + { + "type": "leave_message", + "user_id": recipient_user_id, + "conversation_id": conversation_id, + "message_id": message_id, + "child_id": child_id, + "child_name": self._target_str(target, "child_name"), + "device_id": self._target_str(target, "device_id"), + } + ) + page = (settings.wechat_mp_notification_page or "").strip() or "pages/notification/index" + return f"{page}?token={token}" + + def _create_notification_token(self, payload: Mapping[str, Any]) -> str: + now = int(time.time()) + ttl_seconds = max(60, int(settings.wechat_mp_notification_token_expire_minutes or 0) * 60) + normalized_payload = { + "v": WECHAT_MP_NOTIFICATION_TOKEN_VERSION, + "iat": now, + "exp": now + ttl_seconds, + **dict(payload), + } + payload_bytes = json.dumps(normalized_payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + payload_part = self._b64url_encode(payload_bytes) + signature = self._sign_notification_payload(payload_part) + return f"{payload_part}.{signature}" + + def _decode_notification_token(self, token: str) -> Mapping[str, Any]: + normalized_token = str(token or "").strip() + if not normalized_token or "." not in normalized_token: + raise HTTPException(status_code=400, detail="invalid notification token") + payload_part, signature = normalized_token.rsplit(".", 1) + expected_signature = self._sign_notification_payload(payload_part) + if not hmac.compare_digest(signature, expected_signature): + raise HTTPException(status_code=400, detail="invalid notification token") + try: + payload = json.loads(self._b64url_decode(payload_part).decode("utf-8")) + except Exception as exc: + raise HTTPException(status_code=400, detail="invalid notification token") from exc + if not isinstance(payload, dict): + raise HTTPException(status_code=400, detail="invalid notification token") + if int(payload.get("v") or 0) != WECHAT_MP_NOTIFICATION_TOKEN_VERSION: + raise HTTPException(status_code=400, detail="unsupported notification token") + if int(payload.get("exp") or 0) < int(time.time()): + raise HTTPException(status_code=400, detail="notification token expired") + return payload + + def _sign_notification_payload(self, payload_part: str) -> str: + secret = f"wechat-mp-notification:{settings.jwt_secret}".encode("utf-8") + digest = hmac.new(secret, payload_part.encode("ascii"), hashlib.sha256).digest() + return self._b64url_encode(digest) + + @staticmethod + def _b64url_encode(value: bytes) -> str: + return base64.urlsafe_b64encode(value).decode("ascii").rstrip("=") + + @staticmethod + def _b64url_decode(value: str) -> bytes: + padding = "=" * (-len(value) % 4) + return base64.urlsafe_b64decode(f"{value}{padding}") + def _template_id(self, notification_type: str) -> str: default_template_id = (settings.wechat_mp_template_id or "").strip() if notification_type == "leave_message": @@ -579,6 +826,23 @@ class WechatMpNotificationService(DatabaseServiceBase): def _mark_sent(self, *, device_id: str, notification_type: str, openid: str) -> None: self._last_sent_at[(device_id, notification_type, openid)] = time.time() + @staticmethod + def _target_value(target: DeviceParentLeaveMessageTarget | Mapping[str, Any], key: str) -> Any: + if isinstance(target, Mapping): + return target.get(key) + return getattr(target, key) + + @classmethod + def _target_int(cls, target: DeviceParentLeaveMessageTarget | Mapping[str, Any], key: str) -> int: + try: + return int(cls._target_value(target, key) or 0) + except (TypeError, ValueError): + return 0 + + @classmethod + def _target_str(cls, target: DeviceParentLeaveMessageTarget | Mapping[str, Any], key: str) -> str: + return str(cls._target_value(target, key) or "").strip() + @staticmethod def _clip_template_value(value: str, max_len: int) -> str: text = str(value or "").strip() diff --git a/talkingq-url/config.py b/talkingq-url/config.py index bb8de98..a6f9e28 100644 --- a/talkingq-url/config.py +++ b/talkingq-url/config.py @@ -120,6 +120,14 @@ class Settings(BaseSettings): wechat_mp_miniprogram_link_enabled: bool = Field(default=True, validation_alias="WECHAT_MP_MINIPROGRAM_LINK_ENABLED") wechat_mp_miniprogram_appid: str = Field(default="", validation_alias="WECHAT_MP_MINIPROGRAM_APPID") wechat_mp_chat_page: str = Field(default="pages/chat/detail/index", validation_alias="WECHAT_MP_CHAT_PAGE") + wechat_mp_notification_page: str = Field( + default="pages/notification/index", + validation_alias="WECHAT_MP_NOTIFICATION_PAGE", + ) + wechat_mp_notification_token_expire_minutes: int = Field( + default=10080, + validation_alias="WECHAT_MP_NOTIFICATION_TOKEN_EXPIRE_MINUTES", + ) wechat_mp_device_page: str = Field(default="pages/device/index", validation_alias="WECHAT_MP_DEVICE_PAGE") wechat_mp_oauth_redirect_uri: str = Field(default="", validation_alias="WECHAT_MP_OAUTH_REDIRECT_URI") wechat_mp_bind_success_page: str = Field( diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index 658d7a6..e7b24ab 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/talkingq-url/handlers/mqtt_handler.py @@ -492,7 +492,7 @@ class TalkingQMQTTService: media_file_key = str(params.get("media_file_key") or params.get("audio_url") or "").strip() if media_file_key: try: - device_identity, _ = await im_service.create_device_parent_leave_message( + device_identity, _, notification_targets = await im_service.create_device_parent_leave_message( device_id=device_id, media_file_key=media_file_key, media_duration_ms=params.get("media_duration_ms"), @@ -505,6 +505,7 @@ class TalkingQMQTTService: wechat_mp_notification_service.schedule_leave_message_notification( device_id=device_id, child_name=device_identity.child_name, + targets=notification_targets, ) logger.info(device_id, "", f"[短按留言] 设备 {device_id} 留言已写入家长会话") await self._publish( diff --git a/talkingq-url/handlers/websocket_message_handler.py b/talkingq-url/handlers/websocket_message_handler.py index 3501bbe..b165b24 100644 --- a/talkingq-url/handlers/websocket_message_handler.py +++ b/talkingq-url/handlers/websocket_message_handler.py @@ -289,7 +289,7 @@ async def process_parent_leave_message(device_id: str, audio_cache_key: str): content_type=media_mime_type, extension=extension, ) - device_identity, _ = await im_conversation_service.create_device_parent_leave_message( + device_identity, _, notification_targets = await im_conversation_service.create_device_parent_leave_message( device_id=device_id, media_file_key=stored_audio.file_key, media_mime_type=media_mime_type, @@ -305,6 +305,7 @@ async def process_parent_leave_message(device_id: str, audio_cache_key: str): wechat_mp_notification_service.schedule_leave_message_notification( device_id=device_id, child_name=device_identity.child_name, + targets=notification_targets, ) session_logger.info( device_id, diff --git a/talkingq-url/tests/test_wechat_mp_notification.py b/talkingq-url/tests/test_wechat_mp_notification.py index 576f134..c33cb9f 100644 --- a/talkingq-url/tests/test_wechat_mp_notification.py +++ b/talkingq-url/tests/test_wechat_mp_notification.py @@ -1,8 +1,9 @@ from urllib.parse import parse_qs, urlparse import pytest +from fastapi import HTTPException -from banban.service.wechat_mp_notification import WechatMpNotificationService +from banban.service.wechat_mp_notification import WechatMpNotificationService, WechatMpRecipient def configure_enabled_wechat_mp(monkeypatch): @@ -17,6 +18,8 @@ def configure_enabled_wechat_mp(monkeypatch): monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_miniprogram_link_enabled", True) monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_miniprogram_appid", "wx-mini") monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_chat_page", "pages/chat/detail/index") + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_notification_page", "pages/notification/index") + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_notification_token_expire_minutes", 10080) monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_device_page", "pages/device/index") monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_oauth_redirect_uri", "https://example.com/banban/wechat-mp/oauth/callback") monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_leave_message_dedup_seconds", 60) @@ -34,13 +37,83 @@ async def test_notify_leave_message_sends_template_to_unique_subscribed_recipien async def fake_list_recipients(*, device_id): assert device_id == "TalkingQ_XQSN00001007" - return ["openid-1", "openid-2"] + return [ + WechatMpRecipient(user_id=4, openid="openid-1"), + WechatMpRecipient(user_id=5, openid="openid-2"), + ] async def fake_send_template_message(**kwargs): sent.append(kwargs) return {"ok": True, "msgid": "msg-id"} - monkeypatch.setattr(service, "_list_device_family_mp_openids", fake_list_recipients) + monkeypatch.setattr(service, "_list_device_family_mp_recipients", fake_list_recipients) + monkeypatch.setattr(service, "_send_template_message", fake_send_template_message) + + await service.notify_leave_message( + device_id="TalkingQ_XQSN00001007", + child_name="小明", + targets=[ + { + "parent_user_id": 4, + "conversation_id": 6, + "message_id": 209, + "child_id": 8, + "child_name": "小明", + "device_id": "TalkingQ_XQSN00001007", + }, + { + "parent_user_id": 5, + "conversation_id": 7, + "message_id": 210, + "child_id": 8, + "child_name": "小明", + "device_id": "TalkingQ_XQSN00001007", + }, + ], + ) + + assert len(sent) == 2 + assert sent[0] == { + "openid": "openid-1", + "template_id": "template-id", + "title": "小明", + "condition": "收到新的设备留言", + "page": sent[0]["page"], + } + assert sent[1] == { + "openid": "openid-2", + "template_id": "template-id", + "title": "小明", + "condition": "收到新的设备留言", + "page": sent[1]["page"], + } + assert sent[0]["page"].startswith("pages/notification/index?token=") + assert sent[1]["page"].startswith("pages/notification/index?token=") + assert sent[0]["page"] != sent[1]["page"] + + first_token = sent[0]["page"].split("token=", 1)[1] + second_token = sent[1]["page"].split("token=", 1)[1] + assert service._decode_notification_token(first_token)["conversation_id"] == 6 + assert service._decode_notification_token(first_token)["user_id"] == 4 + assert service._decode_notification_token(second_token)["conversation_id"] == 7 + assert service._decode_notification_token(second_token)["user_id"] == 5 + + +@pytest.mark.asyncio +async def test_notify_leave_message_falls_back_to_chat_page_without_targets(monkeypatch): + configure_enabled_wechat_mp(monkeypatch) + service = WechatMpNotificationService() + sent = [] + + async def fake_list_recipients(*, device_id): + del device_id + return [WechatMpRecipient(user_id=4, openid="openid-1")] + + async def fake_send_template_message(**kwargs): + sent.append(kwargs) + return {"ok": True, "msgid": "msg-id"} + + monkeypatch.setattr(service, "_list_device_family_mp_recipients", fake_list_recipients) monkeypatch.setattr(service, "_send_template_message", fake_send_template_message) await service.notify_leave_message(device_id="TalkingQ_XQSN00001007", child_name="小明") @@ -52,14 +125,7 @@ async def test_notify_leave_message_sends_template_to_unique_subscribed_recipien "title": "小明", "condition": "收到新的设备留言", "page": "pages/chat/detail/index", - }, - { - "openid": "openid-2", - "template_id": "template-id", - "title": "小明", - "condition": "收到新的设备留言", - "page": "pages/chat/detail/index", - }, + } ] @@ -71,13 +137,13 @@ async def test_notify_leave_message_dedupes_same_device_and_openid(monkeypatch): async def fake_list_recipients(*, device_id): del device_id - return ["openid-1"] + return [WechatMpRecipient(user_id=4, openid="openid-1")] async def fake_send_template_message(**kwargs): sent.append(kwargs["openid"]) return {"ok": True, "msgid": "msg-id"} - monkeypatch.setattr(service, "_list_device_family_mp_openids", fake_list_recipients) + monkeypatch.setattr(service, "_list_device_family_mp_recipients", fake_list_recipients) monkeypatch.setattr(service, "_send_template_message", fake_send_template_message) await service.notify_leave_message(device_id="TalkingQ_XQSN00001007", child_name="小明") @@ -86,6 +152,91 @@ async def test_notify_leave_message_dedupes_same_device_and_openid(monkeypatch): assert sent == ["openid-1"] +@pytest.mark.asyncio +async def test_resolve_notification_link_returns_chat_detail_params(monkeypatch): + configure_enabled_wechat_mp(monkeypatch) + service = WechatMpNotificationService() + token = service._create_notification_token( + { + "type": "leave_message", + "user_id": 4, + "conversation_id": 6, + "message_id": 209, + "child_id": 8, + "child_name": "小明", + "device_id": "TalkingQ_XQSN00001007", + } + ) + + class FakeResult: + def mappings(self): + return self + + def first(self): + return { + "conversation_id": 6, + "child_id": 8, + "child_name": "小明", + "device_id": "TalkingQ_XQSN00001007", + } + + class FakeSession: + async def execute(self, sql, params): + assert params["message_id"] == 209 + assert params["conversation_id"] == 6 + assert params["child_id"] == 8 + assert params["user_id"] == 4 + return FakeResult() + + async def close(self): + pass + + async def fake_get_session(): + return FakeSession() + + monkeypatch.setattr(service, "get_session", fake_get_session) + + result = await service.resolve_notification_link(token=token, user_id=4) + + assert result["route"] == "pages/chat/detail/index" + assert result["params"] == { + "id": 6, + "source": "im", + "name": "家长沟通", + "peerKind": "parent", + "roleKey": "", + "conversationTypeName": "parent_child", + "peerId": "4", + "childId": 8, + "childName": "小明", + "parentUserId": 4, + "deviceId": "TalkingQ_XQSN00001007", + "channelLabel": "微信小程序", + "canSend": "1", + "messageId": 209, + } + + +@pytest.mark.asyncio +async def test_resolve_notification_link_rejects_other_user(monkeypatch): + configure_enabled_wechat_mp(monkeypatch) + service = WechatMpNotificationService() + token = service._create_notification_token( + { + "type": "leave_message", + "user_id": 4, + "conversation_id": 6, + "message_id": 209, + "child_id": 8, + } + ) + + with pytest.raises(HTTPException) as exc_info: + await service.resolve_notification_link(token=token, user_id=5) + + assert exc_info.value.status_code == 403 + + @pytest.mark.asyncio async def test_bind_url_persists_state_and_builds_oauth_url(monkeypatch): configure_enabled_wechat_mp(monkeypatch)