Python监听Telegram指定聊天消息异常求助

你的代码无法监听新消息，主要是以下几个关键问题导致的，逐一修复即可解决：

1. 未正确配置目标Chat ID #

代码中om_private_alerts_channel_id是空字符串，必须填入真实的目标聊天/频道ID：

• 频道ID通常以-100开头（例如-1001234567890）
• 普通私聊ID为正数，可通过给Bot发消息后调用get_updates接口获取
• 确保Bot已被添加到目标频道并授予读取消息的权限

2. allowed_updates参数设置错误 #

allowed_updates=[]会阻止Telegram服务器推送任何更新类型，需明确指定需要监听的更新类型，比如同时支持频道帖子和普通消息：

updates = await bot.get_updates(
    offset=last_update_id, timeout=30, allowed_updates=["channel_post", "message"]
)

3. 仅处理频道消息，忽略普通聊天消息 #

代码只判断了update.channel_post（频道帖子），如果目标是普通私聊，需要同时检查update.message：

for update in updates:
    message = update.channel_post or update.message
    if message:
        if str(message.chat.id) == om_private_alerts_channel_id:
            # 处理消息

4. 消息实体无法直接序列化JSON #

message.entities是Telegram库的自定义对象，不能直接用json.dumps序列化，需转换为字典格式：

entities = [e.to_dict() for e in message.entities] if message.entities else None

5. 可选：优化异步环境下的文件操作 #

同步文件写入可能阻塞异步事件循环，建议改用aiofiles实现异步写入：
先安装依赖：pip install aiofiles
再修改消息处理函数为异步：

async def process_message(message):
    text = message.text
    entities = [e.to_dict() for e in message.entities] if message.entities else None
    print(text)
    print(entities)
    log_entry = {"text": text, "entities": entities}
    async with aiofiles.open('log.json', 'a') as file:
        await file.write(json.dumps(log_entry) + '\n')

调用时需添加await：await process_message(message)

修复后的完整代码 #

import json
import asyncio
from telegram import Bot
from telegram.error import TelegramError
import nest_asyncio
import aiofiles

nest_asyncio.apply()

# 替换为你的目标Chat/频道ID
om_private_alerts_channel_id = '-1001234567890' 
bot_token = 'your_bot_token_here' # 替换为BotFather提供的Token

# 异步处理消息
async def process_message(message):
    text = message.text
    entities = [e.to_dict() for e in message.entities] if message.entities else None
    # 打印消息到控制台
    print(text)
    print(entities)
    # 异步写入JSON日志
    log_entry = {
        "text": text,
        "entities": entities
    }
    async with aiofiles.open('log.json', 'a') as file:
        await file.write(json.dumps(log_entry) + '\n')

# 异步接收消息
async def receive_messages(bot):
    last_update_id = None
    while True:
        try:
            updates = await bot.get_updates(
                offset=last_update_id, timeout=30, allowed_updates=["channel_post", "message"]
            )
            if updates:
                last_update_id = updates[-1].update_id + 1
                for update in updates:
                    # 同时处理频道帖子和普通消息
                    message = update.channel_post or update.message
                    if message:
                        if str(message.chat.id) == om_private_alerts_channel_id:
                            await process_message(message)
        except TelegramError as e:
            print("接收消息出错:", e)
        await asyncio.sleep(1)

# 主函数
async def main():
    bot = Bot(bot_token)
    try:
        await receive_messages(bot)
    except KeyboardInterrupt:
        print("Bot已停止.")

if __name__ == "__main__":
    asyncio.run(main())

内容的提问来源于stack exchange，提问作者F0l0w