Compare commits

..

5 Commits

Author SHA1 Message Date
stu2not
eebc5a14d8 限制待收听留言周期提醒来源 2026-05-27 17:14:55 +08:00
stu2not
1b75435a3c 告警记录保存位置快照 2026-05-27 17:06:12 +08:00
stu2not
356e333c81 定位信息增加地址解析 2026-05-27 17:05:24 +08:00
stu2not
fb11e48584 定位刷新超时返回最近一次位置 2026-05-27 16:53:03 +08:00
stu2not
bb56cda2e8 待收听留言入库后即时推送提醒 2026-05-27 16:31:44 +08:00
20 changed files with 796 additions and 50 deletions

View File

@@ -336,6 +336,11 @@
} }
} }
.alarm-row-vertical {
align-items: flex-start;
gap: 20px;
}
.alarm-label { .alarm-label {
font-size: 28px; font-size: 28px;
color: #666666; color: #666666;
@@ -349,6 +354,28 @@
color: #1A1A1A; color: #1A1A1A;
} }
.alarm-location-value {
flex: 1;
min-width: 0;
text-align: right;
}
.alarm-location-main {
display: block;
font-size: 28px;
line-height: 1.5;
word-break: break-all;
color: #1A1A1A;
}
.alarm-location-hint {
display: block;
margin-top: 6px;
font-size: 22px;
line-height: 1.5;
color: #9CA3AF;
}
.alarm-empty { .alarm-empty {
padding: 24px; padding: 24px;
} }
@@ -434,6 +461,22 @@
text-align: right; text-align: right;
} }
.alarm-history-location,
.alarm-history-location-hint {
display: block;
margin-top: 8px;
font-size: 24px;
line-height: 1.5;
color: #4B5563;
word-break: break-all;
}
.alarm-history-location-hint {
margin-top: 4px;
font-size: 22px;
color: #9CA3AF;
}
.alarm-history-empty { .alarm-history-empty {
padding: 8px 0 4px; padding: 8px 0 4px;
} }

View File

