d146ad9ebd
主要更新： - ✨ 新增 Telegram Bot 交互界面 - ✨ 新增欧洲账单自动生成功能 - 📦 整理项目结构，部署文件移至 deployment/ 目录 - 📝 完善文档，新增 CHANGELOG 和 Bot 部署指南 - 🔧 统一使用 pyproject.toml 管理依赖（支持 uv） - 🛡️ 增强 .gitignore，防止敏感配置泄露 新增文件： - tg_bot.py: Telegram Bot 主程序 - generate_billing.py: 独立账单生成工具 - modules/billing.py: 欧洲账单生成模块 - deployment/: Docker、systemd 等部署配置 - docs/: 完整的文档和更新日志 Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
428 lines
13 KiB
Python
428 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Telegram Bot for OpenAI Account Registration
|
|
|
|
Features:
|
|
- /start - 开始使用机器人
|
|
- /register - 注册单个账号
|
|
- /batch <count> - 批量注册账号
|
|
- /help - 帮助信息
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import asyncio
|
|
import secrets
|
|
import string
|
|
from typing import Optional
|
|
from datetime import datetime
|
|
|
|
# Telegram Bot
|
|
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
|
|
from telegram.ext import (
|
|
Application,
|
|
CommandHandler,
|
|
CallbackQueryHandler,
|
|
ContextTypes,
|
|
MessageHandler,
|
|
filters
|
|
)
|
|
|
|
# Local modules
|
|
from modules.register import OpenAIRegistrar
|
|
from modules.tempmail import TempMailClient
|
|
from modules.billing import EUBillingGenerator
|
|
from config import TEMPMAIL_CONFIG, DEBUG
|
|
|
|
# Bot configuration
|
|
BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN', '')
|
|
ALLOWED_USER_IDS = os.getenv('ALLOWED_USER_IDS', '') # Comma-separated user IDs
|
|
|
|
# Parse allowed users
|
|
ALLOWED_USERS = set()
|
|
if ALLOWED_USER_IDS:
|
|
try:
|
|
ALLOWED_USERS = set(int(uid.strip()) for uid in ALLOWED_USER_IDS.split(',') if uid.strip())
|
|
except ValueError:
|
|
print("⚠️ Warning: Invalid ALLOWED_USER_IDS format. Bot will be public!")
|
|
|
|
|
|
def generate_random_password(length: int = 16) -> str:
|
|
"""生成随机密码"""
|
|
chars = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789!@#$%^&*'
|
|
return ''.join(secrets.choice(chars) for _ in range(length))
|
|
|
|
|
|
def check_authorization(user_id: int) -> bool:
|
|
"""检查用户是否有权限使用 bot"""
|
|
if not ALLOWED_USERS:
|
|
return True # 如果未配置白名单，允许所有用户
|
|
return user_id in ALLOWED_USERS
|
|
|
|
|
|
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""处理 /start 命令"""
|
|
user = update.effective_user
|
|
|
|
if not check_authorization(user.id):
|
|
await update.message.reply_text(
|
|
"❌ 你没有权限使用此机器人。\n"
|
|
"请联系管理员获取访问权限。"
|
|
)
|
|
return
|
|
|
|
keyboard = [
|
|
[InlineKeyboardButton("📝 注册单个账号", callback_data="register_single")],
|
|
[InlineKeyboardButton("📦 批量注册", callback_data="register_batch")],
|
|
[InlineKeyboardButton("❓ 帮助", callback_data="help")]
|
|
]
|
|
reply_markup = InlineKeyboardMarkup(keyboard)
|
|
|
|
await update.message.reply_text(
|
|
f"👋 你好 {user.first_name}!\n\n"
|
|
"🤖 OpenAI 账号注册机器人\n\n"
|
|
"功能:\n"
|
|
"• 自动注册 OpenAI 账号\n"
|
|
"• 自动验证邮箱\n"
|
|
"• 获取 Access Token\n"
|
|
"• 生成欧洲账单 URL (可选)\n\n"
|
|
"请选择操作:",
|
|
reply_markup=reply_markup
|
|
)
|
|
|
|
|
|
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""处理 /help 命令"""
|
|
help_text = """
|
|
📖 **使用帮助**
|
|
|
|
**命令列表:**
|
|
/start - 开始使用
|
|
/register - 注册单个账号
|
|
/batch <数量> - 批量注册 (例: /batch 5)
|
|
/help - 显示此帮助
|
|
|
|
**注册流程:**
|
|
1. 选择注册模式 (单个/批量)
|
|
2. 选择是否生成账单 URL
|
|
3. 等待注册完成 (通常 30-60秒)
|
|
4. 接收账号信息
|
|
|
|
**账号信息包含:**
|
|
• 邮箱地址
|
|
• 密码
|
|
• Access Token
|
|
• 账单 URL (如已选择)
|
|
|
|
**注意事项:**
|
|
• 账号密码会自动生成
|
|
• 邮箱使用临时邮箱服务
|
|
• 请保存好收到的账号信息
|
|
|
|
如有问题,请联系管理员。
|
|
"""
|
|
|
|
if update.message:
|
|
await update.message.reply_text(help_text, parse_mode='Markdown')
|
|
elif update.callback_query:
|
|
await update.callback_query.message.reply_text(help_text, parse_mode='Markdown')
|
|
|
|
|
|
async def register_single(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""处理单个账号注册"""
|
|
user_id = update.effective_user.id
|
|
|
|
if not check_authorization(user_id):
|
|
await update.message.reply_text("❌ 你没有权限使用此机器人。")
|
|
return
|
|
|
|
# 询问是否需要生成账单
|
|
keyboard = [
|
|
[InlineKeyboardButton("✅ 生成账单 URL", callback_data="reg_single_with_billing")],
|
|
[InlineKeyboardButton("❌ 不生成账单", callback_data="reg_single_no_billing")]
|
|
]
|
|
reply_markup = InlineKeyboardMarkup(keyboard)
|
|
|
|
await update.message.reply_text(
|
|
"🔹 单个账号注册\n\n"
|
|
"是否需要生成欧洲账单 URL?",
|
|
reply_markup=reply_markup
|
|
)
|
|
|
|
|
|
async def batch_register(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""处理批量注册命令"""
|
|
user_id = update.effective_user.id
|
|
|
|
if not check_authorization(user_id):
|
|
await update.message.reply_text("❌ 你没有权限使用此机器人。")
|
|
return
|
|
|
|
# 解析数量
|
|
try:
|
|
count = int(context.args[0]) if context.args else 0
|
|
|
|
if count <= 0 or count > 20:
|
|
await update.message.reply_text(
|
|
"❌ 请提供有效的数量 (1-20)\n\n"
|
|
"示例: /batch 5"
|
|
)
|
|
return
|
|
|
|
# 保存数量到上下文
|
|
context.user_data['batch_count'] = count
|
|
|
|
# 询问是否需要生成账单
|
|
keyboard = [
|
|
[InlineKeyboardButton("✅ 生成账单 URL", callback_data="reg_batch_with_billing")],
|
|
[InlineKeyboardButton("❌ 不生成账单", callback_data="reg_batch_no_billing")]
|
|
]
|
|
reply_markup = InlineKeyboardMarkup(keyboard)
|
|
|
|
await update.message.reply_text(
|
|
f"🔹 批量注册 {count} 个账号\n\n"
|
|
"是否需要生成欧洲账单 URL?",
|
|
reply_markup=reply_markup
|
|
)
|
|
|
|
except (IndexError, ValueError):
|
|
await update.message.reply_text(
|
|
"❌ 请提供有效的数量\n\n"
|
|
"示例: /batch 5"
|
|
)
|
|
|
|
|
|
async def button_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""处理按钮回调"""
|
|
query = update.callback_query
|
|
await query.answer()
|
|
|
|
user_id = query.from_user.id
|
|
|
|
if not check_authorization(user_id):
|
|
await query.message.reply_text("❌ 你没有权限使用此机器人。")
|
|
return
|
|
|
|
data = query.data
|
|
|
|
if data == "register_single":
|
|
keyboard = [
|
|
[InlineKeyboardButton("✅ 生成账单 URL", callback_data="reg_single_with_billing")],
|
|
[InlineKeyboardButton("❌ 不生成账单", callback_data="reg_single_no_billing")]
|
|
]
|
|
reply_markup = InlineKeyboardMarkup(keyboard)
|
|
await query.message.reply_text(
|
|
"🔹 单个账号注册\n\n"
|
|
"是否需要生成欧洲账单 URL?",
|
|
reply_markup=reply_markup
|
|
)
|
|
|
|
elif data == "register_batch":
|
|
await query.message.reply_text(
|
|
"🔹 批量注册\n\n"
|
|
"请使用命令: /batch <数量>\n\n"
|
|
"示例: /batch 5"
|
|
)
|
|
|
|
elif data == "help":
|
|
await help_command(update, context)
|
|
|
|
elif data.startswith("reg_single_"):
|
|
generate_billing = "with_billing" in data
|
|
await perform_registration(query.message, 1, generate_billing)
|
|
|
|
elif data.startswith("reg_batch_"):
|
|
count = context.user_data.get('batch_count', 0)
|
|
if count <= 0:
|
|
await query.message.reply_text(
|
|
"❌ 请先使用 /batch <数量> 命令"
|
|
)
|
|
return
|
|
|
|
generate_billing = "with_billing" in data
|
|
await perform_registration(query.message, count, generate_billing)
|
|
|
|
|
|
async def perform_registration(message, count: int, generate_billing: bool):
|
|
"""执行注册流程"""
|
|
# 发送开始消息
|
|
status_msg = await message.reply_text(
|
|
f"🔄 开始注册 {count} 个账号...\n"
|
|
f"{'✅ 将生成账单 URL' if generate_billing else '❌ 不生成账单'}\n\n"
|
|
"⏳ 请稍候..."
|
|
)
|
|
|
|
# 初始化临时邮箱客户端
|
|
try:
|
|
api_base_url = TEMPMAIL_CONFIG.get('api_base_url')
|
|
username = TEMPMAIL_CONFIG.get('username')
|
|
password_cfg = TEMPMAIL_CONFIG.get('password')
|
|
admin_token = TEMPMAIL_CONFIG.get('admin_token')
|
|
|
|
if username and password_cfg:
|
|
tempmail_client = TempMailClient(
|
|
api_base_url=api_base_url,
|
|
username=username,
|
|
password=password_cfg
|
|
)
|
|
elif admin_token:
|
|
tempmail_client = TempMailClient(
|
|
api_base_url=api_base_url,
|
|
admin_token=admin_token
|
|
)
|
|
else:
|
|
await status_msg.edit_text("❌ 临时邮箱配置错误，请联系管理员")
|
|
return
|
|
|
|
except Exception as e:
|
|
await status_msg.edit_text(f"❌ 初始化失败: {str(e)}")
|
|
return
|
|
|
|
# 注册账号
|
|
success_accounts = []
|
|
failed_accounts = []
|
|
|
|
for i in range(1, count + 1):
|
|
try:
|
|
# 更新状态
|
|
await status_msg.edit_text(
|
|
f"🔄 正在注册第 {i}/{count} 个账号...\n"
|
|
f"✅ 成功: {len(success_accounts)}\n"
|
|
f"❌ 失败: {len(failed_accounts)}"
|
|
)
|
|
|
|
# 生成密码
|
|
password = generate_random_password()
|
|
|
|
# 创建注册器
|
|
registrar = OpenAIRegistrar(tempmail_client=tempmail_client)
|
|
|
|
# 执行注册
|
|
result = registrar.register_with_auto_email(password)
|
|
|
|
if not result.get('success'):
|
|
failed_accounts.append({
|
|
'email': result.get('email', 'N/A'),
|
|
'error': result.get('error', 'Unknown error')
|
|
})
|
|
continue
|
|
|
|
email = result['email']
|
|
account_info = {
|
|
'email': email,
|
|
'password': password,
|
|
'verified': result.get('verified', False)
|
|
}
|
|
|
|
# 如果需要生成账单
|
|
if generate_billing:
|
|
try:
|
|
access_token = registrar._step5_get_access_token()
|
|
billing_gen = EUBillingGenerator()
|
|
billing_result = billing_gen.generate_checkout_url(access_token)
|
|
|
|
if billing_result.success:
|
|
account_info['access_token'] = access_token
|
|
account_info['checkout_url'] = billing_result.checkout_url
|
|
else:
|
|
account_info['billing_error'] = billing_result.error
|
|
except Exception as e:
|
|
account_info['billing_error'] = str(e)
|
|
|
|
success_accounts.append(account_info)
|
|
|
|
except Exception as e:
|
|
failed_accounts.append({
|
|
'email': 'N/A',
|
|
'error': str(e)
|
|
})
|
|
|
|
# 发送结果
|
|
await status_msg.edit_text(
|
|
f"✅ 注册完成!\n\n"
|
|
f"成功: {len(success_accounts)}\n"
|
|
f"失败: {len(failed_accounts)}\n\n"
|
|
"正在发送账号信息..."
|
|
)
|
|
|
|
# 发送每个成功的账号
|
|
for idx, acc in enumerate(success_accounts, 1):
|
|
account_text = (
|
|
f"━━━━━━━━━━━━━━━━\n"
|
|
f"**账号 #{idx}**\n"
|
|
f"━━━━━━━━━━━━━━━━\n"
|
|
f"📧 邮箱: `{acc['email']}`\n"
|
|
f"🔑 密码: `{acc['password']}`\n"
|
|
)
|
|
|
|
if 'access_token' in acc:
|
|
account_text += f"🎫 Token: `{acc['access_token'][:50]}...`\n"
|
|
|
|
if 'checkout_url' in acc:
|
|
account_text += f"💳 [账单链接]({acc['checkout_url']})\n"
|
|
elif 'billing_error' in acc:
|
|
account_text += f"⚠️ 账单生成失败: {acc['billing_error']}\n"
|
|
|
|
await message.reply_text(account_text, parse_mode='Markdown')
|
|
|
|
# 如果有失败的账号
|
|
if failed_accounts:
|
|
failed_text = "❌ **失败的账号:**\n\n"
|
|
for idx, acc in enumerate(failed_accounts, 1):
|
|
failed_text += f"{idx}. {acc['email']}: {acc['error']}\n"
|
|
await message.reply_text(failed_text, parse_mode='Markdown')
|
|
|
|
# 删除状态消息
|
|
await status_msg.delete()
|
|
|
|
|
|
async def error_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""全局错误处理"""
|
|
print(f"❌ Error: {context.error}")
|
|
|
|
if update and update.effective_message:
|
|
await update.effective_message.reply_text(
|
|
f"❌ 发生错误: {str(context.error)}\n\n"
|
|
"请稍后重试或联系管理员。"
|
|
)
|
|
|
|
|
|
def main():
|
|
"""启动 bot"""
|
|
if not BOT_TOKEN:
|
|
print("❌ Error: TELEGRAM_BOT_TOKEN not set!")
|
|
print("Please set environment variable: export TELEGRAM_BOT_TOKEN='your_token'")
|
|
sys.exit(1)
|
|
|
|
# 检查配置
|
|
api_base_url = TEMPMAIL_CONFIG.get('api_base_url')
|
|
if not api_base_url or 'your.tempmail.domain' in api_base_url:
|
|
print("❌ Error: TEMPMAIL_CONFIG not configured in config.py")
|
|
sys.exit(1)
|
|
|
|
print("🤖 Starting Telegram Bot...")
|
|
print(f"📋 Allowed Users: {len(ALLOWED_USERS) if ALLOWED_USERS else 'ALL (Public)'}")
|
|
|
|
# 创建应用
|
|
application = Application.builder().token(BOT_TOKEN).build()
|
|
|
|
# 添加处理器
|
|
application.add_handler(CommandHandler("start", start))
|
|
application.add_handler(CommandHandler("help", help_command))
|
|
application.add_handler(CommandHandler("register", register_single))
|
|
application.add_handler(CommandHandler("batch", batch_register))
|
|
application.add_handler(CallbackQueryHandler(button_callback))
|
|
|
|
# 错误处理
|
|
application.add_error_handler(error_handler)
|
|
|
|
# 启动 bot
|
|
print("✅ Bot started successfully!")
|
|
print("Press Ctrl+C to stop")
|
|
|
|
application.run_polling(allowed_updates=Update.ALL_TYPES)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|