如何通过Telegram API读取投票结果？已实现投票创建

问题根源 #

你调用/pollAnswer出现404是因为它不是可主动调用的API接口，而是Telegram向Bot推送的更新类型。当用户提交投票时，Telegram会发送包含poll_answer字段的更新，你需要通过getUpdates轮询或Webhook接收。

官方Bot API提供了getPoll接口，可通过创建投票时拿到的poll_id直接拉取投票的最新统计结果，这是最直接的方式。

示例代码 #

1. 主动拉取投票结果（推荐） #

通过poll_id直接获取当前投票的完整结果，适合不需要实时监听的场景：

import requests

def get_telegram_poll_results(bot_token, poll_id):
    """获取指定poll_id的投票结果"""
    url = f"https://api.telegram.org/bot{bot_token}/getPoll"
    params = {"poll_id": poll_id}
    
    response = requests.get(url, params=params, timeout=15)
    if response.status_code != 200:
        print(f"请求失败：状态码 {response.status_code}")
        return None
    
    poll = response.json()["result"]
    # 输出投票信息
    print(f"投票问题：{poll['question']}")
    print("投票结果：")
    for idx, opt in enumerate(poll["options"]):
        print(f"{idx+1}. {opt['text']}：{opt['voter_count']} 票")
    
    return poll

# 使用示例：
# bot_token, _ = get_telegram_creds()
# poll_id = "创建投票时返回的poll_id"
# get_telegram_poll_results(bot_token, poll_id)

2. 实时监听用户投票操作 #

如果需要实时获取用户的投票行为，可通过轮询getUpdates接收poll_answer更新：

import requests
import time

def listen_poll_updates(bot_token, offset=0):
    """轮询监听投票更新"""
    url = f"https://api.telegram.org/bot{bot_token}/getUpdates"
    
    while True:
        params = {
            "offset": offset,
            "timeout": 30  # 长轮询，减少请求频率
        }
        response = requests.get(url, params=params, timeout=35)
        
        if response.status_code != 200:
            print(f"请求失败：状态码 {response.status_code}")
            time.sleep(5)
            continue
        
        updates = response.json()["result"]
        for update in updates:
            # 处理投票答案更新
            if "poll_answer" in update:
                pa = update["poll_answer"]
                user_id = pa["user"]["id"]
                poll_id = pa["poll_id"]
                selected_options = pa["option_ids"]
                
                print(f"用户 {user_id} 投票，选择了选项索引：{selected_options}")
                # 可选：立即拉取最新投票结果
                # get_telegram_poll_results(bot_token, poll_id)
            
            # 更新offset，避免重复处理同一更新
            offset = update["update_id"] + 1
        
        time.sleep(1)

# 使用示例：
# bot_token, _ = get_telegram_creds()
# listen_poll_updates(bot_token)

3. 完整流程示例（创建+获取结果） #

结合你已有的创建投票代码，实现完整流程：

import requests
import json
import sys
import time

def get_telegram_creds():
    # 替换为你的凭证获取逻辑
    return "你的BotToken", "目标群组ID"

def new_telegram_poll():
    """创建Telegram投票"""
    telegram_api, telegram_channel_id = get_telegram_creds()
    url = f"https://api.telegram.org/bot{telegram_api}/sendpoll"
    
    params = {
        "chat_id": telegram_channel_id,
        "question": "This is a poll",
        "options": json.dumps(["option1", "option2", "option3"]),
        "is_anonymous": False,
        "allows_multiple_answers": False,
        "type": "regular",
    }
    
    response = requests.get(url, data=params, timeout=15)
    if response.status_code != 200:
        print(f"创建投票失败：状态码 {response.status_code}")
        sys.exit()
    
    result = response.json()["result"]
    message_id = result["message_id"]
    poll_id = result["poll"]["id"]
    print(f"投票创建成功，message_id: {message_id}, poll_id: {poll_id}")
    return message_id, poll_id

if __name__ == "__main__":
    # 创建投票
    _, poll_id = new_telegram_poll()
    bot_token, _ = get_telegram_creds()
    
    # 等待用户投票后获取结果
    print("等待10秒后获取投票结果...")
    time.sleep(10)
    get_telegram_poll_results(bot_token, poll_id)

关键注意事项 #

• 确保Bot在目标群组中拥有读取消息权限，建议设置为管理员
• 使用getUpdates时必须更新offset，否则会重复处理相同的更新
• 高并发场景建议使用Webhook替代轮询，减少服务器资源消耗

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