diff --git a/notifications/telegram.py b/notifications/telegram.py index c1215d3..211d444 100644 --- a/notifications/telegram.py +++ b/notifications/telegram.py @@ -17,6 +17,18 @@ except ImportError: # pragma: no cover - compatibility with older pinned shared wheels _merge_strategy_plugin_i18n = None +_TELEGRAM_MARKET_SYMBOL_LINK_RE = re.compile(r"(? str: + text = str(value or "") + return _TELEGRAM_MARKET_SYMBOL_LINK_RE.sub( + lambda match: f"{match.group(1)}.{_TELEGRAM_MARKET_SYMBOL_LINK_JOINER}{match.group(2)}", + text, + ) + + _PRICE_SOURCE_LABELS = { "longbridge_candlesticks": ("LongBridge 日线K线", "LongBridge daily candlesticks"), "schwab_daily_history_with_live_quote_overlay": ("Schwab 日线历史", "Schwab daily history"), @@ -522,7 +534,11 @@ def send_tg_message(message: str) -> None: return url = f"https://api.telegram.org/bot{token}/sendMessage" try: - requests_module.post(url, json={"chat_id": chat_id, "text": message}, timeout=15) + requests_module.post( + url, + json={"chat_id": chat_id, "text": _break_telegram_market_symbol_auto_links(message)}, + timeout=15, + ) except Exception as exc: print(f"Telegram send failed: {type(exc).__name__}", flush=True) diff --git a/tests/test_notifications.py b/tests/test_notifications.py new file mode 100644 index 0000000..e5b99f2 --- /dev/null +++ b/tests/test_notifications.py @@ -0,0 +1,34 @@ +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) +PLATFORM_KIT_SRC = ROOT.parent / "QuantPlatformKit" / "src" +if str(PLATFORM_KIT_SRC) not in sys.path: + sys.path.insert(0, str(PLATFORM_KIT_SRC)) + +from notifications.telegram import build_sender + + +class FakeRequests: + def __init__(self): + self.calls = [] + + def post(self, url, json, timeout): + self.calls.append((url, json, timeout)) + return object() + + +def test_build_sender_breaks_market_symbol_auto_links(): + fake_requests = FakeRequests() + sender = build_sender("token-1", "chat-1", requests_module=fake_requests) + sender("SOXL.US and 00700.HK") + + assert len(fake_requests.calls) == 1 + url, payload, timeout = fake_requests.calls[0] + assert "token-1" in url + assert payload["chat_id"] == "chat-1" + assert payload["text"] == "SOXL.\u2060US and 00700.\u2060HK" + assert timeout == 15