huangyicao/banban
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(binding): support qr scan and nfc card binding flow

Browse Source
This commit is contained in:
ChengCan
2026-04-30 00:58:56 +08:00
parent 25131adfa2
commit dfb532d883
9 changed files with 984 additions and 489 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
banban-mini/src/pages/bind/index.scss
View File

@@ -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;

299
banban-mini/src/pages/bind/index.tsx
View File

@@ -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<number | null>(null)
const [newChildName, setNewChildName] = useState('')
const [pendingDeviceId, setPendingDeviceId] = useState<string | null>(null)
const [bindToken, setBindToken] = useState('')
const [bindStatus, setBindStatus] = useState<number | null>(null)
const [cardUUID, setCardUUID] = useState('')
const [bindHint, setBindHint] = useState('')
const pollingRef = useRef<ReturnType<typeof setTimeout> | 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 (
<View className='bind-page'>
<View className='bind-header'>
<Text className='title'>{pendingDeviceId ? '补全绑定' : '绑定设备'}</Text>
<Text className='title'>{isPendingBinding ? '补全绑定' : '扫码贴卡绑定设备'}</Text>
<Text className='subtitle'>
{pendingDeviceId ? '设备已绑到当前账号，请补充儿童资料完成配置' : '扫描二维码或手动填写设备号和序列号'}
{isPendingBinding
? '当前设备已存在待补全关系，请先补充儿童资料'
: '先选择儿童，再扫码设备，系统会下发绑卡指令，随后去设备上贴卡确认'}
</Text>
</View>
<View className='scan-area'>
<View className='scan-frame' onClick={handleScanCode}>
<View className='icon-bg orange'>
<Image
className='control-icon-img'
src={require('../../assets/tab-icons/rings.png')}
mode='aspectFit'
/>
<Image className='control-icon-img' src={require('../../assets/tab-icons/rings.png')} mode='aspectFit' />
</View>
<Text className='scan-text'>{isScanning ? '正在扫码...' : '点击扫码'}</Text>
<Text className='scan-hint'>请将二维码放入框内</Text>
<Text className='scan-text'>
{!canEditDeviceFields ? '请先完成儿童资料' : isScanning ? '正在扫码...' : '点击扫码'}
</Text>
<Text className='scan-hint'>
{!canEditDeviceFields ? '先选儿童，再扫设备二维码' : '扫描设备二维码或条码'}
</Text>
</View>
<View className='scan-corners'>
@@ -264,39 +362,7 @@ export default function Bind() {
<View className='bind-form'>
<View className='form-card'>
<Text className='form-title'>设备信息</Text>
<View className='field-item'>
<Text className='field-label'>设备号</Text>
<Input
className='field-input'
value={deviceId}
disabled={!!pendingDeviceId}
placeholder='请输入设备号'
onInput={(event) => setDeviceId(event.detail.value)}
/>
</View>
<View className='field-item'>
<Text className='field-label'>设备序列号</Text>
<Input
className='field-input'
value={serialNumber}
disabled={!!pendingDeviceId}
placeholder='请输入设备序列号'
onInput={(event) => setSerialNumber(event.detail.value)}
/>
</View>
{pendingDeviceId && (
<View className='pending-tip'>
<Text className='pending-tip-text'>当前待完成设备：{pendingDeviceId}</Text>
</View>
)}
</View>
<View className='form-card'>
<Text className='form-title'>儿童信息</Text>
<Text className='form-title'>儿童资料</Text>
{children.length > 0 && (
<View className='child-list'>
@@ -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)
}
}}
/>
</View>
<Text className='field-hint'>优先选择已有儿童，也可以直接输入一个新昵称。</Text>
<Text className='field-hint'>
{children.length > 0
? '可以选已有儿童，也可以在这里新建一个儿童后再绑定设备'
: '当前还没有儿童资料，请先创建一个儿童资料'}
</Text>
</View>
<Button className='submit-btn' loading={submitting} disabled={submitting} onClick={handleSubmit}>
{pendingDeviceId ? '完成儿童关联' : '绑定设备'}
<View className='form-card'>
<Text className='form-title'>设备信息</Text>
<View className='field-item'>
<Text className='field-label'>设备号</Text>
<Input
className='field-input'
value={deviceId}
disabled={!canEditDeviceFields || isPendingBinding}
placeholder={canEditDeviceFields ? '请输入设备号' : '请先选择儿童资料'}
onInput={(event) => setDeviceId(event.detail.value)}
/>
</View>
<View className='field-item'>
<Text className='field-label'>设备序列号</Text>
<Input
className='field-input'
value={serialNumber}
disabled={!canEditDeviceFields || isPendingBinding}
placeholder={canEditDeviceFields ? '请输入设备序列号' : '请先选择儿童资料'}
onInput={(event) => setSerialNumber(event.detail.value)}
/>
</View>
{bindHint ? (
<View className='pending-tip'>
<Text className='pending-tip-text'>{bindHint}</Text>
{cardUUID ? <Text className='pending-tip-subtext'>卡片 UUID：{cardUUID}</Text> : null}
</View>
) : null}
{isPendingBinding && (
<View className='pending-tip'>
<Text className='pending-tip-text'>当前待补全设备：{pendingDeviceId}</Text>
</View>
)}
</View>
<Button
className='submit-btn'
loading={submitting}
disabled={submitting || !hasSelectedOrNewChild || isPollingBind}
onClick={handleSubmit}
>
{isPendingBinding ? '完成儿童关联' : isPollingBind ? '等待贴卡确认' : '发送绑卡指令'}
</Button>
</View>
@@ -342,15 +461,15 @@ export default function Bind() {
<Text className='tips-title'>绑定帮助</Text>
<View className='tip-item'>
<Text className='tip-number'>1</Text>
<Text className='tip-text'>先扫描二维码，自动填入设备号和序列号。</Text>
<Text className='tip-text'>先选择已有儿童，或者在本页新建一个儿童</Text>
</View>
<View className='tip-item'>
<Text className='tip-number'>2</Text>
<Text className='tip-text'>如果二维码里只有设备号，请手动补充序列号。</Text>
<Text className='tip-text'>扫描设备二维码，确认设备号和序列号正确</Text>
</View>
<View className='tip-item'>
<Text className='tip-number'>3</Text>
<Text className='tip-text'>没有儿童资料时，可以直接在本页新建并完成绑定。</Text>
<Text className='tip-text'>点击发送绑卡指令，然后去设备上贴自己的卡完成确认</Text>
</View>
</View>
</View>

114
banban-mini/src/services/binding.ts
View File

@@ -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<T extends { device_id: string }>(items:
export async function getCurrentBinding(): Promise<Binding | null> {
try {
return await request<Binding>('/bindings/current')
return await request<Binding>('/banban/bindings/current')
} catch (error: any) {
if (error?.status === 404) return null
throw error
@@ -75,16 +103,13 @@ export async function getCurrentBinding(): Promise<Binding | null> {
}
export async function getBindings(cursor?: number, limit: number = 20): Promise<BindingListResponse> {
const query = buildQuery({
cursor,
limit,
})
return request<BindingListResponse>(`/bindings?${query}`)
const query = buildQuery({ cursor, limit })
return request<BindingListResponse>(`/banban/bindings?${query}`)
}
export async function getBinding(deviceId: string): Promise<Binding | null> {
try {
return await request<Binding>(`/bindings/${deviceId}`)
return await request<Binding>(`/banban/bindings/${deviceId}`)
} catch (error: any) {
if (error?.status === 404) return null
throw error
@@ -100,7 +125,6 @@ export async function resolveActiveBinding(): Promise<Binding | null> {
setSelectedBindingDeviceId(selectedBinding.device_id)
return selectedBinding
}
clearSelectedBindingDeviceId()
}
@@ -115,24 +139,92 @@ export async function resolveActiveBinding(): Promise<Binding | null> {
}
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<NFCSessionStartResponse> {
return request<NFCSessionStartResponse>('/banban/bindings/start', {
method: 'POST',
data,
})
}
export async function getNFCBindSession(bindToken: string): Promise<NFCSessionStatus> {
return request<NFCSessionStatus>(`/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<void> {
return request(`/bindings/${deviceId}`, {
return request(`/banban/bindings/${deviceId}`, {
method: 'DELETE',
})
}
export function resolveBindingForChild<T extends { child_id: number | null }>(
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,
}
}

43
banban-mini/src/services/child.ts
View File

@@ -1,5 +1,8 @@
import Taro from '@tarojs/taro'
import { request } from './api'
const SELECTED_CHILD_ID_KEY = 'selectedChildId'
function buildQuery(params: Record<string, string | number | undefined | null>): 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<T extends { child_id: number }>(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<Child> {
return request<Child>('/children', {
return request<Child>('/banban/children', {
method: 'POST',
data,
})
@@ -37,18 +72,18 @@ export async function getChildren(cursor?: number, limit: number = 20): Promise<
cursor,
limit,
})
return request<ChildListResponse>(`/children?${query}`)
return request<ChildListResponse>(`/banban/children?${query}`)
}
export async function getChild(childId: number): Promise<Child> {
return request<Child>(`/children/${childId}`)
return request<Child>(`/banban/children/${childId}`)
}
export async function updateChild(
childId: number,
data: { child_name?: string; child_gender?: number; child_birthday?: string }
): Promise<Child> {
return request<Child>(`/children/${childId}`, {
return request<Child>(`/banban/children/${childId}`, {
method: 'PATCH',
data,
})

237
talkingq-url/banban/dao/binding.py
View File

@@ -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()

86
talkingq-url/banban/routers/bindings.py
View File

@@ -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)

98
talkingq-url/banban/service/binding.py
View File

@@ -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:

327
talkingq-url/handlers/mqtt_handler.py
View File

@@ -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"

262
talkingq-url/services/card_service.py
View File

@@ -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()