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

4

Browse Source
This commit is contained in:
kyx236
2026-01-27 09:47:12 +08:00
parent c937bc7356
commit a973343b48
3 changed files with 273 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files

415
browser_automation.py
View File

@@ -2303,81 +2303,108 @@ def perform_cpa_authorization(page, email: str, password: str) -> bool:
log_current_url(page, "CPA-密码步骤完成后")
# 等待授权回调
max_wait = 45
start_time = time.time()
callback_url = None
progress_shown = False
last_url_in_loop = None
log.step(f"等待 CPA 授权回调 (最多 {max_wait}s)...")
# ========== 授权回调阶段 - 需要串行执行 ==========
try:
import run
auth_lock = run._auth_callback_lock
except (ImportError, AttributeError):
auth_lock = None
while time.time() - start_time < max_wait:
try:
current_url = page.url
if auth_lock:
log.step("等待授权回调锁...")
auth_lock.acquire()
log.step("获取授权回调锁，开始授权...")
# 记录 URL 变化
if current_url != last_url_in_loop:
log_current_url(page, "CPA等待回调中")
last_url_in_loop = current_url
try:
# 等待授权回调
max_wait = 45
start_time = time.time()
callback_url = None
progress_shown = False
last_url_in_loop = None
auth_btn_clicked = False # 标记是否已点击授权按钮
log.step(f"等待 CPA 授权回调 (最多 {max_wait}s)...")
# 检查是否到达回调页面 (CPA 使用 localhost:1455)
if is_cpa_callback_url(current_url):
while time.time() - start_time < max_wait:
try:
current_url = page.url
# 记录 URL 变化
if current_url != last_url_in_loop:
log_current_url(page, "CPA等待回调中")
last_url_in_loop = current_url
# URL 变化后重置授权按钮点击标记
auth_btn_clicked = False
# 检查是否到达回调页面 (CPA 使用 localhost:1455)
if is_cpa_callback_url(current_url):
if progress_shown:
log.progress_clear()
log.success("CPA 获取到回调 URL")
log.info(f"[URL] CPA回调地址: {current_url}", icon="browser")
callback_url = current_url
break
# 尝试点击授权按钮 (只点击一次)
if not auth_btn_clicked:
try:
buttons = page.eles('css:button[type="submit"]')
for btn in buttons:
if btn.states.is_displayed and btn.states.is_enabled:
btn_text = btn.text.lower()
if any(x in btn_text for x in ['allow', 'authorize', 'continue', '授权', '允许', '继续', 'accept']):
if progress_shown:
log.progress_clear()
progress_shown = False
log.step(f"点击按钮: {btn.text}")
btn.click()
auth_btn_clicked = True # 标记已点击
time.sleep(2) # 等待页面响应
break
except Exception:
pass
elapsed = int(time.time() - start_time)
log.progress_inline(f"[CPA等待中... {elapsed}s]")
progress_shown = True
time.sleep(1.5)
except Exception as e:
if progress_shown:
log.progress_clear()
log.success("CPA 获取到回调 URL")
log.info(f"[URL] CPA回调地址: {current_url}", icon="browser")
callback_url = current_url
break
progress_shown = False
log.warning(f"CPA检查异常: {e}")
time.sleep(1.5)
# 尝试点击授权按钮
if progress_shown:
log.progress_clear()
if not callback_url:
log.error("CPA 无法获取回调 URL")
return False
# CPA 特有流程: 提交回调 URL
log.step("提交 CPA 回调 URL...")
if not cpa_submit_callback(callback_url):
log.error("CPA 回调 URL 提交失败")
return False
# CPA 特有流程: 轮询授权状态
if cpa_poll_auth_status(state):
log.success("CPA Codex 授权成功")
return True
else:
log.error("CPA 授权状态检查失败")
return False
finally:
# 确保释放锁
if auth_lock:
try:
buttons = page.eles('css:button[type="submit"]')
for btn in buttons:
if btn.states.is_displayed and btn.states.is_enabled:
btn_text = btn.text.lower()
if any(x in btn_text for x in ['allow', 'authorize', 'continue', '授权', '允许', '继续', 'accept']):
if progress_shown:
log.progress_clear()
progress_shown = False
log.step(f"点击按钮: {btn.text}")
btn.click()
time.sleep(1.5)
break
except Exception:
pass
elapsed = int(time.time() - start_time)
log.progress_inline(f"[CPA等待中... {elapsed}s]")
progress_shown = True
time.sleep(1.5)
except Exception as e:
if progress_shown:
log.progress_clear()
progress_shown = False
log.warning(f"CPA检查异常: {e}")
time.sleep(1.5)
if progress_shown:
log.progress_clear()
if not callback_url:
log.error("CPA 无法获取回调 URL")
return False
# CPA 特有流程: 提交回调 URL
log.step("提交 CPA 回调 URL...")
if not cpa_submit_callback(callback_url):
log.error("CPA 回调 URL 提交失败")
return False
# CPA 特有流程: 轮询授权状态
if cpa_poll_auth_status(state):
log.success("CPA Codex 授权成功")
return True
else:
log.error("CPA 授权状态检查失败")
return False
auth_lock.release()
log.step("释放授权回调锁")
except RuntimeError:
pass # 锁可能已经被释放
def perform_cpa_authorization_with_otp(page, email: str) -> bool:
@@ -2537,6 +2564,7 @@ def perform_cpa_authorization_with_otp(page, email: str) -> bool:
callback_url = None
progress_shown = False
last_url_in_loop = None
auth_btn_clicked = False # 标记是否已点击授权按钮
log.step(f"等待 CPA 授权回调 (最多 {max_wait}s)...")
while time.time() - start_time < max_wait:
@@ -2546,6 +2574,8 @@ def perform_cpa_authorization_with_otp(page, email: str) -> bool:
if current_url != last_url_in_loop:
log_current_url(page, "CPA-OTP流程-等待回调中")
last_url_in_loop = current_url
# URL 变化后重置授权按钮点击标记
auth_btn_clicked = False
if is_cpa_callback_url(current_url):
if progress_shown:
@@ -2555,21 +2585,24 @@ def perform_cpa_authorization_with_otp(page, email: str) -> bool:
callback_url = current_url
break
try:
buttons = page.eles('css:button[type="submit"]')
for btn in buttons:
if btn.states.is_displayed and btn.states.is_enabled:
btn_text = btn.text.lower()
if any(x in btn_text for x in ['allow', 'authorize', 'continue', '授权', '允许', '继续', 'accept']):
if progress_shown:
log.progress_clear()
progress_shown = False
log.step(f"点击按钮: {btn.text}")
btn.click()
time.sleep(1.5)
break
except Exception:
pass
# 尝试点击授权按钮 (只点击一次)
if not auth_btn_clicked:
try:
buttons = page.eles('css:button[type="submit"]')
for btn in buttons:
if btn.states.is_displayed and btn.states.is_enabled:
btn_text = btn.text.lower()
if any(x in btn_text for x in ['allow', 'authorize', 'continue', '授权', '允许', '继续', 'accept']):
if progress_shown:
log.progress_clear()
progress_shown = False
log.step(f"点击按钮: {btn.text}")
btn.click()
auth_btn_clicked = True # 标记已点击
time.sleep(2) # 等待页面响应
break
except Exception:
pass
elapsed = int(time.time() - start_time)
log.progress_inline(f"[CPA-OTP等待中... {elapsed}s]")
@@ -2684,86 +2717,162 @@ def perform_s2a_authorization(page, email: str, password: str) -> bool:
log_current_url(page, "S2A-密码步骤完成后")
# 等待授权回调 (S2A 使用 localhost 回调)
max_wait = 45
start_time = time.time()
callback_url = None
progress_shown = False
last_url_in_loop = None
progress_update(phase="授权", step="等待回调...")
log.step(f"等待 S2A 授权回调 (最多 {max_wait}s)...")
# ========== 授权回调阶段 - 需要串行执行 ==========
# 获取授权回调锁，确保同一时间只有一个线程进行授权回调
try:
import run
auth_lock = run._auth_callback_lock
except (ImportError, AttributeError):
auth_lock = None
while time.time() - start_time < max_wait:
try:
current_url = page.url
if auth_lock:
log.step("等待授权回调锁...")
auth_lock.acquire()
log.step("获取授权回调锁，开始授权...")
# 记录 URL 变化
if current_url != last_url_in_loop:
log_current_url(page, "S2A等待回调中")
last_url_in_loop = current_url
try:
# 检查是否在授权确认页面，如果是则点击授权按钮
time.sleep(1)
current_url = page.url
if "/authorize" in current_url or "consent" in current_url:
try:
log.step("检测到授权确认页面，点击授权按钮...")
auth_btn = wait_for_element(page, 'css:button[type="submit"]', timeout=5)
if auth_btn:
btn_text = auth_btn.text.lower() if auth_btn.text else ""
if 'authorize' in btn_text or '授权' in btn_text or 'allow' in btn_text or 'continue' in btn_text:
old_url = page.url
auth_btn.click()
time.sleep(2)
if page.url != old_url:
log_url_change(page, old_url, "S2A-点击授权按钮后")
except Exception as e:
log.warning(f"S2A 授权按钮点击异常: {e}")
# 检查是否到达回调页面 (S2A 使用 localhost:1455 或类似端口)
if "localhost" in current_url and "code=" in current_url:
# 等待授权回调 (S2A 使用 localhost 回调)
max_wait = 45
start_time = time.time()
callback_url = None
progress_shown = False
last_url_in_loop = None
auth_btn_clicked = False # 标记是否已点击授权按钮
error_detected = False # 标记是否检测到错误
progress_update(phase="授权", step="等待回调...")
log.step(f"等待 S2A 授权回调 (最多 {max_wait}s)...")
while time.time() - start_time < max_wait:
try:
current_url = page.url
# 记录 URL 变化
if current_url != last_url_in_loop:
log_current_url(page, "S2A等待回调中")
last_url_in_loop = current_url
# URL 变化后重置授权按钮点击标记
auth_btn_clicked = False
error_detected = False
# 检查是否到达回调页面 (S2A 使用 localhost:1455 或类似端口)
if "localhost" in current_url and "code=" in current_url:
if progress_shown:
log.progress_clear()
progress_shown = False
callback_url = current_url
log.success(f"捕获 S2A 回调 URL")
break
# 检测 OpenAI 错误页面
if not error_detected:
try:
error_elem = page.ele('text:Something went wrong', timeout=0.3) or \
page.ele('text:出错了', timeout=0.3) or \
page.ele('text:Access denied', timeout=0.3) or \
page.ele('text:rate limit', timeout=0.3)
if error_elem:
log.warning(f"检测到错误页面: {error_elem.text[:50] if error_elem.text else 'unknown'}")
error_detected = True
# 尝试刷新页面
page.refresh()
time.sleep(3)
continue
except Exception:
pass
# 检测错误
check_and_handle_error(page)
# 检查是否需要点击 Authorize (只点击一次)
if not auth_btn_clicked:
try:
auth_btn = page.ele('css:button[type="submit"]', timeout=0.5)
if auth_btn:
btn_text = auth_btn.text.lower() if auth_btn.text else ""
btn_enabled = auth_btn.states.is_enabled if hasattr(auth_btn, 'states') else True
btn_displayed = auth_btn.states.is_displayed if hasattr(auth_btn, 'states') else True
if 'authorize' in btn_text or '授权' in btn_text or 'continue' in btn_text or 'allow' in btn_text:
if btn_enabled and btn_displayed:
log.step(f"点击授权按钮: '{auth_btn.text}'")
old_url = page.url
auth_btn.click()
auth_btn_clicked = True # 标记已点击
time.sleep(3) # 增加等待时间
new_url = page.url
if new_url != old_url:
log_url_change(page, old_url, "S2A点击授权按钮后")
else:
log.warning(f"点击授权按钮后 URL 未变化，当前: {new_url[:80]}...")
else:
log.warning(f"授权按钮不可用: enabled={btn_enabled}, displayed={btn_displayed}")
except Exception as e:
log.warning(f"检查授权按钮异常: {e}")
elapsed = int(time.time() - start_time)
log.progress_inline(f"[S2A等待中... {elapsed}s]")
progress_shown = True
time.sleep(1.5)
except Exception as e:
if progress_shown:
log.progress_clear()
progress_shown = False
log.warning(f"S2A检查异常: {e}")
time.sleep(1.5)
callback_url = current_url
log.success(f"捕获 S2A 回调 URL")
break
if progress_shown:
log.progress_clear()
# 检测错误
check_and_handle_error(page)
if not callback_url:
log.error("S2A 无法获取回调链接")
return False
# 检查是否需要点击 Authorize
# 从回调 URL 中提取 code
code = extract_code_from_url(callback_url)
if not code:
log.error("S2A 无法从回调链接提取授权码")
return False
# S2A 特有流程: 用授权码创建账号 (传入完整邮箱用于验证)
progress_update(phase="授权", step="提交授权码...")
log.step("正在提交 S2A 授权码...")
result = s2a_create_account_from_oauth(code, session_id, name=email)
if result:
log.success("S2A 授权流程完成")
return True
else:
log.error("S2A 账号入库失败")
return False
finally:
# 确保释放锁
if auth_lock:
try:
auth_btn = page.ele('css:button[type="submit"]', timeout=0.5)
if auth_btn:
btn_text = auth_btn.text.lower() if auth_btn.text else ""
if 'authorize' in btn_text or '授权' in btn_text or 'continue' in btn_text:
log.step("点击授权按钮...")
old_url = page.url
auth_btn.click()
wait_for_url_change(page, old_url, timeout=5)
log_url_change(page, old_url, "S2A点击授权按钮后")
except Exception:
pass
elapsed = int(time.time() - start_time)
log.progress_inline(f"[S2A等待中... {elapsed}s]")
progress_shown = True
time.sleep(1.5)
except Exception as e:
if progress_shown:
log.progress_clear()
progress_shown = False
log.warning(f"S2A检查异常: {e}")
time.sleep(1.5)
if progress_shown:
log.progress_clear()
if not callback_url:
log.error("S2A 无法获取回调链接")
return False
# 从回调 URL 中提取 code
code = extract_code_from_url(callback_url)
if not code:
log.error("S2A 无法从回调链接提取授权码")
return False
# S2A 特有流程: 用授权码创建账号 (传入完整邮箱用于验证)
progress_update(phase="授权", step="提交授权码...")
log.step("正在提交 S2A 授权码...")
result = s2a_create_account_from_oauth(code, session_id, name=email)
if result:
log.success("S2A 授权流程完成")
return True
else:
log.error("S2A 账号入库失败")
return False
auth_lock.release()
log.step("释放授权回调锁")
except RuntimeError:
pass # 锁可能已经被释放
# ==================== 格式3专用: 登录获取 Session ====================