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

a

Browse Source
This commit is contained in:
kyx236
2026-01-24 08:06:56 +08:00
parent 970340fbd4
commit af161cca4f
1 changed files with 138 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

141
telegram_bot.py
View File

@@ -3141,9 +3141,9 @@ class ProvisionerBot:
elif action == "test_api":
await self._test_autogptplus_api(query)
elif action == "domains":
await self._show_autogptplus_domains(query, sub_action)
await self._show_autogptplus_domains(query, sub_action, context)
elif action == "ibans":
await self._show_autogptplus_ibans(query, sub_action)
await self._show_autogptplus_ibans(query, sub_action, context)
elif action == "fingerprint":
await self._toggle_autogptplus_fingerprint(query)
elif action == "stats":
@@ -3503,7 +3503,7 @@ class ProvisionerBot:
reply_markup=reply_markup
)
async def _show_autogptplus_domains(self, query, sub_action: str = ""):
async def _show_autogptplus_domains(self, query, sub_action: str = "", context=None):
"""显示/管理域名"""
try:
from auto_gpt_team import (
@@ -3526,6 +3526,25 @@ class ProvisionerBot:
)
return
if sub_action == "add":
# 提示添加域名
keyboard = [[InlineKeyboardButton("❌ 取消", callback_data="autogptplus:domains")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"<b>➕ 添加邮箱域名</b>\n\n"
"请直接发送要添加的域名:\n\n"
"<b>支持格式:</b>\n"
"• 单个: <code>@example.com</code>\n"
"• 多个: <code>@a.com @b.com</code>\n"
"• 逗号分隔: <code>@a.com,@b.com</code>\n\n"
"<i>@ 符号可省略，会自动添加</i>",
parse_mode="HTML",
reply_markup=reply_markup
)
if context:
context.user_data["autogptplus_waiting_domain"] = True
return
# 显示域名列表
domains = get_email_domains()
file_count = get_file_domains_count()
@@ -3545,6 +3564,7 @@ class ProvisionerBot:
keyboard = [
[
InlineKeyboardButton("➕ 添加域名", callback_data="autogptplus:domains:add"),
InlineKeyboardButton("🗑️ 清空txt域名", callback_data="autogptplus:domains:clear"),
],
[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")],
@@ -3574,7 +3594,7 @@ class ProvisionerBot:
reply_markup=reply_markup
)
async def _show_autogptplus_ibans(self, query, sub_action: str = ""):
async def _show_autogptplus_ibans(self, query, sub_action: str = "", context=None):
"""显示/管理 IBAN"""
try:
from auto_gpt_team import get_sepa_ibans, load_ibans_from_file, SEPA_IBANS, clear_sepa_ibans
@@ -3591,6 +3611,24 @@ class ProvisionerBot:
)
return
if sub_action == "add":
# 提示添加 IBAN
keyboard = [[InlineKeyboardButton("❌ 取消", callback_data="autogptplus:ibans")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"<b>➕ 添加 IBAN</b>\n\n"
"请直接发送要添加的 IBAN:\n\n"
"<b>支持格式:</b>\n"
"• 单个: <code>DE89370400440532013000</code>\n"
"• 多个: 每行一个或逗号分隔\n\n"
"<i>只支持德国 IBAN (DE开头)</i>",
parse_mode="HTML",
reply_markup=reply_markup
)
if context:
context.user_data["autogptplus_waiting_iban"] = True
return
# 显示 IBAN 列表
ibans = get_sepa_ibans()
file_ibans = load_ibans_from_file()
@@ -3612,6 +3650,7 @@ class ProvisionerBot:
keyboard = [
[
InlineKeyboardButton("➕ 添加IBAN", callback_data="autogptplus:ibans:add"),
InlineKeyboardButton("🗑️ 清空IBAN", callback_data="autogptplus:ibans:clear"),
],
[InlineKeyboardButton("◀️ 返回", callback_data="autogptplus:back")],
@@ -3838,7 +3877,7 @@ class ProvisionerBot:
@admin_only
async def handle_team_custom_count(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""处理文本输入 (GPT Team 自定义数量 / AutoGPTPlus Token)"""
"""处理文本输入 (GPT Team 自定义数量 / AutoGPTPlus Token / 域名 / IBAN)"""
# 处理 AutoGPTPlus Token 输入
if context.user_data.get("autogptplus_waiting_token"):
@@ -3846,6 +3885,18 @@ class ProvisionerBot:
await self._handle_autogptplus_token_input(update, context)
return
# 处理域名添加输入
if context.user_data.get("autogptplus_waiting_domain"):
context.user_data["autogptplus_waiting_domain"] = False
await self._handle_autogptplus_domain_input(update, context)
return
# 处理 IBAN 添加输入
if context.user_data.get("autogptplus_waiting_iban"):
context.user_data["autogptplus_waiting_iban"] = False
await self._handle_autogptplus_iban_input(update, context)
return
# 处理 GPT Team 自定义数量输入
if not context.user_data.get("team_waiting_count"):
return # 不在等待状态，忽略消息
@@ -4031,6 +4082,86 @@ class ProvisionerBot:
reply_markup=reply_markup
)
async def _handle_autogptplus_domain_input(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""处理域名添加输入"""
try:
from auto_gpt_team import add_email_domains
text = update.message.text.strip()
if not text:
await update.message.reply_text("❌ 域名不能为空")
return
# 解析输入 (支持空格、逗号、换行分隔)
domains = [s.strip() for s in text.replace(",", " ").replace("\n", " ").split() if s.strip()]
if not domains:
await update.message.reply_text("❌ 未提供有效的域名")
return
added, skipped, invalid, total = add_email_domains(domains)
# 构建响应消息
lines = ["<b>✅ 域名导入完成</b>\n"]
lines.append(f"新增: {added}")
lines.append(f"跳过 (重复): {skipped}")
if invalid > 0:
lines.append(f"无效 (格式错误): {invalid}")
lines.append(f"当前总数: {total}")
keyboard = [[InlineKeyboardButton("📧 域名管理", callback_data="autogptplus:domains")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(
"\n".join(lines),
parse_mode="HTML",
reply_markup=reply_markup
)
except ImportError:
await update.message.reply_text("❌ auto_gpt_team 模块未找到")
except Exception as e:
await update.message.reply_text(f"❌ 添加域名失败: {e}")
async def _handle_autogptplus_iban_input(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""处理 IBAN 添加输入"""
try:
from auto_gpt_team import add_sepa_ibans
text = update.message.text.strip()
if not text:
await update.message.reply_text("❌ IBAN 不能为空")
return
# 解析输入 (支持空格、逗号、换行分隔)
ibans = [s.strip() for s in text.replace(",", " ").replace("\n", " ").split() if s.strip()]
if not ibans:
await update.message.reply_text("❌ 未提供有效的 IBAN")
return
added, skipped, total = add_sepa_ibans(ibans)
# 构建响应消息
lines = ["<b>✅ IBAN 导入完成</b>\n"]
lines.append(f"新增: {added}")
lines.append(f"跳过 (重复/无效): {skipped}")
lines.append(f"当前总数: {total}")
keyboard = [[InlineKeyboardButton("💳 IBAN 管理", callback_data="autogptplus:ibans")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(
"\n".join(lines),
parse_mode="HTML",
reply_markup=reply_markup
)
except ImportError:
await update.message.reply_text("❌ auto_gpt_team 模块未找到")
except Exception as e:
await update.message.reply_text(f"❌ 添加 IBAN 失败: {e}")
async def main():
"""主函数"""