855 lines
33 KiB
Python
855 lines
33 KiB
Python
import asyncio
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import secrets
|
||
import time
|
||
from collections.abc import Mapping
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timedelta
|
||
from typing import Any
|
||
from urllib.parse import urlencode
|
||
|
||
import httpx
|
||
from fastapi import HTTPException
|
||
from sqlalchemy import text
|
||
|
||
from banban.service.im import DeviceParentLeaveMessageTarget
|
||
from config import settings
|
||
from services.database_service_base import DatabaseServiceBase
|
||
from utils.logger import session_logger
|
||
|
||
|
||
WECHAT_MP_ACCOUNT_TYPE = "service_account"
|
||
WECHAT_MP_NOTIFICATION_TOKEN_VERSION = 1
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class WechatMpRecipient:
|
||
user_id: int
|
||
openid: str
|
||
|
||
|
||
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()
|
||
|
||
async def resolve_notification_link(self, *, token: str, user_id: int) -> Mapping[str, Any]:
|
||
payload = self._decode_notification_token(token)
|
||
if payload.get("type") != "leave_message":
|
||
raise HTTPException(status_code=400, detail="unsupported notification type")
|
||
if int(payload.get("user_id") or 0) != user_id:
|
||
raise HTTPException(status_code=403, detail="notification does not belong to current user")
|
||
|
||
conversation_id = int(payload.get("conversation_id") or 0)
|
||
message_id = int(payload.get("message_id") or 0)
|
||
child_id = int(payload.get("child_id") or 0)
|
||
device_id = str(payload.get("device_id") or "").strip()
|
||
if conversation_id <= 0 or message_id <= 0 or child_id <= 0:
|
||
raise HTTPException(status_code=400, detail="invalid notification token")
|
||
|
||
db_session = await self.get_session()
|
||
try:
|
||
user_id_str = str(user_id)
|
||
child_id_str = str(child_id)
|
||
result = await db_session.execute(
|
||
text(
|
||
"""
|
||
SELECT
|
||
c.id AS conversation_id,
|
||
ch.child_id,
|
||
ch.child_name,
|
||
db.device_id
|
||
FROM im_conversations AS c
|
||
JOIN im_messages AS m
|
||
ON m.id = :message_id
|
||
AND m.conversation_id = c.id
|
||
AND m.deleted_at IS NULL
|
||
JOIN children AS ch
|
||
ON ch.child_id = :child_id
|
||
AND ch.status = 1
|
||
LEFT JOIN parent_child_relations AS pcr
|
||
ON pcr.child_id = ch.child_id
|
||
AND pcr.user_id = :user_id
|
||
AND pcr.status = 1
|
||
LEFT JOIN device_bindings AS db
|
||
ON db.child_id = ch.child_id
|
||
AND db.status = 1
|
||
LEFT JOIN device_family_members AS dfm
|
||
ON dfm.device_id = db.device_id
|
||
AND dfm.user_id = :user_id
|
||
AND dfm.status = 1
|
||
WHERE c.id = :conversation_id
|
||
AND c.status = 1
|
||
AND c.conversation_type = 2
|
||
AND (
|
||
(
|
||
c.participant_a_type = 2
|
||
AND c.participant_a_id = :child_id_str
|
||
AND c.participant_b_type = 1
|
||
AND c.participant_b_id = :user_id_str
|
||
)
|
||
OR (
|
||
c.participant_b_type = 2
|
||
AND c.participant_b_id = :child_id_str
|
||
AND c.participant_a_type = 1
|
||
AND c.participant_a_id = :user_id_str
|
||
)
|
||
)
|
||
AND (
|
||
pcr.id IS NOT NULL
|
||
OR db.owner_user_id = :user_id
|
||
OR dfm.id IS NOT NULL
|
||
)
|
||
LIMIT 1
|
||
"""
|
||
),
|
||
{
|
||
"message_id": message_id,
|
||
"conversation_id": conversation_id,
|
||
"child_id": child_id,
|
||
"child_id_str": child_id_str,
|
||
"user_id": user_id,
|
||
"user_id_str": user_id_str,
|
||
},
|
||
)
|
||
row = result.mappings().first()
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="notification target not found")
|
||
|
||
resolved_child_name = str(row.get("child_name") or payload.get("child_name") or "").strip()
|
||
resolved_device_id = str(row.get("device_id") or device_id).strip()
|
||
return {
|
||
"route": "pages/chat/detail/index",
|
||
"params": {
|
||
"id": conversation_id,
|
||
"source": "im",
|
||
"name": "家长沟通",
|
||
"peerKind": "parent",
|
||
"roleKey": "",
|
||
"conversationTypeName": "parent_child",
|
||
"peerId": user_id_str,
|
||
"childId": child_id,
|
||
"childName": resolved_child_name,
|
||
"parentUserId": user_id,
|
||
"deviceId": resolved_device_id,
|
||
"channelLabel": "微信小程序",
|
||
"canSend": "1",
|
||
"messageId": message_id,
|
||
},
|
||
}
|
||
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,
|
||
targets: list[DeviceParentLeaveMessageTarget | Mapping[str, Any]] | None = None,
|
||
) -> None:
|
||
if not self.is_configured():
|
||
return
|
||
recipients = await self._list_device_family_mp_recipients(device_id=device_id)
|
||
if not recipients:
|
||
session_logger.info(device_id, "wechat_mp", "留言公众号推送跳过：没有绑定服务号openid的家长")
|
||
return
|
||
target_by_user_id = {
|
||
self._target_int(target, "parent_user_id"): target
|
||
for target in targets or []
|
||
if self._target_int(target, "parent_user_id") > 0
|
||
}
|
||
dedup_seconds = max(0, int(settings.wechat_mp_leave_message_dedup_seconds or 0))
|
||
|
||
for recipient in recipients:
|
||
page = self._build_leave_message_page(
|
||
recipient_user_id=recipient.user_id,
|
||
target=target_by_user_id.get(recipient.user_id),
|
||
)
|
||
await self._send_to_recipient(
|
||
device_id=device_id,
|
||
notification_type="leave_message",
|
||
openid=recipient.openid,
|
||
title=child_name or device_id,
|
||
condition="收到新的设备留言",
|
||
page=page,
|
||
dedup_seconds=dedup_seconds,
|
||
)
|
||
|
||
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:
|
||
await self._send_to_recipient(
|
||
device_id=device_id,
|
||
notification_type=notification_type,
|
||
openid=openid,
|
||
title=title,
|
||
condition=condition,
|
||
page=page,
|
||
dedup_seconds=dedup_seconds,
|
||
template_id=template_id,
|
||
)
|
||
|
||
async def _send_to_recipient(
|
||
self,
|
||
*,
|
||
device_id: str,
|
||
notification_type: str,
|
||
openid: str,
|
||
title: str,
|
||
condition: str,
|
||
page: str,
|
||
dedup_seconds: int,
|
||
template_id: str | None = None,
|
||
) -> None:
|
||
resolved_template_id = template_id or self._template_id(notification_type)
|
||
if not resolved_template_id:
|
||
return
|
||
|
||
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}")
|
||
return
|
||
|
||
result = await self._send_template_message(
|
||
openid=openid,
|
||
template_id=resolved_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 settings.wechat_mp_miniprogram_link_enabled and 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]:
|
||
recipients = await self._list_device_family_mp_recipients(device_id=device_id)
|
||
return [recipient.openid for recipient in recipients]
|
||
|
||
async def _list_device_family_mp_recipients(self, *, device_id: str) -> list[WechatMpRecipient]:
|
||
db_session = await self.get_session()
|
||
try:
|
||
result = await db_session.execute(
|
||
text(
|
||
"""
|
||
SELECT recipients.user_id, 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,
|
||
},
|
||
)
|
||
recipients: list[WechatMpRecipient] = []
|
||
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)
|
||
recipients.append(
|
||
WechatMpRecipient(
|
||
user_id=int(row["user_id"]),
|
||
openid=openid,
|
||
)
|
||
)
|
||
return recipients
|
||
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 _build_leave_message_page(
|
||
self,
|
||
*,
|
||
recipient_user_id: int,
|
||
target: DeviceParentLeaveMessageTarget | Mapping[str, Any] | None,
|
||
) -> str:
|
||
if target is None:
|
||
return settings.wechat_mp_chat_page
|
||
|
||
child_id = self._target_int(target, "child_id")
|
||
conversation_id = self._target_int(target, "conversation_id")
|
||
message_id = self._target_int(target, "message_id")
|
||
if child_id <= 0 or conversation_id <= 0 or message_id <= 0:
|
||
return settings.wechat_mp_chat_page
|
||
|
||
token = self._create_notification_token(
|
||
{
|
||
"type": "leave_message",
|
||
"user_id": recipient_user_id,
|
||
"conversation_id": conversation_id,
|
||
"message_id": message_id,
|
||
"child_id": child_id,
|
||
"child_name": self._target_str(target, "child_name"),
|
||
"device_id": self._target_str(target, "device_id"),
|
||
}
|
||
)
|
||
page = (settings.wechat_mp_notification_page or "").strip() or "pages/notification/index"
|
||
return f"{page}?token={token}"
|
||
|
||
def _create_notification_token(self, payload: Mapping[str, Any]) -> str:
|
||
now = int(time.time())
|
||
ttl_seconds = max(60, int(settings.wechat_mp_notification_token_expire_minutes or 0) * 60)
|
||
normalized_payload = {
|
||
"v": WECHAT_MP_NOTIFICATION_TOKEN_VERSION,
|
||
"iat": now,
|
||
"exp": now + ttl_seconds,
|
||
**dict(payload),
|
||
}
|
||
payload_bytes = json.dumps(normalized_payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
||
payload_part = self._b64url_encode(payload_bytes)
|
||
signature = self._sign_notification_payload(payload_part)
|
||
return f"{payload_part}.{signature}"
|
||
|
||
def _decode_notification_token(self, token: str) -> Mapping[str, Any]:
|
||
normalized_token = str(token or "").strip()
|
||
if not normalized_token or "." not in normalized_token:
|
||
raise HTTPException(status_code=400, detail="invalid notification token")
|
||
payload_part, signature = normalized_token.rsplit(".", 1)
|
||
expected_signature = self._sign_notification_payload(payload_part)
|
||
if not hmac.compare_digest(signature, expected_signature):
|
||
raise HTTPException(status_code=400, detail="invalid notification token")
|
||
try:
|
||
payload = json.loads(self._b64url_decode(payload_part).decode("utf-8"))
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=400, detail="invalid notification token") from exc
|
||
if not isinstance(payload, dict):
|
||
raise HTTPException(status_code=400, detail="invalid notification token")
|
||
if int(payload.get("v") or 0) != WECHAT_MP_NOTIFICATION_TOKEN_VERSION:
|
||
raise HTTPException(status_code=400, detail="unsupported notification token")
|
||
if int(payload.get("exp") or 0) < int(time.time()):
|
||
raise HTTPException(status_code=400, detail="notification token expired")
|
||
return payload
|
||
|
||
def _sign_notification_payload(self, payload_part: str) -> str:
|
||
secret = f"wechat-mp-notification:{settings.jwt_secret}".encode("utf-8")
|
||
digest = hmac.new(secret, payload_part.encode("ascii"), hashlib.sha256).digest()
|
||
return self._b64url_encode(digest)
|
||
|
||
@staticmethod
|
||
def _b64url_encode(value: bytes) -> str:
|
||
return base64.urlsafe_b64encode(value).decode("ascii").rstrip("=")
|
||
|
||
@staticmethod
|
||
def _b64url_decode(value: str) -> bytes:
|
||
padding = "=" * (-len(value) % 4)
|
||
return base64.urlsafe_b64decode(f"{value}{padding}")
|
||
|
||
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 _target_value(target: DeviceParentLeaveMessageTarget | Mapping[str, Any], key: str) -> Any:
|
||
if isinstance(target, Mapping):
|
||
return target.get(key)
|
||
return getattr(target, key)
|
||
|
||
@classmethod
|
||
def _target_int(cls, target: DeviceParentLeaveMessageTarget | Mapping[str, Any], key: str) -> int:
|
||
try:
|
||
return int(cls._target_value(target, key) or 0)
|
||
except (TypeError, ValueError):
|
||
return 0
|
||
|
||
@classmethod
|
||
def _target_str(cls, target: DeviceParentLeaveMessageTarget | Mapping[str, Any], key: str) -> str:
|
||
return str(cls._target_value(target, key) or "").strip()
|
||
|
||
@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()
|