Python Telegram Bot如何存储/读取用户Chat ID以实现主动消息推送

嘿，我来帮你搞定这个问题！核心就是先持久化存储用户的Chat ID，然后在特定条件触发时通过Bot API主动发消息，而且完全适配打包成.exe的场景，下面一步步给你拆解：

一、选个适合exe场景的存储方案 #

因为要打包成单文件exe，得用不需要额外服务的轻量存储，推荐这两种：

1. JSON文件（最适合轻量需求） #

纯文本格式，读写简单，Python自带json库，打包后exe会自动在同目录创建/读取文件，完全不需要额外依赖。

2. SQLite数据库（用户多了也能用） #

Python内置sqlite3库，文件型数据库，同样不需要服务器，适合用户量稍大的场景。

二、实现Chat ID的自动存储逻辑 #

当用户第一次给Bot发消息时，自动记录他们的Chat ID，以后就能随时推送了：

示例：JSON存储版 #

import json
from telegram import Update
from telegram.ext import ApplicationBuilder, MessageHandler, filters, ContextTypes

# 存储Chat ID的文件，打包后和exe同目录
CHAT_ID_FILE = "chat_ids.json"
# 替换成你的Bot Token
BOT_TOKEN = "你的Bot Token"

# 加载已存的Chat ID
def load_chat_ids():
    try:
        with open(CHAT_ID_FILE, "r", encoding="utf-8") as f:
            return set(json.load(f))
    except FileNotFoundError:
        # 文件不存在就返回空集合
        return set()

# 保存Chat ID到文件
def save_chat_ids(chat_ids):
    with open(CHAT_ID_FILE, "w", encoding="utf-8") as f:
        json.dump(list(chat_ids), f, ensure_ascii=False)

# 处理用户消息，自动存储Chat ID
async def handle_user_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id
    chat_ids = load_chat_ids()
    
    if chat_id not in chat_ids:
        chat_ids.add(chat_id)
        save_chat_ids(chat_ids)
        await update.message.reply_text("已成功注册，后续会给你推送消息哦！")
    else:
        await update.message.reply_text("你已经在推送列表里啦～")

def start_bot():
    application = ApplicationBuilder().token(BOT_TOKEN).build()
    # 监听所有用户文本消息（除了命令）
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_user_message))
    # 启动Bot轮询
    application.run_polling()

if __name__ == "__main__":
    start_bot()

三、实现主动推送消息的逻辑 #

现在有了存储的Chat ID，就可以在任何条件触发时遍历这些ID发消息了。关键是要让Bot的实例能被全局访问，或者单独写一个发送函数：

整合推送的完整示例 #

import json
import asyncio
from telegram import Update
from telegram.ext import ApplicationBuilder, MessageHandler, filters, ContextTypes

CHAT_ID_FILE = "chat_ids.json"
BOT_TOKEN = "你的Bot Token"
# 全局保存Bot应用实例，方便推送时调用
bot_application = None

def load_chat_ids():
    try:
        with open(CHAT_ID_FILE, "r", encoding="utf-8") as f:
            return set(json.load(f))
    except FileNotFoundError:
        return set()

def save_chat_ids(chat_ids):
    with open(CHAT_ID_FILE, "w", encoding="utf-8") as f:
        json.dump(list(chat_ids), f, ensure_ascii=False)

async def handle_user_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id
    chat_ids = load_chat_ids()
    
    if chat_id not in chat_ids:
        chat_ids.add(chat_id)
        save_chat_ids(chat_ids)
        await update.message.reply_text("注册成功！")
    else:
        await update.message.reply_text("你已经在列表里啦～")

# 主动推送消息的核心函数
async def push_notification():
    chat_ids = load_chat_ids()
    if not chat_ids:
        print("没有注册用户，跳过推送")
        return
    
    # 你的条件判断逻辑，这里用示例的a和b
    a = 1
    b = 2
    if a > b:
        message_content = "触发条件：a > b，给你发送这条消息！"
    else:
        message_content = "触发条件：a ≤ b，给你发送这条消息！"
    
    # 遍历所有用户发送消息
    for chat_id in chat_ids:
        try:
            await bot_application.bot.send_message(chat_id=chat_id, text=message_content)
            print(f"成功给用户 {chat_id} 推送消息")
        except Exception as e:
            # 处理发送失败的情况（比如用户拉黑了Bot）
            print(f"给用户 {chat_id} 推送失败：{str(e)}")

async def main():
    global bot_application
    bot_application = ApplicationBuilder().token(BOT_TOKEN).build()
    bot_application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_user_message))
    
    # 启动Bot的轮询任务
    polling_task = asyncio.create_task(bot_application.run_polling())
    
    # 模拟每30秒检查一次条件（你可以替换成自己的触发逻辑，比如某个事件发生时调用push_notification）
    while True:
        await push_notification()
        await asyncio.sleep(30)

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

四、打包成exe的注意事项 #

用pyinstaller打包时，记住这几点：

1. 文件路径问题：代码里的CHAT_ID_FILE = "chat_ids.json"不需要改，exe运行时会自动在当前目录创建/读取这个文件。

2. 打包命令：

pyinstaller --onefile --windowed your_script_name.py

• --onefile：打包成单个exe文件，方便分享
• --windowed：运行时不显示控制台窗口（如果需要看日志可以去掉这个参数）

3. 依赖处理：python-telegram-bot库会被pyinstaller自动打包进去，不需要额外操作。

五、一些关键提醒 #

• Bot Token安全：别把Token硬编码在代码里分享给朋友，可以改成让用户运行时输入，或者从环境变量读取：

import os
BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") or input("请输入你的Bot Token：")

• 异常处理：发送消息时可能遇到用户拉黑Bot、账号注销等情况，一定要加异常捕获，避免程序崩溃。
• 轮询 vs Webhook：如果Bot需要长期运行，Webhook更高效，但打包成exe用轮询更简单，不需要服务器，适合个人分享场景。

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