Python开发Telegram bot实现/add_user命令写入角色txt文件咨询

前置说明 #

原有代码存在2个可优化点：

• 启动时一次性读取用户列表，后续添加新用户后权限不会实时生效，建议改为每次调用命令时动态校验或者添加后刷新缓存
• 代码中absensak变量未定义，需要提前补充对应角色的用户列表读取逻辑

/add_user 命令实现逻辑 #

核心步骤如下：

1. 接收命令传入的参数，校验格式是否符合/add_user 用户名 as 角色的要求
2. 校验角色是否仅为允许的admin/user
3. 检查目标用户是否已经存在于对应角色文件中，避免重复写入
4. 写入用户到对应角色的txt文件
5. 返回操作结果给调用者

完整实现代码 #

from telegram.ext import *
import os

API_KEY = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
# 提前定义角色文件路径常量，方便后续维护
ROLE_FILE_PATH = {
    "admin": "./userlist/admin.txt",
    "user": "./userlist/user.txt"
}

# 新增通用工具函数：读取指定角色的用户列表
def load_role_users(role: str) -> list:
    file_path = ROLE_FILE_PATH.get(role)
    if not os.path.exists(file_path):
        # 如果文件不存在先创建空文件
        os.makedirs(os.path.dirname(file_path), exist_ok=True)
        open(file_path, 'w', encoding='utf-8').close()
        return []
    with open(file_path, 'r', encoding='utf-8') as f:
        # 按行读取，去掉空行和前后空白字符
        return [u.strip() for u in f.readlines() if u.strip()]

# 新增通用工具函数：写入用户到指定角色文件
def add_user_to_role(username: str, role: str) -> bool:
    file_path = ROLE_FILE_PATH.get(role)
    if not file_path:
        return False
    # 先检查用户是否已经存在
    existing_users = load_role_users(role)
    if username in existing_users:
        return False
    # 追加写入
    with open(file_path, 'a', encoding='utf-8') as f:
        f.write(f"{username}\n")
    return True

def start_command(update, context):
    update.message.reply_text('Hello there! I\'m a very useful bot. What\'s up?')
    
def abs_command(update, context):
    update.message.reply_text('ABS command is started . .')
    os.system('python abs.py')
    update.message.reply_text('ABS command is Finished . .')    
    
def add_user_command(update, context):
    # 校验参数数量是否正确
    args = context.args
    if len(args) != 3 or args[1].lower() != 'as':
        update.message.reply_text("命令格式错误，请使用：/add_user 用户名 as 角色，支持的角色为admin/user")
        return
    target_username = args[0].strip()
    target_role = args[2].strip().lower()
    # 校验角色合法性
    if target_role not in ROLE_FILE_PATH.keys():
        update.message.reply_text("不支持的角色，仅允许添加admin/user角色")
        return
    # 执行写入
    add_result = add_user_to_role(target_username, target_role)
    if add_result:
        update.message.reply_text(f"成功将用户 {target_username} 添加为 {target_role} 角色")
    else:
        update.message.reply_text(f"添加失败：用户 {target_username} 已经是 {target_role} 角色，或角色不存在")
    

# Run the programme
if __name__ == '__main__':
    updater = Updater(API_KEY, use_context=True)
    dp = updater.dispatcher

    # 启动时预加载用户列表
    user = load_role_users("user")
    admin = load_role_users("admin")
    # 注意这里原代码的absensak变量未定义，请自行补充对应角色的读取逻辑
    absensak = [] # 这里替换为你实际的absensak角色用户列表读取逻辑
    
    dp.add_handler(CommandHandler('start', start_command, Filters.user(username=user)))
    dp.add_handler(CommandHandler('start', start_command, Filters.user(username=admin)))    
    dp.add_handler(CommandHandler('abs', abs_command, Filters.user(username=absensak)))
    dp.add_handler(CommandHandler('add_user', add_user_command, Filters.user(username=admin)))
    
    # Run the bot
    updater.start_polling(1.0)
    updater.idle()

额外优化建议 #

如果需要添加用户后立即生效不需要重启bot，可以改为每次调用命令时动态读取用户列表校验权限，不需要在启动时一次性绑定Filters。

内容的提问来源于stack exchange，提问作者Kautsar Shabil Yudhany