更新autodone 和 德国iban

This commit is contained in:
dela
2026-01-11 18:31:12 +08:00
parent 25e676012b
commit d31ac8628f
8 changed files with 1854 additions and 41 deletions

419
PAYMENT_GUIDE.md Normal file
View File

@@ -0,0 +1,419 @@
# OpenAI 自动注册 + 支付集成指南
## 🎯 功能概述
现在 `auto_register.py` 已经完全集成了支付功能,可以一键完成:
1.**注册 OpenAI 账号**(自动生成临时邮箱 + 邮箱验证)
2.**生成欧洲账单 URL**EU Billing
3.**自动添加支付方式**SEPA自动生成德国IBAN
---
## 📋 核心模块
### 1. **IBAN 生成器** (`modules/iban_generator.py`)
- 自动生成符合 ISO 7064 Mod 97-10 标准的德国IBAN
- 使用真实的德国银行代码BLZ
- 每个账号生成唯一的IBAN
### 2. **Stripe 支付处理器** (`modules/stripe_payment.py`)
- 使用 `curl_cffi` 模拟真实浏览器
- 支持 SEPA 支付方式
- 完整的支付流程:创建支付方式 → 确认支付 → 轮询状态
### 3. **注册器集成** (`modules/register.py`)
- 新增 `add_payment_method()` 方法
- 共享 HTTPClient 实例,保持 session 一致性
---
## 🚀 使用方法
### **基础注册(不含支付)**
```bash
# 注册1个账号
python auto_register.py
# 注册10个账号
python auto_register.py --count 10
# 指定密码(所有账号使用相同密码)
python auto_register.py -c 5 -p "MyPassword123!"
# 保存到指定文件
python auto_register.py -c 10 -o my_accounts.json
```
---
### **注册 + 生成账单 URL**
```bash
# 注册并生成欧洲账单URL
python auto_register.py --eu-billing
# 批量注册10个账号并生成账单
python auto_register.py -c 10 --eu-billing
```
---
### **注册 + 账单 + 自动添加支付(推荐)**
```bash
# 🔥 完整流程:注册 + 账单 + 支付
python auto_register.py --eu-billing --add-payment
# 🔥 批量10个账号完整流程
python auto_register.py -c 10 --eu-billing --add-payment
# 🔥 自定义支付信息
python auto_register.py \
--eu-billing \
--add-payment \
--name "Hans Mueller" \
--address "Hauptstraße 123" \
--city "Berlin" \
--postal-code "10115" \
--state "BE" \
--country "DE"
```
---
## 📝 命令行参数详解
### **基础参数**
| 参数 | 说明 | 默认值 |
|------|------|--------|
| `-c, --count` | 注册账号数量 | `1` |
| `-p, --password` | 指定密码(所有账号相同) | 自动生成随机密码 |
| `-o, --output` | 保存账号的JSON文件路径 | `registered_accounts.json` |
### **功能开关**
| 参数 | 说明 |
|------|------|
| `--eu-billing` | 启用欧洲账单生成 |
| `--add-payment` | 启用自动添加支付方式(需要 `--eu-billing` |
### **支付信息参数**(仅在 `--add-payment` 时生效)
| 参数 | 说明 | 默认值 |
|------|------|--------|
| `--name` | 持卡人姓名 | `John Doe` |
| `--address` | 街道地址 | `123 Main Street` |
| `--city` | 城市 | `New York` |
| `--postal-code` | 邮编 | `10001` |
| `--state` | 州/省 | `NY` |
| `--country` | 国家代码 | `US` |
**注意:** IBAN 会自动生成每个账号使用不同的德国IBAN。
---
## 💡 实际使用示例
### **示例 1: 快速注册 1 个账号(含支付)**
```bash
python auto_register.py --eu-billing --add-payment
```
**输出:**
```
============================================================
Starting batch registration: 1 accounts
EU Billing: Enabled
Payment: Enabled (auto-generating IBANs)
============================================================
✅ IBAN Generator initialized
────────────────────────────────────────────────────────────
[1/1] Registering account #1...
────────────────────────────────────────────────────────────
🔢 Generated IBAN: DE22700201007412345678
✅ [1.1] Visited ChatGPT (200)
✅ [1.2] CSRF token extracted
✅ [1.4] Got OAuth URL
...
✅ [3] Registration successful!
✅ [4.3] Got verification code: 123456
✅ [4.4] Email verified successfully!
✅ [5] Access token retrieved
✅ EU billing URL generated
URL: https://pay.openai.com/c/pay/cs_live_xxx...
🔐 Adding payment method...
INFO:modules.stripe_payment:Creating payment method with IBAN: DE227002****5678
INFO:modules.stripe_payment:✅ Payment method created: pm_xxx
INFO:modules.stripe_payment:Confirming payment with method: pm_xxx
INFO:modules.stripe_payment:✅ Payment confirmation response state: succeeded
✅ Payment method added successfully
✅ Account #1 registered successfully!
Email: abc123@temp.mail
Password: Xy9$zK2@pQ4!mN8&
Checkout URL: https://pay.openai.com/c/pay/cs_live_xxx...
Payment: ✅ Added
IBAN: DE22700201007412345678
```
---
### **示例 2: 批量注册 10 个账号(含支付)**
```bash
python auto_register.py -c 10 --eu-billing --add-payment -o team_accounts.json
```
**结果:**
- 注册 10 个 OpenAI 账号
- 每个账号都有欧洲账单 URL
- 每个账号都已添加支付方式不同的IBAN
- 保存到 `team_accounts.json`
**JSON 输出格式:**
```json
[
{
"email": "abc123@temp.mail",
"password": "Xy9$zK2@pQ4!mN8&",
"verified": true,
"checkout_url": "https://pay.openai.com/c/pay/cs_live_xxx...",
"payment_added": true,
"iban": "DE22700201007412345678"
},
{
"email": "def456@temp.mail",
"password": "Zw3@hJ5!tL9$xM2&",
"verified": true,
"checkout_url": "https://pay.openai.com/c/pay/cs_live_yyy...",
"payment_added": true,
"iban": "DE84601202002216834329"
}
...
]
```
---
### **示例 3: 使用德国地址信息**
```bash
python auto_register.py \
-c 5 \
--eu-billing \
--add-payment \
--name "Hans Mueller" \
--address "Hauptstraße 123" \
--city "Berlin" \
--postal-code "10115" \
--country "DE"
```
---
## 🔧 技术细节
### **IBAN 生成算法**
```python
# 生成流程(自动执行):
1. 随机选择德国银行代码BLZ 60120200
2. 生成 10 位随机账号 2216834329
3. 拼接 BBAN60120200 + 2216834329
4. 计算 ISO 7064 Mod 97-10 校验位
5. 最终 IBANDE84601202002216834329
```
### **支付流程3步**
```
Step 1: 创建支付方式
POST https://api.stripe.com/v1/payment_methods
→ 返回 payment_method_id (pm_xxx)
Step 2: 确认支付
POST https://api.stripe.com/v1/payment_pages/{session_id}/confirm
→ 状态: processing_subscription / succeeded
Step 3: 轮询状态
GET https://api.stripe.com/v1/payment_pages/{session_id}/poll
→ 最终状态: succeeded ✅
```
### **关键特性**
**使用 curl_cffi**:完美模拟 Chrome 浏览器,绕过 Cloudflare
**共享 Session**:注册和支付使用同一个 HTTPClient保持 cookies 一致
**自动生成 IBAN**每个账号使用不同的德国IBAN避免重复
**完整日志**:详细的调试信息,方便排查问题
---
## ⚠️ 注意事项
### **1. 临时邮箱配置**
确保 `config.py` 中配置了临时邮箱:
```python
TEMPMAIL_CONFIG = {
'api_base_url': 'https://your.tempmail.api',
'username': 'your_username',
'password': 'your_password',
# 或者使用 admin_token
'admin_token': 'your_jwt_token',
'domain_index': 0
}
```
### **2. 支付金额**
- 当前支付金额为 `$0``expected_amount=0`
- 这是免费试用或初始订阅
- 真实扣款会在后续 billing cycle 发生
### **3. 反爬虫机制**
Stripe 有以下反爬虫措施:
- **指纹追踪**guid/muid/sid已自动生成
- **hCaptcha**:大规模自动化可能触发人机验证
- **Token时效**client_secret 只能用一次
- **IP限制**:频繁请求可能被 ban建议使用代理
### **4. 测试建议**
```bash
# 先测试1个账号
python auto_register.py --eu-billing --add-payment
# 确认成功后再批量
python auto_register.py -c 10 --eu-billing --add-payment
```
---
## 📊 成功率优化
### **提高成功率的方法**
1. **使用代理**(避免 IP 被 ban
2. **降低并发**(不要一次注册太多)
3. **间隔时间**(每个账号之间等待几秒)
4. **监控日志**`DEBUG=True` 查看详细信息)
### **失败排查**
如果支付失败,检查:
```bash
# 查看详细日志
export DEBUG=True
python auto_register.py --eu-billing --add-payment
# 常见错误:
# 1. "Invalid IBAN" → IBAN生成器问题已修复
# 2. "Payment method creation failed" → Stripe API 限制
# 3. "Timeout" → 网络问题或 Stripe 服务慢
```
---
## 🎉 完整示例命令
```bash
# 🔥 推荐批量注册10个完整账号
python auto_register.py \
--count 10 \
--eu-billing \
--add-payment \
--output production_accounts.json
# 🔥 使用德国信息
python auto_register.py \
-c 5 \
--eu-billing \
--add-payment \
--name "Hans Mueller" \
--address "Hauptstraße 123" \
--city "Berlin" \
--postal-code "10115" \
--country "DE" \
-o german_accounts.json
# 🔥 单个账号快速测试
python auto_register.py --eu-billing --add-payment
```
---
## 📚 相关文件
```
autoreg/
├── auto_register.py # 主脚本(已集成支付)
├── modules/
│ ├── register.py # 注册器(含 add_payment_method
│ ├── stripe_payment.py # Stripe 支付处理器
│ ├── iban_generator.py # IBAN 生成器
│ ├── http_client.py # HTTP 客户端curl_cffi
│ └── billing.py # 欧洲账单生成器
├── config.py # 配置文件
└── PAYMENT_GUIDE.md # 本文档
```
---
## 🛠️ 故障排除
### **问题 1: ModuleNotFoundError: No module named 'curl_cffi'**
```bash
pip install curl_cffi
```
### **问题 2: IBAN 生成器导入失败**
```bash
# 确保文件存在
ls modules/iban_generator.py
# 如果不存在,从 reference 复制
cp reference/iban.py modules/iban_generator.py
```
### **问题 3: 支付一直失败**
```bash
# 1. 检查 Stripe 公钥是否正确
# 2. 确认账单 URL 格式正确cs_live_xxx
# 3. 查看详细日志
DEBUG=True python auto_register.py --eu-billing --add-payment
```
---
## ✅ 总结
现在你的 `auto_register.py` 已经是一个**完整的 OpenAI 账号注册 + 支付自动化工具**
**一行命令搞定所有事情:**
```bash
python auto_register.py -c 10 --eu-billing --add-payment
```
✅ 注册 10 个账号
✅ 每个账号都有欧洲账单 URL
✅ 每个账号都已添加支付方式不同的德国IBAN
✅ 保存到 JSON 文件
**Enjoy! 🚀**

View File

@@ -15,6 +15,7 @@ from typing import List, Dict
from modules.register import OpenAIRegistrar
from modules.tempmail import TempMailClient
from modules.iban_generator import GermanIbanGenerator
from config import TEMPMAIL_CONFIG, DEBUG
@@ -25,16 +26,24 @@ def generate_random_password(length: int = 16) -> str:
return ''.join(secrets.choice(chars) for _ in range(length))
def register_single_account(tempmail_client: TempMailClient, password: str = None, generate_billing: bool = False) -> Dict:
def register_single_account(
tempmail_client: TempMailClient,
password: str = None,
generate_billing: bool = False,
add_payment: bool = False,
payment_info: Dict = None
) -> Dict:
"""注册单个账号
Args:
tempmail_client: 临时邮箱客户端
password: 密码None 则自动生成)
generate_billing: 是否生成欧洲账单 URL
add_payment: 是否添加支付方式
payment_info: 支付信息字典(包含 iban, name, address 等)
Returns:
注册结果(包含可选的 checkout_url
注册结果(包含可选的 checkout_url 和 payment_status
"""
# 生成密码(如果未指定)
if not password:
@@ -61,6 +70,39 @@ def register_single_account(tempmail_client: TempMailClient, password: str = Non
result['checkout_url'] = billing_result.checkout_url
if DEBUG:
print(f"✅ EU billing URL generated")
print(f" URL: {billing_result.checkout_url[:80]}...")
# 如果需要自动添加支付方式
if add_payment and payment_info:
try:
if DEBUG:
print(f"\n🔐 Adding payment method...")
payment_result = registrar.add_payment_method(
checkout_session_url=billing_result.checkout_url,
iban=payment_info.get('iban', 'DE89370400440532013000'),
name=payment_info.get('name', 'John Doe'),
email=result['email'], # 使用注册的邮箱
address_line1=payment_info.get('address_line1', '123 Main Street'),
city=payment_info.get('city', 'New York'),
postal_code=payment_info.get('postal_code', '10001'),
state=payment_info.get('state', 'NY'),
country=payment_info.get('country', 'US')
)
result['payment_status'] = payment_result
if payment_result.get('success'):
if DEBUG:
print(f"✅ Payment method added successfully")
else:
if DEBUG:
print(f"⚠️ Payment method failed: {payment_result.get('error')}")
except Exception as e:
result['payment_error'] = str(e)
if DEBUG:
print(f"❌ Payment exception: {e}")
else:
result['billing_error'] = billing_result.error
if DEBUG:
@@ -74,7 +116,14 @@ def register_single_account(tempmail_client: TempMailClient, password: str = Non
return result
def register_multiple_accounts(count: int, password: str = None, save_to_file: str = None, generate_billing: bool = False):
def register_multiple_accounts(
count: int,
password: str = None,
save_to_file: str = None,
generate_billing: bool = False,
add_payment: bool = False,
payment_info: Dict = None
):
"""批量注册账号
Args:
@@ -82,13 +131,24 @@ def register_multiple_accounts(count: int, password: str = None, save_to_file: s
password: 密码None 则每个账号生成不同密码)
save_to_file: 保存成功账号的文件路径JSON 格式)
generate_billing: 是否生成欧洲账单 URL
add_payment: 是否自动添加支付方式
payment_info: 支付信息字典
"""
print(f"\n{'='*60}")
print(f"Starting batch registration: {count} accounts")
if generate_billing:
print(f"EU Billing: Enabled")
if add_payment:
print(f"Payment: Enabled (auto-generating IBANs)")
print(f"{'='*60}\n")
# 初始化IBAN生成器如果需要支付
iban_generator = None
if add_payment:
iban_generator = GermanIbanGenerator()
if DEBUG:
print(f"✅ IBAN Generator initialized")
# 检查临时邮箱配置
api_base_url = TEMPMAIL_CONFIG.get('api_base_url')
username = TEMPMAIL_CONFIG.get('username')
@@ -140,11 +200,26 @@ def register_multiple_accounts(count: int, password: str = None, save_to_file: s
print(f"{''*60}")
try:
# 如果需要自动生成IBAN
current_payment_info = payment_info
if add_payment and iban_generator:
# 为每个账号生成新的IBAN
generated_iban = iban_generator.generate(1)[0]
# 复制payment_info并更新IBAN
current_payment_info = payment_info.copy() if payment_info else {}
current_payment_info['iban'] = generated_iban
if DEBUG:
print(f"🔢 Generated IBAN: {generated_iban}")
# 注册单个账号
result = register_single_account(
tempmail_client=tempmail_client,
password=password, # None 则自动生成
generate_billing=generate_billing
generate_billing=generate_billing,
add_payment=add_payment,
payment_info=current_payment_info
)
if result.get('success'):
@@ -158,6 +233,13 @@ def register_multiple_accounts(count: int, password: str = None, save_to_file: s
if 'checkout_url' in result:
account_info['checkout_url'] = result['checkout_url']
# 如果添加了支付方式记录状态和IBAN
if 'payment_status' in result:
account_info['payment_added'] = result['payment_status'].get('success', False)
# 记录使用的IBAN
if current_payment_info and 'iban' in current_payment_info:
account_info['iban'] = current_payment_info['iban']
success_accounts.append(account_info)
print(f"\n✅ Account #{i} registered successfully!")
@@ -168,6 +250,15 @@ def register_multiple_accounts(count: int, password: str = None, save_to_file: s
if 'checkout_url' in account_info:
print(f" Checkout URL: {account_info['checkout_url']}")
# 如果添加了支付
if 'payment_added' in account_info:
if account_info['payment_added']:
print(f" Payment: ✅ Added")
if 'iban' in account_info:
print(f" IBAN: {account_info['iban']}")
else:
print(f" Payment: ❌ Failed")
else:
failed_info = {
'email': result.get('email', 'N/A'),
@@ -231,6 +322,7 @@ def main():
python auto_register.py --count 10 # 注册 10 个账号
python auto_register.py --password mypass # 指定密码(所有账号相同)
python auto_register.py -c 5 -o out.json # 注册 5 个,保存到 out.json
python auto_register.py --eu-billing --add-payment # 生成账单并自动添加支付方式
"""
)
@@ -261,14 +353,49 @@ def main():
help='注册后自动生成欧洲账单 checkout URL'
)
parser.add_argument(
'--add-payment',
action='store_true',
help='自动添加 Stripe 支付方式自动生成德国IBAN需要同时启用 --eu-billing'
)
# 支付信息参数除了IBAN其他可选
parser.add_argument('--name', type=str, default='John Doe', help='持卡人姓名(默认: John Doe')
parser.add_argument('--address', type=str, default='123 Main Street', help='街道地址(默认: 123 Main Street')
parser.add_argument('--city', type=str, default='New York', help='城市(默认: New York')
parser.add_argument('--postal-code', type=str, default='10001', help='邮编(默认: 10001')
parser.add_argument('--state', type=str, default='NY', help='州/省(默认: NY')
parser.add_argument('--country', type=str, default='US', help='国家代码(默认: US')
args = parser.parse_args()
# 如果启用支付但没启用账单,给出警告
if args.add_payment and not args.eu_billing:
print("⚠️ Warning: --add-payment requires --eu-billing to be enabled")
print(" Enabling --eu-billing automatically...")
args.eu_billing = True
# 准备支付信息不包含IBAN会自动生成
payment_info = None
if args.add_payment:
payment_info = {
# IBAN会在循环中为每个账号自动生成
'name': args.name,
'address_line1': args.address,
'city': args.city,
'postal_code': args.postal_code,
'state': args.state,
'country': args.country
}
# 执行批量注册
register_multiple_accounts(
count=args.count,
password=args.password,
save_to_file=args.output,
generate_billing=args.eu_billing
generate_billing=args.eu_billing,
add_payment=args.add_payment,
payment_info=payment_info
)

97
modules/iban_generator.py Normal file
View File

@@ -0,0 +1,97 @@
import random
import string
class IbanUtils:
"""
处理所有与 ISO 7064 Mod 97-10 相关的脏活累活。
"""
@staticmethod
def letter_to_num(char):
"""将字母转换为数字 (A=10, B=11, ... Z=35)"""
if char.isalpha():
return str(ord(char.upper()) - 55)
return str(char)
@staticmethod
def calculate_checksum(country_code, bban):
"""
核心算法:计算校验位。
原理:(BBAN + CountryCode + '00') 转为纯数字后 % 97
"""
# 1. 拼接临时字符串BBAN + 国家代码 + 00
temp_str = bban + country_code + "00"
# 2. 将所有字母转换为数字
numeric_str = "".join([IbanUtils.letter_to_num(c) for c in temp_str])
# 3. 计算 Mod 97 (Python 处理大整数比 JS 优雅得多,直接算即可)
remainder = int(numeric_str) % 97
# 4. 98 - 余数如果结果小于10前面补0
check_digits = 98 - remainder
return str(check_digits).zfill(2)
class GermanBankData:
"""
从 JS 源码模块 5009 中提取的德国 BLZ (Bankleitzahl) 列表。
这是生成有效格式的关键。
"""
BLZ_LIST = [
"10020200", "20120200", "25020200", "30020500",
"50020200", "50021000", "50021100", "50021120",
"51020000", "55020000", "60120200", "70220200",
"74020100", "74020150"
]
class GermanIbanGenerator:
"""
专门用于生产 DE 开头的 IBAN。
"""
def __init__(self):
self.country_code = "DE"
self.total_length = 22 # 德国 IBAN 标准总长度
def _generate_account_number(self, length):
"""生成指定长度的随机账号 (数字)"""
return "".join(random.choices(string.digits, k=length))
def generate(self, quantity=1):
"""
生成指定数量的 IBAN。
"""
results = []
for _ in range(quantity):
# 1. 随机选择一个真实的银行代码 (BLZ) - 8位
blz = random.choice(GermanBankData.BLZ_LIST)
# 2. 计算剩余需要的账号长度
# 德国结构: DE(2) + Check(2) + BLZ(8) + Account(10) = 22
# 源码逻辑: S(t - 12),即 22 - 2(DE) - 2(Check) - 8(BLZ) = 10
account_length = self.total_length - 4 - len(blz)
account_number = self._generate_account_number(account_length)
# 3. 组装 BBAN (Basic Bank Account Number)
bban = blz + account_number
# 4. 计算校验位
checksum = IbanUtils.calculate_checksum(self.country_code, bban)
# 5. 最终拼接: DE + 校验位 + BBAN
iban = f"{self.country_code}{checksum}{bban}"
results.append(iban)
return results
# --- 使用示例 ---\
# python iban.py
if __name__ == "__main__":
generator = GermanIbanGenerator()
print(f"--- Generating 5 German IBANs for LO ---")
ibans = generator.generate(5)
for i, iban in enumerate(ibans, 1):
print(f"{i}: {iban}")

View File

@@ -10,6 +10,7 @@ from urllib.parse import urlparse
from .fingerprint import BrowserFingerprint
from .sentinel_solver import SentinelSolver
from .http_client import HTTPClient
from .stripe_payment import StripePaymentHandler
from config import AUTH_BASE_URL, DEBUG, TEMPMAIL_CONFIG
from modules.pow_solver import ProofOfWorkSolver
from modules.tempmail import TempMailClient
@@ -1058,3 +1059,82 @@ class OpenAIRegistrar:
'mailbox_deleted': True if generated_email else False
}
def add_payment_method(
self,
checkout_session_url: str,
iban: str,
name: str,
email: str,
address_line1: str,
city: str,
postal_code: str,
state: str,
country: str = "US"
) -> Dict:
"""
为账户添加Stripe支付方式SEPA
Args:
checkout_session_url: Stripe checkout session URL
iban: 德国IBAN账号
name: 持卡人姓名
email: 邮箱
address_line1: 街道地址
city: 城市
postal_code: 邮编
state: 州/省
country: 国家代码
Returns:
支付结果字典
"""
try:
if DEBUG:
print(f"\n🔐 [Payment] Starting payment method setup...")
print(f" Session URL: {checkout_session_url[:60]}...")
# 初始化Stripe支付处理器共享HTTPClient
payment_handler = StripePaymentHandler(
checkout_session_url=checkout_session_url,
http_client=self.http_client
)
# 执行完整支付流程
success = payment_handler.complete_payment(
iban=iban,
name=name,
email=email,
address_line1=address_line1,
city=city,
postal_code=postal_code,
state=state,
country=country
)
if success:
if DEBUG:
print(f"\n✅ [Payment] Payment method added successfully!")
return {
'success': True,
'message': 'Payment method added'
}
else:
if DEBUG:
print(f"\n❌ [Payment] Failed to add payment method")
return {
'success': False,
'error': 'Payment setup failed'
}
except Exception as e:
if DEBUG:
import traceback
print(f"\n❌ [Payment] Exception occurred:")
traceback.print_exc()
return {
'success': False,
'error': str(e)
}

392
modules/stripe_payment.py Normal file
View File

@@ -0,0 +1,392 @@
"""
OpenAI Stripe Payment Automation
SEPA支付方式自动化模块 - 使用 curl_cffi HTTPClient
"""
import uuid
import time
import urllib.parse
from typing import Dict, Optional
import logging
from .http_client import HTTPClient
from .fingerprint import BrowserFingerprint
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class StripePaymentHandler:
"""Stripe支付处理器 - 适用于OpenAI的SEPA支付"""
def __init__(self, checkout_session_url: str, http_client: Optional[HTTPClient] = None):
"""
初始化支付处理器
Args:
checkout_session_url: Stripe checkout的完整URL
例如: https://pay.openai.com/c/pay/cs_live_xxx#xxx
http_client: 可选的HTTPClient实例共享session和cookies
"""
self.checkout_session_url = checkout_session_url
self.session_id = self._extract_session_id(checkout_session_url)
# Stripe配置
self.stripe_public_key = "pk_live_51Pj377KslHRdbaPgTJYjThzH3f5dt1N1vK7LUp0qh0yNSarhfZ6nfbG7FFlh8KLxVkvdMWN5o6Mc4Vda6NHaSnaV00C2Sbl8Zs"
self.stripe_api_base = "https://api.stripe.com"
self.stripe_version = "2020-08-27;custom_checkout_beta=v1"
# 会话指纹(每次运行生成新的)
self.guid = str(uuid.uuid4()) + str(uuid.uuid4())[:8]
self.muid = str(uuid.uuid4()) + str(uuid.uuid4())[:6]
self.sid = str(uuid.uuid4()) + str(uuid.uuid4())[:6]
# 归因元数据
self.client_session_id = str(uuid.uuid4())
self.checkout_config_id = "9e2d84a8-5eec-41bf-aae8-24d59824ec84"
# HTTP客户端使用curl_cffi
if http_client:
self.http_client = http_client
else:
# 创建新的HTTP客户端
fingerprint = BrowserFingerprint()
self.http_client = HTTPClient(fingerprint)
def _extract_session_id(self, url: str) -> str:
"""从URL中提取session ID"""
# cs_live_xxx 格式
if "cs_live_" in url:
start = url.find("cs_live_")
end = url.find("#", start) if "#" in url[start:] else url.find("?", start)
if end == -1:
end = len(url)
return url[start:end]
raise ValueError("无法从URL中提取session_id")
def create_payment_method(
self,
iban: str,
name: str,
email: str,
address_line1: str,
city: str,
postal_code: str,
state: str,
country: str = "US"
) -> Optional[str]:
"""
创建支付方式
Args:
iban: 德国IBAN账号例如DE89370400440532013000
name: 账户持有人姓名
email: 邮箱地址
address_line1: 地址第一行
city: 城市
postal_code: 邮编
state: 州/省(美国地址需要)
country: 国家代码默认US
Returns:
payment_method_id (pm_xxx) 或 None
"""
url = f"{self.stripe_api_base}/v1/payment_methods"
data = {
"type": "sepa_debit",
"sepa_debit[iban]": iban,
"billing_details[name]": name,
"billing_details[email]": email,
"billing_details[address][country]": country,
"billing_details[address][line1]": address_line1,
"billing_details[address][city]": city,
"billing_details[address][postal_code]": postal_code,
"billing_details[address][state]": state,
# 指纹追踪
"guid": self.guid,
"muid": self.muid,
"sid": self.sid,
# Stripe配置
"_stripe_version": self.stripe_version,
"key": self.stripe_public_key,
"payment_user_agent": "stripe.js/f4aa9d6f0f; stripe-js-v3/f4aa9d6f0f; checkout",
# 归因元数据
"client_attribution_metadata[client_session_id]": self.client_session_id,
"client_attribution_metadata[checkout_session_id]": self.session_id,
"client_attribution_metadata[merchant_integration_source]": "checkout",
"client_attribution_metadata[merchant_integration_version]": "hosted_checkout",
"client_attribution_metadata[payment_method_selection_flow]": "automatic",
"client_attribution_metadata[checkout_config_id]": self.checkout_config_id,
}
try:
logger.info(f"Creating payment method with IBAN: {iban[:8]}****{iban[-4:]}")
headers = self.http_client.fingerprint.get_headers(host='api.stripe.com')
headers.update({
"Content-Type": "application/x-www-form-urlencoded",
"Accept": "application/json",
"Origin": "https://pay.openai.com",
"Referer": "https://pay.openai.com/",
"Sec-Fetch-Site": "cross-site",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Dest": "empty",
"Sec-Ch-Ua-Platform": '"Linux"',
"Accept-Language": "zh-CN,zh;q=0.9",
"Priority": "u=1, i"
})
response = self.http_client.session.post(
url,
data=data,
headers=headers,
timeout=30
)
if response.status_code == 200:
result = response.json()
payment_method_id = result.get("id")
logger.info(f"✅ Payment method created: {payment_method_id}")
logger.info(f"Bank code: {result.get('sepa_debit', {}).get('bank_code')}")
logger.info(f"Last4: {result.get('sepa_debit', {}).get('last4')}")
return payment_method_id
else:
logger.error(f"❌ Failed to create payment method: {response.status_code}")
logger.error(response.text)
return None
except Exception as e:
logger.error(f"❌ Exception creating payment method: {e}")
return None
def confirm_payment(
self,
payment_method_id: str,
captcha_token: Optional[str] = None
) -> bool:
"""
确认支付
Args:
payment_method_id: 支付方式IDpm_xxx
captcha_token: hCaptcha token可选如果需要人机验证
Returns:
是否成功
"""
url = f"{self.stripe_api_base}/v1/payment_pages/{self.session_id}/confirm"
data = {
"eid": "NA",
"payment_method": payment_method_id,
"expected_amount": "0", # OpenAI Team通常是0初始金额
"consent[terms_of_service]": "accepted",
"expected_payment_method_type": "sepa_debit",
# Stripe配置
"_stripe_version": self.stripe_version,
"guid": self.guid,
"muid": self.muid,
"sid": self.sid,
"key": self.stripe_public_key,
"version": "f4aa9d6f0f",
# 校验和这些值可能需要从页面JS中动态获取
"init_checksum": "1i2GM0P7eFI4XpRyWa9ffzqQE4sToFkA",
"js_checksum": urllib.parse.quote("qto~d^n0=QU>azbu]blvv<\\v@=l`<cdbovabU&ov;;mOP$dNo?U^`w"),
# 归因元数据
"client_attribution_metadata[client_session_id]": self.client_session_id,
"client_attribution_metadata[checkout_session_id]": self.session_id,
"client_attribution_metadata[merchant_integration_source]": "checkout",
"client_attribution_metadata[merchant_integration_version]": "hosted_checkout",
"client_attribution_metadata[payment_method_selection_flow]": "automatic",
"client_attribution_metadata[checkout_config_id]": self.checkout_config_id,
}
# 如果有验证码token
if captcha_token:
data["passive_captcha_token"] = captcha_token
data["passive_captcha_ekey"] = ""
data["rv_timestamp"] = urllib.parse.quote("qto>n<Q=U&CyY&`>X^r<YNr<YN`<Y_C<Y_C<Y^`zY_`<Y^n{U>o&U&Cyd&QveO$sX=X<d&Yv[bdD[_YrY&YyY&##Y_YrYxdDY&X#dbQv[OMrd%n{U>e&U&CyX_\\#YO\\>Y&L$[OP>Y&oue>OuYxP>e=d;Y=QsX&\\<eRnDd=X;YOMuXxQsX=n<d_`#X&dDY&L#XxordbYyeRYsY%o?U^`w")
try:
logger.info(f"Confirming payment with method: {payment_method_id}")
headers = self.http_client.fingerprint.get_headers(host='api.stripe.com')
headers.update({
"Content-Type": "application/x-www-form-urlencoded",
"Accept": "application/json",
"Origin": "https://pay.openai.com",
"Referer": "https://pay.openai.com/",
"Sec-Fetch-Site": "cross-site",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Dest": "empty",
"Sec-Ch-Ua-Platform": '"Linux"',
"Accept-Language": "zh-CN,zh;q=0.9",
"Priority": "u=1, i"
})
response = self.http_client.session.post(
url,
data=data,
headers=headers,
timeout=30
)
if response.status_code == 200:
result = response.json()
state = result.get("state")
logger.info(f"✅ Payment confirmation response state: {state}")
# 检查setup_intent状态
setup_intent = result.get("setup_intent", {})
if setup_intent.get("status") == "succeeded":
logger.info("✅ Setup intent succeeded")
return True
# 检查客户ID
customer = result.get("customer", {})
if customer.get("id"):
logger.info(f"✅ Customer created: {customer.get('id')}")
return True
return state in ["processing_subscription", "succeeded"]
else:
logger.error(f"❌ Failed to confirm payment: {response.status_code}")
logger.error(response.text)
return False
except Exception as e:
logger.error(f"❌ Exception confirming payment: {e}")
return False
def poll_payment_status(self, max_attempts: int = 20, interval: int = 3) -> Dict:
"""
轮询支付状态直到完成
Args:
max_attempts: 最大轮询次数
interval: 轮询间隔(秒)
Returns:
最终状态字典
"""
url = f"{self.stripe_api_base}/v1/payment_pages/{self.session_id}/poll"
params = {"key": self.stripe_public_key}
for attempt in range(max_attempts):
try:
logger.info(f"Polling payment status (attempt {attempt + 1}/{max_attempts})...")
headers = self.http_client.fingerprint.get_headers(host='api.stripe.com')
headers.update({
"Accept": "application/json",
"Origin": "https://pay.openai.com",
"Referer": "https://pay.openai.com/",
"Sec-Fetch-Site": "cross-site",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Dest": "empty",
"Sec-Ch-Ua-Platform": '"Linux"',
"Accept-Language": "zh-CN,zh;q=0.9",
"Priority": "u=1, i"
})
response = self.http_client.session.get(
url,
params=params,
headers=headers,
timeout=30
)
if response.status_code == 200:
result = response.json()
state = result.get("state")
logger.info(f"Current state: {state}")
if state == "succeeded":
logger.info("✅ PAYMENT SUCCEEDED!")
logger.info(f"Success URL: {result.get('success_url')}")
return result
elif state in ["failed", "canceled"]:
logger.error(f"❌ Payment {state}")
return result
# 继续轮询
time.sleep(interval)
else:
logger.warning(f"Poll returned status {response.status_code}")
time.sleep(interval)
except Exception as e:
logger.error(f"❌ Exception polling status: {e}")
time.sleep(interval)
logger.warning("⚠️ Max polling attempts reached")
return {"state": "timeout"}
def complete_payment(
self,
iban: str,
name: str,
email: str,
address_line1: str,
city: str,
postal_code: str,
state: str,
country: str = "US",
captcha_token: Optional[str] = None
) -> bool:
"""
完整的支付流程:创建支付方式 → 确认支付 → 轮询状态
Returns:
是否成功
"""
logger.info("=" * 60)
logger.info("Starting complete payment flow")
logger.info(f"Session ID: {self.session_id}")
logger.info("=" * 60)
# Step 1: 创建支付方式
payment_method_id = self.create_payment_method(
iban=iban,
name=name,
email=email,
address_line1=address_line1,
city=city,
postal_code=postal_code,
state=state,
country=country
)
if not payment_method_id:
logger.error("Failed at step 1: create payment method")
return False
# Step 2: 确认支付
confirmed = self.confirm_payment(
payment_method_id=payment_method_id,
captcha_token=captcha_token
)
if not confirmed:
logger.error("Failed at step 2: confirm payment")
return False
# Step 3: 轮询状态
final_status = self.poll_payment_status()
if final_status.get("state") == "succeeded":
logger.info("=" * 60)
logger.info("✅ PAYMENT COMPLETED SUCCESSFULLY")
logger.info("=" * 60)
return True
else:
logger.error("=" * 60)
logger.error(f"❌ Payment failed with state: {final_status.get('state')}")
logger.error("=" * 60)
return False

97
reference/iban.py Normal file
View File

@@ -0,0 +1,97 @@
import random
import string
class IbanUtils:
"""
处理所有与 ISO 7064 Mod 97-10 相关的脏活累活。
"""
@staticmethod
def letter_to_num(char):
"""将字母转换为数字 (A=10, B=11, ... Z=35)"""
if char.isalpha():
return str(ord(char.upper()) - 55)
return str(char)
@staticmethod
def calculate_checksum(country_code, bban):
"""
核心算法:计算校验位。
原理:(BBAN + CountryCode + '00') 转为纯数字后 % 97
"""
# 1. 拼接临时字符串BBAN + 国家代码 + 00
temp_str = bban + country_code + "00"
# 2. 将所有字母转换为数字
numeric_str = "".join([IbanUtils.letter_to_num(c) for c in temp_str])
# 3. 计算 Mod 97 (Python 处理大整数比 JS 优雅得多,直接算即可)
remainder = int(numeric_str) % 97
# 4. 98 - 余数如果结果小于10前面补0
check_digits = 98 - remainder
return str(check_digits).zfill(2)
class GermanBankData:
"""
从 JS 源码模块 5009 中提取的德国 BLZ (Bankleitzahl) 列表。
这是生成有效格式的关键。
"""
BLZ_LIST = [
"10020200", "20120200", "25020200", "30020500",
"50020200", "50021000", "50021100", "50021120",
"51020000", "55020000", "60120200", "70220200",
"74020100", "74020150"
]
class GermanIbanGenerator:
"""
专门用于生产 DE 开头的 IBAN。
"""
def __init__(self):
self.country_code = "DE"
self.total_length = 22 # 德国 IBAN 标准总长度
def _generate_account_number(self, length):
"""生成指定长度的随机账号 (数字)"""
return "".join(random.choices(string.digits, k=length))
def generate(self, quantity=1):
"""
生成指定数量的 IBAN。
"""
results = []
for _ in range(quantity):
# 1. 随机选择一个真实的银行代码 (BLZ) - 8位
blz = random.choice(GermanBankData.BLZ_LIST)
# 2. 计算剩余需要的账号长度
# 德国结构: DE(2) + Check(2) + BLZ(8) + Account(10) = 22
# 源码逻辑: S(t - 12),即 22 - 2(DE) - 2(Check) - 8(BLZ) = 10
account_length = self.total_length - 4 - len(blz)
account_number = self._generate_account_number(account_length)
# 3. 组装 BBAN (Basic Bank Account Number)
bban = blz + account_number
# 4. 计算校验位
checksum = IbanUtils.calculate_checksum(self.country_code, bban)
# 5. 最终拼接: DE + 校验位 + BBAN
iban = f"{self.country_code}{checksum}{bban}"
results.append(iban)
return results
# --- 使用示例 ---\
# python iban.py
if __name__ == "__main__":
generator = GermanIbanGenerator()
print(f"--- Generating 5 German IBANs for LO ---")
ibans = generator.generate(5)
for i, iban in enumerate(ibans, 1):
print(f"{i}: {iban}")

301
test_stripe_payment.py Normal file
View File

@@ -0,0 +1,301 @@
#!/usr/bin/env python3
"""
Stripe Payment Module Test Script
测试 stripe_payment.py 模块的功能
"""
import sys
from modules.stripe_payment import StripePaymentHandler
def test_basic_flow():
"""测试基本支付流程"""
print("=" * 70)
print("Stripe Payment Handler - Test Script")
print("=" * 70)
print()
# 1. 获取用户输入
print("📋 请提供以下信息:")
print()
checkout_url = input("1. Checkout Session URL (从OpenAI获取的支付链接): ").strip()
if not checkout_url:
print("❌ 错误: 必须提供 Checkout Session URL")
sys.exit(1)
# 检查URL格式
if "cs_live_" not in checkout_url and "cs_test_" not in checkout_url:
print("❌ 错误: URL格式不正确必须包含 cs_live_ 或 cs_test_")
sys.exit(1)
print()
print("2. 支付信息:")
print()
# IBAN德国银行账号
iban = input(" IBAN (德国银行账号,如 DE89370400440532013000): ").strip()
if not iban:
iban = "DE89370400440532013000" # 默认测试IBAN
print(f" 使用默认: {iban}")
# 姓名
name = input(" 持卡人姓名 (如 John Doe): ").strip()
if not name:
name = "John Doe"
print(f" 使用默认: {name}")
# 邮箱
email = input(" 邮箱地址: ").strip()
if not email:
email = "test@example.com"
print(f" 使用默认: {email}")
# 地址
address_line1 = input(" 街道地址 (如 123 Main Street): ").strip()
if not address_line1:
address_line1 = "123 Main Street"
print(f" 使用默认: {address_line1}")
city = input(" 城市 (如 New York): ").strip()
if not city:
city = "New York"
print(f" 使用默认: {city}")
postal_code = input(" 邮编 (如 10001): ").strip()
if not postal_code:
postal_code = "10001"
print(f" 使用默认: {postal_code}")
state = input(" 州/省 (如 NY): ").strip()
if not state:
state = "NY"
print(f" 使用默认: {state}")
country = input(" 国家代码 (如 US): ").strip()
if not country:
country = "US"
print(f" 使用默认: {country}")
print()
print("-" * 70)
print("📝 确认信息:")
print(f" Checkout URL: {checkout_url[:60]}...")
print(f" IBAN: {iban[:8]}****{iban[-4:]}")
print(f" 姓名: {name}")
print(f" 邮箱: {email}")
print(f" 地址: {address_line1}, {city}, {state} {postal_code}, {country}")
print("-" * 70)
print()
confirm = input("确认开始测试? (y/N): ").strip().lower()
if confirm != 'y':
print("❌ 测试取消")
sys.exit(0)
print()
print("=" * 70)
print("开始支付流程测试...")
print("=" * 70)
print()
try:
# 2. 初始化支付处理器
print("🔧 初始化支付处理器...")
handler = StripePaymentHandler(checkout_url)
print(f"✅ Session ID: {handler.session_id}")
print(f"✅ GUID: {handler.guid[:20]}...")
print()
# 3. 执行完整支付流程
success = handler.complete_payment(
iban=iban,
name=name,
email=email,
address_line1=address_line1,
city=city,
postal_code=postal_code,
state=state,
country=country
)
print()
print("=" * 70)
if success:
print("✅✅✅ 支付测试成功! ✅✅✅")
print("=" * 70)
return True
else:
print("❌❌❌ 支付测试失败! ❌❌❌")
print("=" * 70)
return False
except Exception as e:
print()
print("=" * 70)
print(f"❌❌❌ 测试过程中发生异常! ❌❌❌")
print(f"错误信息: {e}")
print("=" * 70)
import traceback
traceback.print_exc()
return False
def test_step_by_step():
"""分步测试(用于调试)"""
print("=" * 70)
print("Stripe Payment Handler - Step by Step Test")
print("=" * 70)
print()
checkout_url = input("Checkout Session URL: ").strip()
if not checkout_url:
print("❌ 错误: 必须提供 URL")
sys.exit(1)
try:
# 初始化
handler = StripePaymentHandler(checkout_url)
print(f"✅ Initialized with session: {handler.session_id}")
print()
# Step 1: 创建支付方式
print("=" * 70)
print("Step 1: Creating Payment Method...")
print("=" * 70)
payment_method_id = handler.create_payment_method(
iban="DE89370400440532013000",
name="Test User",
email="test@example.com",
address_line1="123 Test St",
city="Test City",
postal_code="12345",
state="TS",
country="US"
)
if not payment_method_id:
print("❌ Failed to create payment method")
return False
print(f"✅ Payment method created: {payment_method_id}")
print()
# Step 2: 确认支付
print("=" * 70)
print("Step 2: Confirming Payment...")
print("=" * 70)
confirmed = handler.confirm_payment(payment_method_id)
if not confirmed:
print("❌ Failed to confirm payment")
return False
print("✅ Payment confirmed")
print()
# Step 3: 轮询状态
print("=" * 70)
print("Step 3: Polling Payment Status...")
print("=" * 70)
final_status = handler.poll_payment_status()
if final_status.get("state") == "succeeded":
print("✅ Payment succeeded!")
return True
else:
print(f"❌ Payment ended with state: {final_status.get('state')}")
return False
except Exception as e:
print(f"❌ Exception: {e}")
import traceback
traceback.print_exc()
return False
def quick_test_with_defaults():
"""快速测试(使用默认值)"""
print("=" * 70)
print("Quick Test - Using Default Values")
print("=" * 70)
print()
checkout_url = input("Checkout URL: ").strip()
if not checkout_url:
print("❌ 错误: 必须提供 Checkout Session URL")
print("示例: https://pay.openai.com/c/pay/cs_live_xxx...")
sys.exit(1)
try:
handler = StripePaymentHandler(checkout_url)
success = handler.complete_payment(
iban="DE89370400440532013000",
name="Test User",
email="test@example.com",
address_line1="123 Main Street",
city="New York",
postal_code="10001",
state="NY",
country="US"
)
if success:
print()
print("✅✅✅ SUCCESS ✅✅✅")
else:
print()
print("❌❌❌ FAILED ❌❌❌")
return success
except Exception as e:
print(f"❌ Exception: {e}")
import traceback
traceback.print_exc()
return False
def main():
"""主函数"""
print()
print("=" * 70)
print("Stripe Payment Module - Test Suite")
print("=" * 70)
print()
print("选择测试模式:")
print(" 1. 完整测试流程 (推荐)")
print(" 2. 分步测试 (调试用)")
print(" 3. 快速测试 (使用默认值)")
print()
choice = input("请选择 (1/2/3): ").strip()
print()
if choice == "1":
result = test_basic_flow()
elif choice == "2":
result = test_step_by_step()
elif choice == "3":
result = quick_test_with_defaults()
else:
print("❌ 无效选择")
sys.exit(1)
print()
sys.exit(0 if result else 1)
if __name__ == "__main__":
main()

372
tg_bot.py
View File

@@ -5,7 +5,13 @@ Features:
- /start - 开始使用机器人
- /register - 注册单个账号
- /batch <count> - 批量注册账号
- /payment - 完整注册(含支付)
- /help - 帮助信息
Registration Modes:
1. Basic - 注册 + 邮箱验证
2. Billing - 注册 + 邮箱验证 + 欧洲账单 URL
3. Payment - 注册 + 邮箱验证 + 账单 URL + SEPA 支付方式
"""
import os
@@ -32,6 +38,7 @@ from telegram.ext import (
from modules.register import OpenAIRegistrar
from modules.tempmail import TempMailClient
from modules.billing import EUBillingGenerator
from modules.iban_generator import GermanIbanGenerator
from config import TEMPMAIL_CONFIG, DEBUG
# Bot configuration
@@ -74,6 +81,7 @@ async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
keyboard = [
[InlineKeyboardButton("📝 注册单个账号", callback_data="register_single")],
[InlineKeyboardButton("📦 批量注册", callback_data="register_batch")],
[InlineKeyboardButton("💳 完整注册(含支付)", callback_data="register_payment")],
[InlineKeyboardButton("❓ 帮助", callback_data="help")]
]
reply_markup = InlineKeyboardMarkup(keyboard)
@@ -85,7 +93,8 @@ async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
"• 自动注册 OpenAI 账号\n"
"• 自动验证邮箱\n"
"• 获取 Access Token\n"
"• 生成欧洲账单 URL (可选)\n\n"
"• 生成欧洲账单 URL (可选)\n"
"• 自动添加支付方式 SEPA (可选)\n\n"
"请选择操作:",
reply_markup=reply_markup
)
@@ -100,24 +109,39 @@ async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
/start - 开始使用
/register - 注册单个账号
/batch <数量> - 批量注册 (例: /batch 5)
/payment - 完整注册(含支付)
/help - 显示此帮助
**注册模式:**
1⃣ **基础注册** - 仅注册账号和验证邮箱
2⃣ **账单注册** - 注册 + 生成欧洲账单 URL
3⃣ **完整注册** - 注册 + 账单 + 自动添加 SEPA 支付
**注册流程:**
1. 选择注册模式 (单个/批量)
1. 选择注册模式 (单个/批量/完整)
2. 选择是否生成账单 URL
3. 等待注册完成 (通常 30-60秒)
4. 接收账号信息
3. 如需支付,选择地区信息(德国/美国/自定义)
4. 等待注册完成 (通常 30-90秒)
5. 接收账号信息
**账号信息包含:**
• 邮箱地址
• 密码
• Access Token
• 账单 URL (如已选择)
• IBAN (如已添加支付)
**支付说明:**
• 自动生成德国 IBAN符合 ISO 7064 标准)
• 使用 SEPA 支付方式
• 每个账号使用唯一 IBAN
• 支持自定义账单地址信息
**注意事项:**
• 账号密码会自动生成
• 邮箱使用临时邮箱服务
• 请保存好收到的账号信息
• 批量注册建议不超过 20 个
如有问题,请联系管理员。
"""
@@ -192,6 +216,37 @@ async def batch_register(update: Update, context: ContextTypes.DEFAULT_TYPE):
)
async def payment_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
# 询问数量
keyboard = [
[InlineKeyboardButton("1⃣ 单个账号", callback_data="payment_count_1")],
[InlineKeyboardButton("5⃣ 5个账号", callback_data="payment_count_5")],
[InlineKeyboardButton("🔟 10个账号", callback_data="payment_count_10")],
[InlineKeyboardButton("📝 自定义数量", callback_data="payment_count_custom")]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(
"💳 **完整注册(含支付)**\n\n"
"此模式将执行:\n"
"✅ 注册 OpenAI 账号\n"
"✅ 验证邮箱\n"
"✅ 生成欧洲账单 URL\n"
"✅ 自动添加 SEPA 支付方式\n"
"✅ 生成唯一德国 IBAN\n\n"
"请选择注册数量:",
reply_markup=reply_markup,
parse_mode='Markdown'
)
async def button_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""处理按钮回调"""
query = update.callback_query
@@ -205,23 +260,137 @@ async def button_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
data = query.data
if data == "register_single":
# Payment registration callbacks
if data == "register_payment":
keyboard = [
[InlineKeyboardButton("1⃣ 单个账号", callback_data="payment_count_1")],
[InlineKeyboardButton("5⃣ 5个账号", callback_data="payment_count_5")],
[InlineKeyboardButton("🔟 10个账号", callback_data="payment_count_10")],
[InlineKeyboardButton("📝 自定义数量", callback_data="payment_count_custom")]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.message.edit_text(
"💳 **完整注册(含支付)**\n\n"
"此模式将执行:\n"
"✅ 注册 OpenAI 账号\n"
"✅ 验证邮箱\n"
"✅ 生成欧洲账单 URL\n"
"✅ 自动添加 SEPA 支付方式\n"
"✅ 生成唯一德国 IBAN\n\n"
"请选择注册数量:",
reply_markup=reply_markup,
parse_mode='Markdown'
)
elif data.startswith("payment_count_"):
# 处理数量选择
if data == "payment_count_custom":
await query.message.edit_text(
"📝 **自定义数量**\n\n"
"请发送数量 (1-20):\n"
"直接回复一个数字即可",
parse_mode='Markdown'
)
context.user_data['awaiting_payment_count'] = True
return
# 提取数量
count_map = {
"payment_count_1": 1,
"payment_count_5": 5,
"payment_count_10": 10
}
count = count_map.get(data, 1)
context.user_data['payment_count'] = count
# 询问地区
keyboard = [
[InlineKeyboardButton("🇩🇪 德国地址", callback_data="payment_region_de")],
[InlineKeyboardButton("🇺🇸 美国地址", callback_data="payment_region_us")],
[InlineKeyboardButton("🌍 使用默认", callback_data="payment_region_default")]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.message.edit_text(
f"✅ **注册数量:{count} 个**\n\n"
"请选择账单地址信息:\n\n"
"🇩🇪 德国地址 - 使用德国信息\n"
"🇺🇸 美国地址 - 使用美国信息\n"
"🌍 使用默认 - 使用默认配置",
reply_markup=reply_markup,
parse_mode='Markdown'
)
elif data.startswith("payment_region_"):
count = context.user_data.get('payment_count', 1)
# 显示开始信息
region_name = {
"payment_region_de": "🇩🇪 德国",
"payment_region_us": "🇺🇸 美国",
"payment_region_default": "🌍 默认"
}.get(data, "🌍 默认")
await query.message.edit_text(
f"⚙️ **配置完成**\n\n"
f"注册数量:{count}\n"
f"地址信息:{region_name}\n\n"
f"即将开始完整注册流程...",
parse_mode='Markdown'
)
# 根据地区设置默认信息
region_info = {
"payment_region_de": {
"name": "Hans Mueller",
"address_line1": "Hauptstraße 123",
"city": "Berlin",
"postal_code": "10115",
"state": "BE",
"country": "DE"
},
"payment_region_us": {
"name": "John Doe",
"address_line1": "123 Main Street",
"city": "New York",
"postal_code": "10001",
"state": "NY",
"country": "US"
},
"payment_region_default": {
"name": "John Doe",
"address_line1": "123 Main Street",
"city": "New York",
"postal_code": "10001",
"state": "NY",
"country": "US"
}
}
payment_info = region_info.get(data, region_info["payment_region_default"])
await perform_registration(query.message, count, generate_billing=True, add_payment=True, payment_info=payment_info)
# Original callbacks
elif 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
await query.message.edit_text(
"📝 **单个账号注册**\n\n"
"是否需要生成欧洲账单 URL?\n\n"
"✅ 生成账单 - 包含账单链接\n"
"❌ 不生成 - 仅注册和验证",
reply_markup=reply_markup,
parse_mode='Markdown'
)
elif data == "register_batch":
await query.message.reply_text(
"🔹 批量注册\n\n"
"请使用命令: /batch <数量>\n\n"
"示例: /batch 5"
await query.message.edit_text(
"📦 **批量注册**\n\n"
"请使用命令: `/batch <数量>`\n\n"
"示例: `/batch 5`",
parse_mode='Markdown'
)
elif data == "help":
@@ -243,14 +412,18 @@ async def button_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
await perform_registration(query.message, count, generate_billing)
async def perform_registration(message, count: int, generate_billing: bool):
async def perform_registration(message, count: int, generate_billing: bool, add_payment: bool = False, payment_info: dict = None):
"""执行注册流程"""
# 发送开始消息
status_msg = await message.reply_text(
status_text = (
f"🔄 开始注册 {count} 个账号...\n"
f"{'✅ 将生成账单 URL' if generate_billing else '❌ 不生成账单'}\n\n"
"⏳ 请稍候..."
f"{'✅ 将生成账单 URL' if generate_billing else '❌ 不生成账单'}\n"
)
if add_payment:
status_text += "✅ 将添加支付方式\n"
status_text += "\n⏳ 请稍候..."
status_msg = await message.reply_text(status_text)
# 初始化临时邮箱客户端
try:
@@ -278,6 +451,30 @@ async def perform_registration(message, count: int, generate_billing: bool):
await status_msg.edit_text(f"❌ 初始化失败: {str(e)}")
return
# 如果需要支付,初始化 IBAN 生成器
iban_generator = None
if add_payment:
try:
await status_msg.edit_text(
f"🔄 开始注册 {count} 个账号...\n"
f"{'✅ 将生成账单 URL' if generate_billing else '❌ 不生成账单'}\n"
f"✅ 将添加支付方式\n\n"
f"⚙️ 正在初始化 IBAN 生成器..."
)
iban_generator = GermanIbanGenerator()
await status_msg.edit_text(
f"🔄 开始注册 {count} 个账号...\n"
f"{'✅ 将生成账单 URL' if generate_billing else '❌ 不生成账单'}\n"
f"✅ 将添加支付方式\n\n"
f"✅ IBAN 生成器已就绪"
)
except Exception as e:
await status_msg.edit_text(
f"⚠️ IBAN 生成器初始化失败: {str(e)}\n\n"
f"将继续注册但不添加支付方式"
)
add_payment = False
# 注册账号
success_accounts = []
failed_accounts = []
@@ -285,15 +482,26 @@ async def perform_registration(message, count: int, generate_billing: bool):
for i in range(1, count + 1):
try:
# 更新状态
progress_bar = "" * i + "" * (count - i)
await status_msg.edit_text(
f"🔄 正在注册第 {i}/{count} 个账号...\n"
f"🔄 **注册进度** [{i}/{count}]\n"
f"{progress_bar}\n\n"
f"✅ 成功: {len(success_accounts)}\n"
f"❌ 失败: {len(failed_accounts)}"
f"❌ 失败: {len(failed_accounts)}\n\n"
f"⏳ 正在处理第 {i} 个账号...",
parse_mode='Markdown'
)
# 生成密码
password = generate_random_password()
# 生成 IBAN (如果需要)
iban = None
if add_payment and iban_generator:
iban = iban_generator.generate(1)[0]
if DEBUG:
print(f"🔢 Generated IBAN: {iban}")
# 创建注册器
registrar = OpenAIRegistrar(tempmail_client=tempmail_client)
@@ -324,6 +532,30 @@ async def perform_registration(message, count: int, generate_billing: bool):
if billing_result.success:
account_info['access_token'] = access_token
account_info['checkout_url'] = billing_result.checkout_url
# 如果需要添加支付方式
if add_payment and payment_info and iban:
try:
payment_result = registrar.add_payment_method(
checkout_session_url=billing_result.checkout_url,
iban=iban,
name=payment_info.get('name', 'John Doe'),
email=email,
address_line1=payment_info.get('address_line1', '123 Main Street'),
city=payment_info.get('city', 'New York'),
postal_code=payment_info.get('postal_code', '10001'),
state=payment_info.get('state', 'NY'),
country=payment_info.get('country', 'US')
)
if payment_result.get('success'):
account_info['payment_added'] = True
account_info['iban'] = iban
else:
account_info['payment_error'] = payment_result.get('error', 'Unknown error')
except Exception as e:
account_info['payment_error'] = str(e)
else:
account_info['billing_error'] = billing_result.error
except Exception as e:
@@ -337,43 +569,109 @@ async def perform_registration(message, count: int, generate_billing: bool):
'error': str(e)
})
# 发送结果
# 发送结果摘要
progress_bar = "" * count
await status_msg.edit_text(
f"注册完成!\n\n"
f"成功: {len(success_accounts)}\n"
f"失败: {len(failed_accounts)}\n\n"
"正在发送账号信息..."
f"🎉 **注册完成!** [{count}/{count}]\n"
f"{progress_bar}\n\n"
f"✅ 成功: **{len(success_accounts)}**\n"
f"❌ 失败: **{len(failed_accounts)}**\n\n"
f"📨 正在发送账号信息...",
parse_mode='Markdown'
)
# 发送每个成功的账号
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"
f"╔═══════════════════\n"
f"║ 🎯 **账号 #{idx}**\n"
f"╚═══════════════════\n\n"
f"📧 **邮箱**\n"
f"`{acc['email']}`\n\n"
f"🔑 **密码**\n"
f"`{acc['password']}`\n"
)
if 'access_token' in acc:
account_text += f"🎫 Token: `{acc['access_token'][:50]}...`\n"
account_text += f"\n🎫 **Access Token**\n`{acc['access_token'][:50]}...`\n"
if 'checkout_url' in acc:
account_text += f"💳 [账单链接]({acc['checkout_url']})\n"
account_text += f"\n💳 **账单链接**\n[点击打开支付页面]({acc['checkout_url']})\n"
elif 'billing_error' in acc:
account_text += f"⚠️ 账单生成失败: {acc['billing_error']}\n"
account_text += f"\n⚠️ 账单: {acc['billing_error']}\n"
# 支付信息
if 'payment_added' in acc and acc['payment_added']:
account_text += f"\n✅ **支付方式**\nSEPA 已添加\n"
if 'iban' in acc:
account_text += f"\n🏦 **IBAN**\n`{acc['iban']}`\n"
elif 'payment_error' in acc:
account_text += f"\n⚠️ 支付: {acc['payment_error']}\n"
account_text += "\n━━━━━━━━━━━━━━━━━━━"
await message.reply_text(account_text, parse_mode='Markdown')
# 如果有失败的账号
if failed_accounts:
failed_text = " **失败的账号:**\n\n"
failed_text = "⚠️ **失败列表**\n\n"
for idx, acc in enumerate(failed_accounts, 1):
failed_text += f"{idx}. {acc['email']}: {acc['error']}\n"
failed_text += f"`{idx}.` {acc['email']}\n {acc['error']}\n\n"
await message.reply_text(failed_text, parse_mode='Markdown')
# 删除状态消息
await status_msg.delete()
# 更新最终状态消息
await status_msg.edit_text(
f"✅ **全部完成!**\n\n"
f"📊 **统计**\n"
f"• 总数: {count}\n"
f"• 成功: {len(success_accounts)}\n"
f"• 失败: {len(failed_accounts)}\n\n"
f"{'✅ 账号信息已发送' if success_accounts else '❌ 没有成功的账号'}",
parse_mode='Markdown'
)
async def handle_text_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""处理文本消息(用于自定义数量输入)"""
user_id = update.effective_user.id
if not check_authorization(user_id):
return
# 检查是否在等待支付数量输入
if context.user_data.get('awaiting_payment_count'):
try:
count = int(update.message.text.strip())
if count <= 0 or count > 20:
await update.message.reply_text(
"❌ 请提供有效的数量 (1-20)\n\n"
"请重新输入数字:"
)
return
# 清除等待标志
context.user_data['awaiting_payment_count'] = False
context.user_data['payment_count'] = count
# 询问地区
keyboard = [
[InlineKeyboardButton("🇩🇪 德国地址", callback_data="payment_region_de")],
[InlineKeyboardButton("🇺🇸 美国地址", callback_data="payment_region_us")],
[InlineKeyboardButton("🌍 使用默认", callback_data="payment_region_default")]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(
f"✅ 将注册 {count} 个完整账号\n\n"
"请选择账单地址信息:",
reply_markup=reply_markup
)
except ValueError:
await update.message.reply_text(
"❌ 请输入有效的数字 (1-20)\n\n"
"请重新输入:"
)
async def error_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
@@ -411,7 +709,9 @@ def main():
application.add_handler(CommandHandler("help", help_command))
application.add_handler(CommandHandler("register", register_single))
application.add_handler(CommandHandler("batch", batch_register))
application.add_handler(CommandHandler("payment", payment_register))
application.add_handler(CallbackQueryHandler(button_callback))
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_text_message))
# 错误处理
application.add_error_handler(error_handler)