355 lines
11 KiB
Python
355 lines
11 KiB
Python
"""微信消息通知服务
|
|
|
|
支持微信小程序模板消息推送
|
|
|
|
使用方式:
|
|
from app.core.wechat_message import WechatMessageService
|
|
|
|
# 发送订单状态通知
|
|
WechatMessageService.send_order_status(
|
|
openid="user_openid",
|
|
order_no="ORD202603190001",
|
|
status="已支付",
|
|
amount=100.00
|
|
)
|
|
|
|
# 发送配送通知
|
|
WechatMessageService.send_delivery_notice(
|
|
openid="deliver_openid",
|
|
task_no="DLV20260319001",
|
|
address="深圳市南山区...",
|
|
time="尽快取货"
|
|
)
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from typing import Optional, Dict, Any
|
|
from datetime import datetime
|
|
|
|
import requests
|
|
|
|
from app.core.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class WechatMessageService:
|
|
"""微信消息服务"""
|
|
|
|
# 模板消息ID配置
|
|
TEMPLATE_IDS = {
|
|
"order_status": "ORDER_STATUS_TEMPLATE_ID", # 订单状态通知
|
|
"delivery_notice": "DELIVERY_NOTICE_TEMPLATE_ID", # 配送通知
|
|
"payment_success": "PAYMENT_SUCCESS_TEMPLATE_ID", # 支付成功通知
|
|
"refund_result": "REFUND_RESULT_TEMPLATE_ID", # 退款结果通知
|
|
"delivery_warning": "DELIVERY_WARNING_TEMPLATE_ID", # 配送时效预警
|
|
}
|
|
|
|
def __init__(self):
|
|
self.appid = settings.WXPAY_MINI_APPID
|
|
self.secret = settings.WXPAY_SECRET
|
|
self.access_token = None
|
|
self.token_expires_at = None
|
|
|
|
def _get_access_token(self) -> Optional[str]:
|
|
"""获取Access Token"""
|
|
if self.access_token and self.token_expires_at:
|
|
if datetime.now() < self.token_expires_at:
|
|
return self.access_token
|
|
|
|
try:
|
|
url = "https://api.weixin.qq.com/cgi-bin/token"
|
|
params = {
|
|
"grant_type": "client_credential",
|
|
"appid": self.appid,
|
|
"secret": self.secret
|
|
}
|
|
|
|
response = requests.get(url, params=params, timeout=10)
|
|
result = response.json()
|
|
|
|
if "access_token" in result:
|
|
self.access_token = result["access_token"]
|
|
expires_in = result.get("expires_in", 7200)
|
|
self.token_expires_at = datetime.now().timestamp() + expires_in - 300
|
|
return self.access_token
|
|
else:
|
|
logger.error(f"获取Access Token失败: {result}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"请求Access Token异常: {str(e)}")
|
|
return None
|
|
|
|
def _send_template_message(self, openid: str, template_id: str, data: Dict, page: str = None, color: str = None) -> Dict:
|
|
"""发送模板消息
|
|
|
|
Args:
|
|
openid: 用户openid
|
|
template_id: 模板ID
|
|
data: 模板数据
|
|
page: 跳转页面
|
|
color: 字体颜色
|
|
|
|
Returns:
|
|
Dict: 发送结果
|
|
"""
|
|
access_token = self._get_access_token()
|
|
if not access_token:
|
|
return {"success": False, "error": "获取access_token失败"}
|
|
|
|
try:
|
|
url = f"https://api.weixin.qq.com/cgi-bin/message/template/send?access_token={access_token}"
|
|
|
|
payload = {
|
|
"touser": openid,
|
|
"template_id": template_id,
|
|
"data": data
|
|
}
|
|
|
|
if page:
|
|
payload["miniprogram"] = {
|
|
"appid": self.appid,
|
|
"pagepath": page
|
|
}
|
|
|
|
if color:
|
|
payload["color"] = color
|
|
|
|
response = requests.post(url, json=payload, timeout=10)
|
|
result = response.json()
|
|
|
|
if result.get("errcode") == 0:
|
|
logger.info(f"模板消息发送成功: {openid}, msgid: {result.get('msgid')}")
|
|
return {"success": True, "msgid": result.get("msgid")}
|
|
else:
|
|
logger.error(f"模板消息发送失败: {result}")
|
|
return {"success": False, "error": result.get("errmsg", "发送失败")}
|
|
|
|
except Exception as e:
|
|
logger.error(f"发送模板消息异常: {str(e)}")
|
|
return {"success": False, "error": str(e)}
|
|
|
|
@classmethod
|
|
def send_order_status(
|
|
cls,
|
|
openid: str,
|
|
order_no: str,
|
|
status: str,
|
|
amount: float,
|
|
remark: str = None
|
|
) -> Dict:
|
|
"""发送订单状态通知
|
|
|
|
Args:
|
|
openid: 用户openid
|
|
order_no: 订单号
|
|
status: 订单状态
|
|
amount: 订单金额
|
|
remark: 备注
|
|
"""
|
|
template_id = cls.TEMPLATE_IDS.get("order_status")
|
|
if not template_id:
|
|
return {"success": False, "error": "未配置订单状态模板ID"}
|
|
|
|
data = {
|
|
"order_no": {"value": order_no, "color": "#173177"},
|
|
"status": {"value": status, "color": "#07c160"},
|
|
"amount": {"value": f"¥{amount:.2f}", "color": "#ff6600"},
|
|
"remark": {"value": remark or "感谢您的使用", "color": "#888888"}
|
|
}
|
|
|
|
service = cls()
|
|
return service._send_template_message(
|
|
openid=openid,
|
|
template_id=template_id,
|
|
data=data,
|
|
page="pages/client/order/detail/detail"
|
|
)
|
|
|
|
@classmethod
|
|
def send_payment_success(
|
|
cls,
|
|
openid: str,
|
|
order_no: str,
|
|
amount: float,
|
|
payment_method: str = "微信支付"
|
|
) -> Dict:
|
|
"""发送支付成功通知
|
|
|
|
Args:
|
|
openid: 用户openid
|
|
order_no: 订单号
|
|
amount: 支付金额
|
|
payment_method: 支付方式
|
|
"""
|
|
template_id = cls.TEMPLATE_IDS.get("payment_success")
|
|
if not template_id:
|
|
return {"success": False, "error": "未配置支付成功模板ID"}
|
|
|
|
data = {
|
|
"order_no": {"value": order_no, "color": "#173177"},
|
|
"amount": {"value": f"¥{amount:.2f}", "color": "#ff6600"},
|
|
"payment_method": {"value": payment_method, "color": "#07c160"},
|
|
"time": {"value": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "color": "#888888"}
|
|
}
|
|
|
|
service = cls()
|
|
return service._send_template_message(
|
|
openid=openid,
|
|
template_id=template_id,
|
|
data=data,
|
|
page="pages/client/order/detail/detail"
|
|
)
|
|
|
|
@classmethod
|
|
def send_delivery_notice(
|
|
cls,
|
|
openid: str,
|
|
task_no: str,
|
|
supplier_name: str,
|
|
address: str,
|
|
remark: str = None
|
|
) -> Dict:
|
|
"""发送配送通知
|
|
|
|
Args:
|
|
openid: 配送员openid
|
|
task_no: 配送单号
|
|
supplier_name: 供应商名称
|
|
address: 取货地址
|
|
remark: 备注
|
|
"""
|
|
template_id = cls.TEMPLATE_IDS.get("delivery_notice")
|
|
if not template_id:
|
|
return {"success": False, "error": "未配置配送通知模板ID"}
|
|
|
|
data = {
|
|
"task_no": {"value": task_no, "color": "#173177"},
|
|
"supplier": {"value": supplier_name, "color": "#07c160"},
|
|
"address": {"value": address, "color": "#173177"},
|
|
"remark": {"value": remark or "请尽快前往取货", "color": "#ff6600"}
|
|
}
|
|
|
|
service = cls()
|
|
return service._send_template_message(
|
|
openid=openid,
|
|
template_id=template_id,
|
|
data=data,
|
|
page="pages/deliver/task/detail"
|
|
)
|
|
|
|
@classmethod
|
|
def send_delivery_warning(
|
|
cls,
|
|
openid: str,
|
|
task_no: str,
|
|
remaining_minutes: int,
|
|
urgency_level: str = "紧急"
|
|
) -> Dict:
|
|
"""发送配送时效预警
|
|
|
|
Args:
|
|
openid: 配送员openid
|
|
task_no: 配送单号
|
|
remaining_minutes: 剩余分钟数
|
|
urgency_level: 紧急程度
|
|
"""
|
|
template_id = cls.TEMPLATE_IDS.get("delivery_warning")
|
|
if not template_id:
|
|
return {"success": False, "error": "未配置配送预警模板ID"}
|
|
|
|
color = "#ff0000" if urgency_level == "紧急" else "#ff6600"
|
|
|
|
data = {
|
|
"task_no": {"value": task_no, "color": "#173177"},
|
|
"remaining": {"value": f"剩余{remaining_minutes}分钟", "color": color},
|
|
"level": {"value": urgency_level, "color": color},
|
|
"time": {"value": datetime.now().strftime("%H:%M"), "color": "#888888"}
|
|
}
|
|
|
|
service = cls()
|
|
return service._send_template_message(
|
|
openid=openid,
|
|
template_id=template_id,
|
|
data=data,
|
|
page="pages/deliver/task/detail"
|
|
)
|
|
|
|
@classmethod
|
|
def send_refund_result(
|
|
cls,
|
|
openid: str,
|
|
order_no: str,
|
|
refund_amount: float,
|
|
refund_time: str = None,
|
|
reason: str = None
|
|
) -> Dict:
|
|
"""发送退款结果通知
|
|
|
|
Args:
|
|
openid: 用户openid
|
|
order_no: 订单号
|
|
refund_amount: 退款金额
|
|
refund_time: 退款时间
|
|
reason: 退款原因
|
|
"""
|
|
template_id = cls.TEMPLATE_IDS.get("refund_result")
|
|
if not template_id:
|
|
return {"success": False, "error": "未配置退款结果模板ID"}
|
|
|
|
data = {
|
|
"order_no": {"value": order_no, "color": "#173177"},
|
|
"amount": {"value": f"¥{refund_amount:.2f}", "color": "#ff6600"},
|
|
"time": {"value": refund_time or datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "color": "#888888"},
|
|
"reason": {"value": reason or "原路退回", "color": "#07c160"}
|
|
}
|
|
|
|
service = cls()
|
|
return service._send_template_message(
|
|
openid=openid,
|
|
template_id=template_id,
|
|
data=data,
|
|
page="pages/client/refund/refund"
|
|
)
|
|
|
|
@classmethod
|
|
def send_urge_notification(
|
|
cls,
|
|
deliver_openid: str,
|
|
order_no: str,
|
|
client_name: str,
|
|
address: str
|
|
) -> Dict:
|
|
"""发送催单通知给配送员
|
|
|
|
Args:
|
|
deliver_openid: 配送员openid
|
|
order_no: 订单号
|
|
client_name: 客户名称
|
|
address: 配送地址
|
|
"""
|
|
template_id = cls.TEMPLATE_IDS.get("delivery_notice")
|
|
if not template_id:
|
|
return {"success": False, "error": "未配置配送通知模板ID"}
|
|
|
|
data = {
|
|
"task_no": {"value": order_no, "color": "#173177"},
|
|
"supplier": {"value": f"客户:{client_name}", "color": "#07c160"},
|
|
"address": {"value": address, "color": "#173177"},
|
|
"remark": {"value": "⚠️ 客户催单，请尽快处理", "color": "#ff0000"}
|
|
}
|
|
|
|
service = cls()
|
|
return service._send_template_message(
|
|
openid=deliver_openid,
|
|
template_id=template_id,
|
|
data=data,
|
|
page="pages/deliver/task/detail"
|
|
)
|
|
|
|
|
|
# 全局实例
|
|
wechat_message = WechatMessageService()
|