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>
272 lines
9.3 KiB
Python
272 lines
9.3 KiB
Python
# modules/billing.py
|
||
"""EU Billing Generator for ChatGPT Team Plans
|
||
|
||
This module handles the generation of EU billing checkout URLs for registered accounts.
|
||
It requires a valid access token obtained from an authenticated session.
|
||
"""
|
||
|
||
from dataclasses import dataclass
|
||
from typing import Optional, Dict, Any
|
||
import random
|
||
import requests
|
||
from config import EU_BILLING_CONFIG, FINGERPRINT_CONFIG, DEBUG
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class BillingResult:
|
||
"""Result of billing checkout generation
|
||
|
||
Attributes:
|
||
success: Whether billing generation succeeded
|
||
checkout_url: Generated checkout URL (if successful)
|
||
error: Error message (if failed)
|
||
raw_response: Raw API response data (for debugging)
|
||
"""
|
||
success: bool
|
||
checkout_url: Optional[str] = None
|
||
error: Optional[str] = None
|
||
raw_response: Optional[Dict[str, Any]] = None
|
||
|
||
|
||
class DeviceIDGenerator:
|
||
"""Generate device IDs for billing requests"""
|
||
|
||
@staticmethod
|
||
def generate() -> str:
|
||
"""Generate UUID-like device ID
|
||
|
||
Format: 8-4-4-4-12 hex characters
|
||
Example: a1b2c3d4-e5f6-7890-abcd-ef1234567890
|
||
|
||
Returns:
|
||
Device ID string
|
||
"""
|
||
def random_hex(length: int) -> str:
|
||
return ''.join(random.choices('0123456789abcdef', k=length))
|
||
|
||
return f"{random_hex(8)}-{random_hex(4)}-{random_hex(4)}-{random_hex(4)}-{random_hex(12)}"
|
||
|
||
|
||
class EUBillingGenerator:
|
||
"""Generate EU billing checkout URLs for ChatGPT Team Plans"""
|
||
|
||
def __init__(self, user_agent: Optional[str] = None):
|
||
"""Initialize billing generator
|
||
|
||
Args:
|
||
user_agent: Custom user agent (defaults to FINGERPRINT_CONFIG if None)
|
||
"""
|
||
self.user_agent = user_agent or FINGERPRINT_CONFIG.get('user_agent',
|
||
'Mozilla/5.0 (X11; Linux x86_64; rv:146.0) Gecko/20100101 Firefox/146.0')
|
||
|
||
def generate_checkout_url(self, access_token: str, timeout: int = 30) -> BillingResult:
|
||
"""Generate EU billing checkout URL
|
||
|
||
Args:
|
||
access_token: Valid ChatGPT access token
|
||
timeout: Request timeout in seconds
|
||
|
||
Returns:
|
||
BillingResult with checkout URL or error
|
||
"""
|
||
# Generate unique device ID for this request
|
||
device_id = DeviceIDGenerator.generate()
|
||
|
||
# Build request components
|
||
headers = self._build_headers(access_token, device_id)
|
||
payload = self._build_payload()
|
||
url = EU_BILLING_CONFIG.get('checkout_endpoint',
|
||
'https://chatgpt.com/backend-api/payments/checkout')
|
||
|
||
try:
|
||
# Send POST request to checkout endpoint
|
||
response = requests.post(
|
||
url,
|
||
headers=headers,
|
||
json=payload,
|
||
timeout=timeout,
|
||
allow_redirects=False
|
||
)
|
||
|
||
# Parse JSON response
|
||
try:
|
||
result_data = response.json()
|
||
except requests.exceptions.JSONDecodeError:
|
||
if DEBUG:
|
||
print(f"❌ [Billing] Failed to parse JSON response")
|
||
print(f" Status Code: {response.status_code}")
|
||
print(f" Response Text: {response.text[:500]}...")
|
||
print(f" Response Headers: {dict(response.headers)}")
|
||
return BillingResult(
|
||
success=False,
|
||
error=f"Invalid JSON response (HTTP {response.status_code})",
|
||
raw_response={'status_code': response.status_code, 'text': response.text[:500]}
|
||
)
|
||
|
||
# Handle successful response
|
||
if response.status_code == 200:
|
||
checkout_url = self._construct_checkout_url(result_data)
|
||
|
||
if checkout_url:
|
||
if DEBUG:
|
||
print(f"✅ [Billing] Checkout URL generated")
|
||
print(f" URL: {checkout_url}")
|
||
|
||
return BillingResult(
|
||
success=True,
|
||
checkout_url=checkout_url,
|
||
raw_response=result_data
|
||
)
|
||
else:
|
||
error_msg = f"No checkout URL in response: {result_data}"
|
||
if DEBUG:
|
||
print(f"❌ [Billing] {error_msg}")
|
||
|
||
return BillingResult(
|
||
success=False,
|
||
error=error_msg,
|
||
raw_response=result_data
|
||
)
|
||
|
||
# Handle error responses
|
||
else:
|
||
detail = result_data.get('detail', result_data)
|
||
error_msg = f"HTTP {response.status_code}: {detail}"
|
||
|
||
if DEBUG:
|
||
print(f"❌ [Billing] API error: {error_msg}")
|
||
|
||
return BillingResult(
|
||
success=False,
|
||
error=error_msg,
|
||
raw_response=result_data
|
||
)
|
||
|
||
except requests.RequestException as e:
|
||
error_msg = f"Network error: {str(e)}"
|
||
if DEBUG:
|
||
print(f"❌ [Billing] {error_msg}")
|
||
|
||
return BillingResult(
|
||
success=False,
|
||
error=error_msg
|
||
)
|
||
|
||
except Exception as e:
|
||
error_msg = f"Unexpected error: {str(e)}"
|
||
if DEBUG:
|
||
print(f"❌ [Billing] {error_msg}")
|
||
|
||
return BillingResult(
|
||
success=False,
|
||
error=error_msg
|
||
)
|
||
|
||
def _build_headers(self, access_token: str, device_id: str) -> Dict[str, str]:
|
||
"""Build request headers with all required fields
|
||
|
||
Args:
|
||
access_token: ChatGPT access token
|
||
device_id: Generated device ID
|
||
|
||
Returns:
|
||
Complete headers dictionary
|
||
"""
|
||
# Get OAI client version from config
|
||
oai_version = EU_BILLING_CONFIG.get('oai_client_version',
|
||
'prod-04eaaa443c69cfc8b46b5d52d2b61dbceba21862')
|
||
oai_build = EU_BILLING_CONFIG.get('oai_client_build_number', '4053703')
|
||
|
||
headers = {
|
||
'User-Agent': self.user_agent,
|
||
'Accept': '*/*',
|
||
'Accept-Language': 'en-US,en;q=0.5',
|
||
'Accept-Encoding': 'gzip, deflate', # 移除 br，让 requests 自动处理 gzip
|
||
'Referer': 'https://chatgpt.com/',
|
||
'OAI-Language': 'en-US',
|
||
'OAI-Device-Id': device_id,
|
||
'OAI-Client-Version': oai_version,
|
||
'OAI-Client-Build-Number': str(oai_build),
|
||
'Authorization': f'Bearer {access_token}',
|
||
'Content-Type': 'application/json',
|
||
'Origin': 'https://chatgpt.com',
|
||
'Connection': 'keep-alive',
|
||
'Sec-Fetch-Dest': 'empty',
|
||
'Sec-Fetch-Mode': 'cors',
|
||
'Sec-Fetch-Site': 'same-origin',
|
||
'Priority': 'u=4',
|
||
'TE': 'trailers'
|
||
}
|
||
|
||
return headers
|
||
|
||
def _build_payload(self) -> Dict[str, Any]:
|
||
"""Build checkout payload from EU_BILLING_CONFIG
|
||
|
||
Returns:
|
||
Complete request payload
|
||
"""
|
||
payload = {
|
||
"plan_name": EU_BILLING_CONFIG.get('plan_name', 'chatgptteamplan'),
|
||
"team_plan_data": EU_BILLING_CONFIG.get('team_plan_data', {
|
||
'workspace_name': 'Sepa',
|
||
'price_interval': 'month',
|
||
'seat_quantity': 5,
|
||
}),
|
||
"billing_details": EU_BILLING_CONFIG.get('billing_details', {
|
||
'country': 'DE',
|
||
'currency': 'EUR',
|
||
}),
|
||
"promo_campaign": EU_BILLING_CONFIG.get('promo_campaign', {
|
||
'promo_campaign_id': 'team-1-month-free',
|
||
'is_coupon_from_query_param': False,
|
||
}),
|
||
"checkout_ui_mode": EU_BILLING_CONFIG.get('checkout_ui_mode', 'redirect'),
|
||
}
|
||
|
||
return payload
|
||
|
||
def _construct_checkout_url(self, response_data: Dict[str, Any]) -> Optional[str]:
|
||
"""Construct checkout URL from API response
|
||
|
||
Handles two cases:
|
||
1. Direct 'url' field in response
|
||
2. Manual construction from 'checkout_session_id' + 'processor_entity'
|
||
|
||
Args:
|
||
response_data: API response JSON
|
||
|
||
Returns:
|
||
Checkout URL or None if construction fails
|
||
"""
|
||
# Case 1: Direct URL provided
|
||
checkout_url = response_data.get('url')
|
||
if checkout_url:
|
||
return checkout_url
|
||
|
||
# Case 2: Construct from session ID and processor entity
|
||
checkout_session_id = response_data.get('checkout_session_id')
|
||
|
||
if checkout_session_id:
|
||
processor_entity = response_data.get('processor_entity', 'openai_llc')
|
||
checkout_url = f"https://chatgpt.com/checkout/{processor_entity}/{checkout_session_id}"
|
||
return checkout_url
|
||
|
||
# No URL could be constructed
|
||
return None
|
||
|
||
|
||
# Convenience function for direct usage
|
||
def create_eu_billing(access_token: str, timeout: int = 30) -> BillingResult:
|
||
"""Convenience function to generate EU billing checkout URL
|
||
|
||
Args:
|
||
access_token: Valid ChatGPT access token
|
||
timeout: Request timeout in seconds
|
||
|
||
Returns:
|
||
BillingResult with checkout URL or error
|
||
"""
|
||
generator = EUBillingGenerator()
|
||
return generator.generate_checkout_url(access_token, timeout)
|