From efb13de8ed17a01469e96bd8295ed17d0189a59c Mon Sep 17 00:00:00 2001 From: ChengCan <783785929@qq.com> Date: Thu, 30 Apr 2026 01:03:34 +0800 Subject: [PATCH] feat(voice): upload device audio to cos and enable playback --- banban-mini/src/pages/chat/detail/index.scss | 27 ++ banban-mini/src/pages/chat/detail/index.tsx | 74 ++++- banban-mini/src/pages/chat/index.tsx | 2 +- banban-mini/src/pages/device/index.scss | 51 ++++ banban-mini/src/pages/device/index.tsx | 77 +++-- banban-mini/src/pages/location/index.tsx | 4 +- banban-mini/src/pages/sleep/index.tsx | 174 +++++------ banban-mini/src/services/chat.ts | 43 +-- banban-mini/src/services/location.ts | 12 +- talkingq-url/banban/dao/__init__.py | 3 +- talkingq-url/banban/dao/im.py | 30 +- talkingq-url/banban/routers/im.py | 6 +- talkingq-url/banban/service/im.py | 274 ++++++++++-------- .../banban/service/message_audio_storage.py | 129 +++++++++ talkingq-url/config.py | 1 + talkingq-url/handlers/audio_file_handler.py | 43 +-- .../handlers/websocket_message_handler.py | 13 +- 17 files changed, 640 insertions(+), 323 deletions(-) create mode 100644 talkingq-url/banban/service/message_audio_storage.py diff --git a/banban-mini/src/pages/chat/detail/index.scss b/banban-mini/src/pages/chat/detail/index.scss index d10d348..ad04f3d 100644 --- a/banban-mini/src/pages/chat/detail/index.scss +++ b/banban-mini/src/pages/chat/detail/index.scss @@ -174,6 +174,33 @@ } } +.audio-bubble { + min-width: 220px; + display: flex; + align-items: center; + gap: 14px; + + &.playing { + opacity: 0.82; + } +} + +.audio-bubble-icon { + font-size: 24px; + font-weight: 700; + line-height: 1; +} + +.audio-bubble-text { + flex: 1; + font-size: 28px; +} + +.audio-bubble-duration { + font-size: 24px; + opacity: 0.8; +} + .empty-chat { min-height: 320px; display: flex; diff --git a/banban-mini/src/pages/chat/detail/index.tsx b/banban-mini/src/pages/chat/detail/index.tsx index 9e8d4f5..aa9d399 100644 --- a/banban-mini/src/pages/chat/detail/index.tsx +++ b/banban-mini/src/pages/chat/detail/index.tsx @@ -1,5 +1,5 @@ import { View, Text, ScrollView, Image, Input } from '@tarojs/components' -import { useEffect, useState } from 'react' +import { useEffect, useRef, useState } from 'react' import Taro, { useRouter } from '@tarojs/taro' import { getMessages, @@ -47,6 +47,11 @@ function formatParticipantLabel(message: ChatMessage): string { return `${senderType} ${senderId}` } +function formatAudioDuration(durationMs?: number | null): string { + const totalSeconds = Math.max(1, Math.round((durationMs || 0) / 1000)) + return `${totalSeconds}s` +} + export default function ChatDetail() { const router = useRouter() const params = router.params @@ -85,6 +90,29 @@ export default function ChatDetail() { const [loading, setLoading] = useState(true) const [inputText, setInputText] = useState('') const [sending, setSending] = useState(false) + const [playingMessageId, setPlayingMessageId] = useState(null) + const audioContextRef = useRef(null) + + useEffect(() => { + const audioContext = Taro.createInnerAudioContext() + audioContextRef.current = audioContext + + audioContext.onEnded(() => { + setPlayingMessageId(null) + }) + audioContext.onStop(() => { + setPlayingMessageId(null) + }) + audioContext.onError(() => { + setPlayingMessageId(null) + Taro.showToast({ title: '语音播放失败', icon: 'none' }) + }) + + return () => { + audioContext.destroy() + audioContextRef.current = null + } + }, []) useEffect(() => { void loadChatData(activeConversationId) @@ -157,6 +185,46 @@ export default function ChatDetail() { } } + const handlePlayAudio = (message: ChatMessage) => { + if (message.contentType !== 2 || !message.mediaUrl) { + Taro.showToast({ title: '语音地址不存在', icon: 'none' }) + return + } + + const audioContext = audioContextRef.current + if (!audioContext) { + Taro.showToast({ title: '播放器未就绪', icon: 'none' }) + return + } + + if (playingMessageId === message.id) { + audioContext.stop() + setPlayingMessageId(null) + return + } + + audioContext.stop() + audioContext.src = message.mediaUrl + audioContext.autoplay = true + audioContext.play() + setPlayingMessageId(message.id) + } + + const renderMessageBody = (msg: ChatMessage) => { + if (msg.contentType === 2 && msg.mediaUrl) { + const isPlaying = playingMessageId === msg.id + return ( + handlePlayAudio(msg)}> + {isPlaying ? '[]' : '>'} + {msg.mediaTranscriptText || '点击播放语音'} + {formatAudioDuration(msg.mediaDurationMs)} + + ) + } + + return {msg.content} + } + return ( @@ -199,7 +267,7 @@ export default function ChatDetail() { {formatParticipantLabel(msg)} - {msg.content} + {renderMessageBody(msg)} @@ -214,7 +282,7 @@ export default function ChatDetail() { {formatParticipantLabel(msg)} - {msg.content} + {renderMessageBody(msg)} diff --git a/banban-mini/src/pages/chat/index.tsx b/banban-mini/src/pages/chat/index.tsx index a93fd6a..d6f1a11 100644 --- a/banban-mini/src/pages/chat/index.tsx +++ b/banban-mini/src/pages/chat/index.tsx @@ -65,7 +65,7 @@ export default function Chat() { ) : conversations.length === 0 ? ( - 暂无玩伴会话 + 当前孩子暂无会话 ) : ( diff --git a/banban-mini/src/pages/device/index.scss b/banban-mini/src/pages/device/index.scss index 4cbf338..18dd9a2 100644 --- a/banban-mini/src/pages/device/index.scss +++ b/banban-mini/src/pages/device/index.scss @@ -348,6 +348,14 @@ color: #1A1A1A; } +.empty-subtitle { + display: block; + margin-top: 16px; + font-size: 26px; + line-height: 1.6; + color: #666666; +} + .empty-btn { margin-top: 28px; height: 92px; @@ -378,3 +386,46 @@ font-weight: 600; color: #1A1A1A; } + +.child-focus-card { + display: flex; + align-items: center; + background: #FFFFFF; + border-radius: 24px; + margin: 0 24px 24px; + padding: 28px 24px; + box-shadow: 0 2px 12px rgba(0, 0, 0, 0.04); +} + +.child-focus-avatar { + width: 96px; + height: 96px; + border-radius: 50%; + background: #FEF3E8; + display: flex; + align-items: center; + justify-content: center; + margin-right: 20px; +} + +.child-focus-icon { + font-size: 40px; +} + +.child-focus-info { + flex: 1; +} + +.child-focus-name { + display: block; + font-size: 32px; + font-weight: 700; + color: #1A1A1A; +} + +.child-focus-desc { + display: block; + margin-top: 8px; + font-size: 26px; + color: #666666; +} diff --git a/banban-mini/src/pages/device/index.tsx b/banban-mini/src/pages/device/index.tsx index 492c70c..cd54cc6 100644 --- a/banban-mini/src/pages/device/index.tsx +++ b/banban-mini/src/pages/device/index.tsx @@ -2,8 +2,8 @@ import { View, Text, Switch, Slider, Image } from '@tarojs/components' import { useState } from 'react' import Taro, { useDidShow } from '@tarojs/taro' import { getToken } from '@/services/auth' -import { Binding, resolveActiveBinding } from '@/services/binding' -import { Child, getChild } from '@/services/child' +import { Binding, loadCurrentChildBindingContext } from '@/services/binding' +import { Child } from '@/services/child' import './index.scss' export default function Device() { @@ -25,20 +25,9 @@ export default function Device() { setIsLoading(true) try { - const activeBinding = await resolveActiveBinding() - setBinding(activeBinding) - - if (activeBinding?.child_id) { - try { - const currentChild = await getChild(activeBinding.child_id) - setChild(currentChild) - } catch (error) { - console.error('[device] load child failed:', error) - setChild(null) - } - } else { - setChild(null) - } + const context = await loadCurrentChildBindingContext() + setChild(context.currentChild) + setBinding(context.currentBinding) } catch (error: any) { console.error('[device] load failed:', error) Taro.showToast({ @@ -72,7 +61,7 @@ export default function Device() { ) } - if (!binding) { + if (!child) { return ( @@ -83,7 +72,39 @@ export default function Device() { - 先绑定一台伴伴设备 + 先添加一个儿童资料 + 添加后才能为孩子绑定专属设备并查看聊天和定位信息 + Taro.switchTab({ url: '/pages/sleep/index' })}> + 去添加孩子 + + + + ) + } + + if (!binding) { + return ( + + + {child.child_name} 的伴伴 + + + + + 🧒 + + + {child.child_name} + 当前孩子已选中，还没有绑定设备 + + + + + + + + 给当前孩子绑定一台设备 + 绑定后，这个孩子的聊天记录、定位和设备状态都会显示在首页 Taro.navigateTo({ url: '/pages/bind/index' })}> 去绑定设备 @@ -92,8 +113,8 @@ export default function Device() { ) } - const childName = child?.child_name || '未设置' - const pageTitle = child?.child_name ? `${child.child_name} 的伴伴` : '伴伴设备' + const childName = child.child_name || '未设置' + const pageTitle = `${childName} 的伴伴` const battery = 82 const daysLeft = 4 @@ -106,7 +127,7 @@ export default function Device() { - {binding.child_id ? '已完成绑定' : '待关联儿童'} + 已完成绑定 @@ -124,12 +145,12 @@ export default function Device() { - 设备号 - {binding.device_id} + 当前儿童 + {childName} - 儿童昵称 - {childName} + 当前设备 + {binding.device_id} 绑定时间 @@ -195,12 +216,6 @@ export default function Device() { - - {!binding.child_id && ( - Taro.navigateTo({ url: '/pages/bind/index' })}> - 继续完成儿童关联 - - )} ) } diff --git a/banban-mini/src/pages/location/index.tsx b/banban-mini/src/pages/location/index.tsx index 207aa17..16526b1 100644 --- a/banban-mini/src/pages/location/index.tsx +++ b/banban-mini/src/pages/location/index.tsx @@ -339,7 +339,7 @@ export default function Location() { ) : ( - 暂无设备位置 + 当前孩子暂无设备位置 ) ) : trajectory.length > 0 ? ( @@ -384,7 +384,7 @@ export default function Location() { 暂无轨迹 - {trajectoryMode === 'today' ? '今天还没有新的历史点位。' : '最近时间范围内没有新的历史点位。'} + {trajectoryMode === 'today' ? '当前孩子今天还没有新的历史点位。' : '当前孩子最近没有新的历史点位。'} )} diff --git a/banban-mini/src/pages/sleep/index.tsx b/banban-mini/src/pages/sleep/index.tsx index a788c25..6f55381 100644 --- a/banban-mini/src/pages/sleep/index.tsx +++ b/banban-mini/src/pages/sleep/index.tsx @@ -5,13 +5,11 @@ import { clearToken, getCurrentUserId, getToken } from '@/services/auth' import { BindingListItem, clearSelectedBindingDeviceId, - getBindings, - resolveBindingSelection, - setBindingChild, + loadCurrentChildBindingContext, setSelectedBindingDeviceId, unbindDevice, } from '@/services/binding' -import { Child, createChild, getChildren, updateChild } from '@/services/child' +import { Child, clearSelectedChildId, createChild, setSelectedChildId, updateChild } from '@/services/child' import './index.scss' interface MenuItem { @@ -28,8 +26,9 @@ export default function Sleep() { const [children, setChildren] = useState([]) const [bindings, setBindings] = useState([]) const [binding, setBinding] = useState(null) + const [currentChild, setCurrentChild] = useState(null) const [showModal, setShowModal] = useState(false) - const [showDeviceModal, setShowDeviceModal] = useState(false) + const [showChildModal, setShowChildModal] = useState(false) const [modalType, setModalType] = useState<'add' | 'edit'>('add') const [childName, setChildName] = useState('') const [editingChildId, setEditingChildId] = useState(null) @@ -47,13 +46,11 @@ export default function Sleep() { setLoading(true) try { - const [childResponse, bindingResponse] = await Promise.all([getChildren(), getBindings(undefined, 100)]) - const bindingItems = bindingResponse.items || [] - const activeBinding = resolveBindingSelection(bindingItems) - - setChildren(childResponse.items || []) - setBindings(bindingItems) - setBinding(activeBinding) + const context = await loadCurrentChildBindingContext() + setChildren(context.children) + setBindings(context.bindings as BindingListItem[]) + setCurrentChild(context.currentChild) + setBinding((context.currentBinding as BindingListItem | null) || null) setParentInfo(Taro.getStorageSync('userInfo') || {}) } catch (error: any) { console.error('[manage] load failed:', error) @@ -66,8 +63,7 @@ export default function Sleep() { } } - const currentChild = binding?.child_id ? children.find((item) => item.child_id === binding.child_id) || null : null - const currentChildName = currentChild?.child_name || binding?.child_name || (binding && !binding.child_id ? '待关联' : '未设置') + const currentChildName = currentChild?.child_name || '未设置' const handleOpenModal = (type: 'add' | 'edit', child?: Child) => { setModalType(type) @@ -82,12 +78,19 @@ export default function Sleep() { setEditingChildId(null) } - const handleSelectBinding = (targetBinding: BindingListItem) => { - setSelectedBindingDeviceId(targetBinding.device_id) - setBinding(targetBinding) - setShowDeviceModal(false) + const handleSelectChild = (child: Child) => { + setSelectedChildId(child.child_id) + const nextBinding = bindings.find((item) => item.child_id === child.child_id) || null + if (nextBinding?.device_id) { + setSelectedBindingDeviceId(nextBinding.device_id) + } else { + clearSelectedBindingDeviceId() + } + setCurrentChild(child) + setBinding(nextBinding) + setShowChildModal(false) Taro.showToast({ - title: '已切换设备', + title: '已切换当前孩子', icon: 'success', }) } @@ -102,12 +105,8 @@ export default function Sleep() { try { if (modalType === 'add') { const createdChild = await createChild({ child_name: normalizedName }) - if (binding && !binding.child_id) { - await setBindingChild(binding.device_id, { child_id: createdChild.child_id }) - Taro.showToast({ title: '已创建并关联', icon: 'success' }) - } else { - Taro.showToast({ title: '创建成功', icon: 'success' }) - } + setSelectedChildId(createdChild.child_id) + Taro.showToast({ title: '创建成功', icon: 'success' }) } else if (editingChildId) { await updateChild(editingChildId, { child_name: normalizedName }) Taro.showToast({ title: '修改成功', icon: 'success' }) @@ -124,54 +123,19 @@ export default function Sleep() { } } - const handleChildMenu = () => { - if (binding && !binding.child_id && children.length > 0) { - Taro.showActionSheet({ - itemList: [...children.map((item) => item.child_name), '新建儿童资料'], - success: async (result) => { - if (result.tapIndex === children.length) { - handleOpenModal('add') - return - } - - const targetChild = children[result.tapIndex] - if (!targetChild) return - - try { - await setBindingChild(binding.device_id, { child_id: targetChild.child_id }) - Taro.showToast({ title: '关联成功', icon: 'success' }) - await loadData() - } catch (error: any) { - Taro.showToast({ - title: error?.message || '关联失败，请重试', - icon: 'none', - }) - } - }, - }) - return - } - - if (currentChild) { - handleOpenModal('edit', currentChild) - return - } - - handleOpenModal('add') - } - const handleUnbind = () => { if (!binding) return Taro.showModal({ title: '解除设备绑定', - content: '确定要解除当前设备绑定吗？', + content: `确定要解除 ${currentChildName} 当前绑定的设备吗？`, confirmColor: '#FF8C42', success: async (result) => { if (!result.confirm) return try { await unbindDevice(binding.device_id) + clearSelectedBindingDeviceId() Taro.showToast({ title: '已解绑', icon: 'success' }) await loadData() } catch (error: any) { @@ -193,6 +157,7 @@ export default function Sleep() { if (!result.confirm) return clearToken() clearSelectedBindingDeviceId() + clearSelectedChildId() Taro.removeStorageSync('userInfo') Taro.reLaunch({ url: '/pages/login/index' }) }, @@ -202,21 +167,33 @@ export default function Sleep() { const handleMenuClick = (item: MenuItem) => { if (item.disabled) return - if (item.name === '儿童资料 (用于称呼)') { - handleChildMenu() + if (item.name === '当前孩子') { + setShowChildModal(true) return } - if (item.name === '绑定新设备') { + if (item.name === '编辑当前孩子') { + if (currentChild) { + handleOpenModal('edit', currentChild) + } else { + handleOpenModal('add') + } + return + } + + if (item.name === '绑定设备') { + if (!currentChild) { + handleOpenModal('add') + Taro.showToast({ + title: '请先添加儿童资料', + icon: 'none', + }) + return + } Taro.navigateTo({ url: '/pages/bind/index' }) return } - if (item.name === '切换设备') { - setShowDeviceModal(true) - return - } - if (item.name === '解除设备绑定') { handleUnbind() } @@ -237,24 +214,24 @@ export default function Sleep() { { icon: require('../../assets/tab-icons/orange-robot.png'), iconBgClass: 'orange', - name: '儿童资料 (用于称呼)', + name: '当前孩子', value: currentChildName, arrow: true, }, - { - icon: require('../../assets/tab-icons/rings.png'), - iconBgClass: 'green', - name: '切换设备', - value: bindings.length > 0 ? `共 ${bindings.length} 台` : '暂无设备', - arrow: true, - }, { icon: require('../../assets/tab-icons/orange-robot.png'), iconBgClass: 'orange', - name: '绑定新设备', + name: '编辑当前孩子', value: '', arrow: true, }, + { + icon: require('../../assets/tab-icons/rings.png'), + iconBgClass: 'green', + name: '绑定设备', + value: binding?.device_id ? `当前: ${binding.device_id}` : '未绑定', + arrow: true, + }, { icon: require('../../assets/tab-icons/broken-rings.png'), iconBgClass: 'red', @@ -284,23 +261,23 @@ export default function Sleep() { 用户 ID: {userId || '--'} - {bindings.length > 0 ? `已绑定 ${bindings.length} 台` : '未绑定设备'} + {currentChild ? '已选择当前孩子' : '未选择孩子'} - 已绑定设备数 - {bindings.length} 台 - - - 当前设备 - {binding?.device_id || '未绑定'} + 儿童资料数 + {children.length} 个 当前儿童 {currentChildName} + + 当前设备 + {binding?.device_id || '未绑定'} + @@ -333,39 +310,40 @@ export default function Sleep() { 伴伴 Companion V1.1.0 - {showDeviceModal && ( - setShowDeviceModal(false)}> + {showChildModal && ( + setShowChildModal(false)}> { event.stopPropagation() }} > - 切换设备 - {bindings.length === 0 ? ( - 当前还没有已绑定设备 + 切换当前孩子 + {children.length === 0 ? ( + 当前还没有儿童资料 ) : ( - {bindings.map((item) => { - const isActive = binding?.device_id === item.device_id + {children.map((item) => { + const isActive = currentChild?.child_id === item.child_id + const itemBinding = bindings.find((bindingItem) => bindingItem.child_id === item.child_id) || null return ( handleSelectBinding(item)} + onClick={() => handleSelectChild(item)} > - {item.device_id} + {item.child_name} {isActive && 当前} - {item.child_name || '待关联儿童'} + {itemBinding?.device_id || '未绑定设备'} ) })} )} - setShowDeviceModal(false)}> + setShowChildModal(false)}> 关闭 diff --git a/banban-mini/src/services/chat.ts b/banban-mini/src/services/chat.ts index b5b8505..55d71ab 100644 --- a/banban-mini/src/services/chat.ts +++ b/banban-mini/src/services/chat.ts @@ -1,7 +1,6 @@ import { getCurrentUserId } from './auth' import { request } from './api' -import { resolveActiveBinding } from './binding' -import { getChild } from './child' +import { loadCurrentChildBindingContext } from './binding' export type ConversationSource = 'im' | 'ai' export type PeerKind = 'child' | 'parent' | 'ai' @@ -38,6 +37,11 @@ export interface ChatMessage { content: string time: string conversationId: number + contentType?: number + mediaUrl?: string | null + mediaDurationMs?: number | null + mediaMimeType?: string | null + mediaTranscriptText?: string | null senderType?: string senderId?: string receiverType?: string @@ -240,21 +244,12 @@ function getConversationRank(conversation: ChatConversation): number { } async function getBindingContext(): Promise { - const binding = await resolveActiveBinding() - let childName = String(binding?.child_name || '').trim() - - if (!childName && binding?.child_id) { - try { - const child = await getChild(binding.child_id) - childName = String(child?.child_name || '').trim() - } catch (error) { - console.error('[chat] load child name failed:', error) - } - } + const context = await loadCurrentChildBindingContext() + const childName = String(context.currentChild?.child_name || context.currentBinding?.child_name || '').trim() return { - deviceId: binding?.device_id || '', - childId: binding?.child_id || null, + deviceId: context.currentBinding?.device_id || '', + childId: context.currentChild?.child_id || null, childName, currentUserId: getCurrentUserId(), } @@ -265,7 +260,7 @@ async function getDeviceMessages(context?: BindingContext): Promise(`/devices/${deviceId}/messages?limit=100`) + const response = await request(`/banban/devices/${deviceId}/messages?limit=100`) return response.items || [] } catch (error: any) { if (error?.status === 404) return [] @@ -278,7 +273,7 @@ async function getChildConversations(context?: BindingContext): Promise(`/children/${childId}/conversations?limit=100`) + const response = await request(`/banban/children/${childId}/conversations?limit=100`) return response.items || [] } catch (error: any) { if (error?.status === 404) return [] @@ -294,7 +289,7 @@ async function getChildConversationMessages( if (!childId || conversationId <= 0) return [] const response = await request( - `/children/${childId}/conversations/${conversationId}/messages?limit=100` + `/banban/children/${childId}/conversations/${conversationId}/messages?limit=100` ) return response.items || [] } @@ -437,6 +432,11 @@ export async function getMessages( content: item.content || '暂无内容', time: formatDetailTime(item.created_at), conversationId: item.conversation_id, + contentType: 1, + mediaUrl: null, + mediaDurationMs: null, + mediaMimeType: null, + mediaTranscriptText: null, senderType: item.is_user ? 'device' : 'ai', senderId: item.is_user ? context.deviceId : item.role_key, receiverType: item.is_user ? 'ai' : 'device', @@ -462,6 +462,11 @@ export async function getMessages( content: formatImPreview(item), time: formatDetailTime(item.created_at), conversationId: item.conversation_id, + contentType: item.content_type, + mediaUrl: item.media_file_key || null, + mediaDurationMs: item.media_duration_ms || null, + mediaMimeType: item.media_mime_type || null, + mediaTranscriptText: item.media_transcript_text || null, senderType: item.sender_type, senderId: item.sender_id, receiverType: item.receiver_type, @@ -580,7 +585,7 @@ export async function sendParentMessage(childId: number, content: string): Promi throw new Error('请输入消息内容') } - return request(`/children/${childId}/messages`, { + return request(`/banban/children/${childId}/messages`, { method: 'POST', data: { content_type: 1, diff --git a/banban-mini/src/services/location.ts b/banban-mini/src/services/location.ts index 7375451..2d5948d 100644 --- a/banban-mini/src/services/location.ts +++ b/banban-mini/src/services/location.ts @@ -1,5 +1,5 @@ import { request } from './api' -import { resolveActiveBinding } from './binding' +import { loadCurrentChildBindingContext } from './binding' export interface DeviceLocation { child_id: number @@ -32,11 +32,12 @@ export interface DeviceTrajectoryResponse { } export async function getCurrentDeviceLocation(deviceId?: string): Promise { - const resolvedDeviceId = String(deviceId || '').trim() || (await resolveActiveBinding())?.device_id || '' + const resolvedDeviceId = + String(deviceId || '').trim() || (await loadCurrentChildBindingContext()).currentBinding?.device_id || '' if (!resolvedDeviceId) return null try { - return await request(`/devices/${resolvedDeviceId}/location`) + return await request(`/banban/devices/${resolvedDeviceId}/location`) } catch (error: any) { if (error?.status === 404) return null throw error @@ -49,7 +50,8 @@ export async function getDeviceTrajectory(params?: { endAt?: string limit?: number }): Promise { - const resolvedDeviceId = String(params?.deviceId || '').trim() || (await resolveActiveBinding())?.device_id || '' + const resolvedDeviceId = + String(params?.deviceId || '').trim() || (await loadCurrentChildBindingContext()).currentBinding?.device_id || '' if (!resolvedDeviceId) return null const query = new URLSearchParams() @@ -59,7 +61,7 @@ export async function getDeviceTrajectory(params?: { try { const suffix = query.toString() ? `?${query.toString()}` : '' - return await request(`/devices/${resolvedDeviceId}/trajectory${suffix}`) + return await request(`/banban/devices/${resolvedDeviceId}/trajectory${suffix}`) } catch (error: any) { if (error?.status === 404) return null throw error diff --git a/talkingq-url/banban/dao/__init__.py b/talkingq-url/banban/dao/__init__.py index fd6faa8..793454d 100644 --- a/talkingq-url/banban/dao/__init__.py +++ b/talkingq-url/banban/dao/__init__.py @@ -7,7 +7,8 @@ class BaseDAO: self.db = db async def execute(self, query, params: dict = None): - return await self.db.execute(text(query), params or {}) + statement = query if hasattr(query, "_execute_on_connection") else text(query) + return await self.db.execute(statement, params or {}) async def commit(self): await self.db.commit() diff --git a/talkingq-url/banban/dao/im.py b/talkingq-url/banban/dao/im.py index b4ab869..dcc0447 100644 --- a/talkingq-url/banban/dao/im.py +++ b/talkingq-url/banban/dao/im.py @@ -25,6 +25,14 @@ class ConversationMessageCreateResult: conversation_type: int message: dict + @property + def conversation_type_name(self) -> str: + if self.conversation_type == 1: + return "child_peer" + if self.conversation_type == 2: + return "parent_child" + return f"unknown_{self.conversation_type}" + class ImDAO(BaseDAO): async def assert_parent_child_access(self, *, user_id: int, child_id: int) -> Mapping[str, Any]: @@ -179,6 +187,22 @@ class ImDAO(BaseDAO): return "[image]" return "[json]" + async def ensure_parent_child_conversation(self, *, parent_user_id: int, child_id: int) -> int: + child_row = await self.assert_parent_child_access(user_id=parent_user_id, child_id=child_id) + parent_row = await self._get_parent_row(parent_user_id) + if not parent_row: + from fastapi import HTTPException + raise HTTPException(status_code=404, detail="parent not found") + + return await self._get_or_create_conversation( + conversation_type=2, + participant_a_type=2, + participant_a_id=str(child_id), + participant_b_type=1, + participant_b_id=str(parent_user_id), + pair_key=f"{child_id}:{parent_user_id}", + ) + async def create_message( self, *, @@ -428,8 +452,8 @@ class ImDAO(BaseDAO): FROM im_conversations WHERE conversation_type = :conversation_type AND pair_key = :pair_key - {lock_clause} LIMIT 1 + {lock_clause} """ ), {"conversation_type": conversation_type, "pair_key": pair_key}, @@ -449,8 +473,8 @@ class ImDAO(BaseDAO): SELECT id, conversation_type, last_seq, status FROM im_conversations WHERE id = :conversation_id - {lock_clause} LIMIT 1 + {lock_clause} """ ), {"conversation_id": conversation_id}, @@ -484,4 +508,4 @@ class ImDAO(BaseDAO): async def _next_primary_key(self, table_name: str) -> int | None: result = await self.execute(text(f"SELECT 1")) - return None \ No newline at end of file + return None diff --git a/talkingq-url/banban/routers/im.py b/talkingq-url/banban/routers/im.py index 757d1a2..958973d 100644 --- a/talkingq-url/banban/routers/im.py +++ b/talkingq-url/banban/routers/im.py @@ -16,7 +16,7 @@ try: ConversationMessageCreateResponse, ParentChildMessageCreateRequest, ) - from banban.service.im import ImService, im_service + from banban.service.im import ImService, im_service, present_message_item except ModuleNotFoundError: from banban.security import get_current_user_id from banban.schemas.im import ( @@ -27,7 +27,7 @@ except ModuleNotFoundError: ConversationMessageCreateResponse, ParentChildMessageCreateRequest, ) - from banban.service.im import ImService, im_service + from banban.service.im import ImService, im_service, present_message_item router = APIRouter(prefix="/children", tags=["im"]) @@ -506,7 +506,7 @@ async def list_child_conversation_messages( rows = rows[:limit] rows = list(rows) rows.reverse() - items = [_row_to_message_item(row) for row in rows] + items = [await present_message_item(row, audio_storage=im_service.audio_storage) for row in rows] next_cursor_seq = items[0].seq if has_more and items else None logger.info( diff --git a/talkingq-url/banban/service/im.py b/talkingq-url/banban/service/im.py index 9c5574d..74d3b86 100644 --- a/talkingq-url/banban/service/im.py +++ b/talkingq-url/banban/service/im.py @@ -1,10 +1,12 @@ from dataclasses import dataclass +import hashlib import json from collections.abc import Mapping from typing import Any -from fastapi import HTTPException, status +from fastapi import HTTPException from services.database_service_base import DatabaseServiceBase +from banban.service.message_audio_storage import MessageAudioStorageService, MessageAudioStorageError try: from banban.dao.im import ImDAO, DeviceIdentity, ConversationMessageCreateResult @@ -43,6 +45,12 @@ def participant_type_name(participant_type: int) -> str: return PARTICIPANT_TYPE_NAMES.get(participant_type, f"unknown_{participant_type}") +def build_device_audio_client_msg_id(*, device_id: str, target_device_id: str, audio_url: str) -> str: + raw = f"{device_id}|{target_device_id}|{audio_url}" + digest = hashlib.sha256(raw.encode("utf-8")).hexdigest() + return f"device-audio-{digest[:32]}" + + def normalize_content_json(value: Any) -> dict[str, Any] | None: if value is None: return None @@ -85,9 +93,24 @@ def row_to_message_item(row: Mapping[str, Any]) -> ChildConversationMessageItem: ) +async def present_message_item( + row: Mapping[str, Any], + *, + audio_storage: MessageAudioStorageService, +) -> ChildConversationMessageItem: + item = row_to_message_item(row) + if item.content_type == 2 and item.media_file_key: + try: + item.media_file_key = await audio_storage.get_audio_url(item.media_file_key) + except MessageAudioStorageError: + pass + return item + + class ImService(DatabaseServiceBase): def __init__(self): super().__init__(service_name="im_service") + self.audio_storage = MessageAudioStorageService() async def assert_parent_child_access(self, *, user_id: int, child_id: int) -> Mapping[str, Any]: db_session = await self.get_session() @@ -105,6 +128,34 @@ class ImService(DatabaseServiceBase): finally: await db_session.close() + async def _build_message_create_result( + self, + *, + dao: ImDAO, + idempotent: bool, + conversation_id: int, + conversation_type: int, + client_msg_id: str, + ) -> ConversationMessageCreateResult: + message_row = await dao._get_message_by_conversation_client_id( + conversation_id=conversation_id, + client_msg_id=client_msg_id, + ) + if not message_row: + raise RuntimeError("message was not found after insert") + + presented_message = await present_message_item( + message_row, + audio_storage=self.audio_storage, + ) + + return ConversationMessageCreateResult( + idempotent=idempotent, + conversation_id=conversation_id, + conversation_type=conversation_type, + message=presented_message, + ) + async def create_parent_child_message( self, *, @@ -137,18 +188,12 @@ class ImService(DatabaseServiceBase): receiver_avatar_snapshot=None, payload=payload, ) - message_row = await dao._get_message_by_conversation_client_id( - conversation_id=conversation_id, - client_msg_id=payload.client_msg_id, - ) - if not message_row: - raise RuntimeError("message was not found after insert") - - return ConversationMessageCreateResult( + return await self._build_message_create_result( + dao=dao, idempotent=idempotent, conversation_id=conversation_id, conversation_type=PARENT_CHILD_CONVERSATION_TYPE, - message=row_to_message_item(message_row), + client_msg_id=payload.client_msg_id, ) except Exception: await db_session.rollback() @@ -156,13 +201,87 @@ class ImService(DatabaseServiceBase): finally: await db_session.close() + async def _create_device_message_with_payload( + self, + *, + dao: ImDAO, + device_identity: DeviceIdentity, + payload: DeviceMessageCreateRequest, + ) -> ConversationMessageCreateResult: + if payload.conversation_type == CHILD_PEER_CONVERSATION_TYPE: + if payload.peer_child_id == device_identity.child_id: + raise HTTPException(status_code=400, detail="peer_child_id must be different from current child") + sender_child_row = await dao.assert_child_exists(child_id=device_identity.child_id) + receiver_child_row = await dao.assert_child_exists(child_id=payload.peer_child_id) + participant_a_id, participant_b_id, pair_key = dao._build_child_peer_pair( + device_identity.child_id, + payload.peer_child_id, + ) + + conversation_id, idempotent = await dao.create_message( + conversation_type=CHILD_PEER_CONVERSATION_TYPE, + participant_a_type=CHILD_PARTICIPANT_TYPE, + participant_a_id=participant_a_id, + participant_b_type=CHILD_PARTICIPANT_TYPE, + participant_b_id=participant_b_id, + pair_key=pair_key, + sender_type=CHILD_PARTICIPANT_TYPE, + sender_id=str(device_identity.child_id), + receiver_type=CHILD_PARTICIPANT_TYPE, + receiver_id=str(payload.peer_child_id), + sender_name_snapshot=sender_child_row["child_name"], + sender_avatar_snapshot=None, + receiver_name_snapshot=receiver_child_row["child_name"], + receiver_avatar_snapshot=None, + payload=payload, + ) + else: + sender_child_row = await dao.assert_child_exists(child_id=device_identity.child_id) + parent_row = await dao._get_parent_row(payload.parent_user_id) + if not parent_row: + raise HTTPException(status_code=404, detail="parent not found") + await dao.assert_parent_child_access(user_id=payload.parent_user_id, child_id=device_identity.child_id) + + conversation_id, idempotent = await dao.create_message( + conversation_type=PARENT_CHILD_CONVERSATION_TYPE, + participant_a_type=CHILD_PARTICIPANT_TYPE, + participant_a_id=str(device_identity.child_id), + participant_b_type=PARENT_PARTICIPANT_TYPE, + participant_b_id=str(payload.parent_user_id), + pair_key=f"{device_identity.child_id}:{payload.parent_user_id}", + sender_type=CHILD_PARTICIPANT_TYPE, + sender_id=str(device_identity.child_id), + receiver_type=PARENT_PARTICIPANT_TYPE, + receiver_id=str(payload.parent_user_id), + sender_name_snapshot=sender_child_row["child_name"], + sender_avatar_snapshot=None, + receiver_name_snapshot=parent_row["nickname"], + receiver_avatar_snapshot=parent_row["avatar_url"], + payload=payload, + ) + + return await self._build_message_create_result( + dao=dao, + idempotent=idempotent, + conversation_id=conversation_id, + conversation_type=payload.conversation_type, + client_msg_id=payload.client_msg_id, + ) + async def create_device_message( self, *, device_id: str, serial_number: str, - payload: DeviceMessageCreateRequest, + payload: DeviceMessageCreateRequest | None = None, + target_device_id: str | None = None, + audio_url: str | None = None, ) -> tuple[DeviceIdentity, ConversationMessageCreateResult]: + if payload is None and (not target_device_id or not audio_url): + raise ValueError("payload or target_device_id/audio_url is required") + if payload is not None and (target_device_id is not None or audio_url is not None): + raise ValueError("payload and target_device_id/audio_url cannot be used together") + db_session = await self.get_session() try: dao = ImDAO(db_session) @@ -171,70 +290,31 @@ class ImService(DatabaseServiceBase): serial_number=serial_number, ) - if payload.conversation_type == CHILD_PEER_CONVERSATION_TYPE: - if payload.peer_child_id == device_identity.child_id: - raise HTTPException(status_code=400, detail="peer_child_id must be different from current child") - sender_child_row = await dao.assert_child_exists(child_id=device_identity.child_id) - receiver_child_row = await dao.assert_child_exists(child_id=payload.peer_child_id) - participant_a_id, participant_b_id, pair_key = dao._build_child_peer_pair( - device_identity.child_id, - payload.peer_child_id, - ) - - conversation_id, idempotent = await dao.create_message( + resolved_payload = payload + if resolved_payload is None: + target_device_identity = await dao.get_device_by_id(device_id=target_device_id) + resolved_payload = DeviceMessageCreateRequest( conversation_type=CHILD_PEER_CONVERSATION_TYPE, - participant_a_type=CHILD_PARTICIPANT_TYPE, - participant_a_id=participant_a_id, - participant_b_type=CHILD_PARTICIPANT_TYPE, - participant_b_id=participant_b_id, - pair_key=pair_key, - sender_type=CHILD_PARTICIPANT_TYPE, - sender_id=str(device_identity.child_id), - receiver_type=CHILD_PARTICIPANT_TYPE, - receiver_id=str(payload.peer_child_id), - sender_name_snapshot=sender_child_row["child_name"], - sender_avatar_snapshot=None, - receiver_name_snapshot=receiver_child_row["child_name"], - receiver_avatar_snapshot=None, - payload=payload, - ) - else: - sender_child_row = await dao.assert_child_exists(child_id=device_identity.child_id) - parent_row = await dao._get_parent_row(payload.parent_user_id) - if not parent_row: - raise HTTPException(status_code=404, detail="parent not found") - await dao.assert_parent_child_access(user_id=payload.parent_user_id, child_id=device_identity.child_id) - - conversation_id, idempotent = await dao.create_message( - conversation_type=PARENT_CHILD_CONVERSATION_TYPE, - participant_a_type=CHILD_PARTICIPANT_TYPE, - participant_a_id=str(device_identity.child_id), - participant_b_type=PARENT_PARTICIPANT_TYPE, - participant_b_id=str(payload.parent_user_id), - pair_key=f"{device_identity.child_id}:{payload.parent_user_id}", - sender_type=CHILD_PARTICIPANT_TYPE, - sender_id=str(device_identity.child_id), - receiver_type=PARENT_PARTICIPANT_TYPE, - receiver_id=str(payload.parent_user_id), - sender_name_snapshot=sender_child_row["child_name"], - sender_avatar_snapshot=None, - receiver_name_snapshot=parent_row["nickname"], - receiver_avatar_snapshot=parent_row["avatar_url"], - payload=payload, + peer_child_id=target_device_identity.child_id, + content_type=2, + media_file_key=audio_url, + media_mime_type="audio/mpeg", + client_msg_id=build_device_audio_client_msg_id( + device_id=device_id, + target_device_id=target_device_id, + audio_url=audio_url, + ), + ext_json={ + "source": "device_audio_message", + "source_device_id": device_id, + "target_device_id": target_device_id, + }, ) - message_row = await dao._get_message_by_conversation_client_id( - conversation_id=conversation_id, - client_msg_id=payload.client_msg_id, - ) - if not message_row: - raise RuntimeError("message was not found after insert") - - result = ConversationMessageCreateResult( - idempotent=idempotent, - conversation_id=conversation_id, - conversation_type=payload.conversation_type, - message=row_to_message_item(message_row), + result = await self._create_device_message_with_payload( + dao=dao, + device_identity=device_identity, + payload=resolved_payload, ) return device_identity, result except Exception: @@ -251,55 +331,5 @@ class ImService(DatabaseServiceBase): finally: await db_session.close() - ''' - Todo 创建设备消息, 还不完善 - ''' - async def create_device_message( - self, - *, - device_id: str, - serial_number: str, - target_device_id: str, - audio_url: str, - ): - db_session = await self.get_session() - try: - dao = ImDAO(db_session) - device_identity = await dao.authenticate_device_identity( - device_id=device_id, - serial_number=serial_number, - ) - target_device_identity = await dao.get_device_by_id(device_id=target_device_id) - sender_child_row = await dao.assert_child_exists(child_id=device_identity.child_id) - receiver_child_row = await dao.assert_child_exists(child_id=target_device_identity.child_id) - - conversation_id, idempotent = await dao.create_message( - conversation_type=PARENT_CHILD_CONVERSATION_TYPE, - participant_a_type=CHILD_PARTICIPANT_TYPE, - participant_a_id=str(device_identity.child_id), - participant_b_type=PARENT_PARTICIPANT_TYPE, - participant_b_id=str(receiver_child_row.child_id), - pair_key=f"{device_identity.child_id}:{target_device_identity.child_id}", - sender_type=CHILD_PARTICIPANT_TYPE, - sender_id=str(device_identity.child_id), - receiver_type=PARENT_PARTICIPANT_TYPE, - receiver_id=str(receiver_child_row.child_id), - sender_name_snapshot=sender_child_row["child_name"], - sender_avatar_snapshot=None, - receiver_name_snapshot=receiver_child_row["child_name"], - receiver_avatar_snapshot=None, - payload=None, - ) - - - return device_identity - except Exception: - await db_session.rollback() - raise - finally: - await db_session.close() - - -# 创建全局 ImService 实例 im_service = ImService() diff --git a/talkingq-url/banban/service/message_audio_storage.py b/talkingq-url/banban/service/message_audio_storage.py new file mode 100644 index 0000000..eb749d3 --- /dev/null +++ b/talkingq-url/banban/service/message_audio_storage.py @@ -0,0 +1,129 @@ +import asyncio +from dataclasses import dataclass +from datetime import UTC, datetime +from urllib.parse import urlparse +from uuid import uuid4 + +try: + from qcloud_cos import CosConfig, CosS3Client +except ModuleNotFoundError: # pragma: no cover - exercised in runtime env + CosConfig = None + CosS3Client = None + +from config import settings + + +class MessageAudioStorageError(Exception): + pass + + +@dataclass(frozen=True) +class StoredMessageAudio: + file_key: str + public_url: str + + +class MessageAudioStorageService: + def __init__(self) -> None: + self._client = None + + def _assert_ready(self) -> None: + if CosConfig is None or CosS3Client is None: + raise MessageAudioStorageError("COS SDK is not installed") + + required_pairs = { + "COS_SECRET_ID": settings.cos_secret_id, + "COS_SECRET_KEY": settings.cos_secret_key, + "COS_REGION": settings.cos_region, + "COS_BUCKET_MESSAGE": settings.cos_bucket_message, + } + missing = [key for key, value in required_pairs.items() if not value] + if missing: + raise MessageAudioStorageError(f"missing COS message config: {', '.join(missing)}") + + def _get_client(self): + if self._client is None: + config = CosConfig( + Region=settings.cos_region, + SecretId=settings.cos_secret_id, + SecretKey=settings.cos_secret_key, + Scheme="https", + ) + self._client = CosS3Client(config) + return self._client + + def _build_key(self, *, device_id: str, extension: str) -> str: + prefix = settings.cos_message_prefix.strip("/") or "messages/audio" + now = datetime.now(UTC) + return ( + f"{prefix}/{device_id}/{now.strftime('%Y/%m/%d')}/" + f"{uuid4().hex}.{extension}" + ) + + def _build_public_url(self, *, file_key: str) -> str: + base_url = settings.cos_public_base_url.strip().rstrip("/") + if not base_url: + base_url = f"https://{settings.cos_bucket_message}.cos.{settings.cos_region}.myqcloud.com" + return f"{base_url}/{file_key.lstrip('/')}" + + def _normalize_file_key(self, file_key_or_url: str) -> str: + value = (file_key_or_url or "").strip() + if not value: + raise MessageAudioStorageError("audio file key is required") + if value.startswith("http://") or value.startswith("https://"): + parsed = urlparse(value) + path = parsed.path.lstrip("/") + if not path: + raise MessageAudioStorageError("audio file key is invalid") + return path + return value.lstrip("/") + + async def upload_audio( + self, + *, + device_id: str, + content: bytes, + content_type: str = "audio/mpeg", + extension: str = "mp3", + ) -> StoredMessageAudio: + self._assert_ready() + if not content: + raise MessageAudioStorageError("audio content is empty") + + file_key = self._build_key(device_id=device_id, extension=extension) + await asyncio.to_thread( + self._upload_audio_sync, + file_key=file_key, + content=content, + content_type=content_type, + ) + return StoredMessageAudio( + file_key=file_key, + public_url=self._build_public_url(file_key=file_key), + ) + + async def get_audio_url(self, file_key_or_url: str) -> str: + self._assert_ready() + normalized_key = self._normalize_file_key(file_key_or_url) + return await asyncio.to_thread( + self._get_client().get_presigned_url, + Bucket=settings.cos_bucket_message, + Key=normalized_key, + Method="GET", + Expired=settings.cos_avatar_url_expire_seconds, + ) + + def _upload_audio_sync( + self, + *, + file_key: str, + content: bytes, + content_type: str, + ) -> None: + self._get_client().put_object( + Bucket=settings.cos_bucket_message, + Body=content, + Key=file_key, + ContentType=content_type, + EnableMD5=False, + ) diff --git a/talkingq-url/config.py b/talkingq-url/config.py index 28ed176..7b175e9 100644 --- a/talkingq-url/config.py +++ b/talkingq-url/config.py @@ -89,6 +89,7 @@ class Settings(BaseSettings): cos_bucket_message: str = Field(default="", validation_alias="COS_BUCKET_MESSAGE") cos_bucket_ava: str = Field(default="", validation_alias="COS_BUCKET_AVA") cos_public_base_url: str = Field(default="", validation_alias="COS_PUBLIC_BASE_URL") + cos_message_prefix: str = Field(default="messages/audio/", validation_alias="COS_MESSAGE_PREFIX") cos_avatar_prefix: str = Field(default="avatars/", validation_alias="COS_AVATAR_PREFIX") cos_avatar_url_expire_seconds: int = Field( default=86400, diff --git a/talkingq-url/handlers/audio_file_handler.py b/talkingq-url/handlers/audio_file_handler.py index efd34ab..e695baf 100644 --- a/talkingq-url/handlers/audio_file_handler.py +++ b/talkingq-url/handlers/audio_file_handler.py @@ -1,36 +1,19 @@ -import os -import uuid -from config import settings +from banban.service.message_audio_storage import MessageAudioStorageService from utils.logger import session_logger -# from utils.audio_denoiser import reduce_background_noise + + +message_audio_storage_service = MessageAudioStorageService() async def save_audio_file(audio_data: bytes, device_id: str) -> str: - """ - 保存音频数据到 assets/audio 目录 - - Args: - audio_data: 音频二进制数据 - device_id: 设备ID - - Returns: - 音频文件的相对路径 - """ + """Upload device audio to COS and return its object key.""" try: - audio_dir = os.path.join(settings.assets_dir, "audio") - os.makedirs(audio_dir, exist_ok=True) - - filename = f"{device_id}_{uuid.uuid4().hex[:8]}.mp3" - filepath = os.path.join(audio_dir, filename) - - with open(filepath, 'wb') as f: - f.write(audio_data) - - # relative_path = f"assets/audio/{filename}" - session_logger.info(device_id, "audio", f"音频文件已保存: {filepath}") - - # reduce_background_noise(filepath, relative_path,noise_path='assets/audio/noise_sample.wav',normalize_volume=True) - return filepath + stored = await message_audio_storage_service.upload_audio( + device_id=device_id, + content=audio_data, + ) + session_logger.info(device_id, "audio", f"audio uploaded to COS: {stored.file_key}") + return stored.file_key except Exception as e: - session_logger.error(device_id, "audio", f"保存音频文件时出错: {e}", exc_info=True) - raise \ No newline at end of file + session_logger.error(device_id, "audio", f"failed to store audio: {e}", exc_info=True) + raise diff --git a/talkingq-url/handlers/websocket_message_handler.py b/talkingq-url/handlers/websocket_message_handler.py index c3c4025..de684e4 100644 --- a/talkingq-url/handlers/websocket_message_handler.py +++ b/talkingq-url/handlers/websocket_message_handler.py @@ -4,7 +4,7 @@ import asyncio from fastapi import WebSocket from handlers.audio_packet_parser import parse_packet from handlers.audio_session_handler import handle_websocket_data -from handlers.audio_file_handler import save_audio_file +from handlers.audio_file_handler import message_audio_storage_service, save_audio_file from services.audio_session import audio_session_manager from services.interrupt_handler import interrupt_handler from services.task_manager import task_manager @@ -265,11 +265,14 @@ async def process_cached_audio(device_id: str, target_device_id: str, serial_num return # 保存音频文件 - audio_path = await save_audio_file(cached_audio, device_id) - audio_url = f"http://{settings.server_host}:{settings.server_port}/{audio_path}" + audio_file_key = await save_audio_file(cached_audio, device_id) # 将音频URL保存到数据库 im_conversation和im_message - await im_conversation_service.create_device_message(device_id=device_id, serial_number=serial_number, target_device_id=target_device_id, audio_url=audio_url) - + await im_conversation_service.create_device_message(device_id=device_id, serial_number=serial_number, target_device_id=target_device_id, audio_url=audio_file_key) + try: + audio_url = await message_audio_storage_service.get_audio_url(audio_file_key) + except Exception: + audio_url = audio_file_key + # 发送URL给目标设备 # target_websocket = await connection_manager.get_connection(target_device_id) # if target_websocket and target_websocket.client_state.name == "CONNECTED":