diff --git a/banban-mini/src/app.config.ts b/banban-mini/src/app.config.ts index cdd6667..4947029 100644 --- a/banban-mini/src/app.config.ts +++ b/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", diff --git a/banban-mini/src/pages/sleep/index.tsx b/banban-mini/src/pages/sleep/index.tsx index 8a85dbe..7f6117f 100644 --- a/banban-mini/src/pages/sleep/index.tsx +++ b/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(null) const [familyMemberDisplayName, setFamilyMemberDisplayName] = useState('') const [parentInfo, setParentInfo] = useState(null) + const [wechatMpStatus, setWechatMpStatus] = useState(null) + const [isLoadingWechatMp, setIsLoadingWechatMp] = useState(false) + const [isStartingWechatMpBind, setIsStartingWechatMpBind] = useState(false) const [isAuthorizingPhone, setIsAuthorizingPhone] = useState(false) const firmwareStatusPollTimerRef = useRef | 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 => { 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', diff --git a/banban-mini/src/pages/webview/index.config.ts b/banban-mini/src/pages/webview/index.config.ts new file mode 100644 index 0000000..9cd43e3 --- /dev/null +++ b/banban-mini/src/pages/webview/index.config.ts @@ -0,0 +1,3 @@ +export default definePageConfig({ + navigationBarTitleText: '服务号通知', +}) diff --git a/banban-mini/src/pages/webview/index.scss b/banban-mini/src/pages/webview/index.scss new file mode 100644 index 0000000..3ae85e6 --- /dev/null +++ b/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; +} diff --git a/banban-mini/src/pages/webview/index.tsx b/banban-mini/src/pages/webview/index.tsx new file mode 100644 index 0000000..a7f0abb --- /dev/null +++ b/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 ( + + 链接无效 + + ) + } + + return ( + + + + ) +} diff --git a/banban-mini/src/services/wechat-mp.ts b/banban-mini/src/services/wechat-mp.ts new file mode 100644 index 0000000..050ed98 --- /dev/null +++ b/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 { + return request('/banban/wechat-mp/bind/status') +} + +export async function startWechatMpBind(): Promise { + return request('/banban/wechat-mp/bind/start', { + method: 'POST', + }) +} diff --git a/talkingq-url/banban/middleware/auth.py b/talkingq-url/banban/middleware/auth.py index 578adb3..55858ca 100644 --- a/talkingq-url/banban/middleware/auth.py +++ b/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", ) diff --git a/talkingq-url/banban/routers/__init__.py b/talkingq-url/banban/routers/__init__.py index ec6643a..898365c 100644 --- a/talkingq-url/banban/routers/__init__.py +++ b/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"]) diff --git a/talkingq-url/banban/routers/wechat_mp.py b/talkingq-url/banban/routers/wechat_mp.py new file mode 100644 index 0000000..e5e3b6f --- /dev/null +++ b/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""" + + + + + {escaped_title} + + + +
+
+

{escaped_title}

+

{escaped_message}

+

返回路径：{bind_success_page}

+
+
+ +""" + return HTMLResponse(content=body) diff --git a/talkingq-url/banban/service/device.py b/talkingq-url/banban/service/device.py index a77434c..a1fe7cc 100644 --- a/talkingq-url/banban/service/device.py +++ b/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: diff --git a/talkingq-url/banban/service/wechat_mp_notification.py b/talkingq-url/banban/service/wechat_mp_notification.py new file mode 100644 index 0000000..a3f8b28 --- /dev/null +++ b/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() diff --git a/talkingq-url/config.py b/talkingq-url/config.py index b9f00ba..b4c1b48 100644 --- a/talkingq-url/config.py +++ b/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") diff --git a/talkingq-url/database/init_db.py b/talkingq-url/database/init_db.py index b44bcbd..d9101bc 100644 --- a/talkingq-url/database/init_db.py +++ b/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 diff --git a/talkingq-url/database/models.py b/talkingq-url/database/models.py index 59fe8ea..59f811d 100644 --- a/talkingq-url/database/models.py +++ b/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" diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index 3e0c4f8..658d7a6 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/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, ) diff --git a/talkingq-url/handlers/websocket_message_handler.py b/talkingq-url/handlers/websocket_message_handler.py index 797e660..3501bbe 100644 --- a/talkingq-url/handlers/websocket_message_handler.py +++ b/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, )