接入服务号模板推送通知

banban-mini/src/app.config.ts

@@ -9,6 +9,7 @@ export default defineAppConfig({
     "pages/sleep/index",
     "pages/sleep-schedule/index",
     "pages/family-invite/index",
+    "pages/webview/index",
   ],
   window: {
     backgroundTextStyle: "light",

banban-mini/src/pages/location/index.tsx

@@ -56,9 +56,7 @@ function formatOptionalNumber(value?: number | null, digits = 1): string | null
 function getLocationAddress(point: DeviceLocation | DeviceTrajectoryPoint): string {
   const address = String(point.address || '').trim()
   if (address) return address
-  if (point.address_resolve_status === 0) return '地点解析中'
+  return '地址解析中'
-  if (point.address_resolve_status === 2) return '地点暂时无法解析'
-  return '当前地点未知'
 }
 
 function buildLocationDetailLines(

banban-mini/src/pages/sleep/index.tsx

@@ -36,6 +36,7 @@ import {
   startDeviceFirmwareUpdate,
   updateDeviceRole,
 } from '@/services/device'
+import { getWechatMpBindStatus, startWechatMpBind, WechatMpBindStatus } from '@/services/wechat-mp'
 import { useSystemBanner } from '@/components/system-banner/use-system-banner'
 import './index.scss'
 
@@ -104,6 +105,9 @@ export default function Sleep() {
   const [editingFamilyMember, setEditingFamilyMember] = useState<FamilyMember | null>(null)
   const [familyMemberDisplayName, setFamilyMemberDisplayName] = useState('')
   const [parentInfo, setParentInfo] = useState<Parent | null>(null)
+  const [wechatMpStatus, setWechatMpStatus] = useState<WechatMpBindStatus | null>(null)
+  const [isLoadingWechatMp, setIsLoadingWechatMp] = useState(false)
+  const [isStartingWechatMpBind, setIsStartingWechatMpBind] = useState(false)
   const [isAuthorizingPhone, setIsAuthorizingPhone] = useState(false)
   const firmwareStatusPollTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null)
   const firmwareStatusPollDeviceIdRef = useRef('')
@@ -150,6 +154,7 @@ export default function Sleep() {
       void loadRoleData(nextBinding?.device_id)
       void loadFamilyData(nextBinding?.device_id)
       void loadCurrentDeviceCards(nextBinding?.device_id)
+      void loadWechatMpStatus()
     } catch (error: any) {
       console.error('[manage] load failed:', error)
       Taro.showToast({
@@ -161,6 +166,18 @@ export default function Sleep() {
     }
   }
 
+  const loadWechatMpStatus = async () => {
+    setIsLoadingWechatMp(true)
+    try {
+      setWechatMpStatus(await getWechatMpBindStatus())
+    } catch (error) {
+      console.warn('[manage] wechat mp bind status load failed:', error)
+      setWechatMpStatus(null)
+    } finally {
+      setIsLoadingWechatMp(false)
+    }
+  }
+
   const loadCurrentParent = async (userId: number): Promise<Parent | null> => {
     try {
       return await getParent(userId)
@@ -825,6 +842,41 @@ export default function Sleep() {
     }
   }
 
+  const handleStartWechatMpBind = async () => {
+    if (isStartingWechatMpBind) return
+
+    setIsStartingWechatMpBind(true)
+    try {
+      const result = await startWechatMpBind()
+      const bindUrl = String(result.url || '').trim()
+      if (!bindUrl) {
+        throw new Error('未获取到服务号绑定链接')
+      }
+      Taro.navigateTo({
+        url: `/pages/webview/index?url=${encodeURIComponent(bindUrl)}`,
+        fail: () => {
+          Taro.setClipboardData({
+            data: bindUrl,
+            success: () => {
+              Taro.showModal({
+                title: '服务号通知',
+                content: '绑定链接已复制，请在微信里打开完成授权。',
+                showCancel: false,
+              })
+            },
+          })
+        },
+      })
+    } catch (error: any) {
+      Taro.showToast({
+        title: error?.message || '服务号绑定失败',
+        icon: 'none',
+      })
+    } finally {
+      setIsStartingWechatMpBind(false)
+    }
+  }
+
   const handleMenuClick = (item: MenuItem) => {
     if (item.disabled) return
 
@@ -878,6 +930,11 @@ export default function Sleep() {
       return
     }
 
+    if (item.name === '服务号通知') {
+      handleStartWechatMpBind()
+      return
+    }
+
     if (item.name === '解除设备绑定') {
       handleUnbind()
     }
@@ -937,6 +994,13 @@ export default function Sleep() {
     : isLoadingFamily
       ? '查询中'
       : `${familyInfo?.total || 1}/4 人`
+  const wechatMpValue = isStartingWechatMpBind
+    ? '打开中'
+    : isLoadingWechatMp
+      ? '查询中'
+      : wechatMpStatus?.bound
+        ? '已开启'
+        : '未绑定'
   const firmwareSubtitle = !binding?.device_id
     ? '绑定设备后可查看系统版本'
     : isLoadingFirmware
@@ -1018,6 +1082,14 @@ export default function Sleep() {
       arrow: true,
       disabled: !binding,
     },
+    {
+      icon: require('../../assets/tab-icons/rings.png'),
+      iconBgClass: 'green',
+      name: '服务号通知',
+      value: wechatMpValue,
+      arrow: true,
+      disabled: isStartingWechatMpBind,
+    },
     {
       icon: require('../../assets/tab-icons/broken-rings.png'),
       iconBgClass: 'red',

banban-mini/src/pages/webview/index.config.ts

@@ -0,0 +1,3 @@
+export default definePageConfig({
+  navigationBarTitleText: '服务号通知',
+})

banban-mini/src/pages/webview/index.scss

@@ -0,0 +1,21 @@
+.webview-page {
+  width: 100%;
+  height: 100vh;
+  background: #FFFFFF;
+}
+
+.webview-empty {
+  min-height: 100vh;
+  display: flex;
+  align-items: center;
+  justify-content: center;
+  padding: 32px;
+  box-sizing: border-box;
+}
+
+.webview-empty-text {
+  font-size: 28px;
+  line-height: 1.6;
+  color: #6B7280;
+  text-align: center;
+}

banban-mini/src/pages/webview/index.tsx

@@ -0,0 +1,23 @@
+import { Text, View, WebView } from '@tarojs/components'
+import { useRouter } from '@tarojs/taro'
+import './index.scss'
+
+export default function WebviewPage() {
+  const router = useRouter()
+  const encodedUrl = String(router.params?.url || '').trim()
+  const targetUrl = encodedUrl ? decodeURIComponent(encodedUrl) : ''
+
+  if (!targetUrl) {
+    return (
+      <View className='webview-empty'>
+        <Text className='webview-empty-text'>链接无效</Text>
+      </View>
+    )
+  }
+
+  return (
+    <View className='webview-page'>
+      <WebView src={targetUrl} />
+    </View>
+  )
+}

banban-mini/src/services/wechat-mp.ts

@@ -0,0 +1,21 @@
+import { request } from './api'
+
+export interface WechatMpBindStartResponse {
+  url: string
+}
+
+export interface WechatMpBindStatus {
+  bound: boolean
+  subscribed: boolean
+  updated_at?: string | null
+}
+
+export async function getWechatMpBindStatus(): Promise<WechatMpBindStatus> {
+  return request<WechatMpBindStatus>('/banban/wechat-mp/bind/status')
+}
+
+export async function startWechatMpBind(): Promise<WechatMpBindStartResponse> {
+  return request<WechatMpBindStartResponse>('/banban/wechat-mp/bind/start', {
+    method: 'POST',
+  })
+}

talkingq-url/banban/dao/device_alarm.py

@@ -1,4 +1,5 @@
 from collections.abc import Mapping
+from datetime import datetime
 from typing import Any
 
 from sqlalchemy import text
@@ -17,6 +18,12 @@ class DeviceAlarmDAO(BaseDAO):
         location: Mapping[str, Any] | None = None,
     ) -> int:
         location = location or {}
+        address = str(location.get("address") or "").strip() or None
+        address_status = location.get("address_resolve_status")
+        if address and address_status is None:
+            address_status = 1
+        elif address_status is None and location.get("lat") is not None and location.get("lng") is not None:
+            address_status = 0
         result = await self.execute(
             """
                 INSERT INTO device_alarm_events (
@@ -28,6 +35,8 @@ class DeviceAlarmDAO(BaseDAO):
                     lat,
                     lng,
                     location_updated_at,
+                    address,
+                    address_resolved_at,
                     address_resolve_status,
                     created_at
                 )
@@ -40,6 +49,8 @@ class DeviceAlarmDAO(BaseDAO):
                     :lat,
                     :lng,
                     :location_updated_at,
+                    :address,
+                    :address_resolved_at,
                     :address_resolve_status,
                     CURRENT_TIMESTAMP
                 )
@@ -53,7 +64,9 @@ class DeviceAlarmDAO(BaseDAO):
                 "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,
+                "address": address,
+                "address_resolved_at": location.get("address_resolved_at"),
+                "address_resolve_status": address_status,
             },
         )
         await self.commit()
@@ -100,7 +113,10 @@ class DeviceAlarmDAO(BaseDAO):
                   coord_type,
                   lat,
                   lng,
-                  updated_at
+                  updated_at,
+                  address,
+                  address_resolved_at,
+                  address_resolve_status
                 FROM child_location_current
                 WHERE child_id = :child_id
                 LIMIT 1
@@ -144,6 +160,15 @@ class DeviceAlarmDAO(BaseDAO):
     ) -> bool:
         lat = location.get("lat")
         lng = location.get("lng")
+        address = str(location.get("address") or "").strip() or None
+        address_status = location.get("address_resolve_status")
+        address_resolved_at = location.get("address_resolved_at")
+        if address and address_status is None:
+            address_status = 1
+        elif address_status is None and lat is not None and lng is not None:
+            address_status = 0
+        if address and address_resolved_at is None:
+            address_resolved_at = datetime.now()
         result = await self.execute(
             text(
                 """
@@ -152,8 +177,8 @@ class DeviceAlarmDAO(BaseDAO):
                     lat = :lat,
                     lng = :lng,
                     location_updated_at = :location_updated_at,
-                    address = NULL,
+                    address = :address,
-                    address_resolved_at = NULL,
+                    address_resolved_at = :address_resolved_at,
                     address_resolve_status = :address_resolve_status
                 WHERE alarm_id = :alarm_id
                 """
@@ -164,7 +189,9 @@ class DeviceAlarmDAO(BaseDAO):
                 "lat": lat,
                 "lng": lng,
                 "location_updated_at": location.get("updated_at"),
-                "address_resolve_status": 0 if lat is not None and lng is not None else None,
+                "address": address,
+                "address_resolved_at": address_resolved_at,
+                "address_resolve_status": address_status,
             },
         )
         await self.commit()

talkingq-url/banban/middleware/auth.py

@@ -11,6 +11,7 @@ NO_AUTH_PATH_PREFIXES = (
     "/banban/auth/login",
     "/banban/device-im",
     "/banban/device-location",
+    "/banban/wechat-mp/oauth/callback",
 )
 
 

talkingq-url/banban/routers/__init__.py

@@ -10,6 +10,7 @@ from banban.routers.im import router as im_router
 from banban.routers.parents import router as parents_router
 from banban.routers.roles import router as roles_router
 from banban.routers.wechat_auth import router as wechat_auth_router
+from banban.routers.wechat_mp import router as wechat_mp_router
 from banban.routers.mqtt_router import router as mqtt_router
 
 # 统一创建一个主路由，前缀为 /banban
@@ -17,6 +18,7 @@ banban_router = APIRouter(prefix="/banban")
 
 # 将所有子路由添加到主路由
 banban_router.include_router(wechat_auth_router, tags=["banban-auth"])
+banban_router.include_router(wechat_mp_router, tags=["banban-wechat-mp"])
 banban_router.include_router(bindings_router, tags=["banban-bindings"])
 banban_router.include_router(children_router, tags=["banban-children"])
 banban_router.include_router(devices_router, tags=["banban-devices"])

talkingq-url/banban/routers/devices.py

@@ -33,6 +33,7 @@ except ModuleNotFoundError:
 
 router = APIRouter(prefix="/devices", tags=["devices"])
 logger = logging.getLogger("app.devices")
+LOCATION_ADDRESS_PENDING_TEXT = "地址解析中"
 
 
 class DeviceMessageItem(BaseModel):
@@ -236,6 +237,11 @@ def _row_to_current_location_response(
     stale: bool = False,
     realtime: bool = True,
 ) -> DeviceLocationCurrentResponse:
+    address = str(row.get("address") or "").strip() or None
+    address_resolve_status = int(row["address_resolve_status"]) if row.get("address_resolve_status") is not None else None
+    if address is None and address_resolve_status != 1:
+        address = LOCATION_ADDRESS_PENDING_TEXT
+
     return DeviceLocationCurrentResponse(
         child_id=int(row["child_id"]),
         child_name=row.get("child_name"),
@@ -252,9 +258,9 @@ def _row_to_current_location_response(
         device_time=row["device_time"],
         server_time=row["server_time"],
         updated_at=row["updated_at"],
-        address=row.get("address"),
+        address=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,
+        address_resolve_status=address_resolve_status,
         stale=stale,
         realtime=realtime,
     )

talkingq-url/banban/routers/wechat_mp.py

@@ -0,0 +1,132 @@
+import html
+import logging
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi.responses import HTMLResponse
+from pydantic import BaseModel
+
+from banban.security import get_current_user_id
+from banban.service.wechat_mp_notification import wechat_mp_notification_service
+from config import settings
+
+
+router = APIRouter(prefix="/wechat-mp", tags=["wechat-mp"])
+logger = logging.getLogger("app.wechat_mp")
+
+
+class WechatMpBindStartResponse(BaseModel):
+    url: str
+
+
+class WechatMpBindStatusResponse(BaseModel):
+    bound: bool
+    subscribed: bool
+    updated_at: str | None = None
+
+
+@router.post("/bind/start", response_model=WechatMpBindStartResponse)
+async def start_wechat_mp_bind(
+    current_user_id: int = Depends(get_current_user_id),
+) -> WechatMpBindStartResponse:
+    try:
+        url = await wechat_mp_notification_service.create_bind_url(user_id=current_user_id)
+    except RuntimeError as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+    return WechatMpBindStartResponse(url=url)
+
+
+@router.get("/bind/status", response_model=WechatMpBindStatusResponse)
+async def get_wechat_mp_bind_status(
+    current_user_id: int = Depends(get_current_user_id),
+) -> WechatMpBindStatusResponse:
+    status = await wechat_mp_notification_service.get_bind_status(user_id=current_user_id)
+    updated_at = status.get("updated_at")
+    return WechatMpBindStatusResponse(
+        bound=bool(status.get("bound")),
+        subscribed=bool(status.get("subscribed")),
+        updated_at=updated_at.isoformat() if hasattr(updated_at, "isoformat") else updated_at,
+    )
+
+
+@router.get("/oauth/callback", response_class=HTMLResponse)
+async def wechat_mp_oauth_callback(
+    code: str = Query(default=""),
+    state: str = Query(default=""),
+) -> HTMLResponse:
+    if not code or not state:
+        return _callback_page("绑定失败", "缺少微信授权参数，请返回小程序重新发起绑定。")
+
+    try:
+        user_id = await wechat_mp_notification_service.bind_oauth_code(code=code, state=state)
+    except Exception as exc:
+        logger.warning("wechat mp oauth bind failed", extra={"event": "wechat_mp_bind_failed"})
+        return _callback_page("绑定失败", f"{exc}")
+
+    logger.info("wechat mp oauth bind succeeded", extra={"event": "wechat_mp_bind_succeeded", "user_id": user_id})
+    return _callback_page("绑定成功", "服务号通知已开启，请返回小程序。")
+
+
+def _callback_page(title: str, message: str) -> HTMLResponse:
+    escaped_title = html.escape(title)
+    escaped_message = html.escape(message)
+    bind_success_page = html.escape(settings.wechat_mp_bind_success_page or "pages/sleep/index")
+    body = f"""<!doctype html>
+<html lang="zh-CN">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width,initial-scale=1,viewport-fit=cover">
+  <title>{escaped_title}</title>
+  <style>
+    body {{
+      margin: 0;
+      font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
+      background: #f5f7fa;
+      color: #1f2937;
+    }}
+    .wrap {{
+      min-height: 100vh;
+      display: flex;
+      align-items: center;
+      justify-content: center;
+      padding: 32px;
+      box-sizing: border-box;
+    }}
+    .panel {{
+      width: 100%;
+      max-width: 420px;
+      background: #fff;
+      border-radius: 16px;
+      padding: 28px;
+      box-shadow: 0 8px 30px rgba(15, 23, 42, .08);
+      box-sizing: border-box;
+    }}
+    h1 {{
+      margin: 0 0 14px;
+      font-size: 24px;
+      line-height: 1.35;
+    }}
+    p {{
+      margin: 0;
+      font-size: 16px;
+      line-height: 1.7;
+      color: #4b5563;
+    }}
+    .hint {{
+      margin-top: 18px;
+      color: #94a3b8;
+      font-size: 13px;
+      word-break: break-all;
+    }}
+  </style>
+</head>
+<body>
+  <div class="wrap">
+    <main class="panel">
+      <h1>{escaped_title}</h1>
+      <p>{escaped_message}</p>
+      <p class="hint">返回路径：{bind_success_page}</p>
+    </main>
+  </div>
+</body>
+</html>"""
+    return HTMLResponse(content=body)

talkingq-url/banban/service/device.py

@@ -8,6 +8,7 @@ from fastapi import HTTPException
 from banban.dao.device import DeviceDAO
 from banban.service.device_setting import device_setting_service
 from banban.service.ota_firmware import ota_firmware_service
+from banban.service.wechat_mp_notification import wechat_mp_notification_service
 from config import settings
 from services.device_update_manager import device_firmware_update_manager
 
@@ -281,6 +282,11 @@ class DeviceService(DatabaseServiceBase):
             schedule_suppressed_until=schedule_suppressed_until,
             clear_schedule_suppression=clear_schedule_suppression,
         )
+        if switch == "off":
+            wechat_mp_notification_service.schedule_sleep_mode_notification(
+                device_id=device_id,
+                child_name=status_row.get("child_name"),
+            )
         return msg_id
 
     def _compare_versions(self, current_version: str | None, latest_version: str | None) -> bool:

talkingq-url/banban/service/device_alarm.py

@@ -3,6 +3,7 @@ from typing import Any
 
 from banban.dao.device_alarm import DeviceAlarmDAO
 from banban.service.amap_geocode import amap_geocode_service
+from banban.service.location import is_valid_coordinate_pair
 from services.task_manager import task_manager
 from services.database_service_base import DatabaseServiceBase
 from utils.logger import session_logger
@@ -34,6 +35,13 @@ class DeviceAlarmService(DatabaseServiceBase):
             location = None
             if child_id is not None:
                 location = await dao.get_current_location_snapshot(child_id=child_id)
+                if location and not is_valid_coordinate_pair(location.get("lat"), location.get("lng")):
+                    session_logger.warning(
+                        device_id,
+                        "alarm_address",
+                        f"忽略无效当前定位快照: child_id={child_id}, lat={location.get('lat')}, lng={location.get('lng')}",
+                    )
+                    location = None
 
             alarm_id = await dao.create(
                 device_id=device_id,
@@ -42,7 +50,13 @@ class DeviceAlarmService(DatabaseServiceBase):
                 source_msg_id=source_msg_id,
                 location=location,
             )
-            if resolve_address and alarm_id and location and location.get("lat") is not None and location.get("lng") is not None:
+            if (
+                resolve_address
+                and alarm_id
+                and location
+                and is_valid_coordinate_pair(location.get("lat"), location.get("lng"))
+                and not str(location.get("address") or "").strip()
+            ):
                 await task_manager.create_task(
                     self.resolve_alarm_address(
                         alarm_id=alarm_id,
@@ -90,7 +104,12 @@ class DeviceAlarmService(DatabaseServiceBase):
         device_id: str,
         location: Mapping[str, Any],
     ) -> bool:
-        if location.get("lat") is None or location.get("lng") is None:
+        if not is_valid_coordinate_pair(location.get("lat"), location.get("lng")):
+            session_logger.warning(
+                device_id,
+                "alarm_address",
+                f"忽略无效告警定位: alarm_id={alarm_id}, lat={location.get('lat')}, lng={location.get('lng')}",
+            )
             return False
 
         updated = await self.update_alarm_location(alarm_id=alarm_id, location=location)
@@ -116,7 +135,13 @@ class DeviceAlarmService(DatabaseServiceBase):
 
     async def resolve_alarm_current_location(self, *, alarm_id: int, device_id: str) -> bool:
         alarm = await self.get_alarm_event(alarm_id=alarm_id, device_id=device_id)
-        if not alarm or alarm.get("lat") is None or alarm.get("lng") is None:
+        if not alarm:
+            return False
+
+        if str(alarm.get("address") or "").strip():
+            return True
+
+        if not is_valid_coordinate_pair(alarm.get("lat"), alarm.get("lng")):
             return False
 
         await self.resolve_alarm_address(
@@ -135,6 +160,8 @@ class DeviceAlarmService(DatabaseServiceBase):
         lat: float,
         lng: float,
     ) -> None:
+        if not is_valid_coordinate_pair(lat, lng):
+            return
         try:
             result = await amap_geocode_service.reverse_geocode(lng=lng, lat=lat)
         except Exception as exc:

talkingq-url/banban/service/location.py

@@ -22,6 +22,20 @@ except ModuleNotFoundError:
 ADDRESS_RESOLVE_PENDING = 0
 ADDRESS_RESOLVE_SUCCESS = 1
 ADDRESS_RESOLVE_FAILED = 2
+INVALID_COORDINATE_EPSILON = 0.0000001
+
+
+def is_valid_coordinate_pair(latitude: float | None, longitude: float | None) -> bool:
+    if latitude is None or longitude is None:
+        return False
+    try:
+        lat = float(latitude)
+        lng = float(longitude)
+    except (TypeError, ValueError):
+        return False
+    if lat < -90 or lat > 90 or lng < -180 or lng > 180:
+        return False
+    return not (abs(lat) < INVALID_COORDINATE_EPSILON and abs(lng) < INVALID_COORDINATE_EPSILON)
 
 
 class LocationService(DatabaseServiceBase):
@@ -42,11 +56,19 @@ class LocationService(DatabaseServiceBase):
         device_id: str,
         serial_number: str,
         payload: DeviceLocationReportRequest,
-    ) -> tuple[Any, Mapping[str, Any]]:
+    ) -> tuple[Any, Mapping[str, Any] | None]:
         device_identity = await im_service.authenticate_device_identity(
             device_id=device_id,
             serial_number=serial_number,
         )
+        if not is_valid_coordinate_pair(payload.lat, payload.lng):
+            session_logger.warning(
+                device_id,
+                "location",
+                f"忽略无效设备定位: lat={payload.lat}, lng={payload.lng}",
+            )
+            return device_identity, None
+
         db_session = await self.get_session()
         try:
             dao = LocationDAO(db_session)
@@ -123,7 +145,13 @@ class LocationService(DatabaseServiceBase):
         battery_pct: int | None = None,
         device_time: datetime | None = None,
     ) -> Mapping[str, Any] | None:
-        if latitude is None or longitude is None:
+        if not is_valid_coordinate_pair(latitude, longitude):
+            if latitude is not None and longitude is not None:
+                session_logger.warning(
+                    device_id,
+                    "location",
+                    f"忽略无效MQTT定位: lat={latitude}, lng={longitude}",
+                )
             return None
 
         @dataclass(frozen=True)
@@ -173,6 +201,8 @@ class LocationService(DatabaseServiceBase):
     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 is_valid_coordinate_pair(row.get("lat"), row.get("lng")):
+            return
         if not row.get("should_resolve_address"):
             return
         await task_manager.create_task(
@@ -194,6 +224,8 @@ class LocationService(DatabaseServiceBase):
         lat: float,
         lng: float,
     ) -> None:
+        if not is_valid_coordinate_pair(lat, lng):
+            return
         try:
             result = await amap_geocode_service.reverse_geocode(lng=lng, lat=lat)
         except Exception as exc:

talkingq-url/banban/service/wechat_mp_notification.py

@@ -0,0 +1,590 @@
+import asyncio
+import secrets
+import time
+from collections.abc import Mapping
+from datetime import datetime, timedelta
+from typing import Any
+from urllib.parse import urlencode
+
+import httpx
+from sqlalchemy import text
+
+from config import settings
+from services.database_service_base import DatabaseServiceBase
+from utils.logger import session_logger
+
+
+WECHAT_MP_ACCOUNT_TYPE = "service_account"
+
+
+class WechatMpNotificationService(DatabaseServiceBase):
+    def __init__(self):
+        super().__init__(service_name="wechat_mp_notification_service")
+        self._access_token: str | None = None
+        self._access_token_expire_at = 0.0
+        self._last_sent_at: dict[tuple[str, str, str], float] = {}
+        self._low_battery_active: set[str] = set()
+
+    def is_configured(self) -> bool:
+        return bool(
+            settings.wechat_mp_push_enabled
+            and settings.wechat_mp_app_id
+            and settings.wechat_mp_app_secret
+            and self._template_id("leave_message")
+        )
+
+    async def create_bind_url(self, *, user_id: int) -> str:
+        if not settings.wechat_mp_app_id or not settings.wechat_mp_oauth_redirect_uri:
+            raise RuntimeError("wechat mp oauth is not configured")
+
+        state = secrets.token_urlsafe(24)
+        await self._create_bind_state(state=state, user_id=user_id)
+        redirect_uri = settings.wechat_mp_oauth_redirect_uri.strip()
+        params = urlencode(
+            {
+                "appid": settings.wechat_mp_app_id,
+                "redirect_uri": redirect_uri,
+                "response_type": "code",
+                "scope": "snsapi_base",
+                "state": state,
+            }
+        )
+        return f"https://open.weixin.qq.com/connect/oauth2/authorize?{params}#wechat_redirect"
+
+    async def bind_oauth_code(self, *, code: str, state: str) -> int:
+        user_id = await self._consume_bind_state(state=state)
+        payload = await self._oauth_access_token(code=code)
+        openid = str(payload.get("openid") or "").strip()
+        if not openid:
+            raise RuntimeError("wechat mp oauth response missing openid")
+        unionid = str(payload.get("unionid") or "").strip() or None
+        user_info = await self._get_mp_user_info(openid=openid)
+        subscribed = 1 if int(user_info.get("subscribe") or 0) == 1 else 0
+        await self._upsert_parent_wechat_account(
+            user_id=user_id,
+            openid=openid,
+            unionid=unionid,
+            subscribed=subscribed,
+        )
+        if not subscribed:
+            session_logger.info(str(user_id), "wechat_mp", "服务号openid已绑定，但用户尚未关注服务号")
+        return user_id
+
+    async def get_bind_status(self, *, user_id: int) -> Mapping[str, Any]:
+        db_session = await self.get_session()
+        try:
+            result = await db_session.execute(
+                text(
+                    """
+                    SELECT openid, unionid, subscribed, updated_at
+                    FROM parent_wechat_accounts
+                    WHERE user_id = :user_id
+                      AND app_id = :app_id
+                      AND account_type = :account_type
+                    LIMIT 1
+                    """
+                ),
+                {
+                    "user_id": user_id,
+                    "app_id": settings.wechat_mp_app_id,
+                    "account_type": WECHAT_MP_ACCOUNT_TYPE,
+                },
+            )
+            row = result.mappings().first()
+            return {
+                "bound": bool(row and row.get("openid")),
+                "subscribed": bool(row and int(row.get("subscribed") or 0) == 1),
+                "updated_at": row.get("updated_at") if row else None,
+            }
+        finally:
+            await db_session.close()
+
+    def schedule_leave_message_notification(self, **kwargs: Any) -> None:
+        self._schedule_best_effort(
+            self.notify_leave_message_best_effort(**kwargs),
+            device_id=kwargs.get("device_id"),
+        )
+
+    async def notify_leave_message_best_effort(self, **kwargs: Any) -> None:
+        try:
+            await self.notify_leave_message(**kwargs)
+        except Exception as exc:
+            session_logger.warning(
+                kwargs.get("device_id") or "",
+                "wechat_mp",
+                f"留言公众号推送失败，不影响主流程: {exc}",
+            )
+
+    async def notify_leave_message(
+        self,
+        *,
+        device_id: str,
+        child_name: str | None = None,
+    ) -> None:
+        if not self.is_configured():
+            return
+        recipients = await self._list_device_family_mp_openids(device_id=device_id)
+        if not recipients:
+            session_logger.info(device_id, "wechat_mp", "留言公众号推送跳过：没有绑定服务号openid的家长")
+            return
+        await self._send_to_recipients(
+            device_id=device_id,
+            notification_type="leave_message",
+            recipients=recipients,
+            title=child_name or device_id,
+            condition="收到新的设备留言",
+            page=settings.wechat_mp_chat_page,
+            dedup_seconds=max(0, int(settings.wechat_mp_leave_message_dedup_seconds or 0)),
+        )
+
+    def schedule_low_battery_notification(self, *, device_id: str, power: Any, child_name: str | None = None) -> None:
+        self._schedule_best_effort(
+            self.notify_low_battery_best_effort(device_id=device_id, power=power, child_name=child_name),
+            device_id=device_id,
+        )
+
+    async def notify_low_battery_best_effort(self, **kwargs: Any) -> None:
+        try:
+            await self.notify_low_battery(**kwargs)
+        except Exception as exc:
+            session_logger.warning(
+                kwargs.get("device_id") or "",
+                "wechat_mp",
+                f"低电量公众号推送失败，不影响主流程: {exc}",
+            )
+
+    async def notify_low_battery(self, *, device_id: str, power: Any, child_name: str | None = None) -> None:
+        if not self.is_configured():
+            return
+        try:
+            power_value = int(power)
+        except (TypeError, ValueError):
+            return
+        if power_value >= int(settings.wechat_mp_low_battery_recovery_threshold or 25):
+            self._low_battery_active.discard(device_id)
+            return
+        if power_value >= int(settings.wechat_mp_low_battery_threshold or 20):
+            return
+        if device_id in self._low_battery_active:
+            return
+        self._low_battery_active.add(device_id)
+
+        recipients = await self._list_device_family_mp_openids(device_id=device_id)
+        if not recipients:
+            return
+        await self._send_to_recipients(
+            device_id=device_id,
+            notification_type="low_battery",
+            recipients=recipients,
+            title=child_name or device_id,
+            condition=f"设备电量低于{settings.wechat_mp_low_battery_threshold}%",
+            page=settings.wechat_mp_device_page,
+            dedup_seconds=max(0, int(settings.wechat_mp_low_battery_dedup_seconds or 0)),
+        )
+
+    def schedule_sleep_mode_notification(self, *, device_id: str, child_name: str | None = None) -> None:
+        self._schedule_best_effort(
+            self.notify_sleep_mode_best_effort(device_id=device_id, child_name=child_name),
+            device_id=device_id,
+        )
+
+    async def notify_sleep_mode_best_effort(self, **kwargs: Any) -> None:
+        try:
+            await self.notify_sleep_mode(**kwargs)
+        except Exception as exc:
+            session_logger.warning(
+                kwargs.get("device_id") or "",
+                "wechat_mp",
+                f"休眠公众号推送失败，不影响主流程: {exc}",
+            )
+
+    async def notify_sleep_mode(self, *, device_id: str, child_name: str | None = None) -> None:
+        if not self.is_configured():
+            return
+        recipients = await self._list_device_family_mp_openids(device_id=device_id)
+        if not recipients:
+            return
+        await self._send_to_recipients(
+            device_id=device_id,
+            notification_type="sleep_mode",
+            recipients=recipients,
+            title=child_name or device_id,
+            condition="设备进入休眠模式",
+            page=settings.wechat_mp_device_page,
+            dedup_seconds=max(0, int(settings.wechat_mp_sleep_mode_dedup_seconds or 0)),
+        )
+
+    def _schedule_best_effort(self, coro: Any, *, device_id: str | None) -> None:
+        try:
+            asyncio.create_task(coro)
+        except RuntimeError as exc:
+            session_logger.warning(device_id or "", "wechat_mp", f"公众号推送后台任务创建失败，不影响主流程: {exc}")
+            try:
+                coro.close()
+            except Exception:
+                pass
+
+    async def _send_to_recipients(
+        self,
+        *,
+        device_id: str,
+        notification_type: str,
+        recipients: list[str],
+        title: str,
+        condition: str,
+        page: str,
+        dedup_seconds: int,
+    ) -> None:
+        template_id = self._template_id(notification_type)
+        if not template_id:
+            return
+
+        for openid in recipients:
+            if self._is_deduped(
+                device_id=device_id,
+                notification_type=notification_type,
+                openid=openid,
+                dedup_seconds=dedup_seconds,
+            ):
+                session_logger.info(device_id, "wechat_mp", f"公众号推送防抖跳过: type={notification_type}")
+                continue
+
+            result = await self._send_template_message(
+                openid=openid,
+                template_id=template_id,
+                title=title,
+                condition=condition,
+                page=page,
+            )
+            if result.get("ok"):
+                self._mark_sent(device_id=device_id, notification_type=notification_type, openid=openid)
+                session_logger.info(
+                    device_id,
+                    "wechat_mp",
+                    f"公众号推送成功: type={notification_type}, msgid={result.get('msgid')}",
+                )
+            else:
+                session_logger.warning(
+                    device_id,
+                    "wechat_mp",
+                    f"公众号推送失败: type={notification_type}, errcode={result.get('errcode')}, errmsg={result.get('errmsg')}",
+                )
+
+    async def _send_template_message(
+        self,
+        *,
+        openid: str,
+        template_id: str,
+        title: str,
+        condition: str,
+        page: str,
+    ) -> Mapping[str, Any]:
+        access_token = await self._get_access_token()
+        payload: dict[str, Any] = {
+            "touser": openid,
+            "template_id": template_id,
+            "data": {
+                "thing13": {"value": self._clip_template_value(title, 20)},
+                "thing5": {"value": self._clip_template_value(condition, 20)},
+            },
+        }
+        miniprogram_appid = (settings.wechat_mp_miniprogram_appid or "").strip()
+        if miniprogram_appid and page:
+            payload["miniprogram"] = {
+                "appid": miniprogram_appid,
+                "pagepath": page,
+            }
+        url = f"{settings.wechat_mp_api_base_url.rstrip('/')}/cgi-bin/message/template/send"
+        params = {"access_token": access_token}
+        async with httpx.AsyncClient(timeout=settings.wechat_mp_http_timeout_seconds) as client:
+            response = await client.post(url, params=params, json=payload)
+            response.raise_for_status()
+            data = response.json()
+        return {
+            **data,
+            "ok": int(data.get("errcode") or 0) == 0,
+        }
+
+    async def _list_device_family_mp_openids(self, *, device_id: str) -> list[str]:
+        db_session = await self.get_session()
+        try:
+            result = await db_session.execute(
+                text(
+                    """
+                    SELECT pwa.openid
+                    FROM (
+                      SELECT
+                        db.owner_user_id AS user_id,
+                        0 AS sort_role,
+                        db.bound_at AS sort_time,
+                        db.id AS sort_id
+                      FROM device_bindings AS db
+                      WHERE db.device_id = :device_id
+                        AND db.status = 1
+                        AND db.owner_user_id IS NOT NULL
+                      UNION ALL
+                      SELECT
+                        dfm.user_id,
+                        dfm.role AS sort_role,
+                        dfm.joined_at AS sort_time,
+                        dfm.id AS sort_id
+                      FROM device_family_members AS dfm
+                      WHERE dfm.device_id = :device_id
+                        AND dfm.status = 1
+                    ) AS recipients
+                    JOIN parents AS p
+                      ON p.user_id = recipients.user_id
+                     AND p.status = 1
+                    JOIN parent_wechat_accounts AS pwa
+                      ON pwa.user_id = p.user_id
+                     AND pwa.app_id = :app_id
+                     AND pwa.account_type = :account_type
+                    WHERE pwa.openid IS NOT NULL
+                      AND pwa.openid <> ''
+                      AND pwa.subscribed = 1
+                    ORDER BY recipients.sort_role ASC, recipients.sort_time ASC, recipients.sort_id ASC
+                    """
+                ),
+                {
+                    "device_id": device_id,
+                    "app_id": settings.wechat_mp_app_id,
+                    "account_type": WECHAT_MP_ACCOUNT_TYPE,
+                },
+            )
+            openids: list[str] = []
+            seen: set[str] = set()
+            for row in result.mappings().all():
+                openid = str(row["openid"]).strip()
+                if not openid or openid in seen:
+                    continue
+                seen.add(openid)
+                openids.append(openid)
+            return openids
+        finally:
+            await db_session.close()
+
+    async def _upsert_parent_wechat_account(
+        self,
+        *,
+        user_id: int,
+        openid: str,
+        unionid: str | None,
+        subscribed: int,
+    ) -> None:
+        db_session = await self.get_session()
+        try:
+            await db_session.execute(
+                text(
+                    """
+                    INSERT INTO parent_wechat_accounts (
+                      user_id,
+                      app_id,
+                      account_type,
+                      openid,
+                      unionid,
+                      subscribed,
+                      created_at,
+                      updated_at
+                    ) VALUES (
+                      :user_id,
+                      :app_id,
+                      :account_type,
+                      :openid,
+                      :unionid,
+                      :subscribed,
+                      CURRENT_TIMESTAMP,
+                      CURRENT_TIMESTAMP
+                    )
+                    ON DUPLICATE KEY UPDATE
+                      user_id = VALUES(user_id),
+                      openid = VALUES(openid),
+                      unionid = COALESCE(VALUES(unionid), unionid),
+                      subscribed = VALUES(subscribed),
+                      updated_at = CURRENT_TIMESTAMP
+                    """
+                ),
+                {
+                    "user_id": user_id,
+                    "app_id": settings.wechat_mp_app_id,
+                    "account_type": WECHAT_MP_ACCOUNT_TYPE,
+                    "openid": openid,
+                    "unionid": unionid,
+                    "subscribed": subscribed,
+                },
+            )
+            await db_session.commit()
+        except Exception:
+            await db_session.rollback()
+            raise
+        finally:
+            await db_session.close()
+
+    async def _create_bind_state(self, *, state: str, user_id: int) -> None:
+        expires_at = datetime.now() + timedelta(minutes=10)
+        db_session = await self.get_session()
+        try:
+            await db_session.execute(
+                text(
+                    """
+                    DELETE FROM wechat_mp_bind_states
+                    WHERE expires_at < CURRENT_TIMESTAMP
+                       OR user_id = :user_id
+                    """
+                ),
+                {"user_id": user_id},
+            )
+            await db_session.execute(
+                text(
+                    """
+                    INSERT INTO wechat_mp_bind_states (
+                      state,
+                      user_id,
+                      expires_at,
+                      created_at,
+                      updated_at
+                    ) VALUES (
+                      :state,
+                      :user_id,
+                      :expires_at,
+                      CURRENT_TIMESTAMP,
+                      CURRENT_TIMESTAMP
+                    )
+                    """
+                ),
+                {
+                    "state": state,
+                    "user_id": user_id,
+                    "expires_at": expires_at,
+                },
+            )
+            await db_session.commit()
+        except Exception:
+            await db_session.rollback()
+            raise
+        finally:
+            await db_session.close()
+
+    async def _consume_bind_state(self, *, state: str) -> int:
+        db_session = await self.get_session()
+        try:
+            result = await db_session.execute(
+                text(
+                    """
+                    SELECT id, user_id
+                    FROM wechat_mp_bind_states
+                    WHERE state = :state
+                      AND consumed_at IS NULL
+                      AND expires_at >= CURRENT_TIMESTAMP
+                    LIMIT 1
+                    """
+                ),
+                {"state": state},
+            )
+            row = result.mappings().first()
+            if not row:
+                raise RuntimeError("invalid or expired wechat mp bind state")
+
+            update_result = await db_session.execute(
+                text(
+                    """
+                    UPDATE wechat_mp_bind_states
+                    SET consumed_at = CURRENT_TIMESTAMP,
+                        updated_at = CURRENT_TIMESTAMP
+                    WHERE id = :id
+                      AND consumed_at IS NULL
+                    """
+                ),
+                {"id": row["id"]},
+            )
+            updated = update_result.rowcount
+            if updated is not None and updated != 1:
+                raise RuntimeError("invalid or expired wechat mp bind state")
+            await db_session.commit()
+            return int(row["user_id"])
+        except Exception:
+            await db_session.rollback()
+            raise
+        finally:
+            await db_session.close()
+
+    async def _get_access_token(self) -> str:
+        now = time.time()
+        if self._access_token and now < self._access_token_expire_at - 60:
+            return self._access_token
+        url = f"{settings.wechat_mp_api_base_url.rstrip('/')}/cgi-bin/token"
+        params = {
+            "grant_type": "client_credential",
+            "appid": settings.wechat_mp_app_id,
+            "secret": settings.wechat_mp_app_secret,
+        }
+        async with httpx.AsyncClient(timeout=settings.wechat_mp_http_timeout_seconds) as client:
+            response = await client.get(url, params=params)
+            response.raise_for_status()
+            data = response.json()
+        access_token = str(data.get("access_token") or "").strip()
+        if not access_token:
+            raise RuntimeError(f"wechat mp access token missing: errcode={data.get('errcode')} errmsg={data.get('errmsg')}")
+        self._access_token = access_token
+        self._access_token_expire_at = now + int(data.get("expires_in") or 7200)
+        return access_token
+
+    async def _oauth_access_token(self, *, code: str) -> Mapping[str, Any]:
+        url = f"{settings.wechat_mp_api_base_url.rstrip('/')}/sns/oauth2/access_token"
+        params = {
+            "appid": settings.wechat_mp_app_id,
+            "secret": settings.wechat_mp_app_secret,
+            "code": code,
+            "grant_type": "authorization_code",
+        }
+        async with httpx.AsyncClient(timeout=settings.wechat_mp_http_timeout_seconds) as client:
+            response = await client.get(url, params=params)
+            response.raise_for_status()
+            data = response.json()
+        if data.get("errcode"):
+            raise RuntimeError(f"wechat mp oauth failed: errcode={data.get('errcode')} errmsg={data.get('errmsg')}")
+        return data
+
+    async def _get_mp_user_info(self, *, openid: str) -> Mapping[str, Any]:
+        access_token = await self._get_access_token()
+        url = f"{settings.wechat_mp_api_base_url.rstrip('/')}/cgi-bin/user/info"
+        params = {
+            "access_token": access_token,
+            "openid": openid,
+            "lang": "zh_CN",
+        }
+        async with httpx.AsyncClient(timeout=settings.wechat_mp_http_timeout_seconds) as client:
+            response = await client.get(url, params=params)
+            response.raise_for_status()
+            data = response.json()
+        if data.get("errcode"):
+            raise RuntimeError(f"wechat mp user info failed: errcode={data.get('errcode')} errmsg={data.get('errmsg')}")
+        return data
+
+    def _template_id(self, notification_type: str) -> str:
+        default_template_id = (settings.wechat_mp_template_id or "").strip()
+        if notification_type == "leave_message":
+            return (settings.wechat_mp_leave_message_template_id or "").strip() or default_template_id
+        if notification_type == "low_battery":
+            return (settings.wechat_mp_low_battery_template_id or "").strip() or default_template_id
+        if notification_type == "sleep_mode":
+            return (settings.wechat_mp_sleep_mode_template_id or "").strip() or default_template_id
+        return default_template_id
+
+    def _is_deduped(self, *, device_id: str, notification_type: str, openid: str, dedup_seconds: int) -> bool:
+        if dedup_seconds <= 0:
+            return False
+        last_sent_at = self._last_sent_at.get((device_id, notification_type, openid))
+        return last_sent_at is not None and time.time() - last_sent_at < dedup_seconds
+
+    def _mark_sent(self, *, device_id: str, notification_type: str, openid: str) -> None:
+        self._last_sent_at[(device_id, notification_type, openid)] = time.time()
+
+    @staticmethod
+    def _clip_template_value(value: str, max_len: int) -> str:
+        text = str(value or "").strip()
+        if len(text) <= max_len:
+            return text
+        return text[:max_len]
+
+
+wechat_mp_notification_service = WechatMpNotificationService()

talkingq-url/config.py

@@ -93,6 +93,58 @@ class Settings(BaseSettings):
         default=5.0,
         validation_alias="WECHAT_HTTP_TIMEOUT_SECONDS",
     )
+    wechat_mp_app_id: str = Field(default="", validation_alias="WECHAT_MP_APP_ID")
+    wechat_mp_app_secret: str = Field(default="", validation_alias="WECHAT_MP_APP_SECRET")
+    wechat_mp_api_base_url: str = Field(
+        default="https://api.weixin.qq.com",
+        validation_alias="WECHAT_MP_API_BASE_URL",
+    )
+    wechat_mp_push_enabled: bool = Field(default=False, validation_alias="WECHAT_MP_PUSH_ENABLED")
+    wechat_mp_http_timeout_seconds: float = Field(
+        default=5.0,
+        validation_alias="WECHAT_MP_HTTP_TIMEOUT_SECONDS",
+    )
+    wechat_mp_template_id: str = Field(default="", validation_alias="WECHAT_MP_TEMPLATE_ID")
+    wechat_mp_leave_message_template_id: str = Field(
+        default="",
+        validation_alias="WECHAT_MP_LEAVE_MESSAGE_TEMPLATE_ID",
+    )
+    wechat_mp_low_battery_template_id: str = Field(
+        default="",
+        validation_alias="WECHAT_MP_LOW_BATTERY_TEMPLATE_ID",
+    )
+    wechat_mp_sleep_mode_template_id: str = Field(
+        default="",
+        validation_alias="WECHAT_MP_SLEEP_MODE_TEMPLATE_ID",
+    )
+    wechat_mp_miniprogram_appid: str = Field(default="", validation_alias="WECHAT_MP_MINIPROGRAM_APPID")
+    wechat_mp_chat_page: str = Field(default="pages/chat/detail/index", validation_alias="WECHAT_MP_CHAT_PAGE")
+    wechat_mp_device_page: str = Field(default="pages/device/index", validation_alias="WECHAT_MP_DEVICE_PAGE")
+    wechat_mp_oauth_redirect_uri: str = Field(default="", validation_alias="WECHAT_MP_OAUTH_REDIRECT_URI")
+    wechat_mp_bind_success_page: str = Field(
+        default="pages/sleep/index",
+        validation_alias="WECHAT_MP_BIND_SUCCESS_PAGE",
+    )
+    wechat_mp_low_battery_threshold: int = Field(
+        default=20,
+        validation_alias="WECHAT_MP_LOW_BATTERY_THRESHOLD",
+    )
+    wechat_mp_low_battery_recovery_threshold: int = Field(
+        default=25,
+        validation_alias="WECHAT_MP_LOW_BATTERY_RECOVERY_THRESHOLD",
+    )
+    wechat_mp_low_battery_dedup_seconds: int = Field(
+        default=21600,
+        validation_alias="WECHAT_MP_LOW_BATTERY_DEDUP_SECONDS",
+    )
+    wechat_mp_sleep_mode_dedup_seconds: int = Field(
+        default=300,
+        validation_alias="WECHAT_MP_SLEEP_MODE_DEDUP_SECONDS",
+    )
+    wechat_mp_leave_message_dedup_seconds: int = Field(
+        default=60,
+        validation_alias="WECHAT_MP_LEAVE_MESSAGE_DEDUP_SECONDS",
+    )
 
     sms_enabled: bool = Field(default=False, validation_alias="SMS_ENABLED")
     sms_provider: str = Field(default="aliyun", validation_alias="SMS_PROVIDER")

talkingq-url/database/init_db.py

@@ -353,6 +353,59 @@ async def _ensure_device_family_tables(conn) -> None:
     )
 
 
+async def _ensure_parent_wechat_accounts_table(conn) -> None:
+    await conn.execute(
+        text(
+            """
+            CREATE TABLE IF NOT EXISTS parent_wechat_accounts (
+              id BIGINT NOT NULL AUTO_INCREMENT,
+              user_id BIGINT NOT NULL,
+              app_id VARCHAR(64) NOT NULL,
+              account_type VARCHAR(32) NOT NULL,
+              openid VARCHAR(64) NOT NULL,
+              unionid VARCHAR(64) NULL,
+              subscribed TINYINT NOT NULL DEFAULT 0,
+              created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+              updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+              PRIMARY KEY (id),
+              UNIQUE KEY uq_parent_wechat_app_openid (app_id, openid),
+              UNIQUE KEY uq_parent_wechat_user_type (user_id, account_type),
+              KEY idx_parent_wechat_user (user_id),
+              KEY idx_parent_wechat_unionid (unionid),
+              CONSTRAINT fk_parent_wechat_user
+                FOREIGN KEY (user_id)
+                REFERENCES parents (user_id)
+            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
+            """
+        )
+    )
+
+
+async def _ensure_wechat_mp_bind_states_table(conn) -> None:
+    await conn.execute(
+        text(
+            """
+            CREATE TABLE IF NOT EXISTS wechat_mp_bind_states (
+              id BIGINT NOT NULL AUTO_INCREMENT,
+              state VARCHAR(128) NOT NULL,
+              user_id BIGINT NOT NULL,
+              expires_at DATETIME NOT NULL,
+              consumed_at DATETIME NULL,
+              created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+              updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+              PRIMARY KEY (id),
+              UNIQUE KEY uq_wechat_mp_bind_state (state),
+              KEY idx_wechat_mp_bind_user (user_id),
+              KEY idx_wechat_mp_bind_expires (expires_at),
+              CONSTRAINT fk_wechat_mp_bind_user
+                FOREIGN KEY (user_id)
+                REFERENCES parents (user_id)
+            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
+            """
+        )
+    )
+
+
 async def init_db():
     """初始化数据库，创建所有表"""
     try:
@@ -374,6 +427,8 @@ async def init_db():
             await _ensure_bind_session_card_columns(conn)
             await _ensure_cards_allow_multiple_per_device(conn)
             await _ensure_device_family_tables(conn)
+            await _ensure_parent_wechat_accounts_table(conn)
+            await _ensure_wechat_mp_bind_states_table(conn)
         await engine.dispose()
         session_logger.info("system", "database", "数据库表已成功创建")
         return True

talkingq-url/database/models.py

@@ -224,6 +224,44 @@ class Parent(Base):
     updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, server_default=text("CURRENT_TIMESTAMP"))
 
 
+class ParentWechatAccount(Base):
+    __tablename__ = "parent_wechat_accounts"
+    __table_args__ = (
+        UniqueConstraint("app_id", "openid", name="uq_parent_wechat_app_openid"),
+        UniqueConstraint("user_id", "account_type", name="uq_parent_wechat_user_type"),
+        Index("idx_parent_wechat_user", "user_id"),
+        Index("idx_parent_wechat_unionid", "unionid"),
+        {'mysql_charset': 'utf8mb4', 'mysql_collate': 'utf8mb4_unicode_ci'}
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    user_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("parents.user_id"), nullable=False)
+    app_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    account_type: Mapped[str] = mapped_column(String(32), nullable=False)
+    openid: Mapped[str] = mapped_column(String(64), nullable=False)
+    unionid: Mapped[Optional[str]] = mapped_column(String(64))
+    subscribed: Mapped[int] = mapped_column(Integer, server_default="0")
+    created_at: Mapped[Optional[datetime]] = mapped_column(DateTime, server_default=text("CURRENT_TIMESTAMP"))
+    updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, server_default=text("CURRENT_TIMESTAMP"))
+
+
+class WechatMpBindState(Base):
+    __tablename__ = "wechat_mp_bind_states"
+    __table_args__ = (
+        Index("idx_wechat_mp_bind_user", "user_id"),
+        Index("idx_wechat_mp_bind_expires", "expires_at"),
+        {'mysql_charset': 'utf8mb4', 'mysql_collate': 'utf8mb4_unicode_ci'}
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    state: Mapped[str] = mapped_column(String(128), unique=True, nullable=False)
+    user_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("parents.user_id"), nullable=False)
+    expires_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
+    consumed_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
+    created_at: Mapped[Optional[datetime]] = mapped_column(DateTime, server_default=text("CURRENT_TIMESTAMP"))
+    updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, server_default=text("CURRENT_TIMESTAMP"))
+
+
 class Child(Base):
     __tablename__ = "children"
 

talkingq-url/handlers/mqtt_handler.py

@@ -13,6 +13,7 @@ from banban.service.im import im_service
 from banban.service.location import location_service
 from banban.service.pending_voice_message import pending_voice_message_service
 from banban.service.sms_notification import sms_notification_service
+from banban.service.wechat_mp_notification import wechat_mp_notification_service
 from config import settings
 from services.card_service import card_service
 from services.device_target_cache import device_target_cache
@@ -184,6 +185,10 @@ class TalkingQMQTTService:
                 device_type=device_type,
             ),
         )
+        wechat_mp_notification_service.schedule_low_battery_notification(
+            device_id=device_id,
+            power=power,
+        )
         if latitude is not None and longitude is not None:
             await self._schedule_persistence(
                 device_id,
@@ -497,7 +502,7 @@ class TalkingQMQTTService:
                     client_msg_id=params.get("client_msg_id"),
                     ext_json=params.get("ext_json") if isinstance(params.get("ext_json"), dict) else None,
                 )
-                sms_notification_service.schedule_leave_message_notification(
+                wechat_mp_notification_service.schedule_leave_message_notification(
                     device_id=device_id,
                     child_name=device_identity.child_name,
                 )

talkingq-url/handlers/websocket_message_handler.py

@@ -22,7 +22,7 @@ from handlers.prompt_sound_handler import handle_prompt_sound_request
 from handlers.session_cleanup_handler import handle_old_session_cleanup
 from config import settings
 from banban.service.im import im_service as im_conversation_service
-from banban.service.sms_notification import sms_notification_service
+from banban.service.wechat_mp_notification import wechat_mp_notification_service
 from utils.audio_format import detect_audio_format, wrap_pcm_as_wav
 from fastapi import HTTPException
 
@@ -302,7 +302,7 @@ async def process_parent_leave_message(device_id: str, audio_cache_key: str):
             },
             audio_content=archive_audio,
         )
-        sms_notification_service.schedule_leave_message_notification(
+        wechat_mp_notification_service.schedule_leave_message_notification(
             device_id=device_id,
             child_name=device_identity.child_name,
         )