如何用Python+Telethon实现Telegram目标消息编辑/删除时更新源消息
实现目标频道消息编辑/删除同步到源频道
要实现目标频道消息编辑或删除时同步更新源频道的对应消息,核心是建立源消息与目标消息的ID映射关系,再通过监听目标频道的编辑/删除事件反向操作源消息。以下是完整的实现方案:
核心思路
- 转发消息时记录映射:从源频道转发消息到目标频道后,保存「目标消息ID → (源频道ID, 源消息ID)」的对应关系。
- 监听目标频道事件:分别监听目标频道的
MessageEdited和MessageDeleted事件,根据映射找到对应的源消息,执行编辑或删除操作。 - 可选持久化映射:用文件存储映射关系,避免脚本重启后丢失数据。
完整代码实现
from telethon import TelegramClient, events import json import os # 配置参数 api_id = 你的API_ID api_hash = "你的API_HASH" SOURCE_CHAT_ID = -1001629488043 TARGET_CHAT_ID = -1001567215170 MAPPING_FILE = "msg_mapping.json" # 加载/保存消息映射的工具函数 def load_mapping(): if os.path.exists(MAPPING_FILE): with open(MAPPING_FILE, "r", encoding="utf-8") as f: return json.load(f) return {} def save_mapping(mapping): with open(MAPPING_FILE, "w", encoding="utf-8") as f: json.dump(mapping, f, indent=2) # 初始化客户端和消息映射 client = TelegramClient('telegramfw', api_id, api_hash) msg_mapping = load_mapping() @client.on(events.NewMessage(chats=SOURCE_CHAT_ID)) @client.on(events.MessageEdited(chats=SOURCE_CHAT_ID)) async def forward_from_source(event): """处理源频道消息,转发到目标频道并记录映射""" sent_msg = None try: if event.photo: sent_msg = await client.send_file(TARGET_CHAT_ID, event.photo, caption=event.raw_text) elif event.video: sent_msg = await client.send_file(TARGET_CHAT_ID, event.video, caption=event.raw_text) else: sent_msg = await client.send_message(TARGET_CHAT_ID, event.raw_text) # 记录映射关系并保存到文件 if sent_msg: msg_mapping[str(sent_msg.id)] = [str(SOURCE_CHAT_ID), str(event.message.id)] save_mapping(msg_mapping) except Exception as e: print(f"转发消息出错: {e}") @client.on(events.MessageEdited(chats=TARGET_CHAT_ID)) async def sync_edit_to_source(event): """监听目标频道消息编辑,同步更新源频道对应消息""" target_msg_id = str(event.message.id) if target_msg_id not in msg_mapping: return # 无对应源消息,跳过 source_chat_id, source_msg_id = msg_mapping[target_msg_id] try: if event.photo: await client.edit_message(int(source_chat_id), int(source_msg_id), caption=event.raw_text, file=event.photo) elif event.video: await client.edit_message(int(source_chat_id), int(source_msg_id), caption=event.raw_text, file=event.video) else: await client.edit_message(int(source_chat_id), int(source_msg_id), event.raw_text) except Exception as e: print(f"同步编辑出错: {e}") @client.on(events.MessageDeleted(chats=TARGET_CHAT_ID)) async def sync_delete_to_source(event): """监听目标频道消息删除,同步删除源频道对应消息""" for target_msg_id in event.deleted_ids: target_msg_id_str = str(target_msg_id) if target_msg_id_str not in msg_mapping: continue source_chat_id, source_msg_id = msg_mapping[target_msg_id_str] try: await client.delete_messages(int(source_chat_id), int(source_msg_id)) # 删除映射并保存 del msg_mapping[target_msg_id_str] save_mapping(msg_mapping) except Exception as e: print(f"同步删除出错: {e}") if __name__ == "__main__": client.start() print("客户端已启动,开始监听消息...") client.run_until_disconnected()
关键说明
- 权限要求:确保你的Telegram账号在源频道和目标频道都拥有「编辑消息」和「删除消息」的权限,否则同步操作会失败。
- 映射持久化:用
msg_mapping.json文件存储映射关系,脚本重启后不会丢失之前的消息对应记录。 - 异常处理:添加了基础的异常捕获,避免单个消息出错导致整个脚本崩溃。
- 消息类型兼容:支持文本、图片、视频类型消息的转发和同步编辑。
内容的提问来源于stack exchange,提问作者MagnidownZ
相关产品推荐
相关产品推荐

