From f6cfbd2b1639e323e41853eebd014930c81ca647 Mon Sep 17 00:00:00 2001 From: WS Date: Thu, 30 Jul 2026 13:35:33 +0300 Subject: [PATCH] fix(telegram): expose hidden text-link URLs --- plugins/platforms/telegram/adapter.py | 64 ++++++++- .../test_telegram_text_link_expansion.py | 133 ++++++++++++++++++ 2 files changed, 194 insertions(+), 3 deletions(-) create mode 100644 tests/gateway/test_telegram_text_link_expansion.py diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index bec5365212..a12865ec6e 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -5574,6 +5574,64 @@ class TelegramAdapter(BasePlatformAdapter): """Guest-mode bypass: explicit bot mention (caller already verified group chat).""" return self._telegram_guest_mode() and self._message_mentions_bot(message) + def _expand_link_entities(self, message: Message) -> str: + """Inline Telegram ``text_link`` URLs into visible message text. + + Telegram stores hidden-link entity offsets as UTF-16 code units, while + Python string indexes are Unicode code points. Convert offsets before + inserting so links still expand correctly when text before the anchor + contains emoji or other non-BMP characters. + """ + text = getattr(message, "text", None) + if text: + entities = getattr(message, "entities", None) or [] + else: + text = getattr(message, "caption", None) or "" + entities = getattr(message, "caption_entities", None) or [] + if not text or not entities: + return text + + def utf16_index(offset: int) -> Optional[int]: + units = 0 + for index, char in enumerate(text): + if units == offset: + return index + units += 2 if ord(char) > 0xFFFF else 1 + if units > offset: + return None + return len(text) if units == offset else None + + utf16_length = sum(2 if ord(char) > 0xFFFF else 1 for char in text) + + links: list[tuple[int, int, str]] = [] + for entity in entities: + entity_type = str(getattr(entity, "type", "")).split(".")[-1].lower() + raw_url = getattr(entity, "url", None) + url = raw_url.strip() if isinstance(raw_url, str) else "" + if entity_type != "text_link" or not url: + continue + try: + offset = int(getattr(entity, "offset", -1)) + length = int(getattr(entity, "length", 0)) + except (TypeError, ValueError): + continue + if offset < 0 or length <= 0 or offset + length > utf16_length: + continue + start, end = utf16_index(offset), utf16_index(offset + length) + if start is None or end is None or end <= start: + continue + links.append((start, end, url)) + + expanded = text + for _start, end, url in sorted(links, reverse=True): + inline = f" ({url})" + # The guard makes repeated processing of an already-expanded event + # harmless without changing the original entity offsets. + if expanded[end:].startswith(inline): + continue + expanded = f"{expanded[:end]}{inline}{expanded[end:]}" + return expanded + def _clean_bot_trigger_text(self, text: Optional[str]) -> Optional[str]: bot_username = self._current_bot_username() if not text or not bot_username: @@ -6224,14 +6282,14 @@ class TelegramAdapter(BasePlatformAdapter): if self._should_observe_unmentioned_group_message(msg): _event = self._build_message_event(msg, self._media_message_type(msg), update_id=update.update_id) if msg.caption: - _event.text = self._clean_bot_trigger_text(msg.caption) + _event.text = self._clean_bot_trigger_text(self._expand_link_entities(msg)) await self._cache_observed_media(msg, _event) self._observe_unmentioned_group_message(msg, _event.message_type, update_id=update.update_id, event=_event) return event = self._build_message_event(msg, self._media_message_type(msg), update_id=update.update_id) if msg.caption: from plugins.platforms.telegram.telegram_context import group_trigger_text - event.text = group_trigger_text(self, msg, msg.caption) + event.text = group_trigger_text(self, msg, self._expand_link_entities(msg)) # Stickers: _handle_sticker overwrites event.text with its vision description, so observe attribution must run after it. if msg.sticker: await self._handle_sticker(msg, event) @@ -6532,7 +6590,7 @@ class TelegramAdapter(BasePlatformAdapter): _chat_id_str = str(chat.id) channel_prompt = resolve_channel_prompt(self.config.extra, thread_id_str or _chat_id_str, _chat_id_str if thread_id_str else None) return MessageEvent( - text=message.text or "", message_type=msg_type, source=source, raw_message=message, + text=self._expand_link_entities(message), message_type=msg_type, source=source, raw_message=message, message_id=str(message.message_id), platform_update_id=update_id, reply_to_message_id=reply_to_id, reply_to_text=reply_to_text, auto_skill=topic_skill, channel_prompt=group_identity_prompt(self, message, channel_prompt), diff --git a/tests/gateway/test_telegram_text_link_expansion.py b/tests/gateway/test_telegram_text_link_expansion.py new file mode 100644 index 0000000000..d20d2905a3 --- /dev/null +++ b/tests/gateway/test_telegram_text_link_expansion.py @@ -0,0 +1,133 @@ +"""Tests for Telegram ``text_link`` entity expansion in inbound messages. + +Telegram delivers a URL attached to a word (e.g. "тут" -> github.com) as a +``text_link`` entity. The visible text carries no URL, so without expansion the +model only ever sees the bare word and cannot fetch the link. ``_expand_link_entities`` +inlines the real URL right after its anchor for both ``text`` and ``caption``. +""" + +import pytest + +from plugins.platforms.telegram.adapter import TelegramAdapter + + +class _Entity: + def __init__(self, type, offset, length, url=None): + self.type = type + self.offset = offset + self.length = length + self.url = url + + +class _Message: + def __init__(self, text=None, caption=None, entities=None, caption_entities=None): + self.text = text + self.caption = caption + self.entities = entities + self.caption_entities = caption_entities + + +@pytest.fixture +def adapter(): + # _expand_link_entities only uses getattr on the message, so an unbound + # instance is enough. + return TelegramAdapter.__new__(TelegramAdapter) + + +def test_hidden_link_in_word_is_inlined(adapter): + msg = _Message( + text="Ссылка: тут\n#tag", + entities=[_Entity("text_link", 8, 3, "https://github.com/Cysharp/R3")], + ) + out = adapter._expand_link_entities(msg) + assert "https://github.com/Cysharp/R3" in out + assert out.startswith("Ссылка: тут (https://github.com/Cysharp/R3)") + + +def test_plain_text_without_entities_is_unchanged(adapter): + msg = _Message(text="просто текст без ссылок") + assert adapter._expand_link_entities(msg) == "просто текст без ссылок" + + +def test_caption_link_on_media_is_inlined(adapter): + msg = _Message( + caption="Смотри тут проект", + caption_entities=[_Entity("text_link", 7, 3, "https://example.com/x")], + ) + assert adapter._expand_link_entities(msg) == "Смотри тут (https://example.com/x) проект" + + +def test_utf16_offset_is_respected_after_emoji(adapter): + # Telegram entity offsets are measured in UTF-16 code units. The emoji is + # two units, so the visible anchor starts at offset 3, not Python index 2. + msg = _Message( + text="🔥 тут", + entities=[_Entity("text_link", 3, 3, "https://example.com/emoji")], + ) + assert adapter._expand_link_entities(msg) == "🔥 тут (https://example.com/emoji)" + + +def test_expansion_is_idempotent(adapter): + msg = _Message( + text="Ссылка: тут\n#tag", + entities=[_Entity("text_link", 8, 3, "https://github.com/Cysharp/R3")], + ) + out = adapter._expand_link_entities(msg) + repeat = _Message(text=out, entities=[_Entity("text_link", 8, 3, "https://github.com/Cysharp/R3")]) + assert adapter._expand_link_entities(repeat) == out + + +def test_multiple_distinct_links(adapter): + msg = _Message( + text="a b", + entities=[ + _Entity("text_link", 0, 1, "https://one.com"), + _Entity("text_link", 2, 1, "https://two.com"), + ], + ) + out = adapter._expand_link_entities(msg) + assert out == "a (https://one.com) b (https://two.com)" + + +def test_non_text_link_entities_are_ignored(adapter): + msg = _Message(text="жирный текст", entities=[_Entity("bold", 0, 6)]) + assert adapter._expand_link_entities(msg) == "жирный текст" + + +def test_anchor_repeats_later_in_text(adapter): + msg = _Message( + text="тут и ещё тут", + entities=[_Entity("text_link", 0, 3, "https://x.com")], + ) + out = adapter._expand_link_entities(msg) + assert out.startswith("тут (https://x.com) и ещё тут") + + +@pytest.mark.parametrize( + "entity", + [ + _Entity("text_link", 1, 99, "https://example.com/past-end"), + _Entity("text_link", 0, 1, 123), + _Entity("text_link", "invalid", 1, "https://example.com/bad-offset"), + ], +) +def test_malformed_link_entities_are_ignored(adapter, entity): + msg = _Message(text="abc", entities=[entity]) + assert adapter._expand_link_entities(msg) == "abc" + + +def test_text_does_not_use_caption_entities(adapter): + msg = _Message( + text="plain text", + caption="linked caption", + caption_entities=[_Entity("text_link", 0, 6, "https://example.com/caption")], + ) + assert adapter._expand_link_entities(msg) == "plain text" + + +def test_offset_inside_utf16_surrogate_pair_is_ignored(adapter): + msg = _Message( + text="🔥 link", + entities=[_Entity("text_link", 1, 1, "https://example.com/mid-surrogate")], + ) + assert adapter._expand_link_entities(msg) == "🔥 link"