如何使用python-telegram-bot向已注册用户广播自定义消息

核心逻辑（全平台通用） #

不管你使用的是Telegram、Discord还是微信生态的机器人框架，实现逻辑都是统一的：

• 持久化存储已注册主体的唯一标识：提前把完成注册流程的用户ID、群聊ID存到本地，推荐新手先用JSON文件存储，无需额外搭建数据库，存储时必须区分private（私聊）和group（群聊）两种类型，避免调用错误的发送接口。
• 统一消息发送入口：把不同场景的发消息API封装成通用函数，不用每次广播时重复写判断逻辑。
• 遍历发送+限流：遍历存储的ID列表逐个发送，必须加发送间隔，避免触发平台反垃圾风控规则。

代码示例 #

首先准备存储注册列表的registered_list.json文件，结构如下：

[
  {"id": "user_12345", "type": "private"},
  {"id": "group_67890", "type": "group"},
  {"id": "user_54321", "type": "private"}
]

你需要先在注册逻辑里把完成注册的用户/群信息按上述格式写入该文件，再调用下面的广播函数：

import json
import time

def load_registered_subjects() -> list:
    """加载所有已注册的用户/群信息"""
    with open("registered_list.json", "r", encoding="utf-8") as f:
        return json.load(f)

def broadcast_msg(bot, custom_content: str) -> tuple:
    """
    跨场景广播自定义消息
    :param bot: 你自己初始化完成的机器人实例
    :param custom_content: 要广播的自定义消息内容
    :return: (成功发送数, 失败发送数)
    """
    subjects = load_registered_subjects()
    success = 0
    fail = 0

    for sub in subjects:
        try:
            if sub["type"] == "private":
                # 替换为你用的框架对应的私聊发消息方法
                bot.send_private_msg(user_id=sub["id"], content=custom_content)
            elif sub["type"] == "group":
                # 替换为你用的框架对应的群聊发消息方法
                bot.send_group_msg(group_id=sub["id"], content=custom_content)
            success += 1
            # 发送间隔可根据平台规则调整，新手建议不低于1秒/条
            time.sleep(1)
        except Exception as e:
            print(f"发送至{sub['type']}:{sub['id']}失败，错误信息：{str(e)}")
            fail += 1
    return success, fail

# 调用示例
if __name__ == "__main__":
    # 替换为你的机器人初始化代码
    my_bot = init_your_bot()
    # 传入自定义广播内容即可触发广播
    suc, fai = broadcast_msg(my_bot, "本次更新已完成，所有功能恢复正常使用~")
    print(f"广播结束：成功{suc}条，失败{fai}条")

必要的优化建议 #

• 增加权限校验：如果广播功能是通过聊天命令触发的，必须校验触发者是否为管理员，避免普通用户随意调用广播接口。
• 增加失败重试机制：单次发送失败后可以间隔3秒重试1次，降低网络波动导致的发送失败率。
• 增加退订入口：允许用户/群主动取消广播订阅，删除对应存储的ID，避免被投诉导致账号封禁。

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