@@ -78,6 +78,32 @@ function getAlarmChildLabel(alarm: DeviceAlarmItem | null, fallbackChildName?: s
return '未关联儿童' return '未关联儿童'
} }
function formatAlarmCoordinates(alarm: DeviceAlarmItem | null): string {
if (!alarm || alarm.lat === null || alarm.lat === undefined || alarm.lng === null || alarm.lng === undefined) {
return ''
}
return `${Number(alarm.lat).toFixed(6)}, ${Number(alarm.lng).toFixed(6)}`
}
function getAlarmLocationLabel(alarm: DeviceAlarmItem | null): string {
const address = String(alarm?.address || '').trim()
if (address) return address
const coordinates = formatAlarmCoordinates(alarm)
if (coordinates) return coordinates
return '暂无告警位置'
}
function getAlarmLocationHint(alarm: DeviceAlarmItem | null): string {
if (!alarm?.location_updated_at) return '设备还没有可关联的位置上报'
const prefix = `位置更新于 ${formatAlarmTime(alarm.location_updated_at)}`
if (alarm.location_stale) return `${prefix},可能不是告警发生时的位置`
if (!String(alarm.address || '').trim() && alarm.address_resolve_status === 0) return `${prefix},地址解析中`
if (!String(alarm.address || '').trim() && alarm.address_resolve_status === 2) return `${prefix},地址解析失败`
return prefix
}
export default function Device() { export default function Device() {
const systemBanner = useSystemBanner() const systemBanner = useSystemBanner()
const [binding, setBinding] = useState<Binding | null>(null) const [binding, setBinding] = useState<Binding | null>(null)
@@ -425,6 +451,13 @@ export default function Device() {
<Text className='alarm-label'></Text> <Text className='alarm-label'></Text>
<Text className='alarm-value'>{getAlarmTypeLabel(latestAlarm.source_msg_id)}</Text> <Text className='alarm-value'>{getAlarmTypeLabel(latestAlarm.source_msg_id)}</Text>
</View> </View>
<View className='alarm-row alarm-row-vertical'>
<Text className='alarm-label'></Text>
<View className='alarm-location-value'>
<Text className='alarm-location-main'>{getAlarmLocationLabel(latestAlarm)}</Text>
<Text className='alarm-location-hint'>{getAlarmLocationHint(latestAlarm)}</Text>
</View>
</View>
</View> </View>
<View className='alarm-history'> <View className='alarm-history'>
@@ -446,6 +479,10 @@ export default function Device() {
</Text> </Text>
<Text className='alarm-history-device'>{alarm.device_id}</Text> <Text className='alarm-history-device'>{alarm.device_id}</Text>
</View> </View>
<Text className='alarm-history-location'>{getAlarmLocationLabel(alarm)}</Text>
{alarm.location_stale ? (
<Text className='alarm-history-location-hint'></Text>
) : null}
</View> </View>
)) ))
) : ( ) : (

View File

@@ -2,9 +2,7 @@ import { View, Text, Map } from '@tarojs/components'
import { useState } from 'react' import { useState } from 'react'
import Taro, { useDidShow } from '@tarojs/taro' import Taro, { useDidShow } from '@tarojs/taro'
import { getToken } from '@/services/auth' import { getToken } from '@/services/auth'
import { DEVICE_UNAVAILABLE_MESSAGE } from '@/services/api'
import { loadCurrentChildBindingContext } from '@/services/binding' import { loadCurrentChildBindingContext } from '@/services/binding'
import { getDeviceOnlineStatus } from '@/services/device'
import { import {
getCurrentDeviceLocation, getCurrentDeviceLocation,
getDeviceTrajectory, getDeviceTrajectory,
@@ -55,6 +53,14 @@ function formatOptionalNumber(value?: number | null, digits = 1): string | null
return Number(value).toFixed(digits) return Number(value).toFixed(digits)
} }
function getLocationAddress(point: DeviceLocation | DeviceTrajectoryPoint): string {
const address = String(point.address || '').trim()
if (address) return address
if (point.address_resolve_status === 0) return '地点解析中'
if (point.address_resolve_status === 2) return '地点暂时无法解析'
return '当前地点未知'
}
function buildLocationDetailLines( function buildLocationDetailLines(
point: DeviceLocation | DeviceTrajectoryPoint, point: DeviceLocation | DeviceTrajectoryPoint,
options?: { includeIdentity?: boolean } options?: { includeIdentity?: boolean }
@@ -187,18 +193,11 @@ export default function Location() {
return return
} }
const onlineStatus = await getDeviceOnlineStatus(resolvedDeviceId)
if (onlineStatus && !onlineStatus.online) {
setDeviceLocation(null)
setTrajectory([])
setSelectedPoint(null)
setCoordinates(DEFAULT_COORDINATES)
setLocationNotice(onlineStatus.message || '设备不在线或暂时无法上报位置')
return
}
const currentLocation = await getCurrentDeviceLocation(resolvedDeviceId) const currentLocation = await getCurrentDeviceLocation(resolvedDeviceId)
setDeviceLocation(currentLocation) setDeviceLocation(currentLocation)
if (currentLocation?.stale || currentLocation?.realtime === false) {
setLocationNotice('已显示最近一次定位')
}
const nextCoordinates = currentLocation const nextCoordinates = currentLocation
? { ? {
@@ -249,9 +248,7 @@ export default function Location() {
} }
setLoading(false) setLoading(false)
console.error('[location] load failed:', error) console.error('[location] load failed:', error)
const message = error?.message === DEVICE_UNAVAILABLE_MESSAGE const message = error?.status === 408 ? '暂时无法刷新定位' : error?.message || '位置更新失败'
? '设备不在线或处于休眠中,暂时无法获取位置'
: error?.message || '位置更新失败'
Taro.showToast({ Taro.showToast({
title: message, title: message,
icon: 'none', icon: 'none',
@@ -455,6 +452,7 @@ export default function Location() {
<Text>📍</Text> <Text>📍</Text>
</View> </View>
<View className='location-info'> <View className='location-info'>
<Text className='location-address'>{getLocationAddress(deviceLocation)}</Text>
<Text className='location-coordinate'> <Text className='location-coordinate'>
{deviceLocation.lat.toFixed(6)}, {deviceLocation.lng.toFixed(6)} {deviceLocation.lat.toFixed(6)}, {deviceLocation.lng.toFixed(6)}
</Text> </Text>

View File

@@ -45,6 +45,14 @@ export interface DeviceAlarmItem {
child_id?: number | null child_id?: number | null
child_name?: string | null child_name?: string | null
source_msg_id: string source_msg_id: string
coord_type?: string | null
lat?: number | null
lng?: number | null
location_updated_at?: string | null
location_stale?: boolean
address?: string | null
address_resolved_at?: string | null
address_resolve_status?: number | null
created_at: string created_at: string
} }

View File

@@ -16,7 +16,12 @@ export interface DeviceLocation {
battery_pct?: number | null battery_pct?: number | null
device_time: string device_time: string
server_time: string server_time: string
address?: string | null
address_resolved_at?: string | null
address_resolve_status?: number | null
updated_at: string updated_at: string
stale?: boolean
realtime?: boolean
} }
export interface DeviceTrajectoryPoint extends DeviceLocation { export interface DeviceTrajectoryPoint extends DeviceLocation {

View File

@@ -448,6 +448,13 @@ CREATE TABLE IF NOT EXISTS `device_alarm_events` (
`owner_user_id` BIGINT NULL, `owner_user_id` BIGINT NULL,
`child_id` BIGINT NULL, `child_id` BIGINT NULL,
`source_msg_id` VARCHAR(8) NOT NULL DEFAULT '010', `source_msg_id` VARCHAR(8) NOT NULL DEFAULT '010',
`coord_type` VARCHAR(16) NULL,
`lat` DECIMAL(10, 7) NULL,
`lng` DECIMAL(10, 7) NULL,
`location_updated_at` DATETIME NULL,
`address` VARCHAR(255) NULL,
`address_resolved_at` DATETIME NULL,
`address_resolve_status` TINYINT NULL COMMENT '0=pending,1=success,2=failed',
`created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`alarm_id`), PRIMARY KEY (`alarm_id`),
KEY `idx_device_alarm_device_created` (`device_id`, `created_at`), KEY `idx_device_alarm_device_created` (`device_id`, `created_at`),
@@ -552,6 +559,9 @@ CREATE TABLE IF NOT EXISTS `child_location_current` (
`device_time` DATETIME NOT NULL, `device_time` DATETIME NOT NULL,
`server_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, `server_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
`updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
`address` VARCHAR(255) NULL,
`address_resolved_at` DATETIME NULL,
`address_resolve_status` TINYINT NULL COMMENT '0=pending,1=success,2=failed',
PRIMARY KEY (`child_id`), PRIMARY KEY (`child_id`),
KEY `idx_child_loc_current_device` (`device_id`), KEY `idx_child_loc_current_device` (`device_id`),
KEY `idx_child_loc_current_server_time` (`server_time`), KEY `idx_child_loc_current_server_time` (`server_time`),

View File

@@ -14,14 +14,34 @@ class DeviceAlarmDAO(BaseDAO):
owner_user_id: int | None, owner_user_id: int | None,
child_id: int | None, child_id: int | None,
source_msg_id: str = "010", source_msg_id: str = "010",
location: Mapping[str, Any] | None = None,
) -> int: ) -> int:
location = location or {}
result = await self.execute( result = await self.execute(
""" """
INSERT INTO device_alarm_events ( INSERT INTO device_alarm_events (
device_id, owner_user_id, child_id, source_msg_id, created_at device_id,
owner_user_id,
child_id,
source_msg_id,
coord_type,
lat,
lng,
location_updated_at,
address_resolve_status,
created_at
) )
VALUES ( VALUES (
:device_id, :owner_user_id, :child_id, :source_msg_id, CURRENT_TIMESTAMP :device_id,
:owner_user_id,
:child_id,
:source_msg_id,
:coord_type,
:lat,
:lng,
:location_updated_at,
:address_resolve_status,
CURRENT_TIMESTAMP
) )
""", """,
{ {
@@ -29,11 +49,60 @@ class DeviceAlarmDAO(BaseDAO):
"owner_user_id": owner_user_id, "owner_user_id": owner_user_id,
"child_id": child_id, "child_id": child_id,
"source_msg_id": source_msg_id, "source_msg_id": source_msg_id,
"coord_type": location.get("coord_type"),
"lat": location.get("lat"),
"lng": location.get("lng"),
"location_updated_at": location.get("updated_at"),
"address_resolve_status": 0 if location.get("lat") is not None and location.get("lng") is not None else None,
}, },
) )
await self.commit() await self.commit()
return int(result.lastrowid) return int(result.lastrowid)
async def get_current_location_snapshot(self, *, child_id: int) -> Mapping[str, Any] | None:
result = await self.execute(
text(
"""
SELECT
coord_type,
lat,
lng,
updated_at
FROM child_location_current
WHERE child_id = :child_id
LIMIT 1
"""
),
{"child_id": child_id},
)
return result.mappings().first()
async def update_resolved_address(
self,
*,
alarm_id: int,
address: str | None,
status: int,
) -> bool:
result = await self.execute(
text(
"""
UPDATE device_alarm_events
SET address = :address,
address_resolve_status = :status,
address_resolved_at = CURRENT_TIMESTAMP
WHERE alarm_id = :alarm_id
"""
),
{
"alarm_id": alarm_id,
"address": address,
"status": status,
},
)
await self.commit()
return bool(result.rowcount)
async def get_active_binding_context(self, *, device_id: str) -> Mapping[str, Any] | None: async def get_active_binding_context(self, *, device_id: str) -> Mapping[str, Any] | None:
result = await self.execute( result = await self.execute(
text( text(
@@ -64,6 +133,13 @@ class DeviceAlarmDAO(BaseDAO):
dae.owner_user_id, dae.owner_user_id,
dae.child_id, dae.child_id,
dae.source_msg_id, dae.source_msg_id,
dae.coord_type,
dae.lat,
dae.lng,
dae.location_updated_at,
dae.address,
dae.address_resolved_at,
dae.address_resolve_status,
dae.created_at, dae.created_at,
c.child_name c.child_name
FROM device_alarm_events AS dae FROM device_alarm_events AS dae

View File

@@ -81,9 +81,16 @@ class LocationDAO(BaseDAO):
payload: Any, payload: Any,
) -> Mapping[str, Any]: ) -> Mapping[str, Any]:
now_sql = "CURRENT_TIMESTAMP(3)" now_sql = "CURRENT_TIMESTAMP(3)"
should_resolve_address = True
try: try:
current_row = await self._get_current_location_row(child_id=child_id, lock=True) current_row = await self._get_current_location_row(child_id=child_id, lock=True)
if current_row and current_row.get("address"):
should_resolve_address = self._is_location_address_stale(
current_row,
lat=float(payload.lat),
lng=float(payload.lng),
)
params = { params = {
"child_id": child_id, "child_id": child_id,
"device_id": device_id, "device_id": device_id,
@@ -115,11 +122,14 @@ class LocationDAO(BaseDAO):
battery_pct = :battery_pct, battery_pct = :battery_pct,
device_time = :device_time, device_time = :device_time,
server_time = {now_sql}, server_time = {now_sql},
updated_at = {now_sql} updated_at = {now_sql},
address = CASE WHEN :should_resolve_address THEN NULL ELSE address END,
address_resolved_at = CASE WHEN :should_resolve_address THEN NULL ELSE address_resolved_at END,
address_resolve_status = CASE WHEN :should_resolve_address THEN 0 ELSE address_resolve_status END
WHERE child_id = :child_id WHERE child_id = :child_id
""" """
), ),
params, {**params, "should_resolve_address": should_resolve_address},
) )
else: else:
await self.execute( await self.execute(
@@ -139,7 +149,8 @@ class LocationDAO(BaseDAO):
battery_pct, battery_pct,
device_time, device_time,
server_time, server_time,
updated_at updated_at,
address_resolve_status
) VALUES ( ) VALUES (
:child_id, :child_id,
:device_id, :device_id,
@@ -154,7 +165,8 @@ class LocationDAO(BaseDAO):
:battery_pct, :battery_pct,
:device_time, :device_time,
{now_sql}, {now_sql},
{now_sql} {now_sql},
0
) )
""" """
), ),
@@ -209,7 +221,7 @@ class LocationDAO(BaseDAO):
current_row = await self._get_current_location_row(child_id=child_id, lock=False) current_row = await self._get_current_location_row(child_id=child_id, lock=False)
if not current_row: if not current_row:
raise RuntimeError("current location not found after report") raise RuntimeError("current location not found after report")
return current_row return {**current_row, "should_resolve_address": should_resolve_address}
async def get_device_current_location(self, *, device_id: str, user_id: int) -> Mapping[str, Any]: async def get_device_current_location(self, *, device_id: str, user_id: int) -> Mapping[str, Any]:
access = await self.assert_parent_device_access(device_id=device_id, user_id=user_id) access = await self.assert_parent_device_access(device_id=device_id, user_id=user_id)
@@ -300,7 +312,10 @@ class LocationDAO(BaseDAO):
battery_pct, battery_pct,
device_time, device_time,
server_time, server_time,
updated_at updated_at,
address,
address_resolved_at,
address_resolve_status
FROM child_location_current FROM child_location_current
WHERE child_id = :child_id WHERE child_id = :child_id
LIMIT 1{lock_clause} LIMIT 1{lock_clause}
@@ -310,6 +325,55 @@ class LocationDAO(BaseDAO):
) )
return result.mappings().first() return result.mappings().first()
@staticmethod
def _is_location_address_stale(
row: Mapping[str, Any],
*,
lat: float,
lng: float,
) -> bool:
try:
previous_lat = float(row["lat"])
previous_lng = float(row["lng"])
except (TypeError, ValueError):
return True
return abs(previous_lat - lat) >= 0.0001 or abs(previous_lng - lng) >= 0.0001
async def update_current_location_address(
self,
*,
child_id: int,
device_id: str,
lat: float,
lng: float,
address: str | None,
status: int,
) -> bool:
result = await self.execute(
text(
"""
UPDATE child_location_current
SET address = :address,
address_resolve_status = :status,
address_resolved_at = CURRENT_TIMESTAMP
WHERE child_id = :child_id
AND device_id = :device_id
AND ABS(lat - :lat) < 0.0000001
AND ABS(lng - :lng) < 0.0000001
"""
),
{
"child_id": child_id,
"device_id": device_id,
"lat": lat,
"lng": lng,
"address": address,
"status": status,
},
)
await self.commit()
return bool(result.rowcount)
async def _next_primary_key(self, table_name: str) -> int | None: async def _next_primary_key(self, table_name: str) -> int | None:
return None return None
@@ -361,7 +425,10 @@ class LocationDAO(BaseDAO):
battery_pct, battery_pct,
device_time, device_time,
server_time, server_time,
updated_at updated_at,
address,
address_resolved_at,
address_resolve_status
FROM child_location_current FROM child_location_current
WHERE device_id = :device_id WHERE device_id = :device_id
LIMIT 1 LIMIT 1

View File

@@ -7,6 +7,7 @@ from sqlalchemy import text
import asyncio import asyncio
from datetime import timedelta from datetime import timedelta
from handlers.mqtt_handler import TalkingQMQTTService from handlers.mqtt_handler import TalkingQMQTTService
from config import settings
from services.connection_manager import connection_manager from services.connection_manager import connection_manager
try: try:
from banban.security import get_current_user_id from banban.security import get_current_user_id
@@ -113,6 +114,14 @@ class DeviceAlarmItem(BaseModel):
child_id: int | None = None child_id: int | None = None
child_name: str | None = None child_name: str | None = None
source_msg_id: str source_msg_id: str
coord_type: str | None = None
lat: float | None = None
lng: float | None = None
location_updated_at: datetime | None = None
location_stale: bool = False
address: str | None = None
address_resolved_at: datetime | None = None
address_resolve_status: int | None = None
created_at: datetime created_at: datetime
@@ -202,7 +211,12 @@ def _row_to_ai_conversation_item(row: Mapping) -> DeviceAiConversationItem:
) )
def _row_to_current_location_response(row: Mapping) -> DeviceLocationCurrentResponse: def _row_to_current_location_response(
row: Mapping,
*,
stale: bool = False,
realtime: bool = True,
) -> DeviceLocationCurrentResponse:
return DeviceLocationCurrentResponse( return DeviceLocationCurrentResponse(
child_id=int(row["child_id"]), child_id=int(row["child_id"]),
child_name=row.get("child_name"), child_name=row.get("child_name"),
@@ -219,6 +233,11 @@ def _row_to_current_location_response(row: Mapping) -> DeviceLocationCurrentResp
device_time=row["device_time"], device_time=row["device_time"],
server_time=row["server_time"], server_time=row["server_time"],
updated_at=row["updated_at"], updated_at=row["updated_at"],
address=row.get("address"),
address_resolved_at=row.get("address_resolved_at"),
address_resolve_status=int(row["address_resolve_status"]) if row.get("address_resolve_status") is not None else None,
stale=stale,
realtime=realtime,
) )
@@ -303,16 +322,47 @@ def _seconds_since(value: datetime | None, now: datetime) -> float | None:
def _row_to_alarm_item(row: Mapping) -> DeviceAlarmItem: def _row_to_alarm_item(row: Mapping) -> DeviceAlarmItem:
location_updated_at = row.get("location_updated_at")
created_at = row["created_at"]
location_stale = False
if location_updated_at is not None:
location_age_seconds = _seconds_between(created_at, location_updated_at)
location_stale = (
location_age_seconds is None
or location_age_seconds < 0
or location_age_seconds > settings.alarm_location_stale_seconds
)
return DeviceAlarmItem( return DeviceAlarmItem(
alarm_id=int(row["alarm_id"]), alarm_id=int(row["alarm_id"]),
device_id=str(row["device_id"]), device_id=str(row["device_id"]),
child_id=int(row["child_id"]) if row["child_id"] is not None else None, child_id=int(row["child_id"]) if row["child_id"] is not None else None,
child_name=row.get("child_name"), child_name=row.get("child_name"),
source_msg_id=str(row["source_msg_id"]), source_msg_id=str(row["source_msg_id"]),
created_at=row["created_at"], coord_type=row.get("coord_type"),
lat=float(row["lat"]) if row.get("lat") is not None else None,
lng=float(row["lng"]) if row.get("lng") is not None else None,
location_updated_at=location_updated_at,
location_stale=location_stale,
address=row.get("address"),
address_resolved_at=row.get("address_resolved_at"),
address_resolve_status=int(row["address_resolve_status"]) if row.get("address_resolve_status") is not None else None,
created_at=created_at,
) )
def _seconds_between(later: datetime | None, earlier: datetime | None) -> float | None:
if later is None or earlier is None:
return None
normalized_later = later
normalized_earlier = earlier
if normalized_later.tzinfo is not None and normalized_earlier.tzinfo is None:
normalized_earlier = normalized_earlier.replace(tzinfo=normalized_later.tzinfo)
elif normalized_later.tzinfo is None and normalized_earlier.tzinfo is not None:
normalized_later = normalized_later.replace(tzinfo=normalized_earlier.tzinfo)
return (normalized_later - normalized_earlier).total_seconds()
@router.get("/{device_id}/messages", response_model=DeviceMessageListResponse) @router.get("/{device_id}/messages", response_model=DeviceMessageListResponse)
async def list_device_messages( async def list_device_messages(
device_id: str, device_id: str,
@@ -681,10 +731,21 @@ async def get_current_device_location(
await asyncio.sleep(1) await asyncio.sleep(1)
if row is None: if row is not None:
raise HTTPException(status_code=408, detail="GPS数据上报超时") logger.info(
"device stale location returned",
extra={
"event": "device_stale_location",
"request_id": getattr(request.state, "request_id", None),
"user_id": current_user_id,
"device_id": device_id,
"child_id": int(row["child_id"]),
"location_updated_at": row.get("updated_at"),
},
)
return _row_to_current_location_response(row, stale=True, realtime=False)
raise HTTPException(status_code=408, detail="GPS数据未刷新") raise HTTPException(status_code=408, detail="GPS数据上报超时")
except Exception as e: except Exception as e:
raise HTTPException(status_code=408, detail=f"GPS数据上报失败: {e}") raise HTTPException(status_code=408, detail=f"GPS数据上报失败: {e}")

View File

@@ -31,10 +31,15 @@ class DeviceLocationPoint(BaseModel):
battery_pct: int | None = None battery_pct: int | None = None
device_time: datetime device_time: datetime
server_time: datetime server_time: datetime
address: str | None = None
address_resolved_at: datetime | None = None
address_resolve_status: int | None = None
class DeviceLocationCurrentResponse(DeviceLocationPoint): class DeviceLocationCurrentResponse(DeviceLocationPoint):
updated_at: datetime updated_at: datetime
stale: bool = False
realtime: bool = True
class DeviceLocationReportResponse(DeviceLocationCurrentResponse): class DeviceLocationReportResponse(DeviceLocationCurrentResponse):

View File

@@ -0,0 +1,53 @@
from __future__ import annotations
from dataclasses import dataclass
import httpx
from config import settings
class AmapGeocodeError(RuntimeError):
pass
@dataclass(frozen=True)
class ReverseGeocodeResult:
address: str
class AmapGeocodeService:
async def reverse_geocode(self, *, lng: float, lat: float) -> ReverseGeocodeResult:
api_key = settings.amap_key.strip()
if not api_key:
raise AmapGeocodeError("Amap_key is not configured")
params = {
"key": api_key,
"location": f"{lng:.7f},{lat:.7f}",
"extensions": "base",
"output": "JSON",
}
try:
async with httpx.AsyncClient(timeout=settings.amap_http_timeout_seconds) as client:
response = await client.get(settings.amap_reverse_geocode_url, params=params)
response.raise_for_status()
payload = response.json()
except httpx.HTTPError as exc:
raise AmapGeocodeError(f"amap request failed: {exc}") from exc
except ValueError as exc:
raise AmapGeocodeError("amap response is not valid json") from exc
if str(payload.get("status")) != "1":
raise AmapGeocodeError(
f"amap reverse geocode failed: infocode={payload.get('infocode')} info={payload.get('info')}"
)
address = str((payload.get("regeocode") or {}).get("formatted_address") or "").strip()
if not address:
raise AmapGeocodeError("amap reverse geocode returned empty address")
return ReverseGeocodeResult(address=address)
amap_geocode_service = AmapGeocodeService()

View File

@@ -2,7 +2,15 @@ from collections.abc import Mapping
from typing import Any from typing import Any
from banban.dao.device_alarm import DeviceAlarmDAO from banban.dao.device_alarm import DeviceAlarmDAO
from banban.service.amap_geocode import amap_geocode_service
from services.task_manager import task_manager
from services.database_service_base import DatabaseServiceBase from services.database_service_base import DatabaseServiceBase
from utils.logger import session_logger
ADDRESS_RESOLVE_PENDING = 0
ADDRESS_RESOLVE_SUCCESS = 1
ADDRESS_RESOLVE_FAILED = 2
class DeviceAlarmService(DatabaseServiceBase): class DeviceAlarmService(DatabaseServiceBase):
@@ -16,11 +24,76 @@ class DeviceAlarmService(DatabaseServiceBase):
binding = await dao.get_active_binding_context(device_id=device_id) binding = await dao.get_active_binding_context(device_id=device_id)
if not binding: if not binding:
return None return None
return await dao.create( child_id = int(binding["child_id"]) if binding.get("child_id") is not None else None
location = None
if child_id is not None:
location = await dao.get_current_location_snapshot(child_id=child_id)
alarm_id = await dao.create(
device_id=device_id, device_id=device_id,
owner_user_id=int(binding["owner_user_id"]) if binding.get("owner_user_id") is not None else None, owner_user_id=int(binding["owner_user_id"]) if binding.get("owner_user_id") is not None else None,
child_id=int(binding["child_id"]) if binding.get("child_id") is not None else None, child_id=child_id,
source_msg_id=source_msg_id, source_msg_id=source_msg_id,
location=location,
)
if alarm_id and location and location.get("lat") is not None and location.get("lng") is not None:
await task_manager.create_task(
self.resolve_alarm_address(
alarm_id=alarm_id,
device_id=device_id,
lat=float(location["lat"]),
lng=float(location["lng"]),
),
device_id=device_id,
task_type="alarm_address",
)
return alarm_id
finally:
await db_session.close()
async def resolve_alarm_address(
self,
*,
alarm_id: int,
device_id: str,
lat: float,
lng: float,
) -> None:
try:
result = await amap_geocode_service.reverse_geocode(lng=lng, lat=lat)
except Exception as exc:
session_logger.warning(
device_id,
"alarm_address",
f"告警地址解析失败: alarm_id={alarm_id}, error={exc}",
)
await self.update_alarm_address(
alarm_id=alarm_id,
address=None,
status=ADDRESS_RESOLVE_FAILED,
)
return
await self.update_alarm_address(
alarm_id=alarm_id,
address=result.address,
status=ADDRESS_RESOLVE_SUCCESS,
)
async def update_alarm_address(
self,
*,
alarm_id: int,
address: str | None,
status: int,
) -> bool:
db_session = await self.get_session()
try:
dao = DeviceAlarmDAO(db_session)
return await dao.update_resolved_address(
alarm_id=alarm_id,
address=address,
status=status,
) )
finally: finally:
await db_session.close() await db_session.close()

View File

@@ -5,16 +5,25 @@ from typing import Any
from services.database_service_base import DatabaseServiceBase from services.database_service_base import DatabaseServiceBase
from database.models import ChildLocationCurrent from database.models import ChildLocationCurrent
from services.task_manager import task_manager
from utils.logger import session_logger
try: try:
from banban.dao.location import LocationDAO, ParentDeviceAccess from banban.dao.location import LocationDAO, ParentDeviceAccess
from banban.schemas.location import DeviceLocationReportRequest from banban.schemas.location import DeviceLocationReportRequest
from banban.service.amap_geocode import amap_geocode_service
from banban.service.im import im_service from banban.service.im import im_service
except ModuleNotFoundError: except ModuleNotFoundError:
from banban.dao.location import LocationDAO, ParentDeviceAccess from banban.dao.location import LocationDAO, ParentDeviceAccess
from banban.schemas.location import DeviceLocationReportRequest from banban.schemas.location import DeviceLocationReportRequest
from banban.service.amap_geocode import amap_geocode_service
from banban.service.im import im_service from banban.service.im import im_service
ADDRESS_RESOLVE_PENDING = 0
ADDRESS_RESOLVE_SUCCESS = 1
ADDRESS_RESOLVE_FAILED = 2
class LocationService(DatabaseServiceBase): class LocationService(DatabaseServiceBase):
def __init__(self): def __init__(self):
super().__init__(service_name="location_service") super().__init__(service_name="location_service")
@@ -46,6 +55,7 @@ class LocationService(DatabaseServiceBase):
child_id=device_identity.child_id, child_id=device_identity.child_id,
payload=payload, payload=payload,
) )
await self._queue_location_address_resolution(current_row)
return device_identity, current_row return device_identity, current_row
finally: finally:
await db_session.close() await db_session.close()
@@ -150,11 +160,88 @@ class LocationService(DatabaseServiceBase):
battery_pct=battery_pct, battery_pct=battery_pct,
device_time=device_time or datetime.now(), device_time=device_time or datetime.now(),
) )
return await dao.report_device_location( current_row = await dao.report_device_location(
device_id=device_id, device_id=device_id,
child_id=int(binding_row["child_id"]), child_id=int(binding_row["child_id"]),
payload=payload, payload=payload,
) )
await self._queue_location_address_resolution(current_row)
return current_row
finally:
await db_session.close()
async def _queue_location_address_resolution(self, row: Mapping[str, Any] | None) -> None:
if not row or row.get("lat") is None or row.get("lng") is None:
return
if not row.get("should_resolve_address"):
return
await task_manager.create_task(
self.resolve_current_location_address(
child_id=int(row["child_id"]),
device_id=str(row["device_id"]),
lat=float(row["lat"]),
lng=float(row["lng"]),
),
device_id=str(row["device_id"]),
task_type="location_address",
)
async def resolve_current_location_address(
self,
*,
child_id: int,
device_id: str,
lat: float,
lng: float,
) -> None:
try:
result = await amap_geocode_service.reverse_geocode(lng=lng, lat=lat)
except Exception as exc:
session_logger.warning(
device_id,
"location_address",
f"当前位置地址解析失败: child_id={child_id}, error={exc}",
)
await self.update_current_location_address(
child_id=child_id,
device_id=device_id,
lat=lat,
lng=lng,
address=None,
status=ADDRESS_RESOLVE_FAILED,
)
return
await self.update_current_location_address(
child_id=child_id,
device_id=device_id,
lat=lat,
lng=lng,
address=result.address,
status=ADDRESS_RESOLVE_SUCCESS,
)
async def update_current_location_address(
self,
*,
child_id: int,
device_id: str,
lat: float,
lng: float,
address: str | None,
status: int,
) -> bool:
db_session = await self.get_session()
try:
dao = LocationDAO(db_session)
return await dao.update_current_location_address(
child_id=child_id,
device_id=device_id,
lat=lat,
lng=lng,
address=address,
status=status,
)
finally: finally:
await db_session.close() await db_session.close()

