feat(telegram_bot): Add update_token command for mail API token management

- Add new /update_token command handler to update mail API token in config.toml
- Register command in handlers list and set_my_commands for bot menu
- Add command documentation to help text with usage example
- Implement token update logic with TOML file read/write operations
- Display masked token values (first 8 and last 4 characters) for security
- Add automatic config reload after token update
- Include error handling for missing tomli_w dependency and file operations
- Restrict command to admin users only via @admin_only decorator
This commit is contained in:
2026-01-30 08:57:50 +08:00
parent 20e2719d0e
commit b7e3cd840b

View File

@@ -160,6 +160,7 @@ class ProvisionerBot:
("clean_teams", self.cmd_clean_teams),
("keys_usage", self.cmd_keys_usage),
("autogptplus", self.cmd_autogptplus),
("update_token", self.cmd_update_token),
]
for cmd, handler in handlers:
self.app.add_handler(CommandHandler(cmd, handler))
@@ -292,6 +293,7 @@ class ProvisionerBot:
BotCommand("team_register", "GPT Team 自动注册"),
# AutoGPTPlus
BotCommand("autogptplus", "AutoGPTPlus 管理面板"),
BotCommand("update_token", "更新邮件 API Token"),
]
try:
await self.app.bot.set_my_commands(commands)
@@ -367,6 +369,7 @@ class ProvisionerBot:
<b>🔧 AutoGPTPlus:</b>
/autogptplus - ChatGPT 订阅自动化管理面板
/update_token &lt;token&gt; - 更新邮件 API Token
<b>💡 示例:</b>
<code>/list</code> - 查看所有待处理账号
@@ -3643,6 +3646,68 @@ class ProvisionerBot:
self.current_task = None
self.current_team = None
@admin_only
async def cmd_update_token(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""更新 config.toml 中的邮件 API Token"""
if not context.args:
await update.message.reply_text(
"<b>📝 更新邮件 API Token</b>\n\n"
"用法: <code>/update_token &lt;new_token&gt;</code>\n\n"
"此命令会更新 config.toml 中的:\n"
"• [autogptplus] mail_api_token",
parse_mode="HTML"
)
return
new_token = context.args[0].strip()
if not new_token:
await update.message.reply_text("❌ Token 不能为空")
return
try:
import tomllib
import tomli_w
# 读取当前配置
with open(CONFIG_FILE, "rb") as f:
config = tomllib.load(f)
# 确保 autogptplus section 存在
if "autogptplus" not in config:
config["autogptplus"] = {}
# 获取旧 token (用于显示)
old_token = config["autogptplus"].get("mail_api_token", "")
old_display = f"{old_token[:8]}...{old_token[-4:]}" if len(old_token) > 12 else old_token or "(空)"
# 更新 token
config["autogptplus"]["mail_api_token"] = new_token
# 写回文件
with open(CONFIG_FILE, "wb") as f:
tomli_w.dump(config, f)
# 重载配置
reload_config()
new_display = f"{new_token[:8]}...{new_token[-4:]}" if len(new_token) > 12 else new_token
await update.message.reply_text(
f"<b>✅ Token 更新成功</b>\n\n"
f"旧 Token: <code>{old_display}</code>\n"
f"新 Token: <code>{new_display}</code>\n\n"
f"配置已自动重载生效",
parse_mode="HTML"
)
except ImportError:
await update.message.reply_text(
"❌ 缺少 tomli_w 依赖\n"
"请运行: uv add tomli_w"
)
except Exception as e:
await update.message.reply_text(f"❌ 更新 Token 失败: {e}")
@admin_only
async def cmd_autogptplus(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""AutoGPTPlus 配置管理 - 交互式菜单"""