Skip to content

Commit 5fb3d08

Browse files
authored
Prevent Telegram market symbol auto-links
1 parent 63334d8 commit 5fb3d08

2 files changed

Lines changed: 51 additions & 1 deletion

File tree

notifications/telegram.py

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,18 @@
1717
except ImportError: # pragma: no cover - compatibility with older pinned shared wheels
1818
_merge_strategy_plugin_i18n = None
1919

20+
_TELEGRAM_MARKET_SYMBOL_LINK_RE = re.compile(r"(?<![A-Za-z0-9_])([A-Z0-9]{1,12})\.([A-Z]{2,4})(?![A-Za-z0-9_])")
21+
_TELEGRAM_MARKET_SYMBOL_LINK_JOINER = "\u2060"
22+
23+
24+
def _break_telegram_market_symbol_auto_links(value: object) -> str:
25+
text = str(value or "")
26+
return _TELEGRAM_MARKET_SYMBOL_LINK_RE.sub(
27+
lambda match: f"{match.group(1)}.{_TELEGRAM_MARKET_SYMBOL_LINK_JOINER}{match.group(2)}",
28+
text,
29+
)
30+
31+
2032
_PRICE_SOURCE_LABELS = {
2133
"longbridge_candlesticks": ("LongBridge 日线K线", "LongBridge daily candlesticks"),
2234
"schwab_daily_history_with_live_quote_overlay": ("Schwab 日线历史", "Schwab daily history"),
@@ -522,7 +534,11 @@ def send_tg_message(message: str) -> None:
522534
return
523535
url = f"https://api.telegram.org/bot{token}/sendMessage"
524536
try:
525-
requests_module.post(url, json={"chat_id": chat_id, "text": message}, timeout=15)
537+
requests_module.post(
538+
url,
539+
json={"chat_id": chat_id, "text": _break_telegram_market_symbol_auto_links(message)},
540+
timeout=15,
541+
)
526542
except Exception as exc:
527543
print(f"Telegram send failed: {type(exc).__name__}", flush=True)
528544

tests/test_notifications.py

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
import sys
2+
from pathlib import Path
3+
4+
5+
ROOT = Path(__file__).resolve().parents[1]
6+
if str(ROOT) not in sys.path:
7+
sys.path.insert(0, str(ROOT))
8+
PLATFORM_KIT_SRC = ROOT.parent / "QuantPlatformKit" / "src"
9+
if str(PLATFORM_KIT_SRC) not in sys.path:
10+
sys.path.insert(0, str(PLATFORM_KIT_SRC))
11+
12+
from notifications.telegram import build_sender
13+
14+
15+
class FakeRequests:
16+
def __init__(self):
17+
self.calls = []
18+
19+
def post(self, url, json, timeout):
20+
self.calls.append((url, json, timeout))
21+
return object()
22+
23+
24+
def test_build_sender_breaks_market_symbol_auto_links():
25+
fake_requests = FakeRequests()
26+
sender = build_sender("token-1", "chat-1", requests_module=fake_requests)
27+
sender("SOXL.US and 00700.HK")
28+
29+
assert len(fake_requests.calls) == 1
30+
url, payload, timeout = fake_requests.calls[0]
31+
assert "token-1" in url
32+
assert payload["chat_id"] == "chat-1"
33+
assert payload["text"] == "SOXL.\u2060US and 00700.\u2060HK"
34+
assert timeout == 15

0 commit comments

Comments
 (0)