diff --git a/talkingq-url/banban/service/sms_notification.py b/talkingq-url/banban/service/sms_notification.py index d07412d..28dacf2 100644 --- a/talkingq-url/banban/service/sms_notification.py +++ b/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") diff --git a/talkingq-url/tests/test_sms_notification.py b/talkingq-url/tests/test_sms_notification.py index 05f99c0..229dc23 100644 --- a/talkingq-url/tests/test_sms_notification.py +++ b/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)