Telegram机器人实现群投票统计：从发送投票到结果聚合的完整指南

在Telegram群组中，投票是收集成员意见、组织活动、快速决策的利器。但默认投票功能只能查看实时票数，无法导出或深度分析。通过自定义机器人，不仅可以灵活控制投票选项，还能自动汇总统计结果，甚至做二次分析。本文将手把手教你实现一个具备投票统计功能的Telegram机器人，让群管理事半功倍。

为什么用机器人实现群投票统计？

Telegram内置投票功能虽然方便，但存在局限：投票结束后无法自动生成详细报表；无法在投票过程中动态调整选项；无法将投票数据与其他业务系统打通。而机器人通过Bot API可以完全掌控投票生命周期，从发送、收集到统计，甚至可以结合数据库存储历史记录，满足企业调研、社群运营等复杂场景需求。

实现思路与核心原理

实现群投票统计的关键在于两点：一是使用sendPoll方法发送投票，二是监听PollAnswer更新来记录每个用户的响应。Telegram服务器会将用户的选择实时推送给机器人（通过Webhook或getUpdates），我们只需在回调中解析数据并累加票数。

前置准备

通过@BotFather创建机器人，获取BOT_TOKEN。
将机器人添加到目标群组，并赋予发送消息的权限。
安装Python库：pip install python-telegram-bot（使用20.x版本）。

核心步骤详解

1. 发送投票

使用send_poll方法，需指定问题、选项数组，还可设置是否匿名、是否允许多选。示例代码：

await context.bot.send_poll(
    chat_id=chat_id,
    question="你最常用的Telegram功能是什么？",
    options=["频道", "机器人", "秘密聊天", "群组"],
    is_anonymous=False,
    allows_multiple_answers=True
)

2. 初始化统计容器

为了跟踪每个投票的统计结果，我们用字典存储投票ID及其选项计数：

poll_results = {}

在发送投票后，我们可以通过返回值获取poll对象，并初始化对应键值：

sent_poll = await context.bot.send_poll(...)
poll_id = sent_poll.poll.id
poll_results[poll_id] = [0] * len(options)

3. 监听PollAnswer更新

在update处理器中，判断是否包含poll_answer字段。提取投票ID、用户ID、所选选项索引，更新计数：

async def poll_answer_handler(update, context):
    poll_answer = update.poll_answer
    poll_id = poll_answer.poll_id
    user = poll_answer.user
    option_ids = poll_answer.option_ids

    if poll_id in poll_results:
        for oid in option_ids:
            poll_results[poll_id][oid] += 1
    await update.effective_user.send_message("感谢投票！")

4. 展示统计结果

当管理员触发某个命令时，可发送当前统计。例如添加/results命令：

async def results_command(update, context):
    poll_id = ... # 从上下文或缓存中获取
    if poll_id not in poll_results:
        await update.message.reply_text("没有找到投票记录")
        return
    counts = poll_results[poll_id]
    lines = [f"选项{i+1}: 票" for i, n in enumerate(counts)]
    await update.message.reply_text("\n".join(lines))

完整代码示例

下面是一个简化但完整的机器人应用，包含发送投票、自动统计和查看结果功能：

import logging
from telegram import Update
from telegram.ext import Application, CommandHandler, PollAnswerHandler

TOKEN = "YOUR_BOT_TOKEN"
poll_results = {}

async def start(update, context):
    await update.message.reply_text("我支持投票统计！发送 /poll 创建投票，/results 查看统计。")

async def send_poll(update, context):
    chat_id = update.effective_chat.id
    options = ["选项A", "选项B", "选项C"]
    sent = await context.bot.send_poll(
        chat_id=chat_id,
        question="测试投票",
        options=options,
        is_anonymous=False
    )
    poll_results[sent.poll.id] = [0] * len(options)

async def poll_answer(update, context):
    pa = update.poll_answer
    if pa.poll_id in poll_results:
        for oid in pa.option_ids:
            poll_results[pa.poll_id][oid] += 1

async def show_results(update, context):
    if not poll_results:
        await update.message.reply_text("暂无投票数据")
        return
    for pid, counts in poll_results.items():
        total = sum(counts)
        lines = [f"投票ID: "]
        for i, c in enumerate(counts):
            lines.append(f"选项{i+1}: 票 ({c/total*100:.1f}%)" if total else f"选项{i+1}: 0票")
        await update.message.reply_text("\n".join(lines))

def main():
    app = Application.builder().token(TOKEN).build()
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("poll", send_poll))
    app.add_handler(CommandHandler("results", show_results))
    app.add_handler(PollAnswerHandler(poll_answer))
    app.run_polling()

if __name__ == "__main__":
    main()

进阶优化与注意事项

持久化存储：上述代码将数据存于内存，机器人重启后丢失。建议将poll_results存入Redis或数据库。
处理多选投票：当allows_multiple_answers=True时，option_ids可能包含多个值，需循环累加。
匿名投票：匿名投票的PollAnswer中不包含用户信息，但更新仍会推送，统计逻辑不变。
多投票管理：如果需要同时管理多个投票，可以用更复杂的数据结构，如嵌套字典。
异常处理：确保网络异常或API错误时程序稳定，可添加try/except。

总结

通过Telegram Bot API的sendPoll和PollAnswer，我们可以轻松实现群投票统计功能。本文的示例代码简洁易懂，开发者可根据实际需求扩展，比如增加结束投票、导出CSV、定时统计等。掌握这一技能，你就能打造出专属于社群的高效决策工具。