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

5

Browse Source
This commit is contained in:
kyx236
2026-01-27 09:59:16 +08:00
parent a973343b48
commit 06eaff03b9
1 changed files with 129 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
browser_automation.py
View File

@@ -2243,6 +2243,19 @@ def perform_cpa_authorization(page, email: str, password: str) -> bool:
""" """
log.info(f"开始 CPA 授权: {email}", icon="code") log.info(f"开始 CPA 授权: {email}", icon="code")
# ========== 授权流程 - 需要串行执行 (避免回调端口冲突) ==========
try:
import run
auth_lock = run._auth_callback_lock
except (ImportError, AttributeError):
auth_lock = None
if auth_lock:
log.step("等待授权锁...")
auth_lock.acquire()
log.step("获取授权锁，开始授权流程...")
try:
# 生成授权 URL # 生成授权 URL
auth_url, state = cpa_generate_auth_url() auth_url, state = cpa_generate_auth_url()
if not auth_url or not state: if not auth_url or not state:
@@ -2303,19 +2316,6 @@ def perform_cpa_authorization(page, email: str, password: str) -> bool:
log_current_url(page, "CPA-密码步骤完成后") log_current_url(page, "CPA-密码步骤完成后")
# ========== 授权回调阶段 - 需要串行执行 ==========
try:
import run
auth_lock = run._auth_callback_lock
except (ImportError, AttributeError):
auth_lock = None
if auth_lock:
log.step("等待授权回调锁...")
auth_lock.acquire()
log.step("获取授权回调锁，开始授权...")
try:
# 等待授权回调 # 等待授权回调
max_wait = 45 max_wait = 45
start_time = time.time() start_time = time.time()
@@ -2654,6 +2654,19 @@ def perform_s2a_authorization(page, email: str, password: str) -> bool:
log.info(f"开始 S2A 授权: {email}", icon="code") log.info(f"开始 S2A 授权: {email}", icon="code")
progress_update(phase="授权", step="开始 S2A 授权...") progress_update(phase="授权", step="开始 S2A 授权...")
# ========== 授权流程 - 需要串行执行 (避免回调端口冲突) ==========
try:
import run
auth_lock = run._auth_callback_lock
except (ImportError, AttributeError):
auth_lock = None
if auth_lock:
log.step("等待授权锁...")
auth_lock.acquire()
log.step("获取授权锁，开始授权流程...")
try:
# 生成授权 URL # 生成授权 URL
auth_url, session_id = s2a_generate_auth_url() auth_url, session_id = s2a_generate_auth_url()
if not auth_url or not session_id: if not auth_url or not session_id:
@@ -2717,20 +2730,6 @@ def perform_s2a_authorization(page, email: str, password: str) -> bool:
log_current_url(page, "S2A-密码步骤完成后") log_current_url(page, "S2A-密码步骤完成后")
# ========== 授权回调阶段 - 需要串行执行 ==========
# 获取授权回调锁，确保同一时间只有一个线程进行授权回调
try:
import run
auth_lock = run._auth_callback_lock
except (ImportError, AttributeError):
auth_lock = None
if auth_lock:
log.step("等待授权回调锁...")
auth_lock.acquire()
log.step("获取授权回调锁，开始授权...")
try:
# 检查是否在授权确认页面，如果是则点击授权按钮 # 检查是否在授权确认页面，如果是则点击授权按钮
time.sleep(1) time.sleep(1)
current_url = page.url current_url = page.url