fix(telegram): move fenced-code-block splitting into Telegram-specific helper

Move the fenced-code-block-aware splitting logic out of the shared
split_message helper (used by Signal, Slack, Discord, Weixin, etc.)
and into a Telegram-specific _split_telegram_markdown function.

The shared split_message remains a plain-text chunker. The Telegram
channel now uses _split_telegram_markdown for its raw Markdown paths
that feed _markdown_to_telegram_html, preventing broken HTML rendering
when splits fall inside fenced code blocks.

Also fixes a regression where content beginning with whitespace before
a fence could emit a whitespace-only chunk.

Addresses review feedback on #4257.
This commit is contained in:
axelray-dev
2026-06-11 13:52:19 +08:00
committed by Xubin Ren
parent 131446fa61
commit a5a816abaf
4 changed files with 138 additions and 101 deletions
+63
View File
@@ -17,6 +17,8 @@ from nanobot.channels.telegram import (
TELEGRAM_REPLY_CONTEXT_MAX_LEN,
TelegramChannel,
TelegramConfig,
_markdown_to_telegram_html,
_split_telegram_markdown,
_StreamBuf,
)
@@ -179,6 +181,67 @@ def _make_telegram_update(
return SimpleNamespace(message=message, effective_user=user)
def _assert_code_blocks_render_balanced(chunks: list[str]) -> None:
for chunk in chunks:
html = _markdown_to_telegram_html(chunk)
assert html.count("<pre><code>") == html.count("</code></pre>")
def test_split_telegram_markdown_inside_code_block_moves_before_fence() -> None:
content = "Intro paragraph.\n```python\nprint('a')\nprint('b')\n```\nDone"
chunks = _split_telegram_markdown(content, max_len=35)
assert chunks[0] == "Intro paragraph.\n"
assert chunks[1].startswith("```python\nprint('a')")
_assert_code_blocks_render_balanced(chunks)
def test_split_telegram_markdown_long_code_block_closes_and_reopens() -> None:
content = "```python\n" + ("print('line one')\n" * 6) + "```\nDone"
chunks = _split_telegram_markdown(content, max_len=60)
assert len(chunks) > 1
assert all(len(chunk) <= 60 for chunk in chunks)
assert chunks[0].startswith("```python\n")
assert chunks[0].endswith("\n```")
assert chunks[1].startswith("```python\n")
_assert_code_blocks_render_balanced(chunks)
def test_split_telegram_markdown_multiple_code_blocks() -> None:
content = (
"First\n"
"```js\n"
"one();\n"
"```\n"
"Middle paragraph here\n"
"```py\n"
"two()\n"
"three()\n"
"```\n"
"End"
)
chunks = _split_telegram_markdown(content, max_len=55)
assert chunks[0].endswith("Middle paragraph here\n")
assert chunks[1].startswith("```py\n")
_assert_code_blocks_render_balanced(chunks)
def test_split_telegram_markdown_leading_whitespace_before_fence() -> None:
content = "\n```python\n" + ("print('line one')\n" * 6) + "```\nDone"
chunks = _split_telegram_markdown(content, max_len=60)
assert chunks
assert all(chunk.strip() for chunk in chunks)
assert chunks[0].startswith("```python\n")
_assert_code_blocks_render_balanced(chunks)
@pytest.mark.asyncio
async def test_start_creates_separate_pools_with_proxy(monkeypatch) -> None:
_FakeHTTPXRequest.clear()
-53
View File
@@ -5,56 +5,3 @@ def test_split_message_no_code_blocks_unchanged():
content = "alpha beta gamma delta"
assert split_message(content, max_len=12) == ["alpha beta", "gamma delta"]
def test_split_message_outside_code_block_unchanged():
content = "alpha beta gamma delta\n```python\nx = 1\n```\ndone"
chunks = split_message(content, max_len=12)
assert chunks[0] == "alpha beta"
assert chunks[1].startswith("gamma")
def test_split_message_inside_code_block_moves_before_fence():
content = "Intro paragraph.\n```python\nprint('a')\nprint('b')\n```\nDone"
chunks = split_message(content, max_len=35)
assert chunks[0] == "Intro paragraph.\n"
assert chunks[1].startswith("```python\nprint('a')")
assert all(chunk.count("```") % 2 == 0 for chunk in chunks[1:])
def test_split_message_code_block_longer_than_max_len_closes_and_reopens():
content = "```python\n" + ("print('line one')\n" * 6) + "```\nDone"
chunks = split_message(content, max_len=60)
assert len(chunks) > 1
assert all(len(chunk) <= 60 for chunk in chunks)
assert all(chunk.count("```") % 2 == 0 for chunk in chunks)
assert chunks[0].startswith("```python\n")
assert chunks[0].endswith("\n```")
assert chunks[1].startswith("```python\n")
def test_split_message_multiple_code_blocks_moves_second_block_to_next_chunk():
content = (
"First\n"
"```js\n"
"one();\n"
"```\n"
"Middle paragraph here\n"
"```py\n"
"two()\n"
"three()\n"
"```\n"
"End"
)
chunks = split_message(content, max_len=55)
assert chunks[0].endswith("Middle paragraph here\n")
assert chunks[1].startswith("```py\n")
assert all(chunk.count("```") % 2 == 0 for chunk in chunks)