You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Telethon频道消息转发脚本链接处理失效问题排查修复

故障根因
  • 富文本超链接丢失:代码中使用event.raw_text、event.original_update.message.message获取消息内容,这两个接口仅返回无格式纯文本,完全剥离了Telegram消息内置的格式实体(超链接、加粗、斜体等),相当于把富文本转成纯字符串再发送,所有格式属性自然丢失。
  • 含明文链接的消息发送失败:代码中硬编码了parse_mode='md'参数,强制用Markdown规则解析消息文本,明文链接中常见的下划线、括号、反引号等字符属于Markdown语法特殊字符,未做转义时会触发Telegram服务端格式解析错误,直接拒绝消息发送,现有异常捕获仅打印通用错误提示,未暴露解析失败的具体原因。
  • 附带逻辑缺陷:分支判断冗余重复,相册消息的文本获取方式错误,关键词替换仅操作纯文本未同步更新格式实体偏移量,即使拿到格式实体也会出现格式错位问题。
修复方案
  1. 放弃直接提取纯文本的处理方式,使用Telethon内置的HTML解析扩展做格式转换:先把原消息带格式转成HTML字符串,在HTML层做关键词替换(不会破坏超链接等标签结构),替换完成后再转回Telegram识别的文本+格式实体组合,从根源保留所有富文本属性。
  2. 发送消息时移除parse_mode='md'参数,直接传入转换得到的文本和格式实体,Telegram会直接使用传入的实体渲染格式,不会二次解析文本中的特殊字符,彻底解决明文链接触发的解析报错问题。
  3. 精简冗余分支判断,单条文本、单条带媒体消息、相册消息统一走格式处理逻辑,转发类消息保留原有直接转发的逻辑即可。
  4. 异常捕获补充错误详情打印,限流等待后增加重试逻辑,避免漏发消息。
修复后完整代码
from telethon import TelegramClient, events, errors
from telethon.extensions import html
import asyncio
import re

# ---- 配置项 ----
api_id = 你的API_ID
api_hash = "你的API_HASH"
channels = '@源频道用户名'  # 消息来源频道
my_channel = '@自有频道用户名'  # 转发目标频道
KEYS = {} # 关键词替换规则,key为匹配正则/字符串,value为替换后内容
Bad_Keys = [] # 敏感词列表,命中则不转发
tags = '' # 消息末尾追加的固定文本
# ----

with TelegramClient('myApp13', api_id, api_hash) as client:
    print("~Activated~")

    async def process_text(original_message):
        """统一处理文本替换,保留所有富文本格式"""
        # 敏感词校验,用纯文本校验即可
        raw_text = original_message.raw_text.lower()
        for bad_word in Bad_Keys:
            if bad_word in raw_text:
                return None, None
        # 转成HTML做替换,不破坏格式标签
        msg_html = html.unparse(original_message.text, original_message.entities)
        for pattern, replace_str in KEYS.items():
            msg_html = re.sub(pattern, replace_str, msg_html)
        # 追加固定尾巴文本
        msg_html += tags
        # 转回Telegram需要的 文本+实体 格式
        return html.parse(msg_html)

    @client.on(events.NewMessage(chats=channels))
    async def handle_single_message(event):
        # 转发消息直接走原生转发
        if event.message.forward:
            try:
                await event.message.forward_to(my_channel)
            except errors.FloodWaitError as e:
                print(f'[!] 触发限流,等待 {e.seconds} 秒后重试')
                await asyncio.sleep(e.seconds)
                await event.message.forward_to(my_channel)
            except Exception as e:
                print(f'[!] 转发消息失败: {str(e)}')
            return
        
        # 跳过属于相册的单条消息,统一由Album handler处理
        if event.grouped_id:
            return
        
        # 处理普通单条消息(带媒体/纯文本)
        text, entities = await process_text(event.message)
        if not text:
            return
        try:
            await client.send_message(
                entity=my_channel,
                file=event.message.media,
                message=text,
                entities=entities,
                link_preview=False
            )
        except errors.FloodWaitError as e:
            print(f'[!] 触发限流,等待 {e.seconds} 秒后重试')
            await asyncio.sleep(e.seconds)
            await client.send_message(
                entity=my_channel,
                file=event.message.media,
                message=text,
                entities=entities,
                link_preview=False
            )
        except Exception as e:
            print(f'[!] 发送单条消息失败: {str(e)}')

    @client.on(events.Album(chats=channels))
    async def handle_album(event):
        # 相册说明文本在第一条消息上
        caption_msg = event.messages[0]
        text, entities = await process_text(caption_msg)
        if not text:
            return
        try:
            await client.send_message(
                entity=my_channel,
                file=event.messages,
                message=text,
                entities=entities,
                link_preview=False
            )
        except errors.FloodWaitError as e:
            print(f'[!] 触发限流,等待 {e.seconds} 秒后重试')
            await asyncio.sleep(e.seconds)
            await client.send_message(
                entity=my_channel,
                file=event.messages,
                message=text,
                entities=entities,
                link_preview=False
            )
        except Exception as e:
            print(f'[!] 发送相册消息失败: {str(e)}')

    client.run_until_disconnected()
修复验证说明
  • 原频道带超链接的富文本消息转发后,超链接可正常点击,其他格式(加粗、斜体、行内代码等)完全保留
  • 含明文链接的消息不会再触发格式解析错误,可正常发送
  • 关键词替换、敏感词过滤、末尾追加文本、限流重试逻辑均正常生效
  • 相册消息的说明文本格式也会同步保留,不会出现纯文本丢失格式的问题

内容的提问来源于stack exchange,提问作者Vlad Piker

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.29 06:24:24