mygo-kyx/codexTool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(telegram_bot): Add AutoGPTPlus management panel with configuration controls

Browse Source 
- Add /autogptplus command with interactive menu for ChatGPT subscription automation
- Implement callback handler for AutoGPTPlus actions (config view, token setup, email/API testing)
- Add _show_autogptplus_config() to display current configuration with masked sensitive data
- Add _prompt_autogptplus_token() to handle Cloud Mail API token input and updates
- Add _test_autogptplus_email() and _test_autogptplus_api() for testing functionality
- Add _update_autogptplus_token() to persist token changes to config.toml
- Register AutoGPTPlus command in bot command list with Chinese description
- Update help text to include AutoGPTPlus management panel section
- Add admin-only access control for all AutoGPTPlus operations
- Provides centralized management interface for email domains, SEPA IBANs, and API configuration
This commit is contained in:
kyx236
2026-01-24 06:35:29 +08:00
parent 7c4688895e
commit 0e8b5ba237
1 changed files with 571 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

574
telegram_bot.py
View File

@@ -148,6 +148,7 @@ class ProvisionerBot:
("clean_errors", self.cmd_clean_errors),
("clean_teams", self.cmd_clean_teams),
("keys_usage", self.cmd_keys_usage),
("autogptplus", self.cmd_autogptplus),
]
for cmd, handler in handlers:
self.app.add_handler(CommandHandler(cmd, handler))
@@ -175,6 +176,10 @@ class ProvisionerBot:
self.callback_team_register,
pattern="^team_reg:"
))
self.app.add_handler(CallbackQueryHandler(
self.callback_autogptplus,
pattern="^autogptplus:"
))
# 注册自定义数量输入处理器 (GPT Team 注册)
self.app.add_handler(MessageHandler(
@@ -251,6 +256,7 @@ class ProvisionerBot:
BotCommand("gptmail_add", "添加 GPTMail API Key"),
BotCommand("gptmail_del", "删除 GPTMail API Key"),
BotCommand("test_email", "测试邮箱创建"),
BotCommand("autogptplus", "AutoGPTPlus 管理面板"),
]
try:
await self.app.bot.set_my_commands(commands)
@@ -320,7 +326,10 @@ class ProvisionerBot:
/team_fingerprint - 开启/关闭随机指纹
/team_register - 开始自动订阅注册
<b>💡 示例:</b>
<b><EFBFBD> AutoGPTPlus:</b>
/autogptplus - ChatGPT 订阅自动化管理面板
<b><3E>💡 示例:</b>
<code>/list</code> - 查看所有待处理账号
<code>/run 0</code> - 处理第一个 Team
<code>/gptmail_add my-api-key</code> - 添加 Key
@@ -2991,10 +3000,428 @@ class ProvisionerBot:
self.current_task = None
self.current_team = None
@admin_only
async def cmd_autogptplus(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""AutoGPTPlus 配置管理 - 交互式菜单"""
keyboard = [
[
InlineKeyboardButton("📋 查看配置", callback_data="autogptplus:config"),
InlineKeyboardButton("<EFBFBD> 设置 Token", callback_data="autogptplus:set_token"),
],
[
InlineKeyboardButton("<EFBFBD> 测试邮件", callback_data="autogptplus:test_email"),
InlineKeyboardButton("🔄 测试 API", callback_data="autogptplus:test_api"),
],
]
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(
"<b>🤖 AutoGPTPlus 管理面板</b>\n\n"
"ChatGPT 订阅自动化配置管理\n\n"
"请选择功能:",
parse_mode="HTML",
reply_markup=reply_markup
)
async def callback_autogptplus(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""处理 AutoGPTPlus 回调"""
query = update.callback_query
# 权限检查
user_id = update.effective_user.id
if user_id not in TELEGRAM_ADMIN_CHAT_IDS:
await query.answer("⛔ 无权限", show_alert=True)
return
await query.answer()
data = query.data.split(":")
action = data[1] if len(data) > 1 else ""
if action == "config":
await self._show_autogptplus_config(query)
elif action == "set_token":
await self._prompt_autogptplus_token(query, context)
elif action == "test_email":
await self._test_autogptplus_email(query)
elif action == "test_api":
await self._test_autogptplus_api(query)
elif action == "back":
# 返回主菜单
keyboard = [
[
InlineKeyboardButton("📋 查看配置", callback_data="autogptplus:config"),
InlineKeyboardButton("<EFBFBD> 设置 Token", callback_data="autogptplus:set_token"),
],
[
InlineKeyboardButton("📧 测试邮件", callback_data="autogptplus:test_email"),
InlineKeyboardButton("🔄 测试 API", callback_data="autogptplus:test_api"),
],
]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"<b>🤖 AutoGPTPlus 管理面板</b>\n\n"
"ChatGPT 订阅自动化配置管理\n\n"
"请选择功能:",
parse_mode="HTML",
reply_markup=reply_markup
)
async def _prompt_autogptplus_token(self, query, context: ContextTypes.DEFAULT_TYPE):
"""提示用户输入 Token"""
try:
from auto_gpt_team import MAIL_API_TOKEN
# 脱敏显示当前 Token
current_display = "未配置"
if MAIL_API_TOKEN:
if len(MAIL_API_TOKEN) > 10:
current_display = f"{MAIL_API_TOKEN[:8]}...{MAIL_API_TOKEN[-4:]}"
else:
current_display = MAIL_API_TOKEN[:4] + "..."
keyboard = [[InlineKeyboardButton("❌ 取消", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"<b>🔑 设置 Cloud Mail API Token</b>\n\n"
f"当前 Token: <code>{current_display}</code>\n\n"
"请直接发送新的 Token:\n"
"(发送后将自动保存到 config.toml)",
parse_mode="HTML",
reply_markup=reply_markup
)
# 设置等待输入状态
context.user_data["autogptplus_waiting_token"] = True
except ImportError:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"<b>❌ 模块未找到</b>\n\n"
"auto_gpt_team 模块未安装或导入失败",
parse_mode="HTML",
reply_markup=reply_markup
)
async def _show_autogptplus_config(self, query):
"""显示 AutoGPTPlus 配置"""
try:
from auto_gpt_team import (
MAIL_API_TOKEN, MAIL_API_BASE, EMAIL_DOMAINS,
SEPA_IBANS, RANDOM_FINGERPRINT, get_email_domains, get_sepa_ibans
)
# 脱敏显示 Token
token_display = "未配置"
if MAIL_API_TOKEN:
if len(MAIL_API_TOKEN) > 10:
token_display = f"{MAIL_API_TOKEN[:8]}...{MAIL_API_TOKEN[-4:]}"
else:
token_display = MAIL_API_TOKEN[:4] + "..."
# 获取域名列表
domains = get_email_domains()
domains_display = ", ".join(domains[:3]) if domains else "未配置"
if len(domains) > 3:
domains_display += f" (+{len(domains) - 3})"
# 获取 IBAN 列表
ibans = get_sepa_ibans()
# 随机指纹状态
fingerprint_status = "✅ 已开启" if RANDOM_FINGERPRINT else "❌ 已关闭"
lines = [
"<b>📋 AutoGPTPlus 配置</b>",
"",
"<b>🔑 Cloud Mail API</b>",
f" Token: <code>{token_display}</code>",
f" 地址: {MAIL_API_BASE or '未配置'}",
"",
"<b>📧 邮箱域名</b>",
f" 数量: {len(domains)} 个",
f" 域名: {domains_display}",
"",
"<b>💳 SEPA IBAN</b>",
f" 数量: {len(ibans)} 个",
"",
"<b>🎭 随机指纹</b>",
f" 状态: {fingerprint_status}",
]
# 配置状态检查
config_ok = bool(MAIL_API_TOKEN and MAIL_API_BASE and domains)
if config_ok:
lines.append("\n<b>✅ 配置完整，可以使用</b>")
else:
lines.append("\n<b>⚠️ 配置不完整:</b>")
if not MAIL_API_TOKEN:
lines.append(" • 缺少 mail_api_token")
if not MAIL_API_BASE:
lines.append(" • 缺少 mail_api_base")
if not domains:
lines.append(" • 缺少 email_domains")
keyboard = [
[
InlineKeyboardButton("🔑 设置 Token", callback_data="autogptplus:set_token"),
InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back"),
]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"\n".join(lines),
parse_mode="HTML",
reply_markup=reply_markup
)
except ImportError:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"<b>❌ 模块未找到</b>\n\n"
"auto_gpt_team 模块未安装或导入失败",
parse_mode="HTML",
reply_markup=reply_markup
)
except Exception as e:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
f"❌ 获取配置失败: {e}",
reply_markup=reply_markup
)
async def _test_autogptplus_email(self, query):
"""测试 AutoGPTPlus 邮件创建"""
await query.edit_message_text("⏳ 正在测试邮件创建...")
try:
from auto_gpt_team import (
MAIL_API_TOKEN, MAIL_API_BASE, get_email_domains
)
import requests
import random
import string
# 检查配置
if not MAIL_API_TOKEN or not MAIL_API_BASE:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"<b>❌ 配置不完整</b>\n\n"
"请先在 config.toml 中配置:\n"
"• mail_api_token\n"
"• mail_api_base",
parse_mode="HTML",
reply_markup=reply_markup
)
return
domains = get_email_domains()
if not domains:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"<b>❌ 邮箱域名未配置</b>\n\n"
"请先在 config.toml 中配置 email_domains\n"
"或使用 /domain_add 添加域名",
parse_mode="HTML",
reply_markup=reply_markup
)
return
# 生成测试邮箱
random_str = ''.join(random.choices(string.ascii_lowercase + string.digits, k=8))
domain = random.choice(domains)
test_email = f"test-{random_str}@{domain.lstrip('@')}"
# 测试创建邮箱 (通过查询邮件列表来验证 API 连接)
url = f"{MAIL_API_BASE}/api/public/emailList"
headers = {
"Authorization": MAIL_API_TOKEN,
"Content-Type": "application/json"
}
payload = {
"toEmail": test_email,
"timeSort": "desc",
"size": 1
}
response = requests.post(url, headers=headers, json=payload, timeout=10)
data = response.json()
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
if data.get("code") == 200:
await query.edit_message_text(
"<b>✅ 邮件 API 测试成功</b>\n\n"
f"测试邮箱: <code>{test_email}</code>\n"
f"API 响应: 正常\n\n"
f"邮件系统已就绪，可以接收验证码",
parse_mode="HTML",
reply_markup=reply_markup
)
else:
error_msg = data.get("message", "未知错误")
await query.edit_message_text(
f"<b>⚠️ API 响应异常</b>\n\n"
f"状态码: {data.get('code')}\n"
f"错误: {error_msg}",
parse_mode="HTML",
reply_markup=reply_markup
)
except ImportError:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"<b>❌ 模块未找到</b>\n\n"
"auto_gpt_team 模块未安装或导入失败",
parse_mode="HTML",
reply_markup=reply_markup
)
except requests.exceptions.Timeout:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"<b>❌ 连接超时</b>\n\n"
"无法连接到邮件 API 服务器\n"
"请检查 mail_api_base 配置",
parse_mode="HTML",
reply_markup=reply_markup
)
except Exception as e:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
f"<b>❌ 测试失败</b>\n\n{e}",
parse_mode="HTML",
reply_markup=reply_markup
)
async def _test_autogptplus_api(self, query):
"""测试 AutoGPTPlus API 连接"""
await query.edit_message_text("⏳ 正在测试 API 连接...")
try:
from auto_gpt_team import MAIL_API_TOKEN, MAIL_API_BASE
import requests
# 检查配置
if not MAIL_API_TOKEN or not MAIL_API_BASE:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"<b>❌ 配置不完整</b>\n\n"
"请先在 config.toml 中配置:\n"
"• mail_api_token\n"
"• mail_api_base",
parse_mode="HTML",
reply_markup=reply_markup
)
return
# 测试 API 连接
url = f"{MAIL_API_BASE}/api/public/emailList"
headers = {
"Authorization": MAIL_API_TOKEN,
"Content-Type": "application/json"
}
payload = {
"toEmail": "test@test.com",
"timeSort": "desc",
"size": 1
}
start_time = time.time()
response = requests.post(url, headers=headers, json=payload, timeout=10)
elapsed = time.time() - start_time
data = response.json()
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
if response.status_code == 200 and data.get("code") == 200:
await query.edit_message_text(
"<b>✅ API 连接测试成功</b>\n\n"
f"<b>服务器:</b> {MAIL_API_BASE}\n"
f"<b>响应时间:</b> {elapsed*1000:.0f}ms\n"
f"<b>状态:</b> 正常\n\n"
"Cloud Mail API 服务运行正常",
parse_mode="HTML",
reply_markup=reply_markup
)
else:
error_msg = data.get("message", "未知错误")
await query.edit_message_text(
f"<b>⚠️ API 响应异常</b>\n\n"
f"<b>HTTP 状态:</b> {response.status_code}\n"
f"<b>API 状态:</b> {data.get('code')}\n"
f"<b>错误:</b> {error_msg}\n"
f"<b>响应时间:</b> {elapsed*1000:.0f}ms",
parse_mode="HTML",
reply_markup=reply_markup
)
except ImportError:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"<b>❌ 模块未找到</b>\n\n"
"auto_gpt_team 模块未安装或导入失败",
parse_mode="HTML",
reply_markup=reply_markup
)
except requests.exceptions.ConnectionError:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"<b>❌ 连接失败</b>\n\n"
"无法连接到邮件 API 服务器\n"
"请检查:\n"
"• mail_api_base 地址是否正确\n"
"• 服务器是否在线\n"
"• 网络连接是否正常",
parse_mode="HTML",
reply_markup=reply_markup
)
except requests.exceptions.Timeout:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"<b>❌ 连接超时</b>\n\n"
"API 服务器响应超时 (>10s)\n"
"请检查服务器状态",
parse_mode="HTML",
reply_markup=reply_markup
)
except Exception as e:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
f"<b>❌ 测试失败</b>\n\n{e}",
parse_mode="HTML",
reply_markup=reply_markup
)
@admin_only
async def handle_team_custom_count(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""处理 GPT Team 自定义数量输入"""
# 检查是否在等待输入状态
"""处理文本输入 (GPT Team 自定义数量 / AutoGPTPlus Token)"""
# 处理 AutoGPTPlus Token 输入
if context.user_data.get("autogptplus_waiting_token"):
context.user_data["autogptplus_waiting_token"] = False
await self._handle_autogptplus_token_input(update, context)
return
# 处理 GPT Team 自定义数量输入
if not context.user_data.get("team_waiting_count"):
return # 不在等待状态，忽略消息
@@ -3038,6 +3465,147 @@ class ProvisionerBot:
reply_markup=reply_markup
)
async def _handle_autogptplus_token_input(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""处理 AutoGPTPlus Token 输入 - 保存后立即测试"""
import tomli_w
import requests
token = update.message.text.strip()
if not token:
await update.message.reply_text("❌ Token 不能为空")
return
# 脱敏显示
if len(token) > 10:
token_display = f"{token[:8]}...{token[-4:]}"
else:
token_display = token[:4] + "..."
# 先发送保存中的消息
status_msg = await update.message.reply_text(
f"⏳ 正在保存并验证 Token...\n\n"
f"Token: <code>{token_display}</code>",
parse_mode="HTML"
)
try:
# 读取当前配置获取 API 地址
with open(CONFIG_FILE, "rb") as f:
import tomllib
config = tomllib.load(f)
mail_api_base = config.get("autogptplus", {}).get("mail_api_base", "")
if not mail_api_base:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await status_msg.edit_text(
"<b>❌ 无法验证 Token</b>\n\n"
"mail_api_base 未配置\n"
"请先在 config.toml 中配置 API 地址",
parse_mode="HTML",
reply_markup=reply_markup
)
return
# 测试 Token 是否有效
url = f"{mail_api_base}/api/public/emailList"
headers = {
"Authorization": token,
"Content-Type": "application/json"
}
payload = {
"toEmail": "test@test.com",
"timeSort": "desc",
"size": 1
}
start_time = time.time()
response = requests.post(url, headers=headers, json=payload, timeout=10)
elapsed = time.time() - start_time
data = response.json()
# 检查 Token 是否有效
if response.status_code == 200 and data.get("code") == 200:
# Token 有效，保存到配置文件
if "autogptplus" not in config:
config["autogptplus"] = {}
config["autogptplus"]["mail_api_token"] = token
with open(CONFIG_FILE, "wb") as f:
tomli_w.dump(config, f)
keyboard = [
[
InlineKeyboardButton("📋 查看配置", callback_data="autogptplus:config"),
InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back"),
],
]
reply_markup = InlineKeyboardMarkup(keyboard)
await status_msg.edit_text(
f"<b>✅ Token 验证成功并已保存</b>\n\n"
f"Token: <code>{token_display}</code>\n"
f"响应时间: {elapsed*1000:.0f}ms\n\n"
f"💡 使用 /reload 重载配置使其生效",
parse_mode="HTML",
reply_markup=reply_markup
)
else:
# Token 无效
error_msg = data.get("message", "未知错误")
keyboard = [
[
InlineKeyboardButton("🔑 重新设置", callback_data="autogptplus:set_token"),
InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back"),
],
]
reply_markup = InlineKeyboardMarkup(keyboard)
await status_msg.edit_text(
f"<b>❌ Token 验证失败</b>\n\n"
f"Token: <code>{token_display}</code>\n"
f"错误: {error_msg}\n\n"
f"Token 未保存，请检查后重试",
parse_mode="HTML",
reply_markup=reply_markup
)
except requests.exceptions.ConnectionError:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await status_msg.edit_text(
"<b>❌ 连接失败</b>\n\n"
"无法连接到 API 服务器\n"
"请检查 mail_api_base 配置",
parse_mode="HTML",
reply_markup=reply_markup
)
except requests.exceptions.Timeout:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await status_msg.edit_text(
"<b>❌ 连接超时</b>\n\n"
"API 服务器响应超时",
parse_mode="HTML",
reply_markup=reply_markup
)
except ImportError:
await status_msg.edit_text(
"❌ 缺少 tomli_w 依赖\n"
"请运行: uv add tomli_w"
)
except Exception as e:
keyboard = [[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await status_msg.edit_text(
f"<b>❌ 验证失败</b>\n\n{e}",
parse_mode="HTML",
reply_markup=reply_markup
)
async def main():
"""主函数"""