From 482fbf351f564120c50e8b0d2736dd569a4c5cf6 Mon Sep 17 00:00:00 2001 From: stu2not Date: Wed, 3 Jun 2026 15:14:01 +0800 Subject: [PATCH] =?UTF-8?q?=E9=9B=86=E6=88=90=E5=AE=B6=E9=95=BF=E7=9F=AD?= =?UTF-8?q?=E4=BF=A1=E9=80=9A=E7=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- banban-mini/src/pages/sleep/index.scss | 36 ++ banban-mini/src/pages/sleep/index.tsx | 82 +++- banban-mini/src/services/auth.ts | 9 +- talkingq-url/banban/routers/parents.py | 31 +- talkingq-url/banban/service/parent.py | 14 +- .../banban/service/sms_notification.py | 385 ++++++++++++++++++ talkingq-url/banban/service/wechat_login.py | 172 +++++++- talkingq-url/config.py | 26 ++ talkingq-url/handlers/mqtt_handler.py | 17 +- .../handlers/websocket_message_handler.py | 7 +- talkingq-url/requirements.txt | 5 +- .../tests/test_parent_phone_authorization.py | 130 ++++++ talkingq-url/tests/test_sms_notification.py | 136 +++++++ 13 files changed, 1027 insertions(+), 23 deletions(-) create mode 100644 talkingq-url/banban/service/sms_notification.py create mode 100644 talkingq-url/tests/test_parent_phone_authorization.py create mode 100644 talkingq-url/tests/test_sms_notification.py diff --git a/banban-mini/src/pages/sleep/index.scss b/banban-mini/src/pages/sleep/index.scss index f45576b..81a90c2 100644 --- a/banban-mini/src/pages/sleep/index.scss +++ b/banban-mini/src/pages/sleep/index.scss @@ -44,6 +44,7 @@ .user-info { flex: 1; + min-width: 0; .user-name { font-size: 34px; @@ -51,6 +52,15 @@ color: #1A1A1A; display: block; } + + .user-phone { + display: block; + margin-top: 8px; + font-size: 24px; + line-height: 1.4; + color: #666666; + word-break: break-all; + } } .verified-badge { @@ -64,6 +74,32 @@ font-weight: 500; } } + + .phone-auth-btn { + display: flex; + align-items: center; + justify-content: center; + flex-shrink: 0; + width: 150px; + height: 60px; + margin: 0; + padding: 0; + border-radius: 14px; + background: #FF8C42; + color: #FFFFFF; + font-size: 24px; + font-weight: 600; + line-height: 60px; + + &::after { + border: none; + } + + &[disabled] { + background: #CBD5E1; + color: #FFFFFF; + } + } } .menu-card { diff --git a/banban-mini/src/pages/sleep/index.tsx b/banban-mini/src/pages/sleep/index.tsx index 41c269a..be07c66 100644 --- a/banban-mini/src/pages/sleep/index.tsx +++ b/banban-mini/src/pages/sleep/index.tsx @@ -1,7 +1,7 @@ import { View, Text, Image, Input, Button } from '@tarojs/components' import { useState } from 'react' import Taro, { useDidShow, useShareAppMessage } from '@tarojs/taro' -import { clearToken, getToken } from '@/services/auth' +import { authorizeParentPhone, clearToken, getParent, getToken, Parent } from '@/services/auth' import { DEVICE_UNAVAILABLE_MESSAGE } from '@/services/api' import { getCurrentUserId } from '@/services/session' import { @@ -72,7 +72,8 @@ export default function Sleep() { const [modalType, setModalType] = useState<'add' | 'edit'>('add') const [childName, setChildName] = useState('') const [editingChildId, setEditingChildId] = useState(null) - const [parentInfo, setParentInfo] = useState<{ nickname?: string; avatar_url?: string }>({}) + const [parentInfo, setParentInfo] = useState(null) + const [isAuthorizingPhone, setIsAuthorizingPhone] = useState(false) useDidShow(() => { void loadData() @@ -95,12 +96,14 @@ export default function Sleep() { setLoading(true) try { const context = await loadCurrentChildBindingContext() + const currentUserId = getCurrentUserId() + const currentParent = currentUserId ? await loadCurrentParent(currentUserId) : null const nextBinding = (context.currentBinding as BindingListItem | null) || null setChildren(context.children) setBindings(context.bindings as BindingListItem[]) setCurrentChild(context.currentChild) setBinding(nextBinding) - setParentInfo(Taro.getStorageSync('userInfo') || {}) + setParentInfo(currentParent) void loadFirmwareStatus(nextBinding?.device_id) void loadRoleData(nextBinding?.device_id) void loadFamilyData(nextBinding?.device_id) @@ -115,6 +118,15 @@ export default function Sleep() { } } + const loadCurrentParent = async (userId: number): Promise => { + try { + return await getParent(userId) + } catch (error) { + console.warn('[manage] parent profile load failed:', error) + return null + } + } + const currentChildName = currentChild?.child_name || '未设置' const loadFirmwareStatus = async (deviceId?: string | null) => { @@ -526,6 +538,45 @@ export default function Sleep() { }) } + const handleAuthorizePhone = async (event: any) => { + if (isAuthorizingPhone) return + const phoneCode = String(event?.detail?.code || '').trim() + if (!phoneCode) { + const errMsg = String(event?.detail?.errMsg || '') + Taro.showToast({ + title: errMsg.includes('deny') || errMsg.includes('fail') ? '未授权手机号' : '未获取到手机号授权', + icon: 'none', + }) + return + } + + setIsAuthorizingPhone(true) + Taro.showLoading({ title: '保存中...' }) + try { + const parent = await authorizeParentPhone(phoneCode) + setParentInfo(parent) + const cachedUserInfo = (Taro.getStorageSync('userInfo') || {}) as { + nickname?: string + avatar_url?: string + } + Taro.setStorageSync('userInfo', { + ...cachedUserInfo, + nickname: parent.nickname || cachedUserInfo.nickname, + avatar_url: parent.avatar_url || cachedUserInfo.avatar_url, + }) + Taro.showToast({ title: '短信通知已开启', icon: 'success' }) + } catch (error: any) { + console.error('[manage] phone authorization failed:', error) + Taro.showToast({ + title: error?.message || '手机号保存失败', + icon: 'none', + }) + } finally { + Taro.hideLoading() + setIsAuthorizingPhone(false) + } + } + const handleMenuClick = (item: MenuItem) => { if (item.disabled) return @@ -589,7 +640,9 @@ export default function Sleep() { ) } - const parentDisplayName = parentInfo.nickname?.trim() || '家长' + const parentDisplayName = parentInfo?.nickname?.trim() || '家长' + const parentPhone = String(parentInfo?.phone || '').trim() + const parentPhoneLabel = parentPhone ? `${parentPhone.slice(0, 3)}****${parentPhone.slice(-4)}` : '未开启' const unassignedBindings = bindings.filter((item) => item.device_id && item.child_id === null) const currentFirmwareLabel = firmwareStatus?.current_version || '--' const latestFirmwareLabel = firmwareStatus?.latest_version || '--' @@ -694,7 +747,7 @@ export default function Sleep() { - {parentInfo.avatar_url ? ( + {parentInfo?.avatar_url ? ( ) : ( 👤 @@ -702,10 +755,23 @@ export default function Sleep() { {parentDisplayName} + 短信通知：{parentPhoneLabel} - - {currentChild ? '已选择当前孩子' : '未选择孩子'} - + {parentPhone ? ( + + 已开启 + + ) : ( + + )} diff --git a/banban-mini/src/services/auth.ts b/banban-mini/src/services/auth.ts index e5f404a..2b4821b 100644 --- a/banban-mini/src/services/auth.ts +++ b/banban-mini/src/services/auth.ts @@ -45,7 +45,7 @@ export async function getParent(userId: number): Promise { export async function updateParent( userId: number, - data: { nickname?: string; avatar_url?: string; phone?: string } + data: { nickname?: string; avatar_url?: string } ): Promise { return request(`/banban/parents/${userId}`, { method: 'PATCH', @@ -53,6 +53,13 @@ export async function updateParent( }) } +export async function authorizeParentPhone(code: string): Promise { + return request('/banban/parents/me/phone', { + method: 'POST', + data: { code }, + }) +} + export function getToken(): string { return getStoredToken() } diff --git a/talkingq-url/banban/routers/parents.py b/talkingq-url/banban/routers/parents.py index a905968..6a1be44 100644 --- a/talkingq-url/banban/routers/parents.py +++ b/talkingq-url/banban/routers/parents.py @@ -7,10 +7,12 @@ try: from banban.service.parent import ParentService from banban.service.avatar_storage import AvatarStorageError from banban.security import get_current_user_id + from banban.service.wechat_login import WechatAuthError, WechatAuthService, get_wechat_auth_service except ModuleNotFoundError: from banban.service.parent import ParentService from banban.service.avatar_storage import AvatarStorageError from banban.security import get_current_user_id + from banban.service.wechat_login import WechatAuthError, WechatAuthService, get_wechat_auth_service router = APIRouter(prefix="/parents", tags=["parents"]) @@ -37,7 +39,10 @@ class ParentResponse(BaseModel): class ParentUpdateRequest(BaseModel): nickname: str | None = None avatar_url: str | None = None - phone: str | None = None + + +class ParentPhoneAuthorizeRequest(BaseModel): + code: str class AvatarDownloadResponse(BaseModel): @@ -78,6 +83,28 @@ async def upload_my_avatar( return ParentResponse(**parent) +@router.post("/me/phone", response_model=ParentResponse) +async def update_my_phone_from_wechat( + payload: ParentPhoneAuthorizeRequest, + request: Request, + current_user_id: int = Depends(get_current_user_id), + wechat_auth_service: WechatAuthService = Depends(get_wechat_auth_service), +) -> ParentResponse: + del request + service = ParentService() + try: + parent = await service.update_phone_from_wechat_code( + user_id=current_user_id, + code=payload.code, + wechat_auth_service=wechat_auth_service, + ) + except WechatAuthError as exc: + raise HTTPException(status_code=exc.status_code, detail=str(exc)) from exc + if not parent: + raise HTTPException(status_code=404, detail="parent not found") + return ParentResponse(**parent) + + @router.get("/{user_id}/avatar", response_model=AvatarDownloadResponse) async def get_parent_avatar(user_id: int, request: Request) -> AvatarDownloadResponse: del request @@ -100,7 +127,7 @@ async def get_parent(user_id: int, request: Request) -> ParentResponse: @router.patch("/{user_id}", response_model=ParentResponse) async def update_parent(user_id: int, payload: ParentUpdateRequest, request: Request) -> ParentResponse: service = ParentService() - parent = await service.update(user_id, payload.nickname, payload.avatar_url, payload.phone) + parent = await service.update(user_id, payload.nickname, payload.avatar_url) if not parent: raise HTTPException(status_code=404, detail="parent not found") return ParentResponse(**parent) diff --git a/talkingq-url/banban/service/parent.py b/talkingq-url/banban/service/parent.py index 2788e82..235dc12 100644 --- a/talkingq-url/banban/service/parent.py +++ b/talkingq-url/banban/service/parent.py @@ -5,6 +5,7 @@ from banban.dao.parent import ParentDAO from banban.service.avatar_storage import AvatarStorageService from config import settings from services.database_service_base import DatabaseServiceBase +from banban.service.wechat_login import WechatAuthService, get_wechat_auth_service class ParentService(DatabaseServiceBase): @@ -50,6 +51,17 @@ class ParentService(DatabaseServiceBase): finally: await db_session.close() + async def update_phone_from_wechat_code( + self, + *, + user_id: int, + code: str, + wechat_auth_service: WechatAuthService | None = None, + ) -> Mapping: + wechat_auth = wechat_auth_service or get_wechat_auth_service() + phone = await wechat_auth.exchange_phone_code(code) + return await self.update(user_id=user_id, phone=phone.pure_phone_number) + async def upload_avatar( self, *, @@ -128,4 +140,4 @@ class ParentService(DatabaseServiceBase): avatar_file_key = data.get("avatar_file_key") if avatar_file_key: data["avatar_url"] = await self.avatar_storage.get_avatar_url(avatar_file_key) - return data \ No newline at end of file + return data diff --git a/talkingq-url/banban/service/sms_notification.py b/talkingq-url/banban/service/sms_notification.py new file mode 100644 index 0000000..d07412d --- /dev/null +++ b/talkingq-url/banban/service/sms_notification.py @@ -0,0 +1,385 @@ +import asyncio +import json +import time +from collections.abc import Mapping +from datetime import datetime +from typing import Any + +from sqlalchemy import text + +from config import settings +from services.database_service_base import DatabaseServiceBase +from utils.logger import session_logger + + +class SmsNotificationService(DatabaseServiceBase): + def __init__(self): + super().__init__(service_name="sms_notification_service") + self._last_sent_at: dict[tuple[str, str, str], float] = {} + + async def notify_alarm( + self, + *, + device_id: str, + alarm_id: int | None = None, + child_name: str | None = None, + address: str | None = None, + ) -> None: + if not self._is_enabled(): + return + + if alarm_id and (not child_name or not address): + context = await self._get_alarm_context(alarm_id=alarm_id, device_id=device_id) + child_name = child_name or context.get("child_name") + address = address or context.get("address") + + recipients = await self._list_device_family_phone_numbers(device_id=device_id) + if not recipients: + session_logger.info(device_id, "sms", "告警短信跳过：没有可用家长手机号") + return + + template_code = self._alarm_template_code() + if not template_code: + session_logger.warning(device_id, "sms", "告警短信跳过：未配置短信模板") + return + + params = self._build_template_params( + child_name=child_name, + address=address, + event_time=self._format_now(), + event_label="设备告警", + ) + await self._send_to_recipients( + device_id=device_id, + notification_type="alarm", + recipients=recipients, + template_code=template_code, + template_params=params, + dedup_seconds=max(0, int(settings.sms_alarm_dedup_seconds or 0)), + ref_id=str(alarm_id or ""), + ) + + async def notify_leave_message( + self, + *, + device_id: str, + child_name: str | None = None, + ) -> None: + if not self._is_enabled(): + return + + recipients = await self._list_device_family_phone_numbers(device_id=device_id) + if not recipients: + session_logger.info(device_id, "sms", "留言短信跳过：没有可用家长手机号") + return + + template_code = self._leave_message_template_code() + if not template_code: + session_logger.warning(device_id, "sms", "留言短信跳过：未配置短信模板") + return + + params = self._build_template_params( + child_name=child_name, + address="小程序", + event_time=self._format_now(), + event_label="设备留言", + ) + await self._send_to_recipients( + device_id=device_id, + notification_type="leave_message", + recipients=recipients, + template_code=template_code, + template_params=params, + dedup_seconds=max(0, int(settings.sms_leave_message_dedup_seconds or 0)), + ref_id="", + ) + + async def notify_alarm_best_effort(self, **kwargs: Any) -> None: + try: + await self.notify_alarm(**kwargs) + except Exception as exc: + session_logger.warning(kwargs.get("device_id") or "", "sms", f"告警短信发送失败，不影响主流程: {exc}") + + 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 "", "sms", f"留言短信发送失败，不影响主流程: {exc}") + + def schedule_alarm_notification(self, **kwargs: Any) -> None: + self._schedule_best_effort(self.notify_alarm_best_effort(**kwargs), device_id=kwargs.get("device_id")) + + 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"), + ) + + 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 "", "sms", f"短信后台任务创建失败，不影响主流程: {exc}") + try: + coro.close() + except Exception: + pass + + async def _send_to_recipients( + self, + *, + device_id: str, + notification_type: str, + recipients: list[str], + template_code: str, + template_params: dict[str, str], + dedup_seconds: int, + ref_id: str, + ) -> None: + for phone_number in recipients: + if self._is_deduped( + device_id=device_id, + notification_type=notification_type, + phone_number=phone_number, + dedup_seconds=dedup_seconds, + ): + session_logger.info(device_id, "sms", f"短信防抖跳过: type={notification_type}, ref={ref_id}") + continue + + result = await self._send_sms( + phone_number=phone_number, + template_code=template_code, + template_params=template_params, + ) + if result.get("ok"): + self._mark_sent( + device_id=device_id, + notification_type=notification_type, + phone_number=phone_number, + ) + session_logger.info( + device_id, + "sms", + f"短信发送成功: type={notification_type}, code={result.get('code')}, request_id={result.get('request_id')}", + ) + else: + session_logger.warning( + device_id, + "sms", + f"短信发送失败: type={notification_type}, code={result.get('code')}, message={result.get('message')}", + ) + + async def _send_sms( + self, + *, + phone_number: str, + template_code: str, + template_params: dict[str, str], + ) -> dict[str, Any]: + if settings.sms_dry_run: + return { + "ok": True, + "code": "DRY_RUN", + "message": "dry run", + "request_id": None, + "biz_id": None, + } + try: + return await asyncio.to_thread( + self._send_aliyun_sms_sync, + phone_number=phone_number, + template_code=template_code, + template_params=template_params, + ) + except Exception as exc: + return { + "ok": False, + "code": getattr(exc, "code", None), + "message": getattr(exc, "message", str(exc)), + "request_id": None, + "biz_id": None, + } + + def _send_aliyun_sms_sync( + self, + *, + phone_number: str, + template_code: str, + template_params: dict[str, str], + ) -> dict[str, Any]: + from alibabacloud_dysmsapi20170525.client import Client as DysmsapiClient + from alibabacloud_dysmsapi20170525 import models as sms_models + from alibabacloud_tea_openapi import models as open_api_models + from alibabacloud_tea_util import models as util_models + + client = DysmsapiClient( + open_api_models.Config( + access_key_id=settings.aliyun_sms_access_key_id, + access_key_secret=settings.aliyun_sms_access_key_secret, + endpoint=settings.aliyun_sms_endpoint or "dysmsapi.aliyuncs.com", + connect_timeout=int(settings.sms_http_timeout_seconds * 1000), + read_timeout=int(settings.sms_http_timeout_seconds * 1000), + ) + ) + request = sms_models.SendSmsRequest( + sign_name=settings.aliyun_sms_sign_name, + template_code=template_code, + phone_numbers=phone_number, + template_param=json.dumps(template_params, ensure_ascii=False), + ) + response = client.send_sms_with_options(request, util_models.RuntimeOptions()) + body = getattr(response, "body", None) + code = getattr(body, "code", None) + return { + "ok": code == "OK", + "code": code, + "message": getattr(body, "message", None), + "request_id": getattr(body, "request_id", None), + "biz_id": getattr(body, "biz_id", None), + } + + async def _list_device_family_phone_numbers(self, *, device_id: str) -> list[str]: + db_session = await self.get_session() + try: + result = await db_session.execute( + text( + """ + SELECT p.phone + 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 + WHERE p.phone IS NOT NULL + AND p.phone <> '' + ORDER BY recipients.sort_role ASC, recipients.sort_time ASC, recipients.sort_id ASC + """ + ), + {"device_id": device_id}, + ) + phone_numbers: list[str] = [] + seen: set[str] = set() + for row in result.mappings().all(): + phone_number = str(row["phone"]).strip() + if not phone_number or phone_number in seen: + continue + seen.add(phone_number) + phone_numbers.append(phone_number) + return phone_numbers + finally: + await db_session.close() + + async def _get_alarm_context(self, *, alarm_id: int, device_id: str) -> Mapping[str, Any]: + db_session = await self.get_session() + try: + result = await db_session.execute( + text( + """ + SELECT + c.child_name, + COALESCE(dae.address, clc.address) AS address, + dae.lat, + dae.lng + FROM device_alarm_events AS dae + LEFT JOIN children AS c + ON c.child_id = dae.child_id + AND c.status = 1 + LEFT JOIN child_location_current AS clc + ON clc.child_id = dae.child_id + WHERE dae.alarm_id = :alarm_id + AND dae.device_id = :device_id + LIMIT 1 + """ + ), + {"alarm_id": alarm_id, "device_id": device_id}, + ) + row = result.mappings().first() + if not row: + return {} + data = dict(row) + if not data.get("address") and data.get("lat") is not None and data.get("lng") is not None: + data["address"] = f"{data['lat']},{data['lng']}" + return data + finally: + await db_session.close() + + def _is_enabled(self) -> bool: + if not settings.sms_enabled: + return False + if (settings.sms_provider or "").strip().lower() != "aliyun": + session_logger.warning("system", "sms", f"短信服务未启用：不支持的 provider={settings.sms_provider}") + return False + required = ( + settings.aliyun_sms_access_key_id, + settings.aliyun_sms_access_key_secret, + settings.aliyun_sms_sign_name, + self._default_template_code() or self._alarm_template_code() or self._leave_message_template_code(), + ) + if any(not str(value or "").strip() for value in required): + session_logger.warning("system", "sms", "短信服务未启用：阿里云短信配置不完整") + return False + return True + + def _default_template_code(self) -> str: + return (settings.aliyun_sms_template_code or "").strip() + + def _alarm_template_code(self) -> str: + return (settings.aliyun_sms_alarm_template_code or "").strip() or self._default_template_code() + + def _leave_message_template_code(self) -> str: + return (settings.aliyun_sms_leave_message_template_code or "").strip() or self._default_template_code() + + def _build_template_params( + self, + *, + child_name: str | None, + address: str | None, + event_time: str, + event_label: str, + ) -> dict[str, str]: + return { + "code": event_label, + "conference": (child_name or "伴伴设备")[:20], + "address": (address or "请打开小程序查看")[:60], + "time": event_time, + } + + def _format_now(self) -> str: + return datetime.now().strftime("%Y-%m-%d %H:%M") + + def _is_deduped( + self, + *, + device_id: str, + notification_type: str, + phone_number: str, + dedup_seconds: int, + ) -> bool: + if dedup_seconds <= 0: + return False + last_sent_at = self._last_sent_at.get((device_id, notification_type, phone_number)) + 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, phone_number: str) -> None: + self._last_sent_at[(device_id, notification_type, phone_number)] = time.time() + + +sms_notification_service = SmsNotificationService() diff --git a/talkingq-url/banban/service/wechat_login.py b/talkingq-url/banban/service/wechat_login.py index f824053..a444189 100644 --- a/talkingq-url/banban/service/wechat_login.py +++ b/talkingq-url/banban/service/wechat_login.py @@ -1,5 +1,5 @@ -import asyncio import logging +import time from dataclasses import dataclass import httpx @@ -14,6 +14,7 @@ logger = logging.getLogger("app.wechat_login") INVALID_CODE_ERRCODES = {40029, 40163} MISCONFIGURED_APP_ERRCODES = {40013, 40125} +ACCESS_TOKEN_EXPIRY_SKEW_SECONDS = 300 @dataclass(frozen=True) @@ -23,6 +24,13 @@ class WechatCodeSession: unionid: str | None = None +@dataclass(frozen=True) +class WechatPhoneNumber: + phone_number: str + pure_phone_number: str + country_code: str | None = None + + class WechatAuthError(Exception): def __init__(self, detail: str, *, status_code: int, errcode: int | None = None): super().__init__(detail) @@ -31,6 +39,9 @@ class WechatAuthError(Exception): class WechatAuthService: + _cached_access_token: str | None = None + _cached_access_token_expires_at: float = 0.0 + def __init__(self, client: httpx.AsyncClient | None = None): self._client = client @@ -67,6 +78,95 @@ class WechatAuthService: return WechatCodeSession(openid=openid, session_key=session_key, unionid=unionid) + async def exchange_phone_code(self, code: str) -> WechatPhoneNumber: + if not settings.wechat_app_id or not settings.wechat_app_secret: + raise WechatAuthError("wechat phone number is not configured", status_code=503) + + normalized_code = str(code or "").strip() + if not normalized_code: + raise WechatAuthError("wechat phone code is required", status_code=400) + + access_token = await self._get_access_token() + response = await self._request_phone_number(access_token=access_token, code=normalized_code) + data = self._parse_response_json(response) + + errcode = self._parse_errcode(data.get("errcode")) + if errcode not in (None, 0): + errmsg = data.get("errmsg") + logger.warning( + "wechat phone number exchange rejected code", + extra={ + "event": "wechat_phone_number_rejected", + "errcode": errcode, + "errmsg": errmsg, + }, + ) + raise self._map_phone_error(errcode) + + phone_info = data.get("phone_info") + if not isinstance(phone_info, dict): + raise WechatAuthError("wechat phone response missing phone_info", status_code=502) + + phone_number = phone_info.get("phoneNumber") + pure_phone_number = phone_info.get("purePhoneNumber") + country_code = phone_info.get("countryCode") + if not isinstance(phone_number, str) or not phone_number: + raise WechatAuthError("wechat phone response missing phone number", status_code=502) + if not isinstance(pure_phone_number, str) or not pure_phone_number: + pure_phone_number = phone_number + if not isinstance(country_code, str) or not country_code: + country_code = None + + return WechatPhoneNumber( + phone_number=phone_number, + pure_phone_number=pure_phone_number, + country_code=country_code, + ) + + async def _get_access_token(self) -> str: + if self._client is None: + cached_token = self._get_cached_access_token() + if cached_token: + return cached_token + + response = await self._request_access_token() + data = self._parse_response_json(response) + errcode = self._parse_errcode(data.get("errcode")) + if errcode not in (None, 0): + errmsg = data.get("errmsg") + logger.warning( + "wechat access token request failed", + extra={ + "event": "wechat_access_token_failed", + "errcode": errcode, + "errmsg": errmsg, + }, + ) + raise self._map_access_token_error(errcode) + + access_token = data.get("access_token") + if not isinstance(access_token, str) or not access_token: + raise WechatAuthError("wechat access token response missing token", status_code=502) + if self._client is None: + type(self)._cached_access_token = access_token + type(self)._cached_access_token_expires_at = self._calculate_access_token_expires_at(data.get("expires_in")) + return access_token + + def _get_cached_access_token(self) -> str | None: + token = type(self)._cached_access_token + if token and time.time() < type(self)._cached_access_token_expires_at: + return token + return None + + def _calculate_access_token_expires_at(self, expires_in: object) -> float: + try: + ttl = int(expires_in) + except (TypeError, ValueError): + return 0.0 + if ttl <= 0: + return 0.0 + return time.time() + max(0, ttl - ACCESS_TOKEN_EXPIRY_SKEW_SECONDS) + async def _request_code2session(self, code: str) -> httpx.Response: params = { "appid": settings.wechat_app_id, @@ -97,6 +197,62 @@ class WechatAuthService: if owns_client: await client.aclose() + async def _request_access_token(self) -> httpx.Response: + params = { + "grant_type": "client_credential", + "appid": settings.wechat_app_id, + "secret": settings.wechat_app_secret, + } + + client = self._client + owns_client = client is None + if client is None: + client = httpx.AsyncClient( + base_url=settings.wechat_api_base_url.rstrip("/"), + timeout=settings.wechat_http_timeout_seconds, + ) + + try: + response = await client.get("/cgi-bin/token", params=params) + response.raise_for_status() + return response + except httpx.HTTPError as exc: + logger.warning( + "wechat access token request failed", + extra={"event": "wechat_access_token_request_failed"}, + ) + raise WechatAuthError("wechat phone number service unavailable", status_code=502) from exc + finally: + if owns_client: + await client.aclose() + + async def _request_phone_number(self, *, access_token: str, code: str) -> httpx.Response: + client = self._client + owns_client = client is None + if client is None: + client = httpx.AsyncClient( + base_url=settings.wechat_api_base_url.rstrip("/"), + timeout=settings.wechat_http_timeout_seconds, + ) + + try: + response = await client.post( + "/wxa/business/getuserphonenumber", + params={"access_token": access_token}, + json={"code": code}, + ) + response.raise_for_status() + return response + except httpx.HTTPError as exc: + logger.warning( + "wechat phone number request failed", + extra={"event": "wechat_phone_number_request_failed"}, + ) + raise WechatAuthError("wechat phone number service unavailable", status_code=502) from exc + finally: + if owns_client: + await client.aclose() + def _parse_response_json(self, response: httpx.Response) -> dict: try: data = response.json() @@ -121,6 +277,18 @@ class WechatAuthService: return WechatAuthError("wechat login is not configured correctly", status_code=503, errcode=errcode) return WechatAuthError("wechat login service unavailable", status_code=502, errcode=errcode) + def _map_access_token_error(self, errcode: int) -> WechatAuthError: + if errcode in MISCONFIGURED_APP_ERRCODES: + return WechatAuthError("wechat phone number is not configured correctly", status_code=503, errcode=errcode) + return WechatAuthError("wechat phone number service unavailable", status_code=502, errcode=errcode) + + def _map_phone_error(self, errcode: int) -> WechatAuthError: + if errcode in INVALID_CODE_ERRCODES: + return WechatAuthError("invalid or expired wechat phone code", status_code=401, errcode=errcode) + if errcode in MISCONFIGURED_APP_ERRCODES: + return WechatAuthError("wechat phone number is not configured correctly", status_code=503, errcode=errcode) + return WechatAuthError("wechat phone number service unavailable", status_code=502, errcode=errcode) + def get_wechat_auth_service() -> WechatAuthService: - return WechatAuthService() \ No newline at end of file + return WechatAuthService() diff --git a/talkingq-url/config.py b/talkingq-url/config.py index b768042..ffe9152 100644 --- a/talkingq-url/config.py +++ b/talkingq-url/config.py @@ -91,6 +91,32 @@ class Settings(BaseSettings): validation_alias="WECHAT_HTTP_TIMEOUT_SECONDS", ) + sms_enabled: bool = Field(default=False, validation_alias="SMS_ENABLED") + sms_provider: str = Field(default="aliyun", validation_alias="SMS_PROVIDER") + sms_dry_run: bool = Field(default=False, validation_alias="SMS_DRY_RUN") + sms_http_timeout_seconds: float = Field(default=5.0, validation_alias="SMS_HTTP_TIMEOUT_SECONDS") + sms_alarm_dedup_seconds: int = Field(default=300, validation_alias="SMS_ALARM_DEDUP_SECONDS") + sms_leave_message_dedup_seconds: int = Field( + default=60, + validation_alias="SMS_LEAVE_MESSAGE_DEDUP_SECONDS", + ) + aliyun_sms_access_key_id: str = Field(default="", validation_alias="ALIYUN_SMS_ACCESS_KEY_ID") + aliyun_sms_access_key_secret: str = Field(default="", validation_alias="ALIYUN_SMS_ACCESS_KEY_SECRET") + aliyun_sms_sign_name: str = Field(default="", validation_alias="ALIYUN_SMS_SIGN_NAME") + aliyun_sms_template_code: str = Field(default="", validation_alias="ALIYUN_SMS_TEMPLATE_CODE") + aliyun_sms_alarm_template_code: str = Field( + default="", + validation_alias="ALIYUN_SMS_ALARM_TEMPLATE_CODE", + ) + aliyun_sms_leave_message_template_code: str = Field( + default="", + validation_alias="ALIYUN_SMS_LEAVE_MESSAGE_TEMPLATE_CODE", + ) + aliyun_sms_endpoint: str = Field( + default="dysmsapi.aliyuncs.com", + validation_alias="ALIYUN_SMS_ENDPOINT", + ) + cos_secret_id: str = Field(default="", validation_alias="COS_SECRET_ID") cos_secret_key: str = Field(default="", validation_alias="COS_SECRET_KEY") cos_region: str = Field(default="", validation_alias="COS_REGION") diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index 28bee31..9a89780 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/talkingq-url/handlers/mqtt_handler.py @@ -12,6 +12,7 @@ from banban.service.device_setting import device_setting_service 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 config import settings from services.card_service import card_service from services.device_target_cache import device_target_cache @@ -367,11 +368,11 @@ class TalkingQMQTTService: async def _handle_alarm_report(self, device_id: str, payload: dict): logger.info(device_id, "", f"[告警] 设备 {device_id} 发送紧急报警") - await self._schedule_persistence( - device_id, - "alarm_event", - device_alarm_service.record_alarm_event(device_id=device_id, source_msg_id="010"), - ) + async def _record_alarm_and_notify(): + alarm_id = await device_alarm_service.record_alarm_event(device_id=device_id, source_msg_id="010") + sms_notification_service.schedule_alarm_notification(device_id=device_id, alarm_id=alarm_id) + + await self._schedule_persistence(device_id, "alarm_event", _record_alarm_and_notify()) await self._publish(f"device/{device_id}/event_resp", {"msg_id": "010", "status": "success"}) async def _handle_short_press_message(self, device_id: str, payload: dict): @@ -394,7 +395,7 @@ class TalkingQMQTTService: media_file_key = str(params.get("media_file_key") or params.get("audio_url") or "").strip() if media_file_key: try: - await im_service.create_device_parent_leave_message( + device_identity, _ = await im_service.create_device_parent_leave_message( device_id=device_id, media_file_key=media_file_key, media_duration_ms=params.get("media_duration_ms"), @@ -404,6 +405,10 @@ 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( + device_id=device_id, + child_name=device_identity.child_name, + ) logger.info(device_id, "", f"[短按留言] 设备 {device_id} 留言已写入家长会话") await self._publish( f"device/{device_id}/event_resp", diff --git a/talkingq-url/handlers/websocket_message_handler.py b/talkingq-url/handlers/websocket_message_handler.py index ccc8e8f..a9222bc 100644 --- a/talkingq-url/handlers/websocket_message_handler.py +++ b/talkingq-url/handlers/websocket_message_handler.py @@ -22,6 +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 utils.audio_format import detect_audio_format, wrap_pcm_as_wav from fastapi import HTTPException @@ -287,7 +288,7 @@ async def process_parent_leave_message(device_id: str, audio_cache_key: str): content_type=media_mime_type, extension=extension, ) - await im_conversation_service.create_device_parent_leave_message( + device_identity, _ = await im_conversation_service.create_device_parent_leave_message( device_id=device_id, media_file_key=stored_audio.file_key, media_mime_type=media_mime_type, @@ -300,6 +301,10 @@ async def process_parent_leave_message(device_id: str, audio_cache_key: str): }, audio_content=archive_audio, ) + sms_notification_service.schedule_leave_message_notification( + device_id=device_id, + child_name=device_identity.child_name, + ) session_logger.info( device_id, "parent", diff --git a/talkingq-url/requirements.txt b/talkingq-url/requirements.txt index cf6ea49..ca09a55 100644 --- a/talkingq-url/requirements.txt +++ b/talkingq-url/requirements.txt @@ -1,4 +1,5 @@ -aiofiles==25.1.0 +aiofiles==24.1.0 +alibabacloud_dysmsapi20170525==4.5.0 aiohttp fastapi==0.136.1 # pycld2==0.41 @@ -24,4 +25,4 @@ PyJWT==2.10.1 httpx==0.28.1 cos-python-sdk-v5==1.9.41 pytest==8.3.4 -pytest-asyncio==0.24.0 \ No newline at end of file +pytest-asyncio==0.24.0 diff --git a/talkingq-url/tests/test_parent_phone_authorization.py b/talkingq-url/tests/test_parent_phone_authorization.py new file mode 100644 index 0000000..1a62144 --- /dev/null +++ b/talkingq-url/tests/test_parent_phone_authorization.py @@ -0,0 +1,130 @@ +from dataclasses import dataclass + +import httpx +import pytest + +from banban.service.parent import ParentService +from banban.service.wechat_login import WechatAuthService + + +class FakeSession: + async def close(self): + pass + + +@dataclass(frozen=True) +class FakeWechatPhone: + phone_number: str + pure_phone_number: str + country_code: str | None = None + + +class FakeWechatAuthService: + def __init__(self): + self.codes = [] + + async def exchange_phone_code(self, code): + self.codes.append(code) + return FakeWechatPhone( + phone_number="+8613800138000", + pure_phone_number="13800138000", + country_code="86", + ) + + +@pytest.mark.asyncio +async def test_update_phone_from_wechat_code_persists_current_parent_phone(monkeypatch): + service = ParentService() + wechat_auth_service = FakeWechatAuthService() + updates = [] + + async def fake_get_session(): + return FakeSession() + + async def fake_update(self, user_id, nickname=None, avatar_url=None, phone=None): + updates.append( + { + "user_id": user_id, + "nickname": nickname, + "avatar_url": avatar_url, + "phone": phone, + } + ) + + async def fake_get(user_id): + return { + "user_id": user_id, + "openid": "openid_demo", + "unionid": None, + "nickname": "家长", + "avatar_url": None, + "phone": "13800138000", + "status": 1, + } + + monkeypatch.setattr(service, "get_session", fake_get_session) + monkeypatch.setattr("banban.service.parent.ParentDAO.update", fake_update) + monkeypatch.setattr(service, "get", fake_get) + + parent = await service.update_phone_from_wechat_code( + user_id=9, + code="phone-code-demo", + wechat_auth_service=wechat_auth_service, + ) + + assert wechat_auth_service.codes == ["phone-code-demo"] + assert updates == [ + { + "user_id": 9, + "nickname": None, + "avatar_url": None, + "phone": "13800138000", + } + ] + assert parent["phone"] == "13800138000" + + +@pytest.mark.asyncio +async def test_exchange_phone_code_reuses_cached_access_token(monkeypatch): + requests = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append((request.method, request.url.path, str(request.url.params))) + if request.url.path == "/cgi-bin/token": + return httpx.Response(200, json={"access_token": "wechat-access-token", "expires_in": 7200}) + if request.url.path == "/wxa/business/getuserphonenumber": + return httpx.Response( + 200, + json={ + "errcode": 0, + "phone_info": { + "phoneNumber": "+8613800138000", + "purePhoneNumber": "13800138000", + "countryCode": "86", + }, + }, + ) + return httpx.Response(404, json={"errcode": 404}) + + monkeypatch.setattr(WechatAuthService, "_cached_access_token", None) + monkeypatch.setattr(WechatAuthService, "_cached_access_token_expires_at", 0) + monkeypatch.setattr("banban.service.wechat_login.settings.wechat_app_id", "wechat-app-id") + monkeypatch.setattr("banban.service.wechat_login.settings.wechat_app_secret", "wechat-app-secret") + monkeypatch.setattr("banban.service.wechat_login.settings.wechat_api_base_url", "https://api.weixin.qq.com") + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="https://api.weixin.qq.com") as client: + uncached_service = WechatAuthService() + uncached_service._request_access_token = WechatAuthService(client)._request_access_token + uncached_service._request_phone_number = WechatAuthService(client)._request_phone_number + + first = await uncached_service.exchange_phone_code("phone-code-1") + second = await uncached_service.exchange_phone_code("phone-code-2") + + assert first.pure_phone_number == "13800138000" + assert second.pure_phone_number == "13800138000" + token_requests = [item for item in requests if item[1] == "/cgi-bin/token"] + phone_requests = [item for item in requests if item[1] == "/wxa/business/getuserphonenumber"] + assert len(token_requests) == 1 + assert len(phone_requests) == 2 + assert all("access_token=wechat-access-token" in item[2] for item in phone_requests) diff --git a/talkingq-url/tests/test_sms_notification.py b/talkingq-url/tests/test_sms_notification.py new file mode 100644 index 0000000..05f99c0 --- /dev/null +++ b/talkingq-url/tests/test_sms_notification.py @@ -0,0 +1,136 @@ +import pytest + +from banban.service.sms_notification import SmsNotificationService + + +class FakeSession: + async def execute(self, statement, params): + del statement, params + return FakeResult() + + async def close(self): + pass + + +class FakeResult: + def mappings(self): + return self + + def all(self): + return [ + {"phone": "13800138000"}, + {"phone": "13800138000"}, + {"phone": ""}, + {"phone": "13900139000"}, + ] + + def first(self): + return { + "child_name": "孩子", + "address": "杭州市", + "lat": None, + "lng": None, + } + + +def configure_enabled_sms(monkeypatch): + monkeypatch.setattr("banban.service.sms_notification.settings.sms_enabled", True) + monkeypatch.setattr("banban.service.sms_notification.settings.sms_provider", "aliyun") + monkeypatch.setattr("banban.service.sms_notification.settings.sms_dry_run", False) + monkeypatch.setattr("banban.service.sms_notification.settings.sms_alarm_dedup_seconds", 300) + monkeypatch.setattr("banban.service.sms_notification.settings.sms_leave_message_dedup_seconds", 60) + monkeypatch.setattr("banban.service.sms_notification.settings.aliyun_sms_access_key_id", "access-key-id") + monkeypatch.setattr("banban.service.sms_notification.settings.aliyun_sms_access_key_secret", "access-key-secret") + monkeypatch.setattr("banban.service.sms_notification.settings.aliyun_sms_sign_name", "短信签名") + monkeypatch.setattr("banban.service.sms_notification.settings.aliyun_sms_template_code", "SMS_507185027") + monkeypatch.setattr("banban.service.sms_notification.settings.aliyun_sms_alarm_template_code", "") + monkeypatch.setattr("banban.service.sms_notification.settings.aliyun_sms_leave_message_template_code", "") + + +@pytest.mark.asyncio +async def test_notify_alarm_sends_to_unique_family_phone_numbers(monkeypatch): + configure_enabled_sms(monkeypatch) + service = SmsNotificationService() + sent = [] + + async def fake_get_session(): + return FakeSession() + + async def fake_send_sms(*, phone_number, template_code, template_params): + sent.append( + { + "phone_number": phone_number, + "template_code": template_code, + "template_params": template_params, + } + ) + return {"ok": True, "code": "OK", "message": "OK", "request_id": "request-id"} + + monkeypatch.setattr(service, "get_session", fake_get_session) + monkeypatch.setattr(service, "_send_sms", fake_send_sms) + + await service.notify_alarm( + device_id="TalkingQ_XQSN00001005", + alarm_id=7, + child_name="孩子", + address="杭州市", + ) + + assert [item["phone_number"] for item in sent] == ["13800138000", "13900139000"] + assert all(item["template_code"] == "SMS_507185027" for item in sent) + assert sent[0]["template_params"]["conference"] == "孩子" + assert sent[0]["template_params"]["address"] == "杭州市" + + +@pytest.mark.asyncio +async def test_notify_alarm_dedupes_same_device_and_phone(monkeypatch): + configure_enabled_sms(monkeypatch) + service = SmsNotificationService() + sent = [] + + async def fake_get_session(): + return FakeSession() + + async def fake_send_sms(*, phone_number, template_code, template_params): + del template_code, template_params + sent.append(phone_number) + return {"ok": True, "code": "OK", "message": "OK", "request_id": "request-id"} + + monkeypatch.setattr(service, "get_session", fake_get_session) + monkeypatch.setattr(service, "_send_sms", fake_send_sms) + + await service.notify_alarm(device_id="TalkingQ_XQSN00001005", alarm_id=7) + await service.notify_alarm(device_id="TalkingQ_XQSN00001005", alarm_id=8) + + assert sent == ["13800138000", "13900139000"] + + +@pytest.mark.asyncio +async def test_notify_leave_message_disabled_skips_database_and_send(monkeypatch): + monkeypatch.setattr("banban.service.sms_notification.settings.sms_enabled", False) + service = SmsNotificationService() + + async def fail_get_session(): + raise AssertionError("disabled SMS should not query database") + + monkeypatch.setattr(service, "get_session", fail_get_session) + + await service.notify_leave_message(device_id="TalkingQ_XQSN00001005", child_name="孩子") + + +@pytest.mark.asyncio +async def test_best_effort_swallows_send_errors(monkeypatch): + configure_enabled_sms(monkeypatch) + service = SmsNotificationService() + + async def fake_get_session(): + return FakeSession() + + async def fake_send_sms(*, phone_number, template_code, template_params): + del phone_number, template_code, template_params + raise RuntimeError("provider failed") + + monkeypatch.setattr(service, "get_session", fake_get_session) + monkeypatch.setattr(service, "_send_sms", fake_send_sms) + + await service.notify_alarm_best_effort(device_id="TalkingQ_XQSN00001005", alarm_id=7)