itgoyo Pyrogram Telegram 机器人开发技巧
1. 解析 Telegram 消息链接
从 Telegram 链接中提取 chat_id、message_id 和 topic_id。
Link 数据结构
from dataclasses import dataclass
from typing import Optional, Union
from urllib.parse import parse_qs, urlparse
@dataclass
class Link:
"""Telegram Link"""
group_id: Union[str, int, None] = None
post_id: Optional[int] = None
comment_id: Optional[int] = None
topic_id: Optional[int] = None
提取链接信息
def extract_info_from_link(link: str) -> Link:
"""Extract info from link"""
if link in ("me", "self"):
return Link(group_id=link)
try:
u = urlparse(link)
paths = [p for p in u.path.split("/") if p]
query = parse_qs(u.query)
except ValueError:
return Link()
result = Link()
if "comment" in query:
result.group_id = paths[0]
result.comment_id = int(query["comment"][0])
# BUG FIX: must also preserve post_id — the original code dropped paths[1]
# (e.g. for https://t.me/zaihuapd/40815?comment=8447515, paths[1]=="40815"
# is the channel post id that owns the comment thread; without it you cannot
# call GetDiscussionMessage to enumerate the full thread).
if len(paths) >= 2:
try:
result.post_id = int(paths[1])
except (ValueError, IndexError):
pass
elif len(paths) == 1 and paths[0] != "c":
result.group_id = paths[0]
elif len(paths) == 2:
if paths[0] == "c":
result.group_id = int(f"-100{paths[1]}")
else:
result.group_id = paths[0]
result.post_id = int(paths[1])
elif len(paths) == 3:
if paths[0] == "c":
result.group_id = int(f"-100{paths[1]}")
result.post_id = int(paths[2])
else:
result.group_id = paths[0]
result.topic_id = int(paths[1])
result.post_id = int(paths[2])
elif len(paths) == 4 and paths[0] == "c":
result.group_id = int(f"-100{paths[1]}")
result.topic_id = int(paths[2])
result.post_id = int(paths[3])
return result
解析链接并处理评论
async def parse_link(client: pyrogram.Client, link_str: str):
"""Parse link - handles both regular messages and comments.
Returns (chat_id, message_id, topic_id).
For comment links returns (discussion_chat_id, comment_id, topic_id).
"""
link = extract_info_from_link(link_str)
if link.comment_id:
# 评论链接需要获取关联群组的 ID
chat = await client.get_chat(link.group_id)
if chat:
return chat.linked_chat.id, link.comment_id, link.topic_id
return link.group_id, link.post_id, link.topic_id
async def parse_comment_link(client: pyrogram.Client, link_str: str):
"""Parse a comment link returning full 4-tuple:
(channel_chat_id, post_id, discussion_chat_id, comment_id)
Use this when you need to know BOTH the original channel post id AND
the specific comment id (e.g. to enumerate the whole thread).
Returns (None, None, None, None) for non-comment links.
"""
link = extract_info_from_link(link_str)
if not link.comment_id:
return None, None, None, None
channel_chat = await client.get_chat(link.group_id)
if not channel_chat or not channel_chat.linked_chat:
return None, None, None, None
discussion_chat_id = channel_chat.linked_chat.id
return channel_chat.id, link.post_id, discussion_chat_id, link.comment_id
支持的链接格式:
https://t.me/channel_name/123- 公开频道消息https://t.me/c/1234567890/123- 私有频道消息 (自动转换为 -100 格式)https://t.me/channel_name/topic_id/123- 话题消息https://t.me/channel_name/123?comment=456- 评论消息 (comment_id=456, post_id=123)
2. 自定义 Session 超时 (HookSession/HookClient)
继承 Pyrogram 的 Session 和 Client 类来自定义连接超时时间。
import pyrogram
class HookSession(pyrogram.session.Session):
"""Hook Session with custom timeout"""
def start_timeout(self: pyrogram.session.Session, start_timeout: int):
"""Set the start timeout for the session."""
self.START_TIMEOUT = start_timeout
class HookClient(pyrogram.Client):
"""Hook Client with custom session timeout"""
START_TIME_OUT = 60 # 默认 60 秒
def __init__(self, name: str, **kwargs):
if "start_timeout" in kwargs:
value = kwargs.get("start_timeout")
if value:
self.START_TIME_OUT = value
kwargs.pop("start_timeout")
super().__init__(name, **kwargs)
async def connect(self) -> bool:
if self.is_connected:
raise ConnectionError("Client is already connected")
await self.load_session()
# 使用自定义的 HookSession
self.session = HookSession(
self,
await self.storage.dc_id(),
await self.storage.auth_key(),
await self.storage.test_mode(),
)
self.session.start_timeout(self.START_TIME_OUT)
await self.session.start()
self.is_connected = True
return bool(await self.storage.user_id())
async def start(self):
is_authorized = await self.connect()
try:
if not is_authorized:
await self.authorize()
if not await self.storage.is_bot() and self.takeout:
self.takeout_id = (
await self.invoke(
pyrogram.raw.functions.account.InitTakeoutSession()
)
).id
await self.invoke(pyrogram.raw.functions.updates.GetState())
except (Exception, KeyboardInterrupt):
await self.disconnect()
raise
else:
self.me = await self.get_me()
await self.initialize()
return self
使用方法:
client = HookClient(
"my_bot",
api_id=API_ID,
api_hash=API_HASH,
start_timeout=120 # 自定义超时时间为 120 秒
)
3. 避免视频上传黑屏
上传视频时必须提供缩略图和视频属性，否则视频可能显示为黑屏。
下载视频缩略图
import os
import time
import secrets
import asyncio
async def download_thumbnail(
client: pyrogram.Client,
temp_path: str,
message: pyrogram.types.Message,
):
"""Downloads the thumbnail of a video message to a temporary file."""
thumbnail_file = None
if message.video.thumbs:
thumbnail = message.video.thumbs[0] if message.video.thumbs else None
unique_name = os.path.join(
temp_path,
"thumbnail",
f"thumb-{int(time.time())}-{secrets.token_hex(8)}.jpg",
)
max_attempts = 3
for attempt in range(1, max_attempts + 1):
try:
thumbnail_file = await client.download_media(
thumbnail, file_name=unique_name
)
# 验证文件大小
if os.path.getsize(thumbnail_file) == thumbnail.file_size:
break
raise ValueError(
f"Thumbnail file size mismatch: {os.path.getsize(thumbnail_file)}"
f" vs expected {thumbnail.file_size}"
)
except Exception as e:
if attempt == max_attempts:
logger.exception(f"Failed to download thumbnail: {e}")
else:
await asyncio.sleep(2)
return thumbnail_file
上传视频时使用完整属性
# 关键：必须提供 thumb、width、height、duration
thumbnail_file = await download_thumbnail(client, temp_save_path, message)
await message.reply_video(
video_file_path,
caption=caption,
thumb=thumbnail_file, # 缩略图
width=message.video.width, # 视频宽度
height=message.video.height, # 视频高度
duration=message.video.duration, # 视频时长
parse_mode=pyrogram.enums.ParseMode.HTML,
)
注意事项:
- 没有缩略图会导致视频显示为黑屏
- 缺少 width/height 会影响视频预览
- 上传后记得删除临时缩略图文件
4. 去除消息中的广告
通过分析消息实体(entities)来提取或过滤广告内容，同时保留原有格式。
MessageProcessor 类
import pyrogram
def get_utf16_length(text: str) -> int:
"""Get UTF-16 length of text (Telegram uses UTF-16 for entity offsets)"""
return len(text.encode('utf-16-le')) // 2
class MessageProcessor:
"""Helper class for processing message captions and entities."""
def __init__(self, raw_message, filter_str):
self.raw_message = raw_message
self.raw_caption = raw_message.caption
self.filter_str = filter_str
self.raw_filter_str = pyrogram.parser.utils.add_surrogates(filter_str)
self.raw_caption_str = pyrogram.parser.utils.add_surrogates(raw_message.caption)
self.idx = self.raw_caption_str.find(self.raw_filter_str)
self.start_offset = self.idx
self.end_offset = self.idx + get_utf16_length(filter_str)
self.filtered_entities = []
def process_entities(self):
"""Process and filter message entities."""
for entity in self.raw_message.caption_entities:
cur_start_offset = entity.offset
cur_end_offset = entity.offset + entity.length
# 检查实体是否在过滤范围内
if (
(cur_start_offset >= self.start_offset and cur_end_offset <= self.end_offset)
or (cur_start_offset < self.start_offset and cur_end_offset > self.start_offset)
or (cur_start_offset < self.end_offset and cur_end_offset > self.end_offset)
):
self.filtered_entities.append(entity)
self.filtered_entities.sort(key=lambda x: x.offset)
def get_total_span(self):
"""Calculate the total span for text extraction."""
if self.filtered_entities:
first_entity = self.filtered_entities[0]
last_entity = self.filtered_entities[-1]
return (
min(self.start_offset, first_entity.offset),
max(self.end_offset, last_entity.offset + last_entity.length),
)
return (self.start_offset, self.end_offset)
def extract_text(self, total_span):
"""Extract and process text with adjusted entity offsets."""
text = self.raw_caption[total_span[0]:total_span[1]]
for entity in self.filtered_entities:
entity.offset -= total_span[0]
return pyrogram.parser.Parser.unparse(text, self.filtered_entities, True)
处理广告替换
async def proc_replace_advertisement(message_link: str, filter_str: str):
"""Process and replace advertisement content in a message."""
chat_id, message_id, _ = await parse_link(client, message_link)
raw_message = await client.get_messages(chat_id, message_id)
processor = MessageProcessor(raw_message, filter_str)
processor.process_entities()
total_span = processor.get_total_span()
return processor.extract_text(total_span)
工作原理:
- 解析消息链接获取原始消息
- 使用 filter_str 定位要保留的文本区域
- 提取该区域内的实体(链接、加粗、斜体等)
- 重新组装文本，保留格式但去除广告部分
使用场景:
- 转发消息时去除频道水印/广告
- 保留原始格式(链接、加粗、斜体等)
- 批量处理时添加到过滤列表
完整使用示例
import pyrogram
# 1. 使用自定义超时的客户端
client = HookClient(
"my_bot",
api_id=API_ID,
api_hash=API_HASH,
start_timeout=120
)
async def main():
await client.start()
# 2. 解析消息链接
link = "https://t.me/channel_name/123"
chat_id, message_id, topic_id = await parse_link(client, link)
# 3. 获取消息并下载视频
message = await client.get_messages(chat_id, message_id)
if message.video:
# 下载缩略图避免黑屏
thumbnail = await download_thumbnail(client, "/tmp", message)
await client.send_video(
chat_id=destination_chat,
video=video_path,
thumb=thumbnail,
width=message.video.width,
height=message.video.height,
duration=message.video.duration,
)
# 4. 去除广告
if message.caption:
clean_caption = await proc_replace_advertisement(
link,
"想要的内容部分"
)
print(clean_caption)
await client.stop()
5. 搬运讨论组评论区 (GetDiscussionMessage)
Telegram 的评论区（讨论组）与频道帖子之间的关系：
- 每个开启评论的频道帖子，在
linked_chat（讨论组）里有一条对应的"根消息" - 所有评论都是在讨论组里 reply 到这条根消息的
- Pyrogram 的
message_thread_id字段等于讨论组根消息的 ID
获取讨论组根消息 ID
from pyrogram import raw
async def get_discussion_root_msg_id(client, channel_chat_id, post_id):
"""Return (discussion_chat_id, root_msg_id) for a channel post's comment thread."""
disc = await client.invoke(
raw.functions.messages.GetDiscussionMessage(
peer=await client.resolve_peer(channel_chat_id),
msg_id=post_id,
)
)
discussion_peer = disc.chats[0]
if isinstance(discussion_peer, raw.types.Channel):
discussion_chat_id = int(f"-100{discussion_peer.id}")
else:
discussion_chat_id = discussion_peer.id
root_msg_id = disc.messages[0].id
return discussion_chat_id, root_msg_id
遍历评论区所有消息
async def get_discussion_replies_v2(client, channel_chat_id, post_id, limit=0, offset_id=0, reverse=True):
"""Async generator yielding all comment messages in a post's discussion thread.
Pitfalls:
- Must use a USER account client (not bot) — bots cannot read discussion group history
- Filter by message_thread_id == root_msg_id OR reply_to_message_id == root_msg_id
- Also propagate thread membership via seen_ids (replies-to-replies)
- The root stub message itself (root_msg_id) is the first yielded message
"""
disc = await client.invoke(
raw.functions.messages.GetDiscussionMessage(
peer=await client.resolve_peer(channel_chat_id),
msg_id=post_id,
)
)
discussion_peer = disc.chats[0]
if isinstance(discussion_peer, raw.types.Channel):
discussion_chat_id = int(f"-100{discussion_peer.id}")
else:
discussion_chat_id = discussion_peer.id
root_msg_id = disc.messages[0].id
total = limit or (1 << 31) - 1
chunk_size = min(100, total)
from_id = offset_id if offset_id else (1 if reverse else 0)
seen_ids = set()
current = 0
while True:
messages = await get_chunk_v2(
client=client, chat_id=discussion_chat_id,
limit=chunk_size, from_message_id=from_id, reverse=reverse,
)
if not messages:
break
for msg in messages:
if msg.id in seen_ids:
continue
seen_ids.add(msg.id)
is_in_thread = (
msg.id == root_msg_id
or getattr(msg, "message_thread_id", None) == root_msg_id
or getattr(msg, "reply_to_message_id", None) == root_msg_id
or (msg.reply_to_message_id and msg.reply_to_message_id in seen_ids)
)
if is_in_thread:
yield msg
current += 1
if current >= total:
return
from_id = messages[-1].id + (1 if reverse else -1)
if not reverse and from_id <= 0:
break
if len(messages) < chunk_size:
break
关键注意事项:
- 必须用用户账号（不能用机器人账号）调用，机器人无法访问讨论组历史
parse_comment_link提供channel_chat_id和post_id，二者缺一不可extract_info_from_link解析?comment=xxx链接时必须同时保存post_id（历史 bug）- 评论区线程成员判断：
message_thread_id == root_msg_id或reply_to_message_id == root_msg_id（或传递性：reply 到已知线程消息）