From 944c9cdbb615e049ec02fde0f6cda1dab802d1af Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Thu, 21 May 2026 11:50:32 +0800 Subject: [PATCH] =?UTF-8?q?feat(channel):=20=E6=B7=BB=E5=8A=A0=E8=85=BE?= =?UTF-8?q?=E8=AE=AF=E7=9F=AD=E4=BF=A1=E6=B8=A0=E9=81=93=E6=89=A9=E5=B1=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增腾讯短信（Tencent SMS）渠道扩展，支持在 Yuxi 平台中集成腾讯云短信渠道。 包含以下功能模块： - client: 腾讯云短信 API 客户端封装 - plugin: 渠道插件核心 - config: 渠道配置管理 - gateway: SSE/WebSocket 网关接入 - webhook: Webhook 事件处理 - outbound: 外发消息管理 - streaming: 流式消息处理 - security: 安全校验 - dedupe: 消息去重 - delivery: 送达状态回调 - compliance: 合规管理 - frequency: 频率控制 - templates: 短信模板 - agent_prompt: Agent 提示词 - types: 类型定义 --- .../extensions/tencent_sms/__init__.py | 4 + .../extensions/tencent_sms/agent_prompt.py | 37 ++ .../channel/extensions/tencent_sms/client.py | 495 ++++++++++++++++++ .../extensions/tencent_sms/compliance.py | 104 ++++ .../channel/extensions/tencent_sms/config.py | 220 ++++++++ .../channel/extensions/tencent_sms/dedupe.py | 52 ++ .../extensions/tencent_sms/delivery.py | 41 ++ .../channel/extensions/tencent_sms/errors.py | 47 ++ .../extensions/tencent_sms/frequency.py | 57 ++ .../channel/extensions/tencent_sms/gateway.py | 171 ++++++ .../extensions/tencent_sms/outbound.py | 159 ++++++ .../extensions/tencent_sms/plugin.json | 31 ++ .../channel/extensions/tencent_sms/plugin.py | 280 ++++++++++ .../extensions/tencent_sms/security.py | 15 + .../extensions/tencent_sms/streaming.py | 93 ++++ .../extensions/tencent_sms/templates.py | 34 ++ .../channel/extensions/tencent_sms/types.py | 163 ++++++ .../channel/extensions/tencent_sms/webhook.py | 124 +++++ 18 files changed, 2127 insertions(+) create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/__init__.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/agent_prompt.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/client.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/compliance.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/config.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/dedupe.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/delivery.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/errors.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/frequency.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/gateway.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/outbound.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/plugin.json create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/plugin.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/security.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/streaming.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/templates.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/types.py create mode 100644 backend/package/yuxi/channel/extensions/tencent_sms/webhook.py diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/__init__.py b/backend/package/yuxi/channel/extensions/tencent_sms/__init__.py new file mode 100644 index 00000000..20edd1f4 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/__init__.py @@ -0,0 +1,4 @@ +from yuxi.channel.extensions.tencent_sms.plugin import TencentSmsPlugin +from yuxi.channel.plugins.registry import ChannelPluginRegistry + +ChannelPluginRegistry.register(TencentSmsPlugin()) diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/agent_prompt.py b/backend/package/yuxi/channel/extensions/tencent_sms/agent_prompt.py new file mode 100644 index 00000000..53364925 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/agent_prompt.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +from .templates import SmsTemplateRegistry + + +class TencentSmsAgentPrompt: + def __init__(self, registry: SmsTemplateRegistry): + self._registry = registry + + def build_system_prompt(self, account_id: str) -> str: + templates = self._registry.list_all(account_id) + if not templates: + return "" + + lines = [] + for t in templates: + lines.append( + f"- \u573a\u666f: {t.scene.value} ({t.description or '\u65e0\u63cf\u8ff0'}) | " + f"\u6a21\u677fID: {t.template_id} | \u7b7e\u540d:\u3010{t.sign_name}\u3011| " + f"\u53d8\u91cf\u6570: {t.param_count} | " + f"\u683c\u5f0f\u793a\u4f8b: {t.template_format or '(\u672a\u63d0\u4f9b)'}" + ) + + prompt = ( + "## \u817e\u8baf\u4e91\u77ed\u4fe1\u53d1\u9001\u6307\u5f15\n\n" + "\u4f60\u662f\u77ed\u4fe1\u53d1\u9001\u52a9\u624b\u3002\u6240\u6709\u77ed\u4fe1\u5185\u5bb9\u5fc5\u987b\u4f7f\u7528\u9884\u5ba1\u6a21\u677f\uff1a\n\n" + + "\n".join(lines) + + "\n\n**\u53d1\u9001\u89c4\u5219\uff1a**\n" + "- \u4e25\u683c\u4f7f\u7528\u4ee5\u4e0a\u6a21\u677f\uff0c\u586b\u5145\u6a21\u677f\u53d8\u91cf ({1}\u3001{2}...)\n" + "- \u9a8c\u8bc1\u7801\u573a\u666f\uff1a\u53d8\u91cf\u586b\u9a8c\u8bc1\u7801 + \u6709\u6548\u671f\u5206\u949f\u6570\n" + "- \u901a\u77e5\u573a\u666f\uff1a\u53d8\u91cf\u586b\u6807\u9898 + \u5185\u5bb9\n" + "- \u8425\u9500\u77ed\u4fe1\uff1a\u7ed3\u5c3e\u5fc5\u987b\u5305\u542b\u9000\u8ba2\u63d0\u793a\uff0c\u53d1\u9001\u65f6\u95f4\u4ec5\u9650 8:00-21:00\n" + "- \u53d8\u91cf\u4e2d\u7981\u6b62\u5305\u542b URL\n" + "- \u7981\u6b62\u5305\u542b\u91d1\u878d/\u6559\u80b2/\u623f\u4ea7\u7b49\u654f\u611f\u5173\u952e\u8bcd\n" + "- \u7528\u6237\u56de\u590d T/TD/\u9000\u8ba2 \u89c6\u4e3a\u9000\u8ba2\uff0c\u540e\u7eed\u4e0d\u518d\u53d1\u9001\n" + ) + return prompt diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/client.py b/backend/package/yuxi/channel/extensions/tencent_sms/client.py new file mode 100644 index 00000000..b106cc13 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/client.py @@ -0,0 +1,495 @@ +from __future__ import annotations + +import asyncio +import logging +from concurrent.futures import ThreadPoolExecutor + +from .types import SendResult, TencentSmsAccount + +logger = logging.getLogger(__name__) + +_sms_executor = ThreadPoolExecutor(max_workers=10, thread_name_prefix="sms-") + +MAX_BATCH_SIZE = 200 + + +class TencentSmsClient: + def __init__(self, account: TencentSmsAccount): + self._account = account + self._client = None + + def _build_client(self): + from tencentcloud.common import credential + from tencentcloud.common.profile.client_profile import ClientProfile + from tencentcloud.common.profile.http_profile import HttpProfile + from tencentcloud.sms.v20210111 import sms_client + + cred = credential.Credential(self._account.secret_id, self._account.secret_key) + http_profile = HttpProfile() + http_profile.endpoint = self._account.endpoint + http_profile.reqTimeout = 30 + client_profile = ClientProfile() + client_profile.httpProfile = http_profile + return sms_client.SmsClient(cred, self._account.region, client_profile) + + def _get_client(self): + if self._client is None: + self._client = self._build_client() + return self._client + + async def send_sms( + self, + phone_numbers: list[str], + template_id: str, + template_params: list[str], + sign_name: str = "", + session_context: str = "", + extend_code: str = "", + sender_id: str = "", + ) -> list[SendResult]: + from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException + from tencentcloud.sms.v20210111 import models + + req = models.SendSmsRequest() + req.SmsSdkAppId = self._account.sms_sdk_app_id + req.SignName = sign_name or self._account.sign_name + req.TemplateId = template_id + req.TemplateParamSet = template_params + req.PhoneNumberSet = phone_numbers + req.SessionContext = session_context + + ext = extend_code or self._account.extend_code + if ext: + req.ExtendCode = ext + sid = sender_id or self._account.sender_id + if sid: + req.SenderId = sid + + loop = asyncio.get_running_loop() + try: + resp = await loop.run_in_executor( + _sms_executor, + self._get_client().SendSms, + req, + ) + except TencentCloudSDKException as e: + logger.error("SendSms SDK \u5f02\u5e38: code=%s, message=%s", e.code, e.message) + raise + + results = [] + for status in resp.SendStatusSet: + results.append( + SendResult( + success=status.Code == "Ok", + serial_no=getattr(status, "SerialNo", ""), + phone_number=getattr(status, "PhoneNumber", ""), + code=status.Code or "", + message=getattr(status, "Message", ""), + fee=getattr(status, "Fee", 0), + iso_code=getattr(status, "IsoCode", ""), + session_context=getattr(status, "SessionContext", session_context), + ) + ) + return results + + async def send_single( + self, + phone_number: str, + template_id: str, + template_params: list[str], + sign_name: str = "", + session_context: str = "", + extend_code: str = "", + sender_id: str = "", + ) -> SendResult: + results = await self.send_sms( + [phone_number], template_id, template_params, sign_name, session_context, extend_code, sender_id + ) + return results[0] + + async def pull_send_status(self, limit: int = 100) -> list[dict]: + from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException + from tencentcloud.sms.v20210111 import models + + req = models.PullSmsSendStatusRequest() + req.SmsSdkAppId = self._account.sms_sdk_app_id + req.Limit = min(limit, 100) + + loop = asyncio.get_running_loop() + try: + resp = await loop.run_in_executor( + _sms_executor, + self._get_client().PullSmsSendStatus, + req, + ) + except TencentCloudSDKException as e: + logger.error("PullSmsSendStatus SDK \u5f02\u5e38: %s", e) + return [] + + result = [] + for item in getattr(resp, "PullSmsSendStatusSet", []) or []: + result.append( + { + "serial_no": getattr(item, "SerialNo", ""), + "phone_number": getattr(item, "PhoneNumber", ""), + "report_status": getattr(item, "ReportStatus", ""), + "description": getattr(item, "Description", ""), + "user_receive_time": getattr(item, "UserReceiveTime", ""), + "session_context": getattr(item, "SessionContext", ""), + } + ) + return result + + async def pull_send_status_by_phone( + self, phone_number: str, begin_time: int = 0, end_time: int = 0, limit: int = 100 + ) -> list[dict]: + from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException + from tencentcloud.sms.v20210111 import models + + req = models.PullSmsSendStatusByPhoneNumberRequest() + req.SmsSdkAppId = self._account.sms_sdk_app_id + req.PhoneNumber = phone_number + req.SendDateTime = begin_time + req.EndDateTime = end_time + req.Limit = min(limit, 100) + + loop = asyncio.get_running_loop() + try: + resp = await loop.run_in_executor( + _sms_executor, + self._get_client().PullSmsSendStatusByPhoneNumber, + req, + ) + except TencentCloudSDKException as e: + logger.error("PullSmsSendStatusByPhoneNumber SDK \u5f02\u5e38: %s", e) + return [] + + result = [] + for item in getattr(resp, "PullSmsSendStatusSet", []) or []: + result.append( + { + "serial_no": getattr(item, "SerialNo", ""), + "phone_number": getattr(item, "PhoneNumber", ""), + "report_status": getattr(item, "ReportStatus", ""), + "description": getattr(item, "Description", ""), + "user_receive_time": getattr(item, "UserReceiveTime", ""), + "session_context": getattr(item, "SessionContext", ""), + } + ) + return result + + async def pull_reply_status_by_phone( + self, phone_number: str, begin_time: int = 0, end_time: int = 0, limit: int = 100 + ) -> list[dict]: + from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException + from tencentcloud.sms.v20210111 import models + + req = models.PullSmsReplyStatusByPhoneNumberRequest() + req.SmsSdkAppId = self._account.sms_sdk_app_id + req.PhoneNumber = phone_number + req.SendDateTime = begin_time + req.EndDateTime = end_time + req.Limit = min(limit, 100) + + loop = asyncio.get_running_loop() + try: + resp = await loop.run_in_executor( + _sms_executor, + self._get_client().PullSmsReplyStatusByPhoneNumber, + req, + ) + except TencentCloudSDKException as e: + logger.error("PullSmsReplyStatusByPhoneNumber SDK \u5f02\u5e38: %s", e) + return [] + + result = [] + for item in getattr(resp, "PullSmsReplyStatusSet", []) or []: + result.append( + { + "phone_number": getattr(item, "PhoneNumber", ""), + "reply_content": getattr(item, "ReplyContent", ""), + "reply_time": getattr(item, "ReplyTime", ""), + "reply_unix_time": getattr(item, "ReplyUnixTime", 0), + "extend_code": getattr(item, "ExtendCode", ""), + } + ) + return result + + async def pull_reply_status(self, limit: int = 100) -> list[dict]: + from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException + from tencentcloud.sms.v20210111 import models + + req = models.PullSmsReplyStatusRequest() + req.SmsSdkAppId = self._account.sms_sdk_app_id + req.Limit = min(limit, 100) + + loop = asyncio.get_running_loop() + try: + resp = await loop.run_in_executor( + _sms_executor, + self._get_client().PullSmsReplyStatus, + req, + ) + except TencentCloudSDKException as e: + logger.error("PullSmsReplyStatus SDK \u5f02\u5e38: %s", e) + return [] + + result = [] + for item in getattr(resp, "PullSmsReplyStatusSet", []) or []: + result.append( + { + "phone_number": getattr(item, "PhoneNumber", ""), + "reply_content": getattr(item, "ReplyContent", ""), + "reply_time": getattr(item, "ReplyTime", ""), + "reply_unix_time": getattr(item, "ReplyUnixTime", 0), + "extend_code": getattr(item, "ExtendCode", ""), + } + ) + return result + + async def get_send_statistics(self, begin_time: str, end_time: str, page_size: int = 10, page_no: int = 1) -> dict: + from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException + from tencentcloud.sms.v20210111 import models + + req = models.SendStatusStatisticsRequest() + req.SmsSdkAppId = self._account.sms_sdk_app_id + req.BeginTime = begin_time + req.EndTime = end_time + req.PageSize = page_size + req.PageNumber = page_no + + loop = asyncio.get_running_loop() + try: + resp = await loop.run_in_executor( + _sms_executor, + self._get_client().SendStatusStatistics, + req, + ) + except TencentCloudSDKException as e: + logger.error("SendStatusStatistics SDK \u5f02\u5e38: %s", e) + return {} + + data = getattr(resp, "Data", None) + if not data: + return {} + return { + "total_count": getattr(data, "TotalCount", 0), + "send_count": getattr(data, "SendCount", 0), + "success_count": getattr(data, "SuccessCount", 0), + "fail_count": getattr(data, "FailCount", 0), + "details": [ + { + "date": getattr(d, "Date", ""), + "send_count": getattr(d, "SendCount", 0), + "success_count": getattr(d, "SuccessCount", 0), + "fail_count": getattr(d, "FailCount", 0), + } + for d in getattr(data, "Details", []) or [] + ], + } + + async def get_callback_statistics( + self, begin_time: str, end_time: str, page_size: int = 10, page_no: int = 1 + ) -> dict: + from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException + from tencentcloud.sms.v20210111 import models + + req = models.CallbackStatusStatisticsRequest() + req.SmsSdkAppId = self._account.sms_sdk_app_id + req.BeginTime = begin_time + req.EndTime = end_time + req.PageSize = page_size + req.PageNumber = page_no + + loop = asyncio.get_running_loop() + try: + resp = await loop.run_in_executor( + _sms_executor, + self._get_client().CallbackStatusStatistics, + req, + ) + except TencentCloudSDKException as e: + logger.error("CallbackStatusStatistics SDK \u5f02\u5e38: %s", e) + return {} + + data = getattr(resp, "Data", None) + if not data: + return {} + return { + "total_count": getattr(data, "TotalCount", 0), + "callback_count": getattr(data, "CallbackCount", 0), + "success_count": getattr(data, "SuccessCount", 0), + "fail_count": getattr(data, "FailCount", 0), + "details": [ + { + "date": getattr(d, "Date", ""), + "callback_count": getattr(d, "CallbackCount", 0), + "success_count": getattr(d, "SuccessCount", 0), + "fail_count": getattr(d, "FailCount", 0), + } + for d in getattr(data, "Details", []) or [] + ], + } + + async def describe_phone_number_info(self, phone_numbers: list[str]) -> list[dict]: + from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException + from tencentcloud.sms.v20210111 import models + + req = models.DescribePhoneNumberInfoRequest() + req.SmsSdkAppId = self._account.sms_sdk_app_id + req.PhoneNumberSet = phone_numbers + + loop = asyncio.get_running_loop() + try: + resp = await loop.run_in_executor( + _sms_executor, + self._get_client().DescribePhoneNumberInfo, + req, + ) + except TencentCloudSDKException as e: + logger.error("DescribePhoneNumberInfo SDK \u5f02\u5e38: %s", e) + return [] + + result = [] + for item in getattr(resp, "PhoneNumberInfoSet", []) or []: + result.append( + { + "phone_number": getattr(item, "PhoneNumber", ""), + "nation_code": getattr(item, "NationCode", ""), + "subscriber_number": getattr(item, "SubscriberNumber", ""), + "iso_code": getattr(item, "IsoCode", ""), + "iso_name": getattr(item, "IsoName", ""), + } + ) + return result + + async def report_conversion( + self, + phone_number: str, + serial_no: str, + conversion_time: int = 0, + remark: str = "", + ) -> bool: + import time + + from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException + from tencentcloud.sms.v20210111 import models + + req = models.ReportConversionRequest() + req.SmsSdkAppId = self._account.sms_sdk_app_id + req.PhoneNumber = phone_number + req.SerialNo = serial_no + req.ConversionTime = conversion_time or int(time.time()) + req.Remark = remark + + loop = asyncio.get_running_loop() + try: + resp = await loop.run_in_executor( + _sms_executor, + self._get_client().ReportConversion, + req, + ) + return getattr(resp, "Result", False) + except TencentCloudSDKException as e: + logger.error("ReportConversion SDK \u5f02\u5e38: %s", e) + return False + + async def describe_sms_sign_list(self, sign_name_list: list[str] | None = None) -> list[dict]: + from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException + from tencentcloud.sms.v20210111 import models + + req = models.DescribeSmsSignListRequest() + req.SmsSdkAppId = self._account.sms_sdk_app_id + req.SignNameList = sign_name_list or [self._account.sign_name] + + loop = asyncio.get_running_loop() + try: + resp = await loop.run_in_executor( + _sms_executor, + self._get_client().DescribeSmsSignList, + req, + ) + except TencentCloudSDKException as e: + logger.error("DescribeSmsSignList SDK \u5f02\u5e38: %s", e) + return [] + + result = [] + for item in getattr(resp, "DescribeSmsSignListStatusSet", []) or []: + result.append( + { + "sign_name": getattr(item, "SignName", ""), + "sign_type": getattr(item, "SignType", 0), + "document": getattr(item, "Document", ""), + "international": getattr(item, "International", 0), + "status_code": getattr(item, "StatusCode", 0), + "review_reply": getattr(item, "ReviewReply", ""), + "create_time": getattr(item, "CreateTime", ""), + } + ) + return result + + async def describe_sms_template_list( + self, template_id_set: list[str] | None = None, international: int = 0 + ) -> list[dict]: + from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException + from tencentcloud.sms.v20210111 import models + + req = models.DescribeSmsTemplateListRequest() + req.SmsSdkAppId = self._account.sms_sdk_app_id + req.International = international + if template_id_set: + req.TemplateIdSet = [int(tid) for tid in template_id_set if tid.isdigit()] + + loop = asyncio.get_running_loop() + try: + resp = await loop.run_in_executor( + _sms_executor, + self._get_client().DescribeSmsTemplateList, + req, + ) + except TencentCloudSDKException as e: + logger.error("DescribeSmsTemplateList SDK \u5f02\u5e38: %s", e) + return [] + + result = [] + for item in getattr(resp, "DescribeSmsTemplateListStatusSet", []) or []: + result.append( + { + "template_id": str(getattr(item, "TemplateId", "")), + "template_name": getattr(item, "TemplateName", ""), + "template_content": getattr(item, "TemplateContent", ""), + "template_type": getattr(item, "TemplateType", 0), + "international": getattr(item, "International", 0), + "status_code": getattr(item, "StatusCode", 0), + "review_reply": getattr(item, "ReviewReply", ""), + "create_time": getattr(item, "CreateTime", ""), + } + ) + return result + + async def probe(self) -> bool: + from tencentcloud.common import credential + from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException + from tencentcloud.sms.v20210111 import models, sms_client + + try: + cred = credential.Credential(self._account.secret_id, self._account.secret_key) + loop = asyncio.get_running_loop() + + def _check(): + try: + client = sms_client.SmsClient(cred, self._account.region) + req = models.DescribeSmsSignListRequest() + req.SmsSdkAppId = self._account.sms_sdk_app_id + req.SignNameList = [self._account.sign_name] + client.DescribeSmsSignList(req) + return True + except TencentCloudSDKException as e: + if "AuthFailure" in (e.code or ""): + return False + return True + + return await loop.run_in_executor(_sms_executor, _check) + except Exception as e: + logger.warning("probe exception: %s", e) + return False diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/compliance.py b/backend/package/yuxi/channel/extensions/tencent_sms/compliance.py new file mode 100644 index 00000000..00a56461 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/compliance.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import datetime +import json +import logging +import os +from pathlib import Path + +from .types import HELP_KEYWORDS, OPT_IN_KEYWORDS, OPT_OUT_KEYWORDS + +logger = logging.getLogger(__name__) + +_unsubscribe_blocklist: set[str] = set() +_unsubscribe_file: Path | None = None + +MARKETING_START_HOUR = 8 +MARKETING_END_HOUR = 21 + + +def _get_unsubscribe_file() -> Path: + global _unsubscribe_file + if _unsubscribe_file is None: + data_dir = Path(os.environ.get("YUXI_DATA_DIR", "data")) + _unsubscribe_file = data_dir / "tencent_sms_unsubscribe.json" + return _unsubscribe_file + + +def _load_blocklist() -> set[str]: + file_path = _get_unsubscribe_file() + try: + if file_path.exists(): + data = json.loads(file_path.read_text(encoding="utf-8")) + return set(data) if isinstance(data, list) else set() + except (json.JSONDecodeError, OSError) as e: + logger.warning("\u52a0\u8f7d\u9000\u8ba2\u9ed1\u540d\u5355\u5931\u8d25: %s", e) + return set() + + +def _save_blocklist() -> None: + file_path = _get_unsubscribe_file() + try: + file_path.parent.mkdir(parents=True, exist_ok=True) + file_path.write_text( + json.dumps(sorted(_unsubscribe_blocklist), ensure_ascii=False, indent=2), + encoding="utf-8", + ) + except OSError as e: + logger.warning("\u4fdd\u5b58\u9000\u8ba2\u9ed1\u540d\u5355\u5931\u8d25: %s", e) + + +def init_blocklist() -> None: + global _unsubscribe_blocklist + _unsubscribe_blocklist = _load_blocklist() + logger.info("\u9000\u8ba2\u9ed1\u540d\u5355\u5df2\u52a0\u8f7d: %d \u6761", len(_unsubscribe_blocklist)) + + +def check_marketing_time_window() -> tuple[bool, str]: + hour = datetime.datetime.now().hour + if MARKETING_START_HOUR <= hour < MARKETING_END_HOUR: + return True, "ok" + return ( + False, + f"\u8425\u9500\u77ed\u4fe1\u4ec5\u5141\u8bb8 {MARKETING_START_HOUR}:00-{MARKETING_END_HOUR}:00 \u53d1\u9001", + ) + + +def detect_opt_out(reply_content: str) -> tuple[bool, str | None]: + text = reply_content.strip().upper() + content_lower = reply_content.strip().lower() + + if content_lower in OPT_OUT_KEYWORDS or text in ("T", "TD", "N", "NO"): + return ( + True, + "\u60a8\u5df2\u6210\u529f\u9000\u8ba2\uff0c\u5c06\u4e0d\u518d\u6536\u5230\u672c\u53f7\u7801\u7684\u6d88\u606f\u3002\u56de\u590d START \u91cd\u65b0\u8ba2\u9605\u3002", + ) + + if content_lower in OPT_IN_KEYWORDS: + return True, "\u60a8\u5df2\u91cd\u65b0\u8ba2\u9605\uff0c\u6b22\u8fce\u56de\u6765\uff01" + + if content_lower in HELP_KEYWORDS: + return ( + True, + "\u6b22\u8fce\u4f7f\u7528 AI \u667a\u80fd\u52a9\u624b\u3002\u56de\u590d TD \u9000\u8ba2\u6d88\u606f\uff0c\u56de\u590d HELP \u83b7\u53d6\u5e2e\u52a9\u3002", + ) + + return False, None + + +def add_unsubscribe(phone: str): + _unsubscribe_blocklist.add(phone) + _save_blocklist() + logger.info("\u7528\u6237 %s \u5df2\u52a0\u5165\u9000\u8ba2\u9ed1\u540d\u5355", phone) + + +def remove_unsubscribe(phone: str): + _unsubscribe_blocklist.discard(phone) + _save_blocklist() + logger.info("\u7528\u6237 %s \u5df2\u79fb\u9664\u9000\u8ba2\u9ed1\u540d\u5355", phone) + + +def is_unsubscribed(phone: str) -> bool: + if not _unsubscribe_blocklist: + init_blocklist() + return phone in _unsubscribe_blocklist diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/config.py b/backend/package/yuxi/channel/extensions/tencent_sms/config.py new file mode 100644 index 00000000..1ba7af44 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/config.py @@ -0,0 +1,220 @@ +from __future__ import annotations + +import os + +from .types import SmsScene, SmsTemplateConfig, TemplateType, TencentSmsAccount + + +class TencentSmsConfigAdapter: + ENV_MAP = { + "secret_id": "TENCENT_SMS_SECRET_ID", + "secret_key": "TENCENT_SMS_SECRET_KEY", + "sms_sdk_app_id": "TENCENT_SMS_SDK_APP_ID", + "sign_name": "TENCENT_SMS_SIGN_NAME", + "region": "TENCENT_SMS_REGION", + } + + def __init__(self): + self._config: dict | None = None + + def set_config(self, config: dict) -> None: + self._config = config + + def list_account_ids(self, config: dict) -> list[str]: + accounts = config.get("accounts", []) + if not accounts: + sid = os.environ.get("TENCENT_SMS_SECRET_ID", "") + if sid: + return ["default"] + return [] + return [a.get("account_id", str(i)) for i, a in enumerate(accounts)] + + async def resolve_account(self, account_id: str) -> dict: + if self._config is None: + return {} + accounts = self._config.get("accounts", []) + if not accounts: + sid = os.environ.get("TENCENT_SMS_SECRET_ID", "") + if sid and account_id in ("default", "0", ""): + return { + "account_id": "default", + "secret_id": sid, + "secret_key": os.environ.get("TENCENT_SMS_SECRET_KEY", ""), + "sms_sdk_app_id": os.environ.get("TENCENT_SMS_SDK_APP_ID", ""), + "sign_name": os.environ.get("TENCENT_SMS_SIGN_NAME", ""), + "region": os.environ.get("TENCENT_SMS_REGION", "ap-guangzhou"), + } + return {} + for a in accounts: + if a.get("account_id", str(accounts.index(a))) == account_id: + return a + return {} + + def _parse_account(self, account: dict) -> TencentSmsAccount: + secret_id = account.get("secret_id") or os.environ.get("TENCENT_SMS_SECRET_ID", "") + secret_key = account.get("secret_key") or os.environ.get("TENCENT_SMS_SECRET_KEY", "") + sms_sdk_app_id = account.get("sms_sdk_app_id") or os.environ.get("TENCENT_SMS_SDK_APP_ID", "") + sign_name = account.get("sign_name") or os.environ.get("TENCENT_SMS_SIGN_NAME", "") + region = account.get("region") or os.environ.get("TENCENT_SMS_REGION", "ap-guangzhou") + + templates = [] + for tpl in account.get("templates", []): + templates.append( + SmsTemplateConfig( + scene=SmsScene(tpl.get("scene", "notification")), + template_id=tpl.get("template_id", ""), + sign_name=tpl.get("sign_name", sign_name), + template_type=TemplateType(tpl.get("template_type", "0")), + param_count=tpl.get("param_count", 1), + max_daily=tpl.get("max_daily", 1000), + description=tpl.get("description", ""), + template_format=tpl.get("template_format", ""), + ) + ) + + return TencentSmsAccount( + account_id=account.get("account_id", "default"), + secret_id=secret_id, + secret_key=secret_key, + sms_sdk_app_id=sms_sdk_app_id, + sign_name=sign_name, + region=region, + endpoint=account.get("endpoint", "sms.tencentcloudapi.com"), + callback_url=account.get("callback_url", "/webhook/tencent-sms/callback"), + reply_callback_url=account.get("reply_callback_url", "/webhook/tencent-sms/reply-callback"), + templates=templates, + dm_policy=account.get("dm_policy", "open"), + allow_from=account.get("allow_from", []), + daily_limit=account.get("daily_limit", 1000), + interval_seconds=account.get("interval_seconds", 30), + phone_daily_limit=account.get("phone_daily_limit", 10), + poll_interval_seconds=account.get("poll_interval_seconds", 120), + enabled=account.get("enabled", True), + extend_code=account.get("extend_code", ""), + sender_id=account.get("sender_id", ""), + allowed_callback_ips=account.get("allowed_callback_ips", []), + ) + + def is_configured(self, account: dict) -> bool: + ta = self._parse_account(account) + return ta.is_configured + + def is_enabled(self, account: dict) -> bool: + ta = self._parse_account(account) + return ta.enabled and ta.is_configured + + def disabled_reason(self, account: dict) -> str: + ta = self._parse_account(account) + if not ta.secret_id: + return "\u7f3a\u5c11 SecretId" + if not ta.secret_key: + return "\u7f3a\u5c11 SecretKey" + if not ta.sms_sdk_app_id: + return "\u7f3a\u5c11 SmsSdkAppId" + if not ta.sign_name: + return "\u672a\u914d\u7f6e\u77ed\u4fe1\u7b7e\u540d (SignName)" + if not ta.templates: + return "\u672a\u914d\u7f6e\u77ed\u4fe1\u6a21\u677f" + if not ta.enabled: + return "\u6e20\u9053\u5df2\u7981\u7528" + return "" + + def describe_account(self, account: dict) -> dict: + ta = self._parse_account(account) + return { + "account_id": ta.account_id, + "sms_sdk_app_id": ta.sms_sdk_app_id, + "sign_name": ta.sign_name, + "region": ta.region, + "dm_policy": ta.dm_policy, + "template_count": len(ta.templates), + "template_scenes": [t.scene.value for t in ta.templates], + } + + def config_schema(self) -> dict: + return { + "type": "object", + "properties": { + "secret_id": { + "type": "string", + "description": "\u817e\u8baf\u4e91 API SecretId\uff08CAM \u63a7\u5236\u53f0\u83b7\u53d6\uff09", + }, + "secret_key": { + "type": "string", + "description": "\u817e\u8baf\u4e91 API SecretKey", + "format": "password", + }, + "sms_sdk_app_id": { + "type": "string", + "description": "\u77ed\u4fe1\u5e94\u7528 ID\uff08\u63a7\u5236\u53f0\u5e94\u7528\u7ba1\u7406 -> SDK AppID\uff09", + }, + "sign_name": { + "type": "string", + "description": "\u77ed\u4fe1\u7b7e\u540d\uff08\u5df2\u5ba1\u6838\u901a\u8fc7\uff0c\u4e0d\u542b\u3010\u3011\uff09", + }, + "region": { + "type": "string", + "default": "ap-guangzhou", + "description": "API \u5730\u57df", + }, + "templates": { + "type": "array", + "items": { + "type": "object", + "properties": { + "scene": { + "type": "string", + "enum": ["verification", "notification", "alert", "marketing"], + }, + "template_id": {"type": "string"}, + "sign_name": {"type": "string"}, + "param_count": {"type": "integer", "minimum": 1, "maximum": 12}, + "description": {"type": "string"}, + "template_format": {"type": "string"}, + }, + "required": ["scene", "template_id", "param_count"], + }, + }, + "dm_policy": { + "type": "string", + "enum": ["open", "pairing", "allowlist", "disabled"], + "default": "open", + }, + "allow_from": { + "type": "array", + "items": {"type": "string"}, + "description": "\u5141\u8bb8\u63a5\u6536\u77ed\u4fe1\u7684\u624b\u673a\u53f7\u767d\u540d\u5355\uff08E.164 \u683c\u5f0f\uff09", + }, + "daily_limit": { + "type": "integer", + "minimum": 1, + "maximum": 100000, + "default": 1000, + }, + "phone_daily_limit": { + "type": "integer", + "minimum": 1, + "maximum": 50, + "default": 10, + }, + "extend_code": { + "type": "string", + "description": "\u77ed\u4fe1\u7801\u53f7\u6269\u5c55\u53f7\uff08\u56fd\u9645/\u6e2f\u6fb3\u53f0\u77ed\u4fe1\u7528\uff09", + }, + "sender_id": { + "type": "string", + "description": "\u56fd\u9645/\u6e2f\u6fb3\u53f0 Sender ID", + }, + "allowed_callback_ips": { + "type": "array", + "items": {"type": "string"}, + "description": "\u5141\u8bb8\u7684 Webhook \u56de\u8c03 IP \u767d\u540d\u5355", + }, + "enabled": {"type": "boolean", "default": True}, + }, + "required": ["secret_id", "secret_key", "sms_sdk_app_id", "sign_name"], + } + + def default_account_id(self, config: dict) -> str: + ids = self.list_account_ids(config) + return ids[0] if ids else "default" diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/dedupe.py b/backend/package/yuxi/channel/extensions/tencent_sms/dedupe.py new file mode 100644 index 00000000..0c00c6c0 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/dedupe.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import time +from collections import OrderedDict + + +class TencentSmsDeduplicator: + def __init__(self, max_size: int = 10000, ttl_seconds: int = 600): + self._cache: OrderedDict[str, float] = OrderedDict() + self._max_size = max_size + self._ttl_seconds = ttl_seconds + + def _evict_expired(self, now: float) -> None: + expired = [k for k, ts in self._cache.items() if now - ts > self._ttl_seconds] + for k in expired: + del self._cache[k] + + def _evict_oldest(self) -> None: + while len(self._cache) > self._max_size: + self._cache.popitem(last=False) + + def is_duplicate(self, dedupe_key: str) -> bool: + now = time.monotonic() + self._evict_expired(now) + if dedupe_key in self._cache: + return True + self._cache[dedupe_key] = now + self._evict_oldest() + return False + + def check(self, dedupe_key: str) -> bool: + now = time.monotonic() + self._evict_expired(now) + return dedupe_key in self._cache + + def mark_seen(self, dedupe_key: str) -> None: + now = time.monotonic() + self._evict_expired(now) + if dedupe_key not in self._cache: + self._cache[dedupe_key] = now + self._evict_oldest() + + @property + def ttl_seconds(self) -> int: + return self._ttl_seconds + + @property + def max_entries(self) -> int: + return self._max_size + + def reset(self) -> None: + self._cache.clear() diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/delivery.py b/backend/package/yuxi/channel/extensions/tencent_sms/delivery.py new file mode 100644 index 00000000..5d52fa3b --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/delivery.py @@ -0,0 +1,41 @@ +from __future__ import annotations + +import logging + +from .dedupe import TencentSmsDeduplicator +from .types import DeliveryStatus + +logger = logging.getLogger(__name__) + + +class SmsDeliveryTracker: + def __init__(self): + self._dedupe = TencentSmsDeduplicator(max_size=50000, ttl_seconds=86400) + self._store: dict[str, DeliveryStatus] = {} + + def handle_status_callback(self, item: dict) -> bool: + serial_no = item.get("SerialNo", "") + report_status = item.get("ReportStatus", "") + key = f"status:{serial_no}:{report_status}" + if self._dedupe.is_duplicate(key): + return False + + status = DeliveryStatus( + serial_no=serial_no, + phone_number=item.get("PhoneNumber", ""), + report_status=report_status, + description=item.get("Description", ""), + report_time=item.get("ReportTime", ""), + session_context=item.get("SessionContext", ""), + ) + self._store[serial_no] = status + logger.info( + "\u9001\u8fbe\u72b6\u6001\u66f4\u65b0: serial=%s, status=%s, phone=%s", + serial_no, + report_status, + status.phone_number, + ) + return True + + def get_status(self, serial_no: str) -> DeliveryStatus | None: + return self._store.get(serial_no) diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/errors.py b/backend/package/yuxi/channel/extensions/tencent_sms/errors.py new file mode 100644 index 00000000..3bc59612 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/errors.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +from enum import StrEnum + +from .types import NON_RETRYABLE_CODES + + +class TencentSmsErrorCode(StrEnum): + RATE_LIMITED = "rate_limited" + NETWORK_ERROR = "network_error" + SERVICE_UNAVAILABLE = "service_unavailable" + DELIVERY_FREQ_LIMIT = "delivery_freq_limit" + INVALID_TEMPLATE = "invalid_template" + INVALID_SIGN = "invalid_sign" + INVALID_PHONE = "invalid_phone" + BLACKLISTED = "blacklisted" + DAILY_LIMIT = "daily_limit" + PHONE_DAILY_LIMIT = "phone_daily_limit" + PHONE_HOURLY_LIMIT = "phone_hourly_limit" + PHONE_30S_LIMIT = "phone_30s_limit" + INSUFFICIENT_BALANCE = "insufficient_balance" + SDK_APPID_FAIL = "sdk_appid_fail" + + +def classify_sms_error(exception: Exception) -> TencentSmsErrorCode: + code = getattr(exception, "code", "") or "" + + if "RequestLimitExceeded" in code: + return TencentSmsErrorCode.RATE_LIMITED + if code in ("InternalError", "ResourceUnavailable", "FailedOperation.ServiceError"): + return TencentSmsErrorCode.SERVICE_UNAVAILABLE + if isinstance(exception, OSError): + return TencentSmsErrorCode.NETWORK_ERROR + + if code in NON_RETRYABLE_CODES: + return TencentSmsErrorCode(code.lower() if hasattr(TencentSmsErrorCode, code.lower()) else "network_error") + + return TencentSmsErrorCode.NETWORK_ERROR + + +def is_retryable(error_code: TencentSmsErrorCode) -> bool: + return error_code in { + TencentSmsErrorCode.RATE_LIMITED, + TencentSmsErrorCode.NETWORK_ERROR, + TencentSmsErrorCode.SERVICE_UNAVAILABLE, + TencentSmsErrorCode.DELIVERY_FREQ_LIMIT, + } diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/frequency.py b/backend/package/yuxi/channel/extensions/tencent_sms/frequency.py new file mode 100644 index 00000000..8afbc456 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/frequency.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +import time +from collections import defaultdict + +from .types import SmsScene, TencentSmsAccount + + +class SmsFrequencyGuard: + def __init__(self, window_seconds: int = 60): + self._window = window_seconds + self._records: dict[str, list[tuple[SmsScene, float]]] = defaultdict(list) + + async def check_and_record( + self, + phone: str, + scene: SmsScene, + account: TencentSmsAccount, + ) -> tuple[bool, str]: + now = time.time() + records = self._records[phone] + + interval_check = [t for t, _ in records if now - t < account.interval_seconds] + if interval_check: + wait = account.interval_seconds - (now - interval_check[-1]) + return False, f"\u53d1\u9001\u95f4\u9694\u4e0d\u8db3\uff0c\u8fd8\u9700\u7b49\u5f85 {wait:.0f}s" + + today_start = now - (now % 86400) + today_records = [(t, s) for t, s in records if t >= today_start] + if len(today_records) >= account.phone_daily_limit: + return ( + False, + f"\u8be5\u53f7\u7801\u5df2\u8fbe\u65e5\u53d1\u9001\u4e0a\u9650 {account.phone_daily_limit} \u6761", + ) + + hours_records = [(t, s) for t, s in today_records if now - t < 3600] + if len(hours_records) >= 5: + return False, f"\u8be5\u53f7\u7801 1 \u5c0f\u65f6\u5185\u5df2\u53d1\u9001 {len(hours_records)} \u6761" + + window_records = [(t, s) for t, s in records if now - t < self._window] + if window_records and any(s == scene for _, s in window_records): + return False, "\u540c\u4e00\u573a\u666f 60s \u5185\u5df2\u53d1\u9001\u8fc7" + + records.append((now, scene)) + expired = [(t, s) for t, s in records if now - t > 86400] + for e in expired: + records.remove(e) + return True, "ok" + + def remaining(self, phone: str, account: TencentSmsAccount) -> int: + now = time.time() + today_start = now - (now % 86400) + today_count = sum(1 for t, _ in self._records.get(phone, []) if t >= today_start) + return max(0, account.phone_daily_limit - today_count) + + def reset(self, phone: str) -> None: + self._records.pop(phone, None) diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/gateway.py b/backend/package/yuxi/channel/extensions/tencent_sms/gateway.py new file mode 100644 index 00000000..7a3b04b7 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/gateway.py @@ -0,0 +1,171 @@ +from __future__ import annotations + +import asyncio +import logging + +from .client import TencentSmsClient +from .config import TencentSmsConfigAdapter +from .delivery import SmsDeliveryTracker +from .types import TencentSmsAccount +from .webhook import TencentSmsWebhookHandler, set_webhook_handler + +logger = logging.getLogger("yuxi.channel.tencent-sms") + + +class TencentSmsGatewayAdapter: + def __init__(self): + self._config = TencentSmsConfigAdapter() + self._running = False + self._tasks: list[asyncio.Task] = [] + self._accounts: dict[str, TencentSmsAccount] = {} + self._clients: dict[str, TencentSmsClient] = {} + self._delivery_tracker = SmsDeliveryTracker() + + async def start(self, ctx) -> dict: + config = ctx.config if hasattr(ctx, "config") else {} + self._config.set_config(config) + account = self._config._parse_account(config) + if not account.is_configured: + raise ValueError( + f"腾讯云短信账户 {account.account_id} 未正确配置" + ) + + self._accounts[account.account_id] = account + + client = TencentSmsClient(account) + self._clients[account.account_id] = client + + probe_ok = await client.probe() + if not probe_ok: + logger.warning("腾讯云短信 Gateway probe 未通过") + + if hasattr(ctx, "outbound") and ctx.outbound: + ctx.outbound.set_client(account.account_id, client, account) + + webhook_handler = TencentSmsWebhookHandler( + allowed_ips=getattr(account, "allowed_callback_ips", None), + delivery_tracker=self._delivery_tracker, + ) + set_webhook_handler(webhook_handler) + + self._running = True + + task = asyncio.create_task(self._periodic_pull(client, account)) + self._tasks.append(task) + + logger.info( + "腾讯云短信 Gateway 启动成功, account=%s, app_id=%s, sign=%s", + account.account_id, + account.sms_sdk_app_id, + account.sign_name, + ) + + return { + "running": True, + "account": account, + "client": client, + "accounts": self._accounts, + "clients": self._clients, + "delivery_tracker": self._delivery_tracker, + } + + async def stop(self, ctx) -> None: + self._running = False + for task in self._tasks: + task.cancel() + self._tasks.clear() + self._accounts.clear() + self._clients.clear() + logger.info("腾讯云短信 Gateway 已停止") + + def get_client(self, account_id: str = "default") -> TencentSmsClient | None: + return self._clients.get(account_id) + + def get_account(self, account_id: str = "default") -> TencentSmsAccount | None: + return self._accounts.get(account_id) + + async def probe(self, account: dict) -> bool: + try: + ta = self._config._parse_account(account) + client = TencentSmsClient(ta) + return await client.probe() + except Exception as e: + logger.warning("腾讯云短信 probe 失败: %s", e) + return False + + async def _periodic_pull( + self, + client: TencentSmsClient, + account: TencentSmsAccount, + ): + interval = max(account.poll_interval_seconds, 30) + while self._running: + try: + await self._pull_all_statuses(client) + await self._pull_all_replies(client) + except asyncio.CancelledError: + break + except Exception as e: + logger.exception("定时拉取任务异常: %s", e) + + await asyncio.sleep(interval) + + async def _pull_all_statuses(self, client: TencentSmsClient) -> int: + total_processed = 0 + batch_count = 0 + max_batches = 10 + + while batch_count < max_batches: + statuses = await client.pull_send_status(limit=100) + if not statuses: + break + + logger.info("Pull 拉取到 %d 条送达状态", len(statuses)) + for item in statuses: + self._delivery_tracker.handle_status_callback( + { + "SerialNo": item["serial_no"], + "ReportStatus": item["report_status"], + "PhoneNumber": item["phone_number"], + "Description": item["description"], + "ReportTime": item["user_receive_time"], + "SessionContext": item["session_context"], + } + ) + total_processed += 1 + + batch_count += 1 + if len(statuses) < 100: + break + + if total_processed > 0: + logger.info("本轮共处理 %d 条送达状态", total_processed) + return total_processed + + async def _pull_all_replies(self, client: TencentSmsClient) -> int: + total_processed = 0 + batch_count = 0 + max_batches = 10 + + while batch_count < max_batches: + replies = await client.pull_reply_status(limit=100) + if not replies: + break + + logger.info("Pull 拉取到 %d 条上行回复", len(replies)) + for item in replies: + logger.info( + "上行回复: phone=%s, content=%s, time=%s", + item["phone_number"], + item["reply_content"], + item["reply_time"], + ) + total_processed += 1 + + batch_count += 1 + if len(replies) < 100: + break + + if total_processed > 0: + logger.info("本轮共处理 %d 条上行回复", total_processed) + return total_processed diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/outbound.py b/backend/package/yuxi/channel/extensions/tencent_sms/outbound.py new file mode 100644 index 00000000..0f83e911 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/outbound.py @@ -0,0 +1,159 @@ +from __future__ import annotations + +import asyncio +import logging + +from .client import TencentSmsClient +from .compliance import check_marketing_time_window, is_unsubscribed +from .frequency import SmsFrequencyGuard +from .security import check_allowlist +from .templates import SmsTemplateRegistry +from .types import NON_RETRYABLE_CODES, SendResult, SmsScene, TencentSmsAccount + +logger = logging.getLogger(__name__) + +MAX_RETRIES = 3 +BASE_DELAY = 1.0 + + +class TencentSmsOutboundAdapter: + def __init__(self): + self._clients: dict[str, TencentSmsClient] = {} + self._accounts: dict[str, TencentSmsAccount] = {} + self._frequency = SmsFrequencyGuard() + self._template_registry = SmsTemplateRegistry() + + def set_client(self, account_id: str, client: TencentSmsClient, account: TencentSmsAccount): + self._clients[account_id] = client + self._accounts[account_id] = account + for tpl in account.templates: + self._template_registry.register(account_id, tpl) + + def _resolve_client(self, account_id: str | None) -> tuple[TencentSmsClient, TencentSmsAccount]: + aid = account_id or next(iter(self._accounts), "default") + client = self._clients.get(aid) + account = self._accounts.get(aid) + if not client or not account: + raise RuntimeError("\u6ca1\u6709\u53ef\u7528\u7684\u817e\u8baf\u4e91\u77ed\u4fe1\u5ba2\u6237\u7aef") + return client, account + + async def send_text( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> SendResult: + logger.warning( + "Tencent SMS \u4e0d\u652f\u6301\u76f4\u63a5 send_text\uff0c\u5c1d\u8bd5\u4f7f\u7528 notification \u573a\u666f\u6a21\u677f\u53d1\u9001" + ) + return await self.send_by_template( + phone_number=target_id, + scene=SmsScene.NOTIFICATION, + template_params=[content], + account_id=account_id, + session_context=thread_id or "", + ) + + async def send_by_template( + self, + phone_number: str, + scene: SmsScene, + template_params: list[str], + account_id: str | None = None, + session_context: str = "", + ) -> SendResult: + client, account = self._resolve_client(account_id) + + normalized = await _normalize_phone(phone_number, client) + if not normalized.startswith("+"): + normalized = f"+86{normalized}" + + try: + check_allowlist(account, normalized) + except PermissionError as e: + return SendResult(success=False, phone_number=normalized, code="ALLOWLIST", message=str(e)) + + allowed, reason = await self._frequency.check_and_record(normalized, scene, account) + if not allowed: + return SendResult(success=False, phone_number=normalized, code="RATE_LIMITED", message=reason) + + if is_unsubscribed(normalized): + return SendResult( + success=False, + phone_number=normalized, + code="UNSUBSCRIBED", + message="\u8be5\u53f7\u7801\u5df2\u9000\u8ba2", + ) + + if scene == SmsScene.MARKETING: + allowed, reason = check_marketing_time_window() + if not allowed: + return SendResult(success=False, phone_number=normalized, code="MARKETING_TIME_LIMIT", message=reason) + + binding = self._template_registry.resolve(account_id or "default", scene) + if not binding: + return SendResult( + success=False, + phone_number=normalized, + code="NO_TEMPLATE", + message=f"\u573a\u666f {scene.value} \u672a\u914d\u7f6e\u6a21\u677f", + ) + + if len(template_params) != binding.param_count: + return SendResult( + success=False, + phone_number=normalized, + code="PARAM_MISMATCH", + message=f"\u6a21\u677f\u9700\u8981 {binding.param_count} \u4e2a\u53d8\u91cf\uff0c\u4f20\u5165 {len(template_params)} \u4e2a", + ) + + for attempt in range(MAX_RETRIES): + try: + results = await client.send_sms( + phone_numbers=[normalized], + template_id=binding.template_id, + template_params=template_params, + sign_name=binding.sign_name, + session_context=session_context, + extend_code=account.extend_code, + sender_id=account.sender_id, + ) + result = results[0] + if result.success: + return result + if result.code in NON_RETRYABLE_CODES: + return result + if attempt < MAX_RETRIES - 1: + await asyncio.sleep(BASE_DELAY * (2**attempt)) + except Exception as e: + if attempt == MAX_RETRIES - 1: + raise + logger.warning("send_by_template retry %d: %s", attempt + 1, e) + await asyncio.sleep(BASE_DELAY * (2**attempt)) + + return SendResult(success=False, code="MAX_RETRIES", message="\u5df2\u8fbe\u6700\u5927\u91cd\u8bd5\u6b21\u6570") + + +async def _normalize_phone(phone: str, client: TencentSmsClient | None = None) -> str: + p = phone.strip().replace(" ", "").replace("-", "") + if not p.startswith("+"): + p = f"+86{p}" + + if client and not p.startswith("+86"): + try: + info_list = await client.describe_phone_number_info([p]) + if info_list: + info = info_list[0] + nation_code = info.get("nation_code", "") + subscriber = info.get("subscriber_number", "") + if nation_code and subscriber: + return f"+{nation_code}{subscriber}" + except Exception as e: + logger.debug( + "\u53f7\u7801\u4fe1\u606f\u67e5\u8be2\u5931\u8d25\uff0c\u4f7f\u7528\u539f\u59cb\u53f7\u7801: %s", e + ) + + return p diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/plugin.json b/backend/package/yuxi/channel/extensions/tencent_sms/plugin.json new file mode 100644 index 00000000..f49a1d5d --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/plugin.json @@ -0,0 +1,31 @@ +{ + "id": "tencent-sms", + "name": "腾讯云短信", + "version": "1.0.0", + "label": "腾讯云短信", + "aliases": ["tencent-sms", "腾讯短信", "tencentcloud-sms"], + "description": "腾讯云 SMS 渠道插件，支持模板化短信发送、送达状态追踪与上行回复处理", + "author": "ForcePilot Team", + "order": 160, + "dependencies": ["tencentcloud-sdk-python-sms>=3.1.92"], + "capabilities": { + "chat_types": ["direct"], + "message_types": ["text"], + "streaming": true, + "streaming_mode": "block", + "block_streaming": true, + "block_streaming_chunk_min_chars": 60, + "block_streaming_chunk_max_chars": 67, + "reactions": false, + "polls": false, + "reply": false, + "media": false, + "voice": false, + "edit": false, + "unsend": false, + "threads": false, + "native_commands": false, + "group_management": false + }, + "python_requires": ">=3.12" +} \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/plugin.py b/backend/package/yuxi/channel/extensions/tencent_sms/plugin.py new file mode 100644 index 00000000..48842425 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/plugin.py @@ -0,0 +1,280 @@ +from __future__ import annotations + +from yuxi.channel.capabilities import ChannelCapabilities +from yuxi.channel.errors import ClassifiedError, ErrorSeverity +from yuxi.channel.extensions.base import BaseChannelPlugin + +from .agent_prompt import TencentSmsAgentPrompt +from .config import TencentSmsConfigAdapter +from .dedupe import TencentSmsDeduplicator +from .errors import TencentSmsErrorCode, classify_sms_error, is_retryable as sms_is_retryable +from .gateway import TencentSmsGatewayAdapter +from .outbound import TencentSmsOutboundAdapter +from .security import check_allowlist as sms_check_allowlist, resolve_dm_policy +from .streaming import TencentSmsBlockStreamer +from .templates import SmsTemplateRegistry +from .webhook import get_webhook_handler + + +class TencentSmsPlugin(BaseChannelPlugin): + id = "tencent-sms" + name = "腾讯云短信" + order = 160 + label = "腾讯云短信" + aliases = ["tencent-sms", "腾讯短信", "tencentcloud-sms"] + + def __init__(self): + self._config = TencentSmsConfigAdapter() + self._gateway = TencentSmsGatewayAdapter() + self._outbound = TencentSmsOutboundAdapter() + self._template_registry = SmsTemplateRegistry() + self._agent_prompt = TencentSmsAgentPrompt(self._template_registry) + self._block_streamer = TencentSmsBlockStreamer() + self._deduplicator = TencentSmsDeduplicator(max_size=10000, ttl_seconds=600) + + @property + def capabilities(self) -> ChannelCapabilities: + return ChannelCapabilities( + chat_types=["direct"], + message_types=["text"], + reactions=False, + typing_indicator=False, + threads=False, + edit=False, + unsend=False, + reply=False, + media=False, + voice=False, + native_commands=False, + polls=False, + streaming=True, + streaming_mode="block", + block_streaming=True, + block_streaming_chunk_min_chars=60, + block_streaming_chunk_max_chars=67, + block_streaming_coalesce_idle_ms=800, + ) + + # ── ConfigProtocol ──────────────────────────────────── + + def list_account_ids(self, config: dict) -> list[str]: + return self._config.list_account_ids(config) + + async def resolve_account(self, account_id: str) -> dict: + return await self._config.resolve_account(account_id) + + def is_configured(self, account: dict) -> bool: + return self._config.is_configured(account) + + def is_enabled(self, account: dict) -> bool: + return self._config.is_enabled(account) + + def disabled_reason(self, account: dict) -> str: + return self._config.disabled_reason(account) + + def describe_account(self, account: dict) -> dict: + return self._config.describe_account(account) + + def config_schema(self) -> dict: + return self._config.config_schema() + + # ── GatewayProtocol ─────────────────────────────────── + + async def start(self, ctx) -> object: + return await self._gateway.start(ctx) + + async def stop(self, ctx) -> None: + await self._gateway.stop(ctx) + + # ── OutboundProtocol ────────────────────────────────── + + async def send_text( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + await self._outbound.send_text( + target_id, + content, + reply_to_id=reply_to_id, + thread_id=thread_id, + account_id=account_id, + ) + + # ── StatusProtocol ──────────────────────────────────── + + async def probe(self, account: dict) -> bool: + return await self._gateway.probe(account) + + # ── SecurityProtocol ────────────────────────────────── + + async def check_allowlist(self, peer_id: str, channel_type: str) -> bool: + account_id = next(iter(self._gateway._accounts), "default") + account = self._gateway.get_account(account_id) + if not account: + return True + try: + sms_check_allowlist(account, peer_id) + return True + except PermissionError: + return False + + def resolve_dm_policy(self) -> dict: + account_id = next(iter(self._gateway._accounts), "default") + account = self._gateway.get_account(account_id) + if account: + return {"mode": resolve_dm_policy(account), "allow_from": account.allow_from} + return {"mode": "open", "allow_from": []} + + # ── AgentPromptProtocol ─────────────────────────────── + + @property + def channel_format_instructions(self) -> str | None: + return ( + "You are connected to Tencent Cloud SMS (腾讯云短信). " + "Messages are sent via carrier SMS, NOT instant messaging. " + "ALL messages MUST use pre-approved SMS templates — you CANNOT send free text. " + "Templates use {1}, {2}, ... {12} as variable placeholders. " + "Tencent Cloud SMS limits: single text ≤ 70 chars (UCS-2) or 160 chars (GSM-7). " + "Each variable counts toward the character limit. " + "Keep responses within template constraints. " + "Phone numbers are in E.164 format (+8618501234444). " + "This channel is best for ONE-TIME notifications, NOT multi-turn conversations. " + "Support opt-out keywords: T, TD, 退订, 取消, N, NO. " + "Support help keywords: HELP, 帮助." + ) + + def build_system_prompt(self, context) -> str | None: + account_id = getattr(context, "account_id", "default") if context else "default" + return self._agent_prompt.build_system_prompt(account_id) + + # ── StreamingProtocol ───────────────────────────────── + + @property + def block_streaming_enabled(self) -> bool: + return True + + @property + def block_streaming_chunk_min_chars(self) -> int: + return 60 + + @property + def block_streaming_chunk_max_chars(self) -> int: + return 67 + + @property + def block_streaming_chunk_break_preference(self) -> str: + return "sentence" + + def create_block_chunker(self) -> object: + return self._block_streamer + + # ── ErrorHandlingProtocol ───────────────────────────── + + def classify_error(self, error: BaseException) -> ClassifiedError: + sms_code = classify_sms_error(error) + severity_map = { + "rate_limited": ErrorSeverity.RATE_LIMITED, + "network_error": ErrorSeverity.NETWORK, + "service_unavailable": ErrorSeverity.RETRYABLE, + "delivery_freq_limit": ErrorSeverity.RATE_LIMITED, + } + severity = severity_map.get(sms_code, ErrorSeverity.FATAL) + return ClassifiedError( + severity=severity, + error_message=f"{sms_code}: {error}", + original_error=error, + ) + + def is_retryable(self, error: BaseException) -> bool: + sms_code = classify_sms_error(error) + try: + return sms_is_retryable(TencentSmsErrorCode(sms_code)) + except ValueError: + return False + + # ── DedupeProtocol ──────────────────────────────────── + + def is_duplicate(self, key: str) -> bool: + return self._deduplicator.check(key) + + def mark_seen(self, key: str) -> None: + self._deduplicator.mark_seen(key) + + @property + def ttl_seconds(self) -> int: + return self._deduplicator.ttl_seconds + + @property + def max_entries(self) -> int: + return self._deduplicator.max_entries + + def reset(self) -> None: + self._deduplicator.reset() + + # ── WebhookProtocol ─────────────────────────────────── + + async def handle_webhook( + self, + request: object, + account_id: str | None = None, + ) -> object: + handler = get_webhook_handler() + path = getattr(request, "url", "") + if "reply-callback" in str(path): + return await handler.handle_reply_callback(request) + return await handler.handle_status_callback(request) + + def verify_signature(self, body: bytes, signature: str, secret: str) -> bool: + import hashlib + import hmac + + if not signature or not secret: + return False + expected = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() + return hmac.compare_digest(expected, signature) + + # ── InboundHandlerProtocol ──────────────────────────── + + async def handle_raw_event( + self, + event: dict, + account: dict, + ) -> object | None: + return self.parse_to_unified(event, account.get("account_id", "default")) + + def parse_to_unified( + self, + raw_event: dict, + account_id: str, + ) -> object | None: + event_type = raw_event.get("Type", -1) + if event_type == 1: + phone = raw_event.get("PhoneNumber", "") + content = raw_event.get("ReplyContent", "") + if not phone or not content: + return None + from yuxi.channel.message.models import PeerInfo, PeerKind, UnifiedMessage + + sender = PeerInfo(id=phone, kind=PeerKind.DIRECT, display_name=phone, raw={}) + return UnifiedMessage( + msg_id=f"sms-reply:{phone}:{raw_event.get('ReplyTime', '')}", + channel_type="tencent-sms", + account_id=account_id, + content=content, + sender=sender, + raw=raw_event, + ) + return None + + def resolve_event_type(self, raw_event: dict) -> str: + event_type = raw_event.get("Type", -1) + if event_type == 0: + return "delivery_status" + if event_type == 1: + return "message" + return "unknown" \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/security.py b/backend/package/yuxi/channel/extensions/tencent_sms/security.py new file mode 100644 index 00000000..95cf91ff --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/security.py @@ -0,0 +1,15 @@ +from __future__ import annotations + +from .types import TencentSmsAccount + + +def check_allowlist(account: TencentSmsAccount, phone: str): + allow = account.allow_from + if not allow: + return + if phone not in allow: + raise PermissionError(f"\u53f7\u7801 {phone[:6]}*** \u4e0d\u5728\u767d\u540d\u5355\u4e2d") + + +def resolve_dm_policy(account: TencentSmsAccount) -> str: + return account.dm_policy or "open" diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/streaming.py b/backend/package/yuxi/channel/extensions/tencent_sms/streaming.py new file mode 100644 index 00000000..1c043bea --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/streaming.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +import asyncio +import logging + +from .types import SmsScene + +logger = logging.getLogger(__name__) + +MAX_SMS_SEGMENTS = 10 +UCS2_CHARS_PER_SEGMENT = 67 + + +class TencentSmsBlockStreamer: + def __init__(self, max_segments: int = MAX_SMS_SEGMENTS, chars_per_segment: int = UCS2_CHARS_PER_SEGMENT): + self._max_segments = max_segments + self._chars_per_segment = chars_per_segment + + def chunk_text(self, text: str) -> list[str]: + if not text: + return [] + segments: list[str] = [] + remaining = text + + while remaining and len(segments) < self._max_segments: + if len(remaining) <= self._chars_per_segment: + segments.append(remaining.strip()) + break + + cut = self._chars_per_segment + for delimiter in ["\u3002", "\n", "\uff01", "\uff1f", "\uff0c", " ", ",", "."]: + idx = remaining[: self._chars_per_segment].rfind(delimiter) + if idx > self._chars_per_segment // 2: + cut = idx + len(delimiter) + break + + segment = remaining[:cut].strip() + if segment: + segments.append(segment) + remaining = remaining[cut:].strip() + + if remaining and len(segments) < self._max_segments: + truncated = remaining[: self._chars_per_segment - 3] + "..." + segments.append(truncated) + + return segments + + async def stream_send( + self, + phone: str, + segments: list[str], + send_func, + interval_ms: int = 500, + ): + for i, segment in enumerate(segments): + prefix = f"[{i + 1}/{len(segments)}] " if i > 0 else "" + content = f"{prefix}{segment}" + try: + result = await send_func(phone, content) + logger.debug( + "stream segment %d/%d: %s", + i + 1, + len(segments), + result.serial_no if result.success else f"FAIL:{result.code}", + ) + except Exception as e: + logger.exception("stream segment %d/%d failed: %s", i + 1, len(segments), e) + if i < len(segments) - 1: + await asyncio.sleep(interval_ms / 1000.0) + + async def stream_send_by_template( + self, + phone: str, + scene: SmsScene, + segments: list[str], + send_by_template_func, + interval_ms: int = 500, + ): + for i, segment in enumerate(segments): + prefix = f"[{i + 1}/{len(segments)}] " if i > 0 else "" + content = f"{prefix}{segment}" + try: + result = await send_by_template_func(phone, scene, [content]) + logger.debug( + "stream segment %d/%d: %s", + i + 1, + len(segments), + result.serial_no if result.success else f"FAIL:{result.code}", + ) + except Exception as e: + logger.exception("stream segment %d/%d failed: %s", i + 1, len(segments), e) + if i < len(segments) - 1: + await asyncio.sleep(interval_ms / 1000.0) diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/templates.py b/backend/package/yuxi/channel/extensions/tencent_sms/templates.py new file mode 100644 index 00000000..925bc0ee --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/templates.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +from .types import SmsScene, SmsTemplateConfig + + +class SmsTemplateRegistry: + def __init__(self): + self._bindings: dict[str, dict[SmsScene, SmsTemplateConfig]] = {} + + def register(self, account_id: str, template: SmsTemplateConfig): + scene_bindings = self._bindings.setdefault(account_id, {}) + scene_bindings[template.scene] = template + + def resolve(self, account_id: str, scene: SmsScene) -> SmsTemplateConfig | None: + return self._bindings.get(account_id, {}).get(scene) + + def list_all(self, account_id: str) -> list[SmsTemplateConfig]: + return list(self._bindings.get(account_id, {}).values()) + + def build_agent_prompt(self, account_id: str) -> str: + templates = self.list_all(account_id) + if not templates: + return "\u5f53\u524d\u65e0\u53ef\u7528\u7684\u77ed\u4fe1\u6a21\u677f\u3002" + lines = [] + for t in templates: + lines.append( + f"- \u573a\u666f:{t.scene.value} | \u6a21\u677fID:{t.template_id} | " + f"\u7b7e\u540d:\u3010{t.sign_name}\u3011| \u53d8\u91cf\u6570:{t.param_count} | {t.description or ''}" + ) + return ( + "\u53ef\u7528\u77ed\u4fe1\u6a21\u677f\uff1a\n" + + "\n".join(lines) + + "\n\n\u53d1\u9001\u77ed\u4fe1\u65f6\u8bf7\u9009\u62e9\u5bf9\u5e94\u573a\u666f\uff0c\u4e25\u683c\u6309\u53d8\u91cf\u4e2a\u6570\u586b\u5145 TemplateParamSet\u3002" + ) diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/types.py b/backend/package/yuxi/channel/extensions/tencent_sms/types.py new file mode 100644 index 00000000..ae422227 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/types.py @@ -0,0 +1,163 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import StrEnum + + +class SmsScene(StrEnum): + VERIFICATION = "verification" + NOTIFICATION = "notification" + ALERT = "alert" + MARKETING = "marketing" + + +class SmsSendCode(StrEnum): + OK = "Ok" + PHONE_BLACKLIST = "FailedOperation.PhoneNumberInBlacklist" + SIGNATURE_INCORRECT = "FailedOperation.SignatureIncorrectOrUnapproved" + TEMPLATE_INCORRECT = "FailedOperation.TemplateIncorrectOrUnapproved" + INSUFFICIENT_BALANCE = "FailedOperation.InsufficientBalanceInSmsPackage" + SDK_APPID_FAIL = "UnauthorizedOperation.SmsSdkAppIdVerifyFail" + DAILY_LIMIT = "LimitExceeded.DailyLimit" + DELIVERY_FREQ_LIMIT = "LimitExceeded.DeliveryFrequencyLimit" + PHONE_DAILY_LIMIT = "LimitExceeded.PhoneNumberDailyLimit" + PHONE_HOURLY_LIMIT = "LimitExceeded.PhoneNumberOneHourLimit" + PHONE_30S_LIMIT = "LimitExceeded.PhoneNumberThirtySecondLimit" + INVALID_PHONE = "InvalidParameterValue.IncorrectPhoneNumber" + + +class ReportStatus(StrEnum): + SUCCESS = "SUCCESS" + FAIL = "FAIL" + + +class TemplateType(StrEnum): + COMMON = "0" + MARKETING = "1" + + +@dataclass +class SmsTemplateConfig: + scene: SmsScene + template_id: str + sign_name: str = "" + template_type: TemplateType = TemplateType.COMMON + param_count: int = 1 + max_daily: int = 1000 + description: str = "" + template_format: str = "" + + +@dataclass +class TencentSmsAccount: + account_id: str + secret_id: str = "" + secret_key: str = "" + sms_sdk_app_id: str = "" + sign_name: str = "" + region: str = "ap-guangzhou" + endpoint: str = "sms.tencentcloudapi.com" + callback_url: str = "/webhook/tencent-sms/callback" + reply_callback_url: str = "/webhook/tencent-sms/reply-callback" + templates: list[SmsTemplateConfig] = field(default_factory=list) + dm_policy: str = "open" + allow_from: list[str] = field(default_factory=list) + daily_limit: int = 1000 + interval_seconds: int = 30 + phone_daily_limit: int = 10 + poll_interval_seconds: int = 120 + enabled: bool = True + extend_code: str = "" + sender_id: str = "" + allowed_callback_ips: list[str] = field(default_factory=list) + + @property + def is_configured(self) -> bool: + return bool(self.secret_id and self.secret_key and self.sms_sdk_app_id and self.sign_name) + + +@dataclass +class SendResult: + success: bool + serial_no: str = "" + phone_number: str = "" + code: str = "" + message: str = "" + fee: int = 0 + iso_code: str = "" + session_context: str = "" + + +@dataclass +class BatchSendResult: + total: int = 0 + succeeded: int = 0 + failed: int = 0 + results: list[SendResult] = field(default_factory=list) + + +@dataclass +class DeliveryStatus: + serial_no: str + phone_number: str + report_status: ReportStatus + description: str = "" + report_time: str = "" + user_receive_time: str = "" + fee: int = 0 + session_context: str = "" + + +@dataclass +class ReplyRecord: + phone_number: str + reply_content: str + reply_time: str = "" + reply_unix_time: int = 0 + extend_code: str = "" + country_code: str = "86" + + +NON_RETRYABLE_CODES: frozenset[str] = frozenset( + { + SmsSendCode.PHONE_BLACKLIST, + SmsSendCode.SIGNATURE_INCORRECT, + SmsSendCode.TEMPLATE_INCORRECT, + SmsSendCode.INSUFFICIENT_BALANCE, + SmsSendCode.SDK_APPID_FAIL, + SmsSendCode.DAILY_LIMIT, + SmsSendCode.PHONE_DAILY_LIMIT, + SmsSendCode.PHONE_HOURLY_LIMIT, + SmsSendCode.PHONE_30S_LIMIT, + SmsSendCode.INVALID_PHONE, + } +) + +OPT_OUT_KEYWORDS: frozenset[str] = frozenset( + { + "t", + "td", + "\u9000\u8ba2", + "\u53d6\u6d88", + "n", + "no", + "qx", + } +) + +HELP_KEYWORDS: frozenset[str] = frozenset( + { + "help", + "\u5e2e\u52a9", + } +) + +OPT_IN_KEYWORDS: frozenset[str] = frozenset( + { + "start", + "\u8ba2\u9605", + "dy", + "y", + "yes", + } +) diff --git a/backend/package/yuxi/channel/extensions/tencent_sms/webhook.py b/backend/package/yuxi/channel/extensions/tencent_sms/webhook.py new file mode 100644 index 00000000..fd930d6e --- /dev/null +++ b/backend/package/yuxi/channel/extensions/tencent_sms/webhook.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +import logging + +from fastapi import APIRouter, Request, Response + +from .compliance import add_unsubscribe, detect_opt_out, remove_unsubscribe +from .dedupe import TencentSmsDeduplicator +from .delivery import SmsDeliveryTracker + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/webhook/tencent-sms", tags=["tencent-sms-webhook"]) + + +class TencentSmsWebhookHandler: + def __init__(self, allowed_ips: list[str] | None = None, delivery_tracker: SmsDeliveryTracker | None = None): + self._allowed_ips = set(allowed_ips or []) + self._delivery_tracker = delivery_tracker + self._status_dedupe = TencentSmsDeduplicator(max_size=10000, ttl_seconds=3600) + self._reply_dedupe = TencentSmsDeduplicator(max_size=10000, ttl_seconds=3600) + + async def handle_status_callback(self, request: Request) -> Response: + client_ip = request.client.host if request.client else "" + if self._allowed_ips and client_ip not in self._allowed_ips: + logger.warning("\u72b6\u6001\u56de\u8c03 IP \u4e0d\u5728\u767d\u540d\u5355: %s", client_ip) + return Response(status_code=403) + + try: + body = await request.json() + except Exception as e: + logger.warning("\u72b6\u6001\u56de\u8c03 JSON \u89e3\u6790\u5931\u8d25: %s", e) + return Response(status_code=400) + + items = body if isinstance(body, list) else [body] + for item in items: + if item.get("Type") != 0: + continue + serial_no = item.get("SerialNo", "") + report_status = item.get("ReportStatus", "") + dedupe_key = f"status:{serial_no}:{report_status}" + if self._status_dedupe.is_duplicate(dedupe_key): + continue + logger.info( + "\u9001\u8fbe\u72b6\u6001: serial=%s, status=%s, phone=%s, desc=%s", + serial_no, + report_status, + item.get("PhoneNumber", ""), + item.get("Description", ""), + ) + if self._delivery_tracker: + self._delivery_tracker.handle_status_callback( + { + "SerialNo": serial_no, + "ReportStatus": report_status, + "PhoneNumber": item.get("PhoneNumber", ""), + "Description": item.get("Description", ""), + "ReportTime": item.get("ReportTime", ""), + "SessionContext": item.get("SessionContext", ""), + } + ) + + return Response(content='{"result":0,"errmsg":"OK"}', media_type="application/json") + + async def handle_reply_callback(self, request: Request) -> Response: + client_ip = request.client.host if request.client else "" + if self._allowed_ips and client_ip not in self._allowed_ips: + logger.warning("\u56de\u590d\u56de\u8c03 IP \u4e0d\u5728\u767d\u540d\u5355: %s", client_ip) + return Response(status_code=403) + + try: + body = await request.json() + except Exception as e: + logger.warning("\u56de\u590d\u56de\u8c03 JSON \u89e3\u6790\u5931\u8d25: %s", e) + return Response(status_code=400) + + items = body if isinstance(body, list) else [body] + for item in items: + if item.get("Type") != 1: + continue + phone = item.get("PhoneNumber", "") + content = item.get("ReplyContent", "") + reply_time = item.get("ReplyTime", "") + + dedupe_key = f"reply:{phone}:{reply_time}" + if self._reply_dedupe.is_duplicate(dedupe_key): + continue + + is_opt, response_text = detect_opt_out(content) + if is_opt: + content_lower = content.strip().lower() + if content_lower in ("t", "td", "\u9000\u8ba2", "\u53d6\u6d88", "n", "no", "qx"): + add_unsubscribe(phone) + elif content_lower in ("start", "\u8ba2\u9605", "dy", "y", "yes"): + remove_unsubscribe(phone) + logger.info("Opt-out/opt-in: phone=%s, content=%s, response=%s", phone, content, response_text) + + logger.info("\u4e0a\u884c\u56de\u590d: phone=%s, content=%s, time=%s", phone, content, reply_time) + + return Response(content='{"result":0,"errmsg":"OK"}', media_type="application/json") + + +_webhook_handler: TencentSmsWebhookHandler | None = None + + +def set_webhook_handler(handler: TencentSmsWebhookHandler) -> None: + global _webhook_handler + _webhook_handler = handler + + +def get_webhook_handler() -> TencentSmsWebhookHandler: + if _webhook_handler is None: + return TencentSmsWebhookHandler() + return _webhook_handler + + +@router.post("/callback") +async def status_callback(request: Request): + return await get_webhook_handler().handle_status_callback(request) + + +@router.post("/reply-callback") +async def reply_callback(request: Request): + return await get_webhook_handler().handle_reply_callback(request)