View File

@@ -3,6 +3,7 @@ from dataclasses import dataclass
from banban.dao.im import ImDAO from banban.dao.im import ImDAO
from banban.dao.pending_voice_message import PendingVoiceMessageDAO from banban.dao.pending_voice_message import PendingVoiceMessageDAO
from banban.service.device_audio_cache import device_audio_cache_service from banban.service.device_audio_cache import device_audio_cache_service
from config import settings
from services.database_service_base import DatabaseServiceBase from services.database_service_base import DatabaseServiceBase
from utils.logger import session_logger from utils.logger import session_logger
@@ -28,6 +29,43 @@ class PendingVoiceMessageService(DatabaseServiceBase):
def __init__(self) -> None: def __init__(self) -> None:
super().__init__(service_name="pending_voice_message") super().__init__(service_name="pending_voice_message")
def get_pending_notice_audio_url(self) -> str:
return f"http://{settings.server_host}:{settings.server_port}/assets/audio/new_message.mp3"
async def notify_device_pending_message(self, *, target_device_id: str) -> bool:
normalized_target_device_id = (target_device_id or "").strip()
if not normalized_target_device_id:
return False
try:
from handlers.mqtt_handler import TalkingQMQTTService
service = await TalkingQMQTTService.get_instance()
if service is None:
session_logger.warning(
normalized_target_device_id,
"pending_voice",
"MQTT服务未初始化跳过待收听留言提醒",
)
return False
audio_url = self.get_pending_notice_audio_url()
await service.send_nfc_notice(normalized_target_device_id, audio_url)
session_logger.info(
normalized_target_device_id,
"pending_voice",
f"待收听留言提醒已发送: audio_url={audio_url}",
)
return True
except Exception as exc:
session_logger.error(
normalized_target_device_id,
"pending_voice",
f"待收听留言提醒发送失败: {exc}",
exc_info=True,
)
return False
async def add_pending_message( async def add_pending_message(
self, self,
*, *,
@@ -69,6 +107,10 @@ class PendingVoiceMessageService(DatabaseServiceBase):
finally: finally:
await db_session.close() await db_session.close()
await self.notify_device_pending_message(
target_device_id=normalized_target_device_id,
)
async def get_playback_items( async def get_playback_items(
self, self,
*, *,

View File

@@ -41,6 +41,14 @@ class Settings(BaseSettings):
validation_alias="MINIMAX_BASE_URL", validation_alias="MINIMAX_BASE_URL",
) )
amap_key: str = Field(default="", validation_alias="Amap_key")
amap_reverse_geocode_url: str = Field(
default="https://restapi.amap.com/v3/geocode/regeo",
validation_alias="AMAP_REVERSE_GEOCODE_URL",
)
amap_http_timeout_seconds: float = Field(default=5.0, validation_alias="AMAP_HTTP_TIMEOUT_SECONDS")
alarm_location_stale_seconds: int = Field(default=600, validation_alias="ALARM_LOCATION_STALE_SECONDS")
aliyun_api_key: str = Field(default="", validation_alias="ALIYUN_API_KEY") aliyun_api_key: str = Field(default="", validation_alias="ALIYUN_API_KEY")
aliyun_vocabulary_id: str = Field(default="", validation_alias="ALIYUN_VOCABULARY_ID") aliyun_vocabulary_id: str = Field(default="", validation_alias="ALIYUN_VOCABULARY_ID")

View File

@@ -59,6 +59,53 @@ async def _ensure_schedule_suppressed_until_column(conn) -> None:
) )
async def _ensure_columns(conn, *, table_name: str, columns: list[tuple[str, str]]) -> None:
for column_name, definition in columns:
result = await conn.execute(
text(
"""
SELECT COUNT(*)
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = DATABASE()
AND TABLE_NAME = :table_name
AND COLUMN_NAME = :column_name
"""
),
{"table_name": table_name, "column_name": column_name},
)
if int(result.scalar() or 0) > 0:
continue
await conn.execute(text(f"ALTER TABLE {table_name} ADD COLUMN {definition}"))
async def _ensure_device_alarm_location_columns(conn) -> None:
await _ensure_columns(
conn,
table_name="device_alarm_events",
columns=[
("coord_type", "coord_type VARCHAR(16) NULL AFTER source_msg_id"),
("lat", "lat DECIMAL(10, 7) NULL AFTER coord_type"),
("lng", "lng DECIMAL(10, 7) NULL AFTER lat"),
("location_updated_at", "location_updated_at DATETIME NULL AFTER lng"),
("address", "address VARCHAR(255) NULL AFTER location_updated_at"),
("address_resolved_at", "address_resolved_at DATETIME NULL AFTER address"),
("address_resolve_status", "address_resolve_status TINYINT NULL AFTER address_resolved_at"),
],
)
async def _ensure_child_location_address_columns(conn) -> None:
await _ensure_columns(
conn,
table_name="child_location_current",
columns=[
("address", "address VARCHAR(255) NULL AFTER updated_at"),
("address_resolved_at", "address_resolved_at DATETIME NULL AFTER address"),
("address_resolve_status", "address_resolve_status TINYINT NULL AFTER address_resolved_at"),
],
)
async def _ensure_device_family_tables(conn) -> None: async def _ensure_device_family_tables(conn) -> None:
await conn.execute( await conn.execute(
text( text(
@@ -173,6 +220,8 @@ async def init_db():
await conn.run_sync(Base.metadata.create_all) await conn.run_sync(Base.metadata.create_all)
await _ensure_manual_sleep_mode_column(conn) await _ensure_manual_sleep_mode_column(conn)
await _ensure_schedule_suppressed_until_column(conn) await _ensure_schedule_suppressed_until_column(conn)
await _ensure_device_alarm_location_columns(conn)
await _ensure_child_location_address_columns(conn)
await _ensure_device_family_tables(conn) await _ensure_device_family_tables(conn)
await engine.dispose() await engine.dispose()
session_logger.info("system", "database", "数据库表已成功创建") session_logger.info("system", "database", "数据库表已成功创建")

View File

@@ -344,6 +344,13 @@ class DeviceAlarmEvent(Base):
owner_user_id: Mapped[Optional[int]] = mapped_column(Integer) owner_user_id: Mapped[Optional[int]] = mapped_column(Integer)
child_id: Mapped[Optional[int]] = mapped_column(Integer) child_id: Mapped[Optional[int]] = mapped_column(Integer)
source_msg_id: Mapped[str] = mapped_column(String(8), server_default=text("'010'")) source_msg_id: Mapped[str] = mapped_column(String(8), server_default=text("'010'"))
coord_type: Mapped[Optional[str]] = mapped_column(String(16))
lat: Mapped[Optional[float]] = mapped_column(Numeric(10, 7))
lng: Mapped[Optional[float]] = mapped_column(Numeric(10, 7))
location_updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
address: Mapped[Optional[str]] = mapped_column(String(255))
address_resolved_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
address_resolve_status: Mapped[Optional[int]] = mapped_column(Integer)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
class IMConversation(Base): class IMConversation(Base):
@@ -455,6 +462,9 @@ class ChildLocationCurrent(Base):
updated_at: Mapped[datetime] = mapped_column( updated_at: Mapped[datetime] = mapped_column(
DateTime, default=datetime.utcnow, onupdate=datetime.utcnow DateTime, default=datetime.utcnow, onupdate=datetime.utcnow
) )
address: Mapped[Optional[str]] = mapped_column(String(255))
address_resolved_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
address_resolve_status: Mapped[Optional[int]] = mapped_column(Integer)
class ChildLocationHistory(Base): class ChildLocationHistory(Base):

View File

@@ -448,6 +448,13 @@ CREATE TABLE IF NOT EXISTS `device_alarm_events` (
`owner_user_id` BIGINT NULL, `owner_user_id` BIGINT NULL,
`child_id` BIGINT NULL, `child_id` BIGINT NULL,
`source_msg_id` VARCHAR(8) NOT NULL DEFAULT '010', `source_msg_id` VARCHAR(8) NOT NULL DEFAULT '010',
`coord_type` VARCHAR(16) NULL,
`lat` DECIMAL(10, 7) NULL,
`lng` DECIMAL(10, 7) NULL,
`location_updated_at` DATETIME NULL,
`address` VARCHAR(255) NULL,
`address_resolved_at` DATETIME NULL,
`address_resolve_status` TINYINT NULL COMMENT '0=pending,1=success,2=failed',
`created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`alarm_id`), PRIMARY KEY (`alarm_id`),
KEY `idx_device_alarm_device_created` (`device_id`, `created_at`), KEY `idx_device_alarm_device_created` (`device_id`, `created_at`),
@@ -552,6 +559,9 @@ CREATE TABLE IF NOT EXISTS `child_location_current` (
`device_time` DATETIME NOT NULL, `device_time` DATETIME NOT NULL,
`server_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, `server_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
`updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
`address` VARCHAR(255) NULL,
`address_resolved_at` DATETIME NULL,
`address_resolve_status` TINYINT NULL COMMENT '0=pending,1=success,2=failed',
PRIMARY KEY (`child_id`), PRIMARY KEY (`child_id`),
KEY `idx_child_loc_current_device` (`device_id`), KEY `idx_child_loc_current_device` (`device_id`),
KEY `idx_child_loc_current_server_time` (`server_time`), KEY `idx_child_loc_current_server_time` (`server_time`),

View File

@@ -1,13 +1,6 @@
import logging
from typing import Optional, Dict
from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.interval import IntervalTrigger from apscheduler.triggers.interval import IntervalTrigger
from banban.service.pending_voice_message import pending_voice_message_service from banban.service.pending_voice_message import pending_voice_message_service
from services.connection_manager import connection_manager
from handlers.mqtt_handler import TalkingQMQTTService
from services.offline_audio_cache import offline_audio_cache
from config import settings
from utils.logger import session_logger as logger from utils.logger import session_logger as logger
class TaskScheduler: class TaskScheduler:
@@ -40,22 +33,19 @@ class TaskScheduler:
async def _execute_task(self): async def _execute_task(self):
try: try:
service = await TalkingQMQTTService.get_instance()
pending_devices = await pending_voice_message_service.list_devices_with_pending() pending_devices = await pending_voice_message_service.list_devices_with_pending()
audio_cache = await offline_audio_cache.get_all_audio_cache() device_ids = list(dict.fromkeys(pending_devices))
fallback_devices = list(audio_cache.keys()) if audio_cache else []
device_ids = list(dict.fromkeys([*pending_devices, *fallback_devices]))
if not device_ids: if not device_ids:
logger.warning("", "", "离线音频缓存,跳过本次执行") logger.warning("", "", "待收听留言,跳过本次执行")
return return
for device_id in device_ids: for device_id in device_ids:
websocket = await connection_manager.get_connection(device_id) sent = await pending_voice_message_service.notify_device_pending_message(
if websocket is not None: target_device_id=device_id,
audio_url = f"http://{settings.server_host}:{settings.server_port}/assets/audio/new_message.mp3" )
await service.send_nfc_notice(device_id, audio_url) if sent:
logger.info(device_id, "", f"[定时任务] 发送音频成功: device={device_id}, audio_url={audio_url}") logger.info(device_id, "", f"[定时任务] 发送待收听留言提醒成功: device={device_id}")
else: else:
logger.info(device_id, "", f"[定时任务] 跳过音频发送: 设备 {device_id} 离线") logger.info(device_id, "", f"[定时任务] 待收听留言提醒未发送: device={device_id}")
except Exception as e: except Exception as e:
logger.error("", "", f"[定时任务] 执行失败: {e}") logger.error("", "", f"[定时任务] 执行失败: {e}")

View File

@@ -0,0 +1,114 @@
from types import SimpleNamespace
import pytest
from banban.service.pending_voice_message import PendingVoiceMessageService
from services.scheduler import TaskScheduler
class FakeSession:
def __init__(self):
self.commits = 0
self.rollbacks = 0
self.closed = False
async def execute(self, statement, params=None):
return SimpleNamespace(mappings=lambda: SimpleNamespace(all=lambda: []))
async def commit(self):
self.commits += 1
async def rollback(self):
self.rollbacks += 1
async def close(self):
self.closed = True
@pytest.mark.asyncio
async def test_add_pending_message_sends_mqtt_notice_after_commit(monkeypatch):
service = PendingVoiceMessageService()
session = FakeSession()
upserts = []
notices = []
async def fake_get_session():
return session
async def fake_upsert_pending(self, **kwargs):
upserts.append(kwargs)
async def fake_notify_device_pending_message(*, target_device_id):
notices.append(target_device_id)
return True
monkeypatch.setattr(service, "get_session", fake_get_session)
monkeypatch.setattr(
"banban.service.pending_voice_message.PendingVoiceMessageDAO.upsert_pending",
fake_upsert_pending,
)
monkeypatch.setattr(service, "notify_device_pending_message", fake_notify_device_pending_message)
await service.add_pending_message(
target_device_id=" TalkingQ_device001 ",
sender_device_id=" TalkingQ_device002 ",
im_message_id=123,
media_file_key="messages/audio/demo.mp3",
audio_url="http://example.test/demo.mp3",
source="parent_child_voice",
)
assert session.commits == 1
assert session.rollbacks == 0
assert session.closed is True
assert upserts == [
{
"target_device_id": "TalkingQ_device001",
"sender_device_id": "TalkingQ_device002",
"im_message_id": 123,
"media_file_key": "messages/audio/demo.mp3",
"audio_url": "http://example.test/demo.mp3",
"source": "parent_child_voice",
}
]
assert notices == ["TalkingQ_device001"]
@pytest.mark.asyncio
async def test_scheduler_sends_pending_notice_without_websocket_check(monkeypatch):
scheduler = TaskScheduler()
calls = []
class FakePendingVoiceService:
async def list_devices_with_pending(self):
return ["TalkingQ_device001"]
async def notify_device_pending_message(self, *, target_device_id):
calls.append(target_device_id)
return True
monkeypatch.setattr("services.scheduler.pending_voice_message_service", FakePendingVoiceService())
await scheduler._execute_task()
assert calls == ["TalkingQ_device001"]
@pytest.mark.asyncio
async def test_scheduler_skips_notice_when_no_pending_messages(monkeypatch):
scheduler = TaskScheduler()
calls = []
class FakePendingVoiceService:
async def list_devices_with_pending(self):
return []
async def notify_device_pending_message(self, *, target_device_id):
calls.append(target_device_id)
return True
monkeypatch.setattr("services.scheduler.pending_voice_message_service", FakePendingVoiceService())
await scheduler._execute_task()
assert calls == []