From 9952fc4365cc95c938c9095321dfb4e5757c2c13 Mon Sep 17 00:00:00 2001 From: ChengCan <783785929@qq.com> Date: Thu, 30 Apr 2026 00:58:56 +0800 Subject: [PATCH] feat(binding): support qr scan and nfc card binding flow --- banban-mini/src/pages/bind/index.scss | 7 + banban-mini/src/pages/bind/index.tsx | 299 +++++++++++++++------- banban-mini/src/services/binding.ts | 114 ++++++++- banban-mini/src/services/child.ts | 43 +++- talkingq-url/banban/dao/binding.py | 237 ++++++++++++----- talkingq-url/banban/routers/bindings.py | 86 +++++-- talkingq-url/banban/service/binding.py | 98 ++++++- talkingq-url/handlers/mqtt_handler.py | 327 ++++++++++-------------- talkingq-url/services/card_service.py | 262 +++++++++++-------- 9 files changed, 984 insertions(+), 489 deletions(-) diff --git a/banban-mini/src/pages/bind/index.scss b/banban-mini/src/pages/bind/index.scss index 007b012..d9cb7c9 100644 --- a/banban-mini/src/pages/bind/index.scss +++ b/banban-mini/src/pages/bind/index.scss @@ -218,6 +218,13 @@ color: #D46B08; } +.pending-tip-subtext { + display: block; + margin-top: 8px; + font-size: 24px; + color: #8C5400; +} + .child-list { display: flex; flex-wrap: wrap; diff --git a/banban-mini/src/pages/bind/index.tsx b/banban-mini/src/pages/bind/index.tsx index aaf8d6f..8abeb44 100644 --- a/banban-mini/src/pages/bind/index.tsx +++ b/banban-mini/src/pages/bind/index.tsx @@ -1,11 +1,31 @@ import { View, Text, Image, Input, Button } from '@tarojs/components' -import { useState } from 'react' -import Taro, { useDidShow } from '@tarojs/taro' +import { useEffect, useRef, useState } from 'react' +import Taro, { useDidShow, useDidHide } from '@tarojs/taro' + import { getToken } from '@/services/auth' -import { directBind, resolveActiveBinding, setBindingChild, setSelectedBindingDeviceId } from '@/services/binding' -import { Child, createChild, getChildren } from '@/services/child' +import { + getNFCBindSession, + resolveActiveBinding, + setBindingChild, + setSelectedBindingDeviceId, + startNFCBind, +} from '@/services/binding' +import { + Child, + createChild, + getChildren, + resolveChildSelection, + setSelectedChildId as setStoredSelectedChildId, +} from '@/services/child' + import './index.scss' +const SESSION_STATUS_PENDING = 1 +const SESSION_STATUS_COMPLETED = 2 +const SESSION_STATUS_EXPIRED = 3 +const SESSION_STATUS_FAILED = 4 +const SESSION_STATUS_CANCELLED = 5 + function parseBindingPayload(rawValue: string): { deviceId: string; serialNumber: string } { const raw = String(rawValue || '').trim() if (!raw) return { deviceId: '', serialNumber: '' } @@ -71,15 +91,99 @@ export default function Bind() { const [selectedChildId, setSelectedChildId] = useState(null) const [newChildName, setNewChildName] = useState('') const [pendingDeviceId, setPendingDeviceId] = useState(null) + const [bindToken, setBindToken] = useState('') + const [bindStatus, setBindStatus] = useState(null) + const [cardUUID, setCardUUID] = useState('') + const [bindHint, setBindHint] = useState('') + const pollingRef = useRef | null>(null) useDidShow(() => { void loadPageData() }) + useDidHide(() => { + stopPolling() + }) + + useEffect(() => () => stopPolling(), []) + + const hasSelectedOrNewChild = selectedChildId !== null || Boolean(newChildName.trim()) + const isPendingBinding = Boolean(pendingDeviceId) + const isPollingBind = bindStatus === SESSION_STATUS_PENDING && Boolean(bindToken) + const canEditDeviceFields = hasSelectedOrNewChild && !isPollingBind + const goDevicePage = () => { Taro.reLaunch({ url: '/pages/device/index' }) } + const stopPolling = () => { + if (pollingRef.current) { + clearTimeout(pollingRef.current) + pollingRef.current = null + } + } + + const schedulePoll = () => { + stopPolling() + pollingRef.current = setTimeout(() => { + void pollBindSession() + }, 1500) + } + + const resetBindSessionState = () => { + stopPolling() + setBindToken('') + setBindStatus(null) + setCardUUID('') + setBindHint('') + } + + const pollBindSession = async () => { + if (!bindToken) return + + try { + const session = await getNFCBindSession(bindToken) + setBindStatus(session.status) + setCardUUID(session.card_uuid || '') + + if (session.status === SESSION_STATUS_PENDING) { + setBindHint('已发送绑卡指令，请拿自己的卡去设备上贴一下') + schedulePoll() + return + } + + stopPolling() + + if (session.status === SESSION_STATUS_COMPLETED) { + setSelectedBindingDeviceId(session.device_id) + setBindHint(session.card_uuid ? `绑定完成，卡号 ${session.card_uuid}` : '绑定完成') + Taro.showToast({ title: '设备绑定成功', icon: 'success' }) + setTimeout(() => { + goDevicePage() + }, 500) + return + } + + if (session.status === SESSION_STATUS_EXPIRED) { + setBindHint('绑定会话已过期，请重新扫码并贴卡') + return + } + + if (session.status === SESSION_STATUS_FAILED) { + setBindHint('贴卡绑定失败，请重试') + return + } + + if (session.status === SESSION_STATUS_CANCELLED) { + setBindHint('当前绑定已被新的绑定流程替代，请重新扫码') + return + } + } catch (error: any) { + console.error('[bind] poll bind session failed:', error) + setBindHint(error?.message || '查询绑定状态失败，请重试') + } + } + const loadPageData = async () => { if (!getToken()) { Taro.reLaunch({ url: '/pages/login/index' }) @@ -90,17 +194,15 @@ export default function Bind() { try { const [childResponse, activeBinding] = await Promise.all([getChildren(), resolveActiveBinding()]) const currentChildren = childResponse.items || [] - + const currentChild = resolveChildSelection(currentChildren) setChildren(currentChildren) - setSelectedChildId((currentSelectedChildId) => currentSelectedChildId || currentChildren[0]?.child_id || null) + setSelectedChildId(currentChild?.child_id || null) if (activeBinding?.device_id && !activeBinding.child_id) { setPendingDeviceId(activeBinding.device_id) setDeviceId(activeBinding.device_id) } else { setPendingDeviceId(null) - setDeviceId('') - setSerialNumber('') } } catch (error: any) { console.error('[bind] load failed:', error) @@ -122,13 +224,21 @@ export default function Bind() { const child = await createChild({ child_name: childName }) setChildren((currentChildren) => [child, ...currentChildren]) setSelectedChildId(child.child_id) + setStoredSelectedChildId(child.child_id) setNewChildName('') return child.child_id } const handleScanCode = () => { - setIsScanning(true) + if (!canEditDeviceFields) { + Taro.showToast({ + title: '请先选择或新建儿童资料', + icon: 'none', + }) + return + } + setIsScanning(true) Taro.scanCode({ onlyFromCamera: true, scanType: ['qrCode', 'barCode'], @@ -145,11 +255,7 @@ export default function Bind() { }, fail: (err) => { setIsScanning(false) - - if (err.errMsg && err.errMsg.includes('cancel')) { - return - } - + if (err.errMsg && err.errMsg.includes('cancel')) return Taro.showToast({ title: '扫码失败，请重试', icon: 'none', @@ -159,22 +265,21 @@ export default function Bind() { } const handleSubmit = async () => { - if (submitting) return + if (submitting || isPollingBind) return setSubmitting(true) - Taro.showLoading({ title: pendingDeviceId ? '关联中...' : '绑定中...' }) + Taro.showLoading({ title: isPendingBinding ? '关联中...' : '发送绑卡指令...' }) try { const childId = await ensureChildId() + if (!childId) { + throw new Error('请先选择儿童或填写新的儿童昵称') + } - if (pendingDeviceId) { - if (!childId) { - throw new Error('请选择儿童或填写新儿童昵称') - } - - await setBindingChild(pendingDeviceId, { child_id: childId }) + if (isPendingBinding) { + await setBindingChild(pendingDeviceId!, { child_id: childId }) setSelectedBindingDeviceId(pendingDeviceId) - Taro.showToast({ title: '绑定完成', icon: 'success' }) + Taro.showToast({ title: '关联完成', icon: 'success' }) setTimeout(() => { goDevicePage() }, 300) @@ -188,30 +293,21 @@ export default function Bind() { throw new Error('请输入设备序列号') } - const result = await directBind({ + resetBindSessionState() + const session = await startNFCBind({ device_id: deviceId.trim(), serial_number: serialNumber.trim(), - ...(childId ? { child_id: childId } : {}), + child_id: childId, }) - setSelectedBindingDeviceId(result.device_id) - - if (childId || result.child_id) { - Taro.showToast({ title: '设备绑定成功', icon: 'success' }) - setTimeout(() => { - goDevicePage() - }, 300) - return - } - - setPendingDeviceId(result.device_id) - setDeviceId(result.device_id) - Taro.showToast({ - title: '设备已绑定，请继续关联儿童', - icon: 'none', - }) + setBindToken(session.bind_token) + setBindStatus(session.status) + setBindHint('已发送绑卡指令，请去设备上贴自己的卡') + schedulePoll() + Taro.showToast({ title: '请去设备上贴卡', icon: 'none' }) } catch (error: any) { console.error('[bind] submit failed:', error) + setBindHint(error?.message || '绑定失败，请重试') Taro.showToast({ title: error?.message || '绑定失败，请重试', icon: 'none', @@ -235,23 +331,25 @@ export default function Bind() { return ( - {pendingDeviceId ? '补全绑定' : '绑定设备'} + {isPendingBinding ? '补全绑定' : '扫码贴卡绑定设备'} - {pendingDeviceId ? '设备已绑到当前账号，请补充儿童资料完成配置' : '扫描二维码或手动填写设备号和序列号'} + {isPendingBinding + ? '当前设备已存在待补全关系，请先补充儿童资料' + : '先选择儿童，再扫码设备，系统会下发绑卡指令，随后去设备上贴卡确认'} - + - {isScanning ? '正在扫码...' : '点击扫码'} - 请将二维码放入框内 + + {!canEditDeviceFields ? '请先完成儿童资料' : isScanning ? '正在扫码...' : '点击扫码'} + + + {!canEditDeviceFields ? '先选儿童，再扫设备二维码' : '扫描设备二维码或条码'} + @@ -264,39 +362,7 @@ export default function Bind() { - 设备信息 - - - 设备号 - setDeviceId(event.detail.value)} - /> - - - - 设备序列号 - setSerialNumber(event.detail.value)} - /> - - - {pendingDeviceId && ( - - 当前待完成设备：{pendingDeviceId} - - )} - - - - 儿童信息 + 儿童资料 {children.length > 0 && ( @@ -305,7 +371,11 @@ export default function Bind() { key={child.child_id} className={`child-item ${selectedChildId === child.child_id ? 'selected' : ''}`} onClick={() => { - setSelectedChildId((currentChildId) => (currentChildId === child.child_id ? null : child.child_id)) + setSelectedChildId((currentChildId) => { + const nextChildId = currentChildId === child.child_id ? null : child.child_id + setStoredSelectedChildId(nextChildId) + return nextChildId + }) setNewChildName('') }} > @@ -325,16 +395,65 @@ export default function Bind() { setNewChildName(event.detail.value) if (event.detail.value) { setSelectedChildId(null) + setStoredSelectedChildId(null) } }} /> - 优先选择已有儿童，也可以直接输入一个新昵称。 + + {children.length > 0 + ? '可以选已有儿童，也可以在这里新建一个儿童后再绑定设备' + : '当前还没有儿童资料，请先创建一个儿童资料'} + - @@ -342,15 +461,15 @@ export default function Bind() { 绑定帮助 1 - 先扫描二维码，自动填入设备号和序列号。 + 先选择已有儿童，或者在本页新建一个儿童 2 - 如果二维码里只有设备号，请手动补充序列号。 + 扫描设备二维码，确认设备号和序列号正确 3 - 没有儿童资料时，可以直接在本页新建并完成绑定。 + 点击发送绑卡指令，然后去设备上贴自己的卡完成确认 diff --git a/banban-mini/src/services/binding.ts b/banban-mini/src/services/binding.ts index e0501a5..97e66db 100644 --- a/banban-mini/src/services/binding.ts +++ b/banban-mini/src/services/binding.ts @@ -1,5 +1,7 @@ import Taro from '@tarojs/taro' + import { request } from './api' +import { Child, clearSelectedChildId, getChildren, resolveChildSelection, setSelectedChildId } from './child' const SELECTED_BINDING_DEVICE_ID_KEY = 'selectedBindingDeviceId' @@ -26,12 +28,38 @@ export interface BindingListResponse { next_cursor?: number | null } +export interface BindingChildContext { + currentChild: Child | null + currentBinding: Binding | null +} + export interface DirectBindPayload { device_id: string serial_number: string child_id?: number } +export interface NFCSessionStartPayload { + device_id: string + serial_number: string + child_id?: number +} + +export interface NFCSessionStartResponse { + bind_token: string + expires_at: string + status: number +} + +export interface NFCSessionStatus { + bind_token: string + device_id: string + child_id: number | null + status: number + expires_at: string + card_uuid?: string | null +} + export function getSelectedBindingDeviceId(): string | null { const value = Taro.getStorageSync(SELECTED_BINDING_DEVICE_ID_KEY) const deviceId = String(value || '').trim() @@ -67,7 +95,7 @@ export function resolveBindingSelection(items: export async function getCurrentBinding(): Promise { try { - return await request('/bindings/current') + return await request('/banban/bindings/current') } catch (error: any) { if (error?.status === 404) return null throw error @@ -75,16 +103,13 @@ export async function getCurrentBinding(): Promise { } export async function getBindings(cursor?: number, limit: number = 20): Promise { - const query = buildQuery({ - cursor, - limit, - }) - return request(`/bindings?${query}`) + const query = buildQuery({ cursor, limit }) + return request(`/banban/bindings?${query}`) } export async function getBinding(deviceId: string): Promise { try { - return await request(`/bindings/${deviceId}`) + return await request(`/banban/bindings/${deviceId}`) } catch (error: any) { if (error?.status === 404) return null throw error @@ -100,7 +125,6 @@ export async function resolveActiveBinding(): Promise { setSelectedBindingDeviceId(selectedBinding.device_id) return selectedBinding } - clearSelectedBindingDeviceId() } @@ -115,24 +139,92 @@ export async function resolveActiveBinding(): Promise { } export async function directBind(data: DirectBindPayload): Promise<{ device_id: string; child_id: number | null }> { - return request('/bindings/direct', { + return request('/banban/bindings/direct', { method: 'POST', data }) +} + +export async function startNFCBind(data: NFCSessionStartPayload): Promise { + return request('/banban/bindings/start', { method: 'POST', data, }) } +export async function getNFCBindSession(bindToken: string): Promise { + return request(`/banban/bindings/sessions/${bindToken}`) +} + export async function setBindingChild( deviceId: string, data: { child_id: number } ): Promise<{ device_id: string; child_id: number | null }> { - return request(`/bindings/${deviceId}/child`, { + return request(`/banban/bindings/${deviceId}/child`, { method: 'PATCH', data, }) } export async function unbindDevice(deviceId: string): Promise { - return request(`/bindings/${deviceId}`, { + return request(`/banban/bindings/${deviceId}`, { method: 'DELETE', }) } + +export function resolveBindingForChild( + items: T[], + childId?: number | null +): T | null { + const normalizedChildId = Number(childId || 0) + if (!Number.isFinite(normalizedChildId) || normalizedChildId <= 0) return null + + const childBindings = items.filter((item) => item.child_id === normalizedChildId) + if (childBindings.length === 0) return null + + const selectedDeviceId = getSelectedBindingDeviceId() + if (selectedDeviceId) { + const selectedBinding = childBindings.find((item: any) => item.device_id === selectedDeviceId) || null + if (selectedBinding) return selectedBinding + } + + return childBindings[0] || null +} + +export function syncChildAndBindingSelection(children: Child[], bindings: Binding[]): BindingChildContext { + const currentChild = resolveChildSelection(children) + const currentBinding = currentChild ? resolveBindingForChild(bindings, currentChild.child_id) : null + + if (currentChild) { + setSelectedChildId(currentChild.child_id) + } else { + clearSelectedChildId() + } + + if (currentBinding?.device_id) { + setSelectedBindingDeviceId(currentBinding.device_id) + } else { + clearSelectedBindingDeviceId() + } + + return { + currentChild, + currentBinding, + } +} + +export async function loadCurrentChildBindingContext(): Promise<{ + children: Child[] + bindings: Binding[] + currentChild: Child | null + currentBinding: Binding | null +}> { + const [childResponse, bindingResponse] = await Promise.all([getChildren(undefined, 100), getBindings(undefined, 100)]) + const children = childResponse.items || [] + const bindings = bindingResponse.items || [] + const { currentChild, currentBinding } = syncChildAndBindingSelection(children, bindings) + + return { + children, + bindings, + currentChild, + currentBinding, + } +} diff --git a/banban-mini/src/services/child.ts b/banban-mini/src/services/child.ts index 67b8f29..f645901 100644 --- a/banban-mini/src/services/child.ts +++ b/banban-mini/src/services/child.ts @@ -1,5 +1,8 @@ +import Taro from '@tarojs/taro' import { request } from './api' +const SELECTED_CHILD_ID_KEY = 'selectedChildId' + function buildQuery(params: Record): string { return Object.entries(params) .filter(([, value]) => value !== undefined && value !== null && value !== '') @@ -21,12 +24,44 @@ export interface ChildListResponse { next_cursor?: number | null } +export function getSelectedChildId(): number | null { + const value = Number(Taro.getStorageSync(SELECTED_CHILD_ID_KEY) || 0) + return Number.isFinite(value) && value > 0 ? value : null +} + +export function setSelectedChildId(childId?: number | null) { + const normalizedChildId = Number(childId || 0) + if (!Number.isFinite(normalizedChildId) || normalizedChildId <= 0) { + Taro.removeStorageSync(SELECTED_CHILD_ID_KEY) + return + } + Taro.setStorageSync(SELECTED_CHILD_ID_KEY, normalizedChildId) +} + +export function clearSelectedChildId() { + Taro.removeStorageSync(SELECTED_CHILD_ID_KEY) +} + +export function resolveChildSelection(items: T[]): T | null { + const selectedChildId = getSelectedChildId() + const selectedChild = selectedChildId ? items.find((item) => item.child_id === selectedChildId) || null : null + const currentChild = selectedChild || items[0] || null + + if (currentChild) { + setSelectedChildId(currentChild.child_id) + } else { + clearSelectedChildId() + } + + return currentChild +} + export async function createChild(data: { child_name: string child_gender?: number child_birthday?: string }): Promise { - return request('/children', { + return request('/banban/children', { method: 'POST', data, }) @@ -37,18 +72,18 @@ export async function getChildren(cursor?: number, limit: number = 20): Promise< cursor, limit, }) - return request(`/children?${query}`) + return request(`/banban/children?${query}`) } export async function getChild(childId: number): Promise { - return request(`/children/${childId}`) + return request(`/banban/children/${childId}`) } export async function updateChild( childId: number, data: { child_name?: string; child_gender?: number; child_birthday?: string } ): Promise { - return request(`/children/${childId}`, { + return request(`/banban/children/${childId}`, { method: 'PATCH', data, }) diff --git a/talkingq-url/banban/dao/binding.py b/talkingq-url/banban/dao/binding.py index 9ef916a..73d5383 100644 --- a/talkingq-url/banban/dao/binding.py +++ b/talkingq-url/banban/dao/binding.py @@ -4,13 +4,21 @@ from collections.abc import Mapping from datetime import datetime, timedelta from typing import Optional -from sqlalchemy import text -from sqlalchemy.exc import IntegrityError - from banban.dao import BaseDAO logger = logging.getLogger("banban.dao.binding") +SESSION_STATUS_PENDING = 1 +SESSION_STATUS_COMPLETED = 2 +SESSION_STATUS_EXPIRED = 3 +SESSION_STATUS_FAILED = 4 +SESSION_STATUS_CANCELLED = 5 + +BIND_SOURCE_SESSION_CONFIRM = 1 +BIND_SOURCE_DIRECT = 2 +BIND_SOURCE_SET_CHILD = 3 +BIND_SOURCE_NFC = 4 + class BindingDAO(BaseDAO): async def get_device_auth(self, device_id: str) -> Optional[Mapping]: @@ -21,7 +29,7 @@ class BindingDAO(BaseDAO): FROM device_auth WHERE device_id = :device_id LIMIT 1 - """, + """, {"device_id": device_id}, ) ).mappings().first() @@ -39,7 +47,7 @@ class BindingDAO(BaseDAO): unbound_at = NULL, updated_at = CURRENT_TIMESTAMP WHERE id = :id - """, + """, {"id": row_id}, ) return @@ -50,51 +58,28 @@ class BindingDAO(BaseDAO): SET child_id = NULL, updated_at = CURRENT_TIMESTAMP WHERE id = :id - """, + """, {"id": row_id}, ) async def _upsert_parent_child_relation(self, user_id: int, child_id: int) -> None: - updated = await self.execute( + await self.execute( """ - UPDATE parent_child_relations - SET status = 1, + INSERT INTO parent_child_relations (user_id, child_id, relation_type, is_primary, status) + VALUES (:user_id, :child_id, 9, 0, 1) + ON DUPLICATE KEY UPDATE + status = VALUES(status), updated_at = CURRENT_TIMESTAMP - WHERE user_id = :user_id - AND child_id = :child_id - """, + """, {"user_id": user_id, "child_id": child_id}, ) - if updated.rowcount and updated.rowcount > 0: - return - - try: - await self.execute( - """ - INSERT INTO parent_child_relations (user_id, child_id, relation_type, is_primary, status) - VALUES (:user_id, :child_id, 9, 0, 1) - """, - {"user_id": user_id, "child_id": child_id}, - ) - except IntegrityError: - await self.db.rollback() - await self.execute( - """ - UPDATE parent_child_relations - SET status = 1, - updated_at = CURRENT_TIMESTAMP - WHERE user_id = :user_id - AND child_id = :child_id - """, - {"user_id": user_id, "child_id": child_id}, - ) async def _insert_bind_history(self, device_id: str, child_id: Optional[int], user_id: int, bind_source: int) -> None: await self.execute( """ INSERT INTO device_bind_history (device_id, child_id, bound_by_user_id, bind_source, bound_at) VALUES (:device_id, :child_id, :user_id, :bind_source, CURRENT_TIMESTAMP) - """, + """, { "device_id": device_id, "child_id": child_id, @@ -122,7 +107,7 @@ class BindingDAO(BaseDAO): unbound_at = NULL, updated_at = CURRENT_TIMESTAMP WHERE device_id = :device_id - """, + """, {"owner_user_id": user_id, "device_id": device_id}, ) else: @@ -130,7 +115,7 @@ class BindingDAO(BaseDAO): """ INSERT INTO device_bindings (device_id, owner_user_id, child_id, status, bound_at) VALUES (:device_id, :owner_user_id, NULL, 1, CURRENT_TIMESTAMP) - """, + """, {"device_id": device_id, "owner_user_id": user_id}, ) return @@ -156,8 +141,8 @@ class BindingDAO(BaseDAO): unbound_at = NULL, updated_at = CURRENT_TIMESTAMP WHERE device_id = :device_id - """, - {"owner_user_id": user_id, "child_id": child_id, "device_id": device_id}, + """, + {"owner_user_id": user_id, "child_id": child_id, "device_id": device_id}, ) return @@ -168,57 +153,160 @@ class BindingDAO(BaseDAO): """ INSERT INTO device_bindings (device_id, owner_user_id, child_id, status, bound_at) VALUES (:device_id, :owner_user_id, :child_id, 1, CURRENT_TIMESTAMP) - """, + """, {"device_id": device_id, "owner_user_id": user_id, "child_id": child_id}, ) - async def start_bind(self, user_id: int, device_id: str, child_id: Optional[int]) -> str: + async def start_bind(self, user_id: int, device_id: str, child_id: Optional[int]) -> tuple[str, datetime]: bind_token = str(uuid.uuid4()) expires_at = datetime.utcnow() + timedelta(minutes=10) + await self.execute( + """ + UPDATE device_bind_sessions + SET status = :cancelled_status, + consumed_at = COALESCE(consumed_at, CURRENT_TIMESTAMP), + updated_at = CURRENT_TIMESTAMP + WHERE device_id = :device_id + AND status = :pending_status + """, + { + "device_id": device_id, + "pending_status": SESSION_STATUS_PENDING, + "cancelled_status": SESSION_STATUS_CANCELLED, + }, + ) + await self.execute( """ INSERT INTO device_bind_sessions (bind_token, device_id, initiator_user_id, target_child_id, expires_at, status) - VALUES (:bind_token, :device_id, :initiator_user_id, :target_child_id, :expires_at, 1) - """, + VALUES (:bind_token, :device_id, :initiator_user_id, :target_child_id, :expires_at, :status) + """, { "bind_token": bind_token, "device_id": device_id, "initiator_user_id": user_id, "target_child_id": child_id, "expires_at": expires_at, + "status": SESSION_STATUS_PENDING, }, ) - await self.commit() - return bind_token + return bind_token, expires_at async def get_session(self, bind_token: str, user_id: int) -> Optional[Mapping]: return ( await self.execute( - "SELECT * FROM device_bind_sessions WHERE bind_token = :bind_token AND initiator_user_id = :user_id", - {"bind_token": bind_token, "user_id": user_id}, + """ + SELECT + s.*, + CASE + WHEN s.status = :completed_status + AND s.confirmed_at IS NOT NULL + AND c.updated_at >= s.confirmed_at + THEN c.card_uuid + ELSE NULL + END AS card_uuid + FROM device_bind_sessions AS s + LEFT JOIN cards AS c + ON c.device_id = s.device_id + WHERE s.bind_token = :bind_token + AND s.initiator_user_id = :user_id + LIMIT 1 + """, + { + "bind_token": bind_token, + "user_id": user_id, + "completed_status": SESSION_STATUS_COMPLETED, + }, ) ).mappings().first() + async def get_latest_pending_session_by_device(self, device_id: str) -> Optional[Mapping]: + return ( + await self.execute( + """ + SELECT * + FROM device_bind_sessions + WHERE device_id = :device_id + AND status = :status + ORDER BY id DESC + LIMIT 1 + """, + {"device_id": device_id, "status": SESSION_STATUS_PENDING}, + ) + ).mappings().first() + + async def mark_session_status(self, session_id: int, status: int) -> None: + await self.execute( + """ + UPDATE device_bind_sessions + SET status = :status, + consumed_at = COALESCE(consumed_at, CURRENT_TIMESTAMP), + updated_at = CURRENT_TIMESTAMP + WHERE id = :id + """, + {"id": session_id, "status": status}, + ) + async def confirm_bind(self, session_id: int, device_id: str, child_id: Optional[int], user_id: int) -> None: if child_id is not None: await self._upsert_parent_child_relation(user_id=user_id, child_id=child_id) await self.execute( - "UPDATE device_bind_sessions SET status = 2, confirmed_at = CURRENT_TIMESTAMP WHERE id = :id", - {"id": session_id}, + """ + UPDATE device_bind_sessions + SET status = :status, + confirmed_at = CURRENT_TIMESTAMP, + consumed_at = CURRENT_TIMESTAMP, + updated_at = CURRENT_TIMESTAMP + WHERE id = :id + """, + {"id": session_id, "status": SESSION_STATUS_COMPLETED}, ) await self._bind_device(device_id=device_id, user_id=user_id, child_id=child_id) - await self._insert_bind_history(device_id=device_id, child_id=child_id, user_id=user_id, bind_source=1) - await self.commit() + await self._insert_bind_history( + device_id=device_id, + child_id=child_id, + user_id=user_id, + bind_source=BIND_SOURCE_SESSION_CONFIRM, + ) + + async def complete_nfc_bind(self, session_id: int, device_id: str, child_id: Optional[int], user_id: int) -> None: + if child_id is not None: + await self._upsert_parent_child_relation(user_id=user_id, child_id=child_id) + + await self.execute( + """ + UPDATE device_bind_sessions + SET status = :status, + confirmed_at = CURRENT_TIMESTAMP, + consumed_at = CURRENT_TIMESTAMP, + updated_at = CURRENT_TIMESTAMP + WHERE id = :id + """, + {"id": session_id, "status": SESSION_STATUS_COMPLETED}, + ) + + await self._bind_device(device_id=device_id, user_id=user_id, child_id=child_id) + await self._insert_bind_history( + device_id=device_id, + child_id=child_id, + user_id=user_id, + bind_source=BIND_SOURCE_NFC, + ) async def direct_bind(self, device_id: str, child_id: Optional[int], user_id: int) -> None: if child_id is not None: await self._upsert_parent_child_relation(user_id=user_id, child_id=child_id) await self._bind_device(device_id=device_id, user_id=user_id, child_id=child_id) - await self._insert_bind_history(device_id=device_id, child_id=child_id, user_id=user_id, bind_source=2) + await self._insert_bind_history( + device_id=device_id, + child_id=child_id, + user_id=user_id, + bind_source=BIND_SOURCE_DIRECT, + ) await self.commit() async def get_current_by_user(self, user_id: int) -> Optional[Mapping]: @@ -231,7 +319,7 @@ class BindingDAO(BaseDAO): AND status = 1 ORDER BY bound_at DESC LIMIT 1 - """, + """, {"user_id": user_id}, ) ).mappings().first() @@ -260,7 +348,7 @@ class BindingDAO(BaseDAO): WHERE {where} ORDER BY db.id DESC LIMIT :limit - """, + """, params, ) ).mappings().all() @@ -275,7 +363,7 @@ class BindingDAO(BaseDAO): WHERE device_id = :device_id AND owner_user_id = :user_id AND status = 1 - """, + """, {"device_id": device_id, "user_id": user_id}, ) ).mappings().first() @@ -287,7 +375,12 @@ class BindingDAO(BaseDAO): await self._upsert_parent_child_relation(user_id=user_id, child_id=child_id) await self._bind_device(device_id=device_id, user_id=user_id, child_id=child_id) - await self._insert_bind_history(device_id=device_id, child_id=child_id, user_id=user_id, bind_source=3) + await self._insert_bind_history( + device_id=device_id, + child_id=child_id, + user_id=user_id, + bind_source=BIND_SOURCE_SET_CHILD, + ) await self.commit() return True @@ -302,7 +395,30 @@ class BindingDAO(BaseDAO): ) await self.execute( - "INSERT INTO device_bind_history (device_id, child_id, bound_by_user_id, unbound_by_user_id, bind_source, bound_at, unbound_at, unbind_reason) SELECT device_id, child_id, bound_by_user_id, :user_id, bind_source, bound_at, CURRENT_TIMESTAMP, 'user_unbind' FROM device_bind_history WHERE device_id = :device_id AND unbound_at IS NULL", + """ + INSERT INTO device_bind_history ( + device_id, + child_id, + bound_by_user_id, + unbound_by_user_id, + bind_source, + bound_at, + unbound_at, + unbind_reason + ) + SELECT + device_id, + child_id, + bound_by_user_id, + :user_id, + bind_source, + bound_at, + CURRENT_TIMESTAMP, + 'user_unbind' + FROM device_bind_history + WHERE device_id = :device_id + AND unbound_at IS NULL + """, {"device_id": device_id, "user_id": user_id}, ) await self.commit() @@ -318,11 +434,12 @@ class BindingDAO(BaseDAO): rows = ( await self.execute( f""" - SELECT * FROM device_bind_history + SELECT * + FROM device_bind_history WHERE {where} ORDER BY bound_at DESC LIMIT :limit - """, + """, params, ) ).mappings().all() diff --git a/talkingq-url/banban/routers/bindings.py b/talkingq-url/banban/routers/bindings.py index 8db2d62..3104c04 100644 --- a/talkingq-url/banban/routers/bindings.py +++ b/talkingq-url/banban/routers/bindings.py @@ -4,12 +4,15 @@ from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from pydantic import BaseModel -try: - from banban.security import get_current_user_id - from banban.service.binding import BindingError, BindingService -except ModuleNotFoundError: - from banban.security import get_current_user_id - from banban.service.binding import BindingError, BindingService +from banban.dao.binding import ( + SESSION_STATUS_CANCELLED, + SESSION_STATUS_COMPLETED, + SESSION_STATUS_EXPIRED, + SESSION_STATUS_FAILED, + SESSION_STATUS_PENDING, +) +from banban.security import get_current_user_id +from banban.service.binding import BindingError, BindingService router = APIRouter(prefix="/bindings", tags=["bindings"]) @@ -25,6 +28,16 @@ class BindStartRequest(BaseModel): class BindStartResponse(BaseModel): bind_token: str expires_at: str + status: int + + +class BindSessionResponse(BaseModel): + bind_token: str + device_id: str + child_id: int | None = None + status: int + expires_at: str + card_uuid: str | None = None class BindConfirmRequest(BaseModel): @@ -77,6 +90,7 @@ async def start_bind( request: Request, current_user_id: int = Depends(get_current_user_id), ) -> BindStartResponse: + del request service = BindingService() try: bind_token, expires_at = await service.start_bind( @@ -85,9 +99,35 @@ async def start_bind( payload.serial_number, payload.child_id, ) - except BindingError as e: - raise HTTPException(status_code=e.status_code, detail=str(e)) - return BindStartResponse(bind_token=bind_token, expires_at=expires_at.isoformat()) + except BindingError as exc: + raise HTTPException(status_code=exc.status_code, detail=str(exc)) + return BindStartResponse( + bind_token=bind_token, + expires_at=expires_at.isoformat(), + status=SESSION_STATUS_PENDING, + ) + + +@router.get("/sessions/{bind_token}", response_model=BindSessionResponse) +async def get_bind_session( + bind_token: str, + request: Request, + current_user_id: int = Depends(get_current_user_id), +) -> BindSessionResponse: + del request + service = BindingService() + session = await service.get_bind_session(bind_token, current_user_id) + if not session: + raise HTTPException(status_code=404, detail="bind session not found") + + return BindSessionResponse( + bind_token=session["bind_token"], + device_id=session["device_id"], + child_id=session["target_child_id"], + status=int(session["status"]), + expires_at=session["expires_at"].isoformat(), + card_uuid=session.get("card_uuid"), + ) @router.post("/confirm", response_model=BindConfirmResponse) @@ -96,13 +136,14 @@ async def confirm_bind( request: Request, current_user_id: int = Depends(get_current_user_id), ) -> BindConfirmResponse: + del request service = BindingService() try: result = await service.confirm_bind(payload.bind_token, current_user_id) - except BindingError as e: - raise HTTPException(status_code=e.status_code, detail=str(e)) - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) + except BindingError as exc: + raise HTTPException(status_code=exc.status_code, detail=str(exc)) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) return BindConfirmResponse(**result) @@ -127,6 +168,7 @@ async def direct_bind( request: Request, current_user_id: int = Depends(get_current_user_id), ) -> DirectBindResponse: + del request service = BindingService() try: result = await service.direct_bind( @@ -135,8 +177,8 @@ async def direct_bind( payload.child_id, current_user_id, ) - except BindingError as e: - raise HTTPException(status_code=e.status_code, detail=str(e)) + except BindingError as exc: + raise HTTPException(status_code=exc.status_code, detail=str(exc)) return DirectBindResponse(**result) @@ -147,13 +189,14 @@ async def set_binding_child( request: Request, current_user_id: int = Depends(get_current_user_id), ) -> DirectBindResponse: + del request service = BindingService() try: result = await service.set_binding_child(device_id=device_id, child_id=payload.child_id, user_id=current_user_id) - except BindingError as e: - raise HTTPException(status_code=e.status_code, detail=str(e)) - except ValueError as e: - raise HTTPException(status_code=404, detail=str(e)) + except BindingError as exc: + raise HTTPException(status_code=exc.status_code, detail=str(exc)) + except ValueError as exc: + raise HTTPException(status_code=404, detail=str(exc)) return DirectBindResponse(**result) @@ -162,6 +205,7 @@ async def get_current_binding( request: Request, current_user_id: int = Depends(get_current_user_id), ): + del request service = BindingService() binding = await service.get_current_binding(current_user_id) if not binding: @@ -176,6 +220,7 @@ async def list_bindings( limit: int = Query(default=20, ge=1, le=100), current_user_id: int = Depends(get_current_user_id), ) -> BindingListResponse: + del request service = BindingService() rows, has_more = await service.list_bindings(current_user_id, limit, cursor) next_cursor = int(rows[-1]["id"]) if has_more and rows else None @@ -198,6 +243,7 @@ async def get_binding( request: Request, current_user_id: int = Depends(get_current_user_id), ) -> BindingGetResponse: + del request service = BindingService() binding = await service.get_binding(device_id, current_user_id) if not binding: @@ -211,6 +257,7 @@ async def unbind_device( request: Request, current_user_id: int = Depends(get_current_user_id), ) -> None: + del request service = BindingService() if not await service.unbind(device_id, current_user_id): raise HTTPException(status_code=404, detail="binding not found") @@ -224,6 +271,7 @@ async def get_bind_history( limit: int = 20, current_user_id: int = Depends(get_current_user_id), ) -> BindHistoryResponse: + del request, current_user_id cursor_dt = datetime.fromisoformat(cursor) if cursor else None service = BindingService() rows, has_more = await service.list_history(device_id, limit, cursor_dt) diff --git a/talkingq-url/banban/service/binding.py b/talkingq-url/banban/service/binding.py index 34bc87a..cc7b364 100644 --- a/talkingq-url/banban/service/binding.py +++ b/talkingq-url/banban/service/binding.py @@ -2,7 +2,15 @@ from collections.abc import Mapping from datetime import datetime from typing import Optional -from banban.dao.binding import BindingDAO +from banban.dao.binding import ( + SESSION_STATUS_CANCELLED, + SESSION_STATUS_COMPLETED, + SESSION_STATUS_EXPIRED, + SESSION_STATUS_FAILED, + SESSION_STATUS_PENDING, + BindingDAO, +) +from services.card_service import card_service from services.database_service_base import DatabaseServiceBase @@ -26,6 +34,12 @@ class BindingService(DatabaseServiceBase): if int(row["is_active"]) != 1: raise BindingError("device is inactive", status_code=400) + def _normalize_session_status(self, session: Mapping) -> int: + status = int(session["status"]) + if status == SESSION_STATUS_PENDING and datetime.utcnow() > session["expires_at"]: + return SESSION_STATUS_EXPIRED + return status + async def start_bind( self, user_id: int, @@ -37,9 +51,15 @@ class BindingService(DatabaseServiceBase): try: await self._ensure_bindable_device(db_session, device_id, serial_number) dao = BindingDAO(db_session) - bind_token = await dao.start_bind(user_id, device_id, child_id) + bind_token, expires_at = await dao.start_bind(user_id, device_id, child_id) await db_session.commit() - return bind_token, datetime.utcnow() + from handlers.mqtt_handler import TalkingQMQTTService + + service = await TalkingQMQTTService.get_instance() + if service is None: + raise BindingError("MQTT service is unavailable", status_code=503) + await service.send_bind_nfc_command(device_id) + return bind_token, expires_at finally: await db_session.close() @@ -52,7 +72,7 @@ class BindingService(DatabaseServiceBase): raise ValueError("Bind session not found") if datetime.utcnow() > session["expires_at"]: raise ValueError("Bind session expired") - if session["status"] != 1: + if int(session["status"]) != SESSION_STATUS_PENDING: raise ValueError("Bind session already processed") await dao.confirm_bind(session["id"], session["device_id"], session["target_child_id"], user_id) @@ -61,6 +81,76 @@ class BindingService(DatabaseServiceBase): finally: await db_session.close() + async def get_bind_session(self, bind_token: str, user_id: int) -> Optional[Mapping]: + db_session = await self.get_session() + try: + dao = BindingDAO(db_session) + session = await dao.get_session(bind_token, user_id) + if session is None: + return None + + normalized_status = self._normalize_session_status(session) + if normalized_status == SESSION_STATUS_EXPIRED and int(session["status"]) != SESSION_STATUS_EXPIRED: + await dao.mark_session_status(int(session["id"]), SESSION_STATUS_EXPIRED) + await db_session.commit() + session = await dao.get_session(bind_token, user_id) + if session is None: + return None + normalized_status = SESSION_STATUS_EXPIRED + + payload = dict(session) + payload["status"] = normalized_status + payload["card_uuid"] = payload.get("card_uuid") + return payload + finally: + await db_session.close() + + async def finalize_nfc_bind(self, device_id: str, card_uuid: str) -> Optional[Mapping]: + db_session = await self.get_session() + try: + dao = BindingDAO(db_session) + session = await dao.get_latest_pending_session_by_device(device_id) + if not session: + await db_session.rollback() + return None + + if datetime.utcnow() > session["expires_at"]: + await dao.mark_session_status(int(session["id"]), SESSION_STATUS_EXPIRED) + await db_session.commit() + return { + "device_id": device_id, + "bind_token": session["bind_token"], + "status": SESSION_STATUS_EXPIRED, + } + + try: + await card_service.activate_card( + card_uuid=card_uuid, + device_id=device_id, + db_session=db_session, + ) + await dao.complete_nfc_bind( + session_id=int(session["id"]), + device_id=device_id, + child_id=session["target_child_id"], + user_id=int(session["initiator_user_id"]), + ) + await db_session.commit() + except Exception: + await dao.mark_session_status(int(session["id"]), SESSION_STATUS_FAILED) + await db_session.commit() + raise + + return { + "device_id": device_id, + "bind_token": session["bind_token"], + "status": SESSION_STATUS_COMPLETED, + "child_id": session["target_child_id"], + "card_uuid": card_uuid, + } + finally: + await db_session.close() + async def get_binding(self, device_id: str, user_id: int) -> Optional[Mapping]: db_session = await self.get_session() try: diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index d70b2fb..b1533f2 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/talkingq-url/handlers/mqtt_handler.py @@ -1,16 +1,18 @@ +import asyncio import json import time -import asyncio -from typing import Optional, Dict, Callable, Awaitable -from banban.service.device_setting import device_setting_service -from config import settings -from services.card_service import card_service -from services.offline_audio_cache import offline_audio_cache -import aiomqtt -from banban.service.location import location_service -from database.models import ChildLocationCurrent +from typing import Awaitable, Callable, Dict, Optional +import aiomqtt + +from banban.service.binding import BindingService +from banban.service.device_setting import device_setting_service +from banban.service.location import location_service +from config import settings +from database.models import ChildLocationCurrent +from services.card_service import card_service from services.device_target_cache import device_target_cache +from services.offline_audio_cache import offline_audio_cache from utils.logger import session_logger as logger @@ -64,130 +66,117 @@ class TalkingQMQTTService: try: topic = str(message.topic) parts = topic.split("/") - if parts[0] == "device": - device_id = parts[1] if len(parts) >= 2 else "unknown" + if not parts or parts[0] != "device": + continue - if not device_id.startswith(f"{self.device_prefix}_"): - continue + device_id = parts[1] if len(parts) >= 2 else "unknown" + if not device_id.startswith(f"{self.device_prefix}_"): + continue - payload = json.loads(message.payload.decode("utf-8")) - - msg_id = payload.get("msg_id") - - handler = self._msg_handlers.get(msg_id) - if handler: - await handler(device_id, payload) - else: - logger.warning(device_id, "", f"未知 msg_id: {msg_id}, 设备: {device_id}") - - except json.JSONDecodeError as e: - logger.error("", "", f"消息解析失败: {e}") - except Exception as e: - logger.error("", "", f"消息处理异常: {e}") + payload = json.loads(message.payload.decode("utf-8")) + msg_id = payload.get("msg_id") + handler = self._msg_handlers.get(msg_id) + if handler: + await handler(device_id, payload) + else: + logger.warning(device_id, "", f"unknown msg_id={msg_id}") + except json.JSONDecodeError as exc: + logger.error("", "", f"invalid mqtt payload: {exc}") + except Exception as exc: + logger.error("", "", f"mqtt message handling failed: {exc}") except asyncio.CancelledError: pass - except Exception as e: - logger.error("", "", f"消息循环异常: {e}") + except Exception as exc: + logger.error("", "", f"mqtt loop failed: {exc}") self._connected = False async def _handle_device_info(self, device_id: str, payload: dict): - # status = payload.get("status") data = payload.get("data", {}) - # if status == "success": - d_id = data.get("id") - power = data.get("power") - signal = data.get("signal") - version = data.get("version") - voice = data.get("voice") - logger.info(device_id, "", f"[设备信息] 设备 {d_id} 信息: 电量={power}, 信号强度={signal}, 版本号={version}, 音量={voice}") - # 插入到数据库 - await device_setting_service.insert_or_update(device_id=device_id, power=power, signal_strength=signal, version_str=version, volume=voice) - # else: - # logger.warning(device_id, "", f"[设备信息] 设备 {device_id} 查询失败: {payload}") + await device_setting_service.insert_or_update( + device_id=device_id, + power=data.get("power"), + signal_strength=data.get("signal"), + version_str=data.get("version"), + volume=data.get("voice"), + ) await self._publish(f"device/{device_id}/event_resp", {"msg_id": "000", "status": "success"}) async def _handle_gps_response(self, device_id: str, payload: dict): - status = payload.get("status") + if payload.get("status") != "success": + logger.warning(device_id, "", f"[GPS] query failed: {payload}") + return + data = payload.get("data", {}) - if status == "success": - lat = data.get("latitude") - lon = data.get("longitude") - logger.info(device_id, "", f"[GPS] 设备 {device_id} 位置: 纬度={lat}, 经度={lon}") - # 插入到数据库 - location = ChildLocationCurrent(device_id=device_id, lat=lat, lon=lon) - await location_service.insert_or_update(device_id=device_id, location=location) - else: - logger.warning(device_id, "", f"[GPS] 设备 {device_id} 查询失败: {payload}") + location = ChildLocationCurrent( + device_id=device_id, + lat=data.get("latitude"), + lon=data.get("longitude"), + ) + await location_service.insert_or_update(device_id=device_id, location=location) async def _handle_volume_response(self, device_id: str, payload: dict): - status = payload.get("status") - data = payload.get("data", {}) - if status == "success": - level = data.get("current_level") - logger.info(device_id, "", f"[音量] 设备 {device_id} 当前音量: {level}") - # 更新设备音量 - await device_setting_service.insert_or_update(device_id=device_id, volume=level) + if payload.get("status") != "success": + logger.warning(device_id, "", f"[volume] command failed: {payload}") + return - else: - logger.warning(device_id, "", f"[音量] 设备 {device_id} 调节失败: {payload}") + data = payload.get("data", {}) + await device_setting_service.insert_or_update(device_id=device_id, volume=data.get("current_level")) async def _handle_ota_response(self, device_id: str, payload: dict): status = payload.get("status") data = payload.get("data", {}) - if status == "accepted": - current = data.get("current_version") - target = data.get("target_version") - logger.info(device_id, "", f"[OTA] 设备 {device_id} 已接受升级: {current} -> {target}") - elif status == "success": - new_ver = data.get("new_version") - logger.info(device_id, "", f"[OTA] 设备 {device_id} 升级完成: {new_ver}") - # 更新设备版本号 - await device_setting_service.insert_or_update(device_id=device_id, version_str=new_ver) - else: - logger.warning(device_id, "", f"[OTA] 设备 {device_id} 升级异常: {payload}") + if status == "success": + await device_setting_service.insert_or_update(device_id=device_id, version_str=data.get("new_version")) + elif status != "accepted": + logger.warning(device_id, "", f"[OTA] command failed: {payload}") async def _handle_nfc_notice_response(self, device_id: str, payload: dict): status = payload.get("status") - if status == "success": - logger.info(device_id, "", f"[NFC通知] 设备 {device_id} 已收到留言提示") - else: - logger.warning(device_id, "", f"[NFC通知] 设备 {device_id} 留言提示失败: {payload}") + if status != "success": + logger.warning(device_id, "", f"[NFC notice] command failed: {payload}") async def _handle_nfc_listen_report(self, device_id: str, payload: dict): params = payload.get("params", {}) nfc_uuid = params.get("uuid") - logger.info(device_id, "", f"[NFC收听] 设备 {device_id} 请求收听留言, UUID={nfc_uuid}") await self._send_nfc_listen_response(device_id, nfc_uuid) - + async def _handle_bind_response(self, device_id: str, payload: dict): params = payload.get("params", {}) - status = params.get("status") - logger.info(device_id, "", f"[NFC绑定] 设备 {device_id} 请求绑定卡片返回状态: {status}") + data = payload.get("data", {}) + status = payload.get("status") or params.get("status") + logger.info(device_id, "", f"[NFC bind] device={device_id} status={status}") - nfc_uuid = params.get("uuid") - # 卡片与设备绑定 - await card_service.activate_card(device_id=device_id, card_uuid=nfc_uuid) - logger.info(device_id, "", f"[NFC绑定] 设备 {device_id} 请求绑定卡片, UUID={nfc_uuid}") + if status not in (None, "success", "accepted"): + logger.warning(device_id, "", f"[NFC bind] bind failed: {payload}") + return + + nfc_uuid = data.get("uuid") or params.get("uuid") + if not nfc_uuid: + logger.warning(device_id, "", f"[NFC bind] missing uuid: {payload}") + return + + service = BindingService() + result = await service.finalize_nfc_bind(device_id=device_id, card_uuid=nfc_uuid) + if result is None: + logger.warning(device_id, "", "[NFC bind] no pending bind session found") + return + + logger.info(device_id, "", f"[NFC bind] bind completed, uuid={nfc_uuid}, result={result}") async def _handle_open_response(self, device_id: str, payload: dict): params = payload.get("params", {}) - status = params.get("status") - logger.info(device_id, "", f"[设备状态] 设备 {device_id} 请求返回设备状态: {status}") - data = params.get("data", {}) - device_status = data.get("status") - logger.info(device_id, "", f"[设备状态] 设备 {device_id} 请求设备状态: {device_status}") + logger.info(device_id, "", f"[device open] response={params}") async def _send_nfc_listen_response(self, device_id: str, nfc_uuid: str): topic = f"device/{device_id}/event_resp" card = await card_service.get_card_by_uuid(nfc_uuid) if not card: - logger.warning("", "", f"[NFC收听] 设备 {device_id} 没有找到卡片{nfc_uuid}，无法发送留言提示") payload = { "msg_id": "005", "type": 0, "params": { "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/error_card.mp3" - } + }, } await self._publish(topic, payload) return @@ -201,75 +190,54 @@ class TalkingQMQTTService: "type": 0, "params": { "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/test_zh.mp3" - } + }, } await self._publish(topic, payload) - return + return + if has_pending: audio_urls = await offline_audio_cache.get_audio_urls(device_id) - # 53D92B6DA20001 测试卡片 - if nfc_uuid == "53D92B6DA20001": - payload = { - "msg_id": "005", - "type": 0, - "params": { - "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/test_zh.mp3" - } - } - await self._publish(topic, payload) - return if len(audio_urls) == 0: payload = { "msg_id": "005", "type": 0, "params": { "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/no_message.mp3" - } + }, } else: - params = {} - for k, audio_url in enumerate(audio_urls, start=1): - params[f"url_{k}"] = audio_url - payload = { - "msg_id": "005", - "type": 0, - "params": params - } + params = {f"url_{k}": audio_url for k, audio_url in enumerate(audio_urls, start=1)} + payload = {"msg_id": "005", "type": 0, "params": params} await offline_audio_cache.clear_audio_urls(device_id) await self._publish(topic, payload) - else: - payload = { - "msg_id": "005", - "type": 0, - "params": { - "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/no_message.mp3" - } - } - await self._publish(topic, payload) - else: - # 检查卡片是否存在 - existing_card = await card_service.get_card_by_uuid(nfc_uuid) - - if existing_card: - # 卡片已存在，使用卡片绑定的设备ID作为目标设备ID - target_device_id = existing_card.device_id - logger.info(device_id, "card", f"卡片已存在，绑定的设备ID: {target_device_id}") - else: - # 卡片不存在，创建新卡片并绑定到当前设备 - new_card = await card_service.activate_card(nfc_uuid, device_id) - logger.info(device_id, "card", f"新卡片{nfc_uuid}已创建并激活，绑定到设备: {device_id}") return - - # 设置目标设备 - await device_target_cache.set_target(device_id, target_device_id) + payload = { "msg_id": "005", - "type": 1, + "type": 0, "params": { - "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/welcome.mp3" - } + "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/no_message.mp3" + }, } await self._publish(topic, payload) + return + + existing_card = await card_service.get_card_by_uuid(nfc_uuid) + if existing_card: + target_device_id = existing_card.device_id + else: + await card_service.activate_card(nfc_uuid, device_id) + return + + await device_target_cache.set_target(device_id, target_device_id) + payload = { + "msg_id": "005", + "type": 1, + "params": { + "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/welcome.mp3" + }, + } + await self._publish(topic, payload) async def connect(self): async with self._connect_lock: @@ -286,19 +254,15 @@ class TalkingQMQTTService: ) await self._client.__aenter__() self._connected = True - logger.system_info("", f"TalkingQ MQTT 已连接: {self.broker}:{self.port}") await self._client.subscribe("device/+/response", qos=self.qos) - logger.system_info("", "已订阅: device/+/response") - await self._client.subscribe("device/+/event", qos=self.qos) - logger.system_info("", "已订阅: device/+/event") self._message_task = asyncio.create_task(self._message_loop()) - logger.info("", "", "TalkingQ MQTT 服务启动中...") - except Exception as e: + logger.info("", "", f"mqtt connected: {self.broker}:{self.port}") + except Exception as exc: self._connected = False - logger.error("", "", f"TalkingQ MQTT 连接失败: {e}") + logger.error("", "", f"mqtt connect failed: {exc}") async def disconnect(self): if self._message_task and not self._message_task.done(): @@ -325,69 +289,48 @@ class TalkingQMQTTService: await self._ensure_connected() try: await self._client.publish(topic, json.dumps(payload, ensure_ascii=False), qos=self.qos) - logger.info("", "", f"下发命令 -> {topic}: {payload}") - except Exception as e: - logger.error("", "", f"命令下发失败: {e}") + logger.info("", "", f"mqtt publish topic={topic} payload={payload}") + except Exception as exc: + logger.error("", "", f"mqtt publish failed: {exc}") async def send_gps_query(self, device_id: str) -> str: - topic = f"device/{device_id}/command" - payload = {"msg_id": "001"} - await self._publish(topic, payload) + await self._publish(f"device/{device_id}/command", {"msg_id": "001"}) return "001" async def send_volume_command(self, device_id: str, level: int) -> str: - topic = f"device/{device_id}/command" - payload = { - "msg_id": "002", - "params": {"level": level} - } - await self._publish(topic, payload) + await self._publish( + f"device/{device_id}/command", + {"msg_id": "002", "params": {"level": level}}, + ) return "002" async def send_ota_command(self, device_id: str, url: str, version: str) -> str: - topic = f"device/{device_id}/command" - payload = { - "msg_id": "003", - "params": { - "url": url, - "version": version, - } - } - await self._publish(topic, payload) + await self._publish( + f"device/{device_id}/command", + {"msg_id": "003", "params": {"url": url, "version": version}}, + ) return "003" async def send_nfc_notice(self, device_id: str, url: str) -> str: - topic = f"device/{device_id}/command" - payload = { - "msg_id": "004", - "params": {"url_1": url} - } - await self._publish(topic, payload) + await self._publish( + f"device/{device_id}/command", + {"msg_id": "004", "params": {"url_1": url}}, + ) return "004" - - async def send_bind_nfc_command(self, device_id: str, uuid: str) -> str: - topic = f"device/{device_id}/command" - payload = { - "msg_id": "006" - } - await self._publish(topic, payload) + async def send_bind_nfc_command(self, device_id: str, uuid: Optional[str] = None) -> str: + del uuid + await self._publish(f"device/{device_id}/command", {"msg_id": "006"}) return "006" async def send_open_command(self, device_id: str, open_type: int, is_open: int) -> str: - topic = f"device/{device_id}/command" if open_type not in [0, 1]: raise ValueError("open_type must be 0 or 1") + if open_type == 0: - payload = { - "msg_id": "007", - "type": open_type - } - elif open_type == 1: - payload = { - "msg_id": "007", - "type": open_type, - "status": is_open - } - await self._publish(topic, payload) + payload = {"msg_id": "007", "type": open_type} + else: + payload = {"msg_id": "007", "type": open_type, "status": is_open} + + await self._publish(f"device/{device_id}/command", payload) return "007" diff --git a/talkingq-url/services/card_service.py b/talkingq-url/services/card_service.py index 737f32b..41957cf 100644 --- a/talkingq-url/services/card_service.py +++ b/talkingq-url/services/card_service.py @@ -1,12 +1,14 @@ import asyncio -from typing import Optional, Dict -import time -from sqlalchemy import select, update, insert -from sqlalchemy.ext.asyncio import AsyncSession -from utils.logger import session_logger +from typing import Dict, Optional + from pydantic import BaseModel +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + from database.models import Card as DBCard from services.database_service_base import DatabaseServiceBase +from utils.logger import session_logger + class Card(BaseModel): card_id: Optional[int] = None @@ -17,10 +19,9 @@ class Card(BaseModel): total_swaps: int = 0 created_at: Optional[float] = None updated_at: Optional[float] = None - + @classmethod def from_db_model(cls, db_model: DBCard): - """从数据库模型创建卡片对象""" return cls( card_id=db_model.card_id, card_uuid=db_model.card_uuid, @@ -29,162 +30,205 @@ class Card(BaseModel): status=db_model.status, total_swaps=db_model.total_swaps, created_at=db_model.created_at.timestamp() if db_model.created_at else None, - updated_at=db_model.updated_at.timestamp() if db_model.updated_at else None + updated_at=db_model.updated_at.timestamp() if db_model.updated_at else None, ) + class CardService(DatabaseServiceBase): def __init__(self): super().__init__(service_name="card") self.cards: Dict[str, Card] = {} self.lock = asyncio.Lock() - + + async def _cache_card(self, card: Card) -> None: + async with self.lock: + self.cards[card.card_uuid] = card + async def _load_card_from_db(self, card_uuid: str, async_session: AsyncSession) -> Optional[Card]: - """从数据库加载卡片信息""" try: query = select(DBCard).where(DBCard.card_uuid == card_uuid) result = await async_session.execute(query) db_card = result.scalar_one_or_none() - - if db_card: - card = Card.from_db_model(db_card) - async with self.lock: - self.cards[card_uuid] = card - return card + + if not db_card: + return None + + card = Card.from_db_model(db_card) + await self._cache_card(card) + return card + except Exception as exc: + session_logger.error("card", "service", f"load card failed: {exc}") return None - except Exception as e: - session_logger.error("card", "service", f"从数据库加载卡片失败: {str(e)}") - return None - - async def _save_card_to_db(self, card: Card, async_session: AsyncSession): - """保存卡片信息到数据库""" + + async def _clear_existing_device_card( + self, + device_id: str, + card_uuid: str, + async_session: AsyncSession, + ) -> None: + query = select(DBCard).where( + DBCard.device_id == device_id, + DBCard.card_uuid != card_uuid, + ) + result = await async_session.execute(query) + existing_cards = result.scalars().all() + + for db_card in existing_cards: + db_card.device_id = None + db_card.status = 0 + + async with self.lock: + cached = self.cards.get(db_card.card_uuid) + if cached is not None: + cached.device_id = None + cached.status = 0 + + if existing_cards: + await async_session.flush() + + async def _save_card_to_db(self, card: Card, async_session: AsyncSession, commit: bool = True) -> None: try: query = select(DBCard).where(DBCard.card_uuid == card.card_uuid) result = await async_session.execute(query) existing_card = result.scalar_one_or_none() - + if existing_card: - stmt = update(DBCard).where( - DBCard.card_uuid == card.card_uuid - ).values( - device_id=card.device_id, - card_name=card.card_name, - status=card.status, - total_swaps=card.total_swaps - ) + existing_card.device_id = card.device_id + existing_card.card_name = card.card_name + existing_card.status = card.status + existing_card.total_swaps = card.total_swaps + db_card = existing_card else: - stmt = insert(DBCard).values( + db_card = DBCard( card_uuid=card.card_uuid, device_id=card.device_id, card_name=card.card_name, status=card.status, - total_swaps=card.total_swaps + total_swaps=card.total_swaps, ) - - await async_session.execute(stmt) - await async_session.commit() - - session_logger.info("card", "service", f"卡片已保存到数据库: {card.card_uuid}") - except Exception as e: - await async_session.rollback() - session_logger.error("card", "service", f"保存卡片到数据库失败: {str(e)}") + async_session.add(db_card) + + await async_session.flush() + if commit: + await async_session.commit() + + card.card_id = db_card.card_id + await self._cache_card(card) + session_logger.info("card", "service", f"card saved: {card.card_uuid}") + except Exception as exc: + if commit: + await async_session.rollback() + session_logger.error("card", "service", f"save card failed: {exc}") raise - - async def get_card_by_uuid(self, card_uuid: str, force_refresh: bool = False) -> Optional[Card]: - """根据UUID获取卡片信息""" + + async def get_card_by_uuid( + self, + card_uuid: str, + force_refresh: bool = False, + db_session: Optional[AsyncSession] = None, + ) -> Optional[Card]: await self._init_database() - + card = None - if not force_refresh: async with self.lock: card = self.cards.get(card_uuid) - - if not card: - db_session = await self.db_manager.get_session() - try: - card = await self._load_card_from_db(card_uuid, db_session) - finally: - await db_session.close() - - return card - + + if card: + return card + + if db_session is not None: + return await self._load_card_from_db(card_uuid, db_session) + + session = await self.db_manager.get_session() + try: + return await self._load_card_from_db(card_uuid, session) + finally: + await session.close() + async def get_card_by_device_id(self, device_id: str) -> list[Card]: - """根据设备ID获取卡片列表""" await self._init_database() - + db_session = await self.db_manager.get_session() try: query = select(DBCard).where(DBCard.device_id == device_id) result = await db_session.execute(query) db_cards = result.scalars().all() - + cards = [] for db_card in db_cards: card = Card.from_db_model(db_card) - async with self.lock: - self.cards[card.card_uuid] = card + await self._cache_card(card) cards.append(card) - return cards - except Exception as e: - session_logger.error(device_id, "card", f"根据设备ID获取卡片失败: {str(e)}") + except Exception as exc: + session_logger.error(device_id, "card", f"load device cards failed: {exc}") return [] finally: await db_session.close() - - async def activate_card(self, card_uuid: str, device_id: str, card_name: Optional[str] = None) -> Card: - """激活卡片并绑定到设备""" + + async def activate_card( + self, + card_uuid: str, + device_id: str, + card_name: Optional[str] = None, + db_session: Optional[AsyncSession] = None, + ) -> Card: await self._init_database() - - db_session = await self.db_manager.get_session() + + owns_session = db_session is None + if db_session is None: + db_session = await self.db_manager.get_session() + try: - # 检查卡片是否已存在 - existing_card = await self.get_card_by_uuid(card_uuid) - + existing_card = await self.get_card_by_uuid(card_uuid, db_session=db_session) + await self._clear_existing_device_card(device_id=device_id, card_uuid=card_uuid, async_session=db_session) + if existing_card: - # 更新现有卡片 existing_card.device_id = device_id existing_card.card_name = card_name - existing_card.status = 1 # 激活状态 - await self._save_card_to_db(existing_card, db_session) - session_logger.info(device_id, "card", f"卡片已激活并绑定到设备: {card_uuid}") + existing_card.status = 1 + await self._save_card_to_db(existing_card, db_session, commit=owns_session) + session_logger.info(device_id, "card", f"card activated: {card_uuid}") return existing_card - else: - # 创建新卡片 - new_card = Card( - card_uuid=card_uuid, - device_id=device_id, - card_name=card_name, - status=1, # 激活状态 - total_swaps=0 - ) - await self._save_card_to_db(new_card, db_session) - session_logger.info(device_id, "card", f"新卡片已创建并激活: {card_uuid}") - return new_card + + new_card = Card( + card_uuid=card_uuid, + device_id=device_id, + card_name=card_name, + status=1, + total_swaps=0, + ) + await self._save_card_to_db(new_card, db_session, commit=owns_session) + session_logger.info(device_id, "card", f"new card activated: {card_uuid}") + return new_card + except Exception: + if owns_session: + await db_session.rollback() + raise + finally: + if owns_session: + await db_session.close() + + async def increment_swap_count(self, card_uuid: str) -> Optional[Card]: + await self._init_database() + + card = await self.get_card_by_uuid(card_uuid) + if not card: + return None + + card.total_swaps += 1 + db_session = await self.db_manager.get_session() + try: + await self._save_card_to_db(card, db_session) + session_logger.info("card", "service", f"swap count incremented: {card_uuid}, total={card.total_swaps}") + return card finally: await db_session.close() - - async def increment_swap_count(self, card_uuid: str) -> Optional[Card]: - """增加卡片交换次数""" - await self._init_database() - - card = await self.get_card_by_uuid(card_uuid) - if card: - card.total_swaps += 1 - db_session = await self.db_manager.get_session() - try: - await self._save_card_to_db(card, db_session) - session_logger.info("card", "service", f"卡片交换次数已增加: {card_uuid}, 总次数: {card.total_swaps}") - return card - finally: - await db_session.close() - return None - + async def check_card_ownership(self, card_uuid: str, device_id: str) -> bool: - """检查卡片是否属于指定设备""" card = await self.get_card_by_uuid(card_uuid) - if card and card.device_id == device_id: - return True - return False + return bool(card and card.device_id == device_id) + card_service = CardService()