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 07:36:18 +08:00
parent d93383fe23
commit 6b914bad41
1 changed files with 97 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

190
auto_gpt_team.py
View File

@@ -161,6 +161,7 @@ def get_random_fingerprint() -> dict:
def inject_fingerprint(page, fingerprint: dict):
"""注入浏览器指纹伪装脚本"""
global _last_log_message, _last_log_time
try:
webgl_vendor = fingerprint.get("webgl_vendor", "Google Inc. (NVIDIA)")
webgl_renderer = fingerprint.get("webgl_renderer", "ANGLE (NVIDIA)")
@@ -204,7 +205,20 @@ def inject_fingerprint(page, fingerprint: dict):
}});
'''
page.run_js(js_script)
log_status("指纹", f"已注入: {webgl_renderer[:40]}...")
# 获取浏览器语言设置
try:
browser_lang = page.run_js('return navigator.language || navigator.userLanguage || "unknown"')
except:
browser_lang = "unknown"
# 避免重复日志：1秒内相同消息不重复输出
current_time = time.time()
log_msg = f"已注入: {webgl_renderer} | {screen['width']}x{screen['height']} | 语言: {browser_lang}"
if log_msg != _last_log_message or current_time - _last_log_time > 1:
log_status("指纹", log_msg)
_last_log_message = log_msg
_last_log_time = current_time
except Exception as e:
log_status("指纹", f"注入失败: {e}")
@@ -561,8 +575,14 @@ LAST_NAMES = [
# ================= 工具函数 =================
# 日志去重缓存
_last_log_message = ""
_last_log_time = 0
def cleanup_chrome_processes():
"""清理残留的 Chrome 进程 (跨平台支持)"""
global _last_log_message, _last_log_time
try:
if platform.system() == "Windows":
# Windows: 使用 taskkill 清理 chromedriver 和 chrome
@@ -573,7 +593,7 @@ def cleanup_chrome_processes():
)
except Exception:
pass
# 清理无头模式的 chrome 进程 (带 --headless 参数的)
try:
result = subprocess.run(
@@ -586,8 +606,6 @@ def cleanup_chrome_processes():
subprocess.run(['taskkill', '/F', '/PID', pid], capture_output=True, timeout=5)
except Exception:
pass
log_status("清理", "已清理 Chrome 残留进程")
else:
# Linux/Mac: 使用 pkill
try:
@@ -597,7 +615,7 @@ def cleanup_chrome_processes():
)
except Exception:
pass
# 清理无头模式的 chrome 进程
try:
subprocess.run(
@@ -606,8 +624,14 @@ def cleanup_chrome_processes():
)
except Exception:
pass
log_status("清理", "已清理 Chrome 残留进程")
# 避免重复日志：1秒内相同消息不重复输出
current_time = time.time()
log_msg = "已清理 Chrome 残留进程"
if log_msg != _last_log_message or current_time - _last_log_time > 1:
log_status("清理", log_msg)
_last_log_message = log_msg
_last_log_time = current_time
except Exception:
pass # 静默处理，不影响主流程
@@ -1059,24 +1083,28 @@ def run_payment_flow(page, email, step_callback=None):
# ========== 步骤 9: 获取 token 和 account_id ==========
step_cb("获取 Token...")
log_status("获取", "正在获取 access token...")
time.sleep(2)
time.sleep(1)
page.get("https://chatgpt.com/api/auth/session")
time.sleep(2)
time.sleep(1)
try:
# 获取页面内容（JSON）
session_text = page.ele('tag:pre', timeout=5).text
import json
session_data = json.loads(session_text)
access_token = session_data.get('accessToken', '')
if access_token:
log_status("成功", f"获取到 token: {access_token[:50]}...")
# 获取 account_id
# 优先从 session 数据直接提取 account_id（最快）
step_cb("获取 Account ID...")
account_id = fetch_account_id(page, access_token)
account_id = fetch_account_id_from_session(session_data)
# 如果 session 中没有，再通过 API 获取
if not account_id:
account_id = fetch_account_id(page, access_token)
return {
"token": access_token,
"account_id": account_id
@@ -1098,30 +1126,25 @@ def run_payment_flow(page, email, step_callback=None):
return None
def fetch_account_id(page, access_token: str) -> str:
"""通过 API 获取 account_id"""
"""通过 API 获取 account_id (使用 requests 直接请求，更快)"""
log_status("获取", "正在获取 account_id...")
try:
page.get("https://chatgpt.com/backend-api/accounts/check/v4-2023-04-27")
time.sleep(2)
# 使用 JS 请求 API
result = page.run_js(f'''
return fetch("https://chatgpt.com/backend-api/accounts/check/v4-2023-04-27", {{
headers: {{
"Authorization": "Bearer {access_token}",
"Content-Type": "application/json"
}}
}})
.then(r => r.json())
.then(data => JSON.stringify(data))
.catch(e => "error:" + e);
''')
if result and not result.startswith("error:"):
import json
data = json.loads(result)
# 直接使用 requests 请求 API，比 page.get() + JS fetch 快得多
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
resp = requests.get(
"https://chatgpt.com/backend-api/accounts/check/v4-2023-04-27",
headers=headers,
timeout=10
)
if resp.status_code == 200:
data = resp.json()
accounts = data.get("accounts", {})
# 优先查找 Team 账户
for acc_id, acc_info in accounts.items():
if acc_id == "default":
@@ -1131,15 +1154,30 @@ def fetch_account_id(page, access_token: str) -> str:
if "team" in plan_type.lower():
log_status("成功", f"获取到 account_id: {acc_id[:8]}...")
return acc_id
# 取第一个非 default 的
for acc_id in accounts.keys():
if acc_id != "default":
log_status("成功", f"获取到 account_id: {acc_id[:8]}...")
return acc_id
else:
log_progress(f"API 请求失败: {resp.status_code}")
except Exception as e:
log_progress(f"获取 account_id 失败: {e}")
return ""
def fetch_account_id_from_session(session_data: dict) -> str:
"""直接从 session 数据中提取 account_id (最快方式)"""
try:
account = session_data.get("account", {})
account_id = account.get("id", "")
if account_id:
log_status("成功", f"获取到 account_id: {account_id[:8]}...")
return account_id
except Exception as e:
log_progress(f"从 session 提取 account_id 失败: {e}")
return ""
@@ -1418,43 +1456,29 @@ def run_main_process():
# =======================================================
log_status("订阅", "等待进入 ChatGPT 主页...")
# 等待页面跳转完成
time.sleep(5)
# 快速检测循环（减少等待时间）
entered_main = False
for i in range(30):
for i in range(20): # 最多等待 10 秒
current_url = page.url
# 更宽松的判断：只要不在 auth/login 页面就认为进入了主页
if 'chatgpt.com' in current_url:
if '/auth' not in current_url and '/login' not in current_url and 'auth0' not in current_url:
log_progress(f"✓ 已进入主页: {current_url[:60]}...")
log_progress(f"✓ 已进入主页: {current_url}")
entered_main = True
break
time.sleep(1)
if i % 10 == 0 and i > 0:
log_progress(f"等待中... ({i}秒)")
time.sleep(0.5) # 减少等待间隔
if not entered_main:
log_progress("⚠ 等待主页超时，尝试继续...")
log_progress(f"当前URL: {page.url}")
# 尝试直接访问主页
page.get("https://chatgpt.com/")
time.sleep(3)
time.sleep(2)
log_progress(f"跳转后URL: {page.url}")
# 额外等待页面稳定
time.sleep(3)
log_status("订阅", "执行 JS 跳转到支付页...")
# 先检查是否已登录
try:
session_check = page.run_js('return fetch("/api/auth/session").then(r => r.json())')
time.sleep(1)
except:
pass
# 直接执行 JS 跳转到支付页
# 直接执行 JS 跳转到支付页（无需额外等待，JS 会自动获取 session）
checkout_js = '''
(async function(){
try {
@@ -1501,10 +1525,6 @@ def run_main_process():
'''
result = page.run_js(checkout_js)
log_progress(f"JS 执行结果: {result}")
# 等待 JS 执行完成
time.sleep(2)
log_progress(f"当前URL: {page.url}")
# 等待跳转到支付页（使用 URL 检测代替固定等待）
try:
@@ -1512,9 +1532,9 @@ def run_main_process():
log_progress("✓ 已跳转到支付页")
log_progress(f"支付页URL: {page.url}")
except:
time.sleep(2)
time.sleep(1)
log_progress(f"当前URL: {page.url}")
# 执行支付流程
result = run_payment_flow(page, email)
@@ -1817,48 +1837,35 @@ def run_single_registration(progress_callback=None, step_callback=None) -> dict:
if not submit_success:
log_progress(f"⚠ 提交后仍在 about-you 页面，当前URL: {page.url}")
# 等待进入主页 - 改进检测逻辑
# 等待进入主页 - 快速检测
step_cb("等待进入主页...")
log_status("订阅", "等待进入 ChatGPT 主页...")
# 等待页面跳转完成
time.sleep(5)
# 快速检测循环（减少等待时间）
entered_main = False
for i in range(30):
for i in range(20): # 最多等待 10 秒
current_url = page.url
# 更宽松的判断：只要不在 auth/login 页面就认为进入了主页
if 'chatgpt.com' in current_url:
if '/auth' not in current_url and '/login' not in current_url and 'auth0' not in current_url:
log_progress(f"✓ 已进入主页: {current_url[:60]}...")
log_progress(f"✓ 已进入主页: {current_url}")
entered_main = True
break
time.sleep(1)
if i % 10 == 0 and i > 0:
log_progress(f"等待中... ({i}秒)")
time.sleep(0.5) # 减少等待间隔
if not entered_main:
log_progress("⚠ 等待主页超时，尝试继续...")
log_progress(f"当前URL: {page.url}")
# 尝试直接访问主页
page.get("https://chatgpt.com/")
time.sleep(3)
time.sleep(2)
log_progress(f"跳转后URL: {page.url}")
# 额外等待页面稳定
time.sleep(3)
# 跳转到支付页
step_cb("跳转到支付页...")
log_status("订阅", "执行 JS 跳转到支付页...")
# 先检查是否已登录
try:
session_check = page.run_js('return fetch("/api/auth/session").then(r => r.json())')
time.sleep(1)
except:
pass
# 直接执行 JS 跳转到支付页（无需额外等待）
checkout_js = '''
(async function(){
try {
@@ -1884,17 +1891,14 @@ def run_single_registration(progress_callback=None, step_callback=None) -> dict:
'''
result = page.run_js(checkout_js)
log_progress(f"JS 执行结果: {result}")
# 等待 JS 执行完成
time.sleep(2)
log_progress(f"当前URL: {page.url}")
# 等待跳转到支付页
try:
page.wait.url_change('pay.openai.com', timeout=15)
log_progress("✓ 已跳转到支付页")
log_progress(f"支付页URL: {page.url}")
except:
time.sleep(2)
time.sleep(1)
log_progress(f"当前URL: {page.url}")
# 执行支付流程