110 lines
4.1 KiB
Python
110 lines
4.1 KiB
Python
import json
|
|
import time
|
|
from urllib import error, request
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.app.core.config import get_settings
|
|
from backend.app.core.error_codes import ErrorCode
|
|
from backend.app.core.exceptions import AppException
|
|
|
|
|
|
class WeChatNotificationService:
|
|
"""微信小程序订阅消息通知服务。
|
|
|
|
通过微信 subscribeMessage.send 接口向用户推送服务提醒。
|
|
需要在小程序后台配置消息模板，并将模板 ID 填入系统配置。
|
|
"""
|
|
|
|
TOKEN_CACHE: dict[str, str] = {"token": "", "expires_at": 0}
|
|
|
|
REMINDER_TYPE_TEMPLATE_MAP: dict[str, str] = {
|
|
"order_status_change": "wechat_template_order_status",
|
|
"order_approval_needed": "wechat_template_approval",
|
|
"task_assigned": "wechat_template_task_assigned",
|
|
"logistics_timeout": "wechat_template_logistics_timeout",
|
|
"arrears": "wechat_template_arrears",
|
|
"inactive_customer": "wechat_template_inactive_customer",
|
|
}
|
|
|
|
def send_subscribe_message(self, open_id: str, reminder_type: str, title: str, content: str, biz_url: str = "") -> bool:
|
|
settings = get_settings()
|
|
if not settings.wechat_app_id or not settings.wechat_app_secret:
|
|
return False
|
|
|
|
template_id = self._get_template_id(settings, reminder_type)
|
|
if not template_id:
|
|
return False
|
|
|
|
access_token = self._get_access_token(settings)
|
|
if not access_token:
|
|
return False
|
|
|
|
payload = {
|
|
"touser": open_id,
|
|
"template_id": template_id,
|
|
"page": biz_url,
|
|
"data": {
|
|
"thing1": {"value": title[:20]},
|
|
"thing2": {"value": content[:100]},
|
|
"time3": {"value": time.strftime("%Y-%m-%d %H:%M")},
|
|
},
|
|
}
|
|
|
|
url = f"https://api.weixin.qq.com/cgi-bin/message/subscribe/send?access_token={access_token}"
|
|
req = request.Request(
|
|
url=url,
|
|
data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
|
|
headers={"Content-Type": "application/json; charset=UTF-8"},
|
|
method="POST",
|
|
)
|
|
try:
|
|
with request.urlopen(req, timeout=10) as response:
|
|
result = json.loads(response.read().decode("utf-8") or "{}")
|
|
return result.get("errcode", -1) == 0
|
|
except (error.HTTPError, error.URLError, json.JSONDecodeError):
|
|
return False
|
|
|
|
def bind_open_id(self, session: Session, user_id: int, open_id: str) -> None:
|
|
from backend.app.models.system import User
|
|
from sqlalchemy import select
|
|
user = session.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if user is not None:
|
|
user.open_id = open_id
|
|
session.add(user)
|
|
session.commit()
|
|
|
|
def _get_template_id(self, settings, reminder_type: str) -> str:
|
|
config_key = self.REMINDER_TYPE_TEMPLATE_MAP.get(reminder_type, "")
|
|
if not config_key:
|
|
return ""
|
|
return getattr(settings, config_key, "")
|
|
|
|
def _get_access_token(self, settings) -> str:
|
|
now = time.time()
|
|
if self.TOKEN_CACHE["token"] and self.TOKEN_CACHE["expires_at"] > now + 60:
|
|
return self.TOKEN_CACHE["token"]
|
|
|
|
url = (
|
|
f"https://api.weixin.qq.com/cgi-bin/token"
|
|
f"?grant_type=client_credential"
|
|
f"&appid={settings.wechat_app_id}"
|
|
f"&secret={settings.wechat_app_secret}"
|
|
)
|
|
req = request.Request(url=url, method="GET")
|
|
try:
|
|
with request.urlopen(req, timeout=10) as response:
|
|
result = json.loads(response.read().decode("utf-8") or "{}")
|
|
token = result.get("access_token", "")
|
|
expires_in = int(result.get("expires_in", 0))
|
|
if token and expires_in > 0:
|
|
self.TOKEN_CACHE["token"] = token
|
|
self.TOKEN_CACHE["expires_at"] = now + expires_in
|
|
return token
|
|
except (error.HTTPError, error.URLError, json.JSONDecodeError):
|
|
pass
|
|
return ""
|
|
|
|
|
|
wechat_notification_service = WeChatNotificationService()
|