Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 17 additions & 1 deletion notifications/telegram.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"(?<![A-Za-z0-9_])([A-Z0-9]{1,12})\.([A-Z]{2,4})(?![A-Za-z0-9_])")

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve URLs when suppressing symbol auto-links

When a Telegram message contains an uppercase URL or domain such as https://SOXL.US/report or https://EXAMPLE.COM, this regex still matches the hostname/path segment and _break_telegram_market_symbol_auto_links inserts \u2060 into the URL. That changes the actual link target and can make real links in arbitrary sender messages unusable, even though only standalone market symbols should be altered.

Useful? React with 👍 / 👎.

_TELEGRAM_MARKET_SYMBOL_LINK_JOINER = "\u2060"


def _break_telegram_market_symbol_auto_links(value: object) -> 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"),
Expand Down Expand Up @@ -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)

Expand Down
34 changes: 34 additions & 0 deletions tests/test_notifications.py
Original file line number Diff line number Diff line change
@@ -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