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
13 changes: 12 additions & 1 deletion src/quant_platform_kit/notifications/telegram.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,13 +3,24 @@
from __future__ import annotations

import json
import re
import urllib.parse
import urllib.request
from collections.abc import Sequence
from typing import Any


DEFAULT_TELEGRAM_BOT_API_BASE_URL = "https://api.telegram.org"
_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 uppercase URL hosts before breaking symbols

When a notification contains an uppercase host/domain, this pattern still matches the domain token because it runs over the whole message and only excludes adjacent alphanumerics; for example https://EXAMPLE.COM/report is rewritten with U+2060 inside the host. That breaks preservation of real links even though the added test only covers lowercase URLs, so guard URL spans or otherwise avoid substituting domain matches before sending.

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,
)


def parse_telegram_chat_ids(raw_value: str | Sequence[str] | None) -> tuple[str, ...]:
Expand Down Expand Up @@ -85,7 +96,7 @@ def send_telegram_message(
for chat_id in resolved_chat_ids:
payload: dict[str, object] = {
"chat_id": chat_id,
"text": message,
"text": _break_telegram_market_symbol_auto_links(message),
"disable_web_page_preview": bool(disable_web_page_preview),
}
text_parse_mode = str(parse_mode or "").strip()
Expand Down
22 changes: 22 additions & 0 deletions tests/test_strategy_plugin_telegram_notifications.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,28 @@ def opener(request, timeout):
self.assertEqual(payload["text"], "危机通知")
self.assertTrue(payload["disable_web_page_preview"])

def test_send_telegram_message_breaks_market_symbol_auto_links(self):
requests = []

def opener(request, timeout):
requests.append((request, timeout))
return _FakeResponse()

sent = send_telegram_message(
text="SOXL.US 预计;00700.HK 持仓;https://example.com 保持原样",
chat_ids=("123",),
bot_token="123456:ABC",
api_base_url="https://telegram.example.test",
opener=opener,
)

self.assertTrue(sent)
payload = json.loads(requests[0][0].data.decode("utf-8"))
self.assertEqual(
payload["text"],
"SOXL.\u2060US 预计;00700.\u2060HK 持仓;https://example.com 保持原样",
)

def test_send_strategy_plugin_telegram_combines_title_and_body(self):
requests = []

Expand Down