清洗短信模板地址参数

talkingq-url/banban/service/sms_notification.py

@@ -1,5 +1,6 @@
 import asyncio
 import json
+import re
 import time
 from collections.abc import Mapping
 from datetime import datetime
@@ -12,6 +13,13 @@ from services.database_service_base import DatabaseServiceBase
 from utils.logger import session_logger
 
 
+_SMS_ADDRESS_FALLBACK = "请打开小程序查看"
+_SMS_ADDRESS_MAX_LENGTH = 30
+_SMS_ADDRESS_ROAD_PATTERN = re.compile(r".*(?:大道|街道|路|街|巷|弄)")
+_SMS_COORDINATE_PATTERN = re.compile(r"^\s*-?\d+(?:\.\d+)?\s*[,，]\s*-?\d+(?:\.\d+)?\s*$")
+_SMS_ADDRESS_ALLOWED_PATTERN = re.compile(r"[^0-9A-Za-z\u4e00-\u9fff]")
+
+
 class SmsNotificationService(DatabaseServiceBase):
     def __init__(self):
         super().__init__(service_name="sms_notification_service")
@@ -358,10 +366,24 @@ class SmsNotificationService(DatabaseServiceBase):
         return {
             "code": event_label,
             "conference": (child_name or "伴伴设备")[:20],
-            "address": (address or "请打开小程序查看")[:60],
+            "address": self._sanitize_sms_address(address),
             "time": event_time,
         }
 
+    def _sanitize_sms_address(self, address: str | None) -> str:
+        raw_address = str(address or "").strip()
+        if not raw_address or _SMS_COORDINATE_PATTERN.match(raw_address):
+            return _SMS_ADDRESS_FALLBACK
+
+        road_match = _SMS_ADDRESS_ROAD_PATTERN.match(raw_address)
+        if road_match and len(road_match.group(0)) < len(raw_address):
+            raw_address = f"{road_match.group(0)}附近"
+
+        sanitized = _SMS_ADDRESS_ALLOWED_PATTERN.sub("", raw_address)
+        if not sanitized:
+            return _SMS_ADDRESS_FALLBACK
+        return sanitized[:_SMS_ADDRESS_MAX_LENGTH]
+
     def _format_now(self) -> str:
         return datetime.now().strftime("%Y-%m-%d %H:%M")
 

talkingq-url/tests/test_sms_notification.py

@@ -82,6 +82,39 @@ async def test_notify_alarm_sends_to_unique_family_phone_numbers(monkeypatch):
     assert sent[0]["template_params"]["address"] == "杭州市"
 
 
+@pytest.mark.asyncio
+async def test_notify_alarm_sanitizes_address_for_sms_template(monkeypatch):
+    configure_enabled_sms(monkeypatch)
+    service = SmsNotificationService()
+    sent = []
+
+    async def fake_get_session():
+        return FakeSession()
+
+    async def fake_send_sms(*, phone_number, template_code, template_params):
+        del phone_number, template_code
+        sent.append(template_params)
+        return {"ok": True, "code": "OK", "message": "OK", "request_id": "request-id"}
+
+    monkeypatch.setattr(service, "get_session", fake_get_session)
+    monkeypatch.setattr(service, "_send_sms", fake_send_sms)
+
+    await service.notify_alarm(
+        device_id="TalkingQ_XQSN00001005",
+        alarm_id=7,
+        child_name="孩子",
+        address="浙江省杭州市萧山区盈丰街道扬帆路顺发·美哉美城",
+    )
+
+    assert sent[0]["address"] == "浙江省杭州市萧山区盈丰街道扬帆路附近"
+
+
+def test_sms_address_sanitizer_falls_back_for_coordinates():
+    service = SmsNotificationService()
+
+    assert service._sanitize_sms_address("30.245,120.215") == "请打开小程序查看"
+
+
 @pytest.mark.asyncio
 async def test_notify_alarm_dedupes_same_device_and_phone(monkeypatch):
     configure_enabled_sms(monkeypatch)