fix(chanmama): 导出确认弹窗被浮层清理误删与代理出口拉黑修复

浮层清理只删可见浮层——页面常驻 90+ 个 display:none 弹窗模板(含导出确认框
模板),误删后 Vue 只在游离节点上"打开"弹窗,页面上仅剩灰色遮罩与滚动锁,
导出等待 1320s 必超时;导出等待改为逐秒弹窗状态机(可点即点/点不到先关/卡死
强删并补点导出),导出失败落截图与 HTML 诊断;浏览器强制直连并切换真 Chrome
非 stealthy 模式(StealthySession 下蝉妈妈前端会静默压制弹窗渲染);导出按钮
原生可信点击优先,防 @click 绑定在内层元素时 JS 合成点击无效。
This commit is contained in:
wangyunlong committed 2026-10-09 13:45:29 +08:00
1 parent 130f06b320
commit 41f448420d
3 files changed
+687 -55

No files matched your search

@@ -52,6 +52,11 @@ REFRESH_WAIT_SECONDS = int(os.getenv("CHANMAMA_REFRESH_WAIT_SECONDS", "600"))
CHANMAMA_LIST_READY_TIMEOUT_SECONDS = int( CHANMAMA_LIST_READY_TIMEOUT_SECONDS = int(
os.getenv("CHANMAMA_LIST_READY_TIMEOUT_SECONDS", "45") os.getenv("CHANMAMA_LIST_READY_TIMEOUT_SECONDS", "45")
) )
# 公告/活动浮层可能盖住视频列表导致渲染判定一直不满足;渲染等待期间按
# 该间隔无交互清理浮层(派发 ESC + 移除 fixed/absolute 浮层节点)。
CHANMAMA_OVERLAY_DISMISS_INTERVAL_SECONDS = float(
os.getenv("CHANMAMA_OVERLAY_DISMISS_INTERVAL_SECONDS", "10")
)
# 蝉妈妈导出接口实测可能在 15 分钟边界后才把 Excel 写入下载目录;把轮询 # 蝉妈妈导出接口实测可能在 15 分钟边界后才把 Excel 写入下载目录;把轮询
# 上限做成可配置,并默认留出 20 分钟,避免文件已生成却被脚本判定为失败。 # 上限做成可配置,并默认留出 20 分钟,避免文件已生成却被脚本判定为失败。
CHANMAMA_EXPORT_WAIT_SECONDS = int( CHANMAMA_EXPORT_WAIT_SECONDS = int(
@@ -99,16 +104,20 @@ def create_driver():
profile_dir = str(PATHS.browser_profile_dir) profile_dir = str(PATHS.browser_profile_dir)
os.makedirs(profile_dir, exist_ok=True) os.makedirs(profile_dir, exist_ok=True)
cdp_url = os.getenv("GYXX_BROWSER_CDP_URL") or None cdp_url = os.getenv("GYXX_BROWSER_CDP_URL") or None
# 蝉妈妈是国内站且其 WAF 会拉黑代理出口(Clash 节点 TLS 直接 reset,
# 导出 API 全断、确认弹窗永远不弹);强制浏览器直连。
direct_flags = ["--no-proxy-server"]
# StealthySession(隐藏指纹补丁)下蝉妈妈前端会静默压制功能: 页面能开
# 但导出确认弹窗不渲染(只剩空壳遮罩)。真 Chrome + 非 stealthy + 直连
# 已实测弹窗完整渲染(含取消/导出按钮),不再使用 stealthy。
try: try:
return ScraplingDriver.start( return ScraplingDriver.start(
profile_dir=profile_dir, profile_dir=profile_dir,
download_dir=DOWNLOAD_DIR, download_dir=DOWNLOAD_DIR,
cdp_url=cdp_url, cdp_url=cdp_url,
stealthy=True, stealthy=False,
# 蝉妈妈新版 Vue 页面与 StealthySession + 系统 Chrome 组合 real_chrome=True,
# 不兼容:HTML 能返回,但 #app 不会挂载。使用 Scrapling extra_flags=direct_flags,
# 自带 Chromium 仍保留隐藏指纹会话,且不影响已连接的 CDP。
**({"real_chrome": False} if cdp_url is None else {}),
) )
except Exception as exc: except Exception as exc:
if ( if (
@@ -125,8 +134,9 @@ def create_driver():
profile_dir=profile_dir, profile_dir=profile_dir,
download_dir=DOWNLOAD_DIR, download_dir=DOWNLOAD_DIR,
cdp_url=None, cdp_url=None,
stealthy=True, stealthy=False,
real_chrome=False, real_chrome=True,
extra_flags=direct_flags,
) )
@@ -297,7 +307,10 @@ def handle_captcha(driver) -> bool:
def verify_login(driver): def verify_login(driver):
"""验证是否已登录 """验证是否已登录
通过访问目标页面,检查是否出现登录相关的元素 通过访问目标页面,检查是否出现登录相关的元素。
cookie 失效时蝉妈妈会把目标页重定向到 login.html,但重定向常在
首屏渲染后数秒才发生——等待 URL 稳定后再判定,避免误判已登录
(误判会跳过重新登录,导致后续导航一直超时)。
""" """
try: try:
# 跳转到目标页面 # 跳转到目标页面
@@ -305,10 +318,12 @@ def verify_login(driver):
WebDriverWait(driver, 12).until( WebDriverWait(driver, 12).until(
EC.presence_of_element_located((By.TAG_NAME, "body")) EC.presence_of_element_located((By.TAG_NAME, "body"))
) )
time.sleep(3) # cookie 失效时的重定向在首屏渲染后 3~20 秒才发生,固定等满 10 秒
# 再判定,不能因 URL 短暂未变而提前判为已登录
time.sleep(10)
cur_lower = driver.current_url.lower()
# 检查是否跳转回登录页 / 注册页(cookie 失效时蝉妈妈会重定向到 login.html 或 register.html) # 检查是否跳转回登录页 / 注册页(cookie 失效时蝉妈妈会重定向到 login.html 或 register.html)
cur_lower = driver.current_url.lower()
if "login" in cur_lower or "register" in cur_lower: if "login" in cur_lower or "register" in cur_lower:
print(f"🔒 检测到被重定向到登录/注册页 ({driver.current_url}),cookie 已失效") print(f"🔒 检测到被重定向到登录/注册页 ({driver.current_url}),cookie 已失效")
return False return False
@@ -458,6 +473,180 @@ def login(driver):
return False return False
def _dismiss_overlays(driver) -> int:
"""无交互清理遮挡浮层:派发 ESC 并移除 fixed/absolute 定位的弹层。"""
try:
removed = int(driver.execute_script(_OVERLAY_DISMISS_SCRIPT) or 0)
except Exception as error:
print(f"⚠️ 浮层清理脚本执行失败(忽略): {error}")
return 0
if removed:
print(f"🧹 已清理 {removed} 个遮挡浮层")
return removed
def _dismiss_export_stage_overlays(driver) -> int:
"""导出阶段清理弹窗(ESC + 移除弹层节点)。
只在弹窗不可确认时调用;返回移除的弹窗数。
"""
try:
removed = int(driver.execute_script(_EXPORT_STAGE_DISMISS_SCRIPT) or 0)
except Exception as error:
print(f"⚠️ 导出阶段弹窗清理脚本执行失败(忽略): {error}")
return 0
if removed > 0:
print(f"🧹 导出阶段清理 {removed} 个遮挡弹窗")
return removed
def _export_dialog_state(driver) -> str:
"""探测导出阶段弹窗状态: confirmable / present / none。"""
try:
state = driver.execute_script(_EXPORT_DIALOG_STATE_SCRIPT)
except Exception:
return "none"
return state if state in {"confirmable", "present", "none"} else "none"
def _log_export_dialog_texts(driver) -> None:
"""把当前可见弹窗的类名与文字打进日志,用于定位弹窗身份。"""
try:
texts = driver.execute_script(_EXPORT_DIALOG_TEXT_SCRIPT) or []
except Exception:
return
for i, text in enumerate(texts[:4], 1):
print(f"🪟 弹窗{i}内容: {text}")
def _click_export_confirm(driver) -> bool:
"""点击导出确认弹窗中的【导出】按钮,成功返回 True。"""
# 关键:用 contains(.,'导出') 匹配后代文本("导出"在 <span> 内)
confirm_selectors = [
# 精确:主按钮(primary) + 含"导出"文字
"//div[contains(@class,'el-dialog') or @role='dialog']//button[contains(.,'导出')]",
# 弹窗内任意含"导出"的按钮
"//div[contains(@class,'el-dialog') or @role='dialog']//button[contains(@class,'el-button--primary')][contains(.,'导出')]",
# 兼容"确定"/"确认"
"//div[contains(@class,'el-dialog') or @role='dialog']//button[contains(.,'确定')]",
"//div[contains(@class,'el-dialog') or @role='dialog']//button[contains(.,'确认')]",
]
for sel in confirm_selectors:
try:
btn = driver.find_element(By.XPATH, sel)
if btn.is_displayed():
# 原生点击优先(可触发内层元素的 Vue @click),失败再退 JS 合成点击
try:
btn.click()
except Exception:
driver.execute_script("arguments[0].click();", btn)
print(f"✅ 已在弹窗中点击导出按钮: {sel}")
return True
except Exception:
continue
return False
def _reclick_export_control(driver) -> bool:
"""重新触发一次导出按钮(第一次点击可能已被弹窗吞掉)。
优先 Playwright 原生点击(在元素命中点派发可信鼠标事件,能触发绑定
在按钮内层 span/div 上的 Vue @click);JS 合成 click 的目标是按钮自身,
内层监听器不会触发——蝉妈妈改版后导出点击无声失效就是这个原因。
"""
if _native_click_export_button(driver):
return True
try:
return bool(driver.execute_script(_CLICK_EXPORT_CONTROL_SCRIPT))
except Exception:
return False
_EXPORT_BUTTON_XPATHS = [
"//button[contains(.,'导出数据')]",
"//button[contains(.,'批量导出')]",
"//button[contains(.,'导出视频')]",
"//*[contains(@class,'export') and (self::button or self::a or @role='button')]",
]
def _native_click_export_button(driver) -> bool:
"""用原生可信点击触发导出按钮,成功返回 True。"""
for xp in _EXPORT_BUTTON_XPATHS:
try:
elements = driver.find_elements(By.XPATH, xp)
except Exception:
continue
for el in elements:
try:
if el.is_displayed():
el.click()
return True
except Exception:
continue
return False
def _wait_video_list_ready(driver, timeout_seconds: float) -> None:
"""等待视频列表渲染就绪,期间周期性清理遮挡浮层。
语义与 WebDriverWait.until 保持一致:浏览器侧的暂时性异常只记录
不中断,到达 deadline 后抛 TimeoutException。
"""
timeout = max(0.0, float(timeout_seconds))
deadline = time.monotonic() + timeout
next_dismiss_at = 0.0
last_error: Exception | None = None
while True:
try:
if driver.execute_script(_VIDEO_LIST_READY_SCRIPT):
return
except Exception as error:
last_error = error
now = time.monotonic()
if now >= deadline:
detail = f": {last_error}" if last_error else ""
raise TimeoutException(
f"condition was not met within {timeout}s{detail}"
) from last_error
if now >= next_dismiss_at:
_dismiss_overlays(driver)
next_dismiss_at = now + CHANMAMA_OVERLAY_DISMISS_INTERVAL_SECONDS
time.sleep(0.5)
def _dump_render_failure(driver, url: str) -> None:
"""渲染失败时落截图与页面 HTML,供事后定位是浮层/验证码还是改版。"""
try:
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
diag_dir = Path(DOWNLOAD_DIR) / "_diag"
diag_dir.mkdir(parents=True, exist_ok=True)
screenshot = diag_dir / f"render_failure_{stamp}.png"
driver.save_screenshot(str(screenshot))
(diag_dir / f"render_failure_{stamp}.html").write_text(
driver.page_source, encoding="utf-8"
)
print(f"🩺 渲染失败诊断已保存: {screenshot}")
except Exception as error:
print(f"⚠️ 渲染失败诊断保存失败: {error}")
def _dump_export_failure(driver) -> None:
"""导出失败时落截图与页面 HTML,供事后定位弹窗/按钮的实际形态。"""
try:
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
diag_dir = Path(DOWNLOAD_DIR) / "_diag"
diag_dir.mkdir(parents=True, exist_ok=True)
screenshot = diag_dir / f"export_failure_{stamp}.png"
driver.save_screenshot(str(screenshot))
(diag_dir / f"export_failure_{stamp}.html").write_text(
driver.page_source, encoding="utf-8"
)
print(f"🩺 导出失败诊断已保存: {screenshot}")
except Exception as error:
print(f"⚠️ 导出失败诊断保存失败: {error}")
def navigate_to_target(driver, url): def navigate_to_target(driver, url):
"""跳转到目标博主视频页面""" """跳转到目标博主视频页面"""
last_error = None last_error = None
@@ -475,12 +664,11 @@ def navigate_to_target(driver, url):
) )
time.sleep(3) # 等待页面数据加载 time.sleep(3) # 等待页面数据加载
# 截图中的页面虽然 URL 正确,但正文尚未渲染;在交给导出逻辑 # 截图中的页面虽然 URL 正确,但正文尚未渲染;在交给导出逻辑
# 前确认虚拟视频列表已经出现,空白页则有限重载一次。 # 前确认虚拟视频列表已经出现,空白页则有限重载一次。浮层多在
WebDriverWait(driver, CHANMAMA_LIST_READY_TIMEOUT_SECONDS).until( # 首屏挂载后才弹出:进等待前先清一轮,等待期间按间隔再清,
lambda current: bool( # 避免延迟弹出的公告/活动浮层永久挡住渲染判定。
current.execute_script(_VIDEO_LIST_READY_SCRIPT) _dismiss_overlays(driver)
) _wait_video_list_ready(driver, CHANMAMA_LIST_READY_TIMEOUT_SECONDS)
)
print(f"✅ 已到达并完成渲染目标页面: {current_url}") print(f"✅ 已到达并完成渲染目标页面: {current_url}")
return True return True
except Exception as exc: except Exception as exc:
@@ -499,6 +687,7 @@ def navigate_to_target(driver, url):
except Exception as reset_exc: except Exception as reset_exc:
print(f"⚠️ 蝉妈妈页面重置失败,继续重试目标页: {reset_exc}") print(f"⚠️ 蝉妈妈页面重置失败,继续重试目标页: {reset_exc}")
time.sleep(1) time.sleep(1)
_dump_render_failure(driver, url)
raise RuntimeError(f"蝉妈妈目标页加载失败: {url}; {last_error}") from last_error raise RuntimeError(f"蝉妈妈目标页加载失败: {url}; {last_error}") from last_error
@@ -539,6 +728,206 @@ def _find_video_table(driver):
return WebDriverWait(driver, 15).until(locate) return WebDriverWait(driver, 15).until(locate)
# 蝉妈妈公告/活动弹窗以 fixed/absolute 浮层盖住页面,且类名不稳定、未必
# 有关闭按钮。这里不点叉:派发 ESC 键事件(多数组件库监听 document 关闭
# 弹窗),再移除定位浮层节点并解除其对 html/body 的滚动锁;已知类名未
# 命中时兜底移除 z-index 高且覆盖大半屏的 fixed 层。列表区
# (.aweme-record-wrapper/.list-box/.list-auth-wrap-block) 内的元素永不命中。
# .v-modal 是 ElementUI 直接挂在 <body> 下的灰色遮罩,与弹窗本体分离,
# 必须显式清理,否则弹窗删掉后页面仍然灰屏、无法滚动;滚动锁 class
# (el-popup-parent--hidden 等)同理。
# 关键约束:只移除"可见"的浮层。页面常驻 90+ 个 display:none 的隐藏弹窗
# 模板(含导出确认框的模板)——删掉它们后 Vue 只会在游离节点上"打开"弹窗,
# 页面上永远渲染不出来(导出只剩灰遮罩),因此 display:none/零尺寸节点
# 一律跳过。
_OVERLAY_DISMISS_SCRIPT = r"""
() => {
const esc = new KeyboardEvent('keydown', {
key: 'Escape', code: 'Escape', keyCode: 27, which: 27, bubbles: true,
});
document.dispatchEvent(esc);
window.dispatchEvent(esc);
let removed = 0;
const seen = new Set();
const isOverlay = (el) => {
const style = window.getComputedStyle(el);
return style.position === 'fixed' || style.position === 'absolute';
};
const drop = (el) => {
if (seen.has(el) || !isOverlay(el)) return;
const style = window.getComputedStyle(el);
if (style.display === 'none') return;
const rect = el.getBoundingClientRect();
if (rect.width === 0 || rect.height === 0) return;
if (el.closest('.aweme-record-wrapper,.list-box,.list-auth-wrap-block')) return;
seen.add(el);
el.remove();
removed += 1;
};
const selectors = [
'.el-dialog__wrapper', '.el-overlay', '.el-message-box__wrapper',
'.ant-modal-mask', '.ant-modal-wrap', '.modal-mask', '.modal-dialog',
'.v-modal', '.v-modal-inner',
'[class*="popup"]', '[class*="dialog"]', '[class*="overlay"]', '[class*="mask"]',
];
for (const selector of selectors) {
document.querySelectorAll(selector).forEach(drop);
}
if (removed === 0) {
document.querySelectorAll('body *').forEach((el) => {
const style = window.getComputedStyle(el);
const zIndex = Number.parseInt(style.zIndex, 10);
const rect = el.getBoundingClientRect();
if (Number.isFinite(zIndex) && zIndex > 500
&& rect.height > window.innerHeight * 0.5) {
drop(el);
}
});
}
document.body.classList.remove(
'el-popup-parent--hidden', 'el-popup-parent--overflow-hidden', 'overflow-hidden'
);
document.documentElement.classList.remove(
'el-popup-parent--hidden', 'el-popup-parent--overflow-hidden', 'overflow-hidden'
);
document.documentElement.style.overflow = '';
document.body.style.overflow = '';
document.body.style.paddingRight = '';
return removed;
}
"""
# 导出阶段专用清理:与 _OVERLAY_DISMISS_SCRIPT 同一套清理思路,但
# 调用时机由 _export_dialog_state 决定——只在弹窗"没有可点的确认按钮"时
# 才执行(确认框可点时直接点导出,绝不清)。三件事:
# 1) 先 ESC + 点弹窗自带的关闭(×)按钮,走框架的正常关闭路径;
# 2) 再移除残留的弹层节点——必须包含 .v-modal:ElementUI 的灰色遮罩直接
# 挂在 <body> 下、与弹窗本体分离,漏掉它就是"删了弹窗页面仍灰屏";
# 3) 最后解除滚动锁:清掉 body/html 上的 el-popup-parent--hidden 等
# 框架加的 overflow 锁 class 与内联样式,否则页面删完遮罩也不能滚动。
_EXPORT_STAGE_DISMISS_SCRIPT = r"""
() => {
const visible = (el) => {
const style = window.getComputedStyle(el);
const rect = el.getBoundingClientRect();
return style.display !== 'none' && style.visibility !== 'hidden'
&& style.opacity !== '0' && rect.width > 0 && rect.height > 0;
};
const esc = new KeyboardEvent('keydown', {
key: 'Escape', code: 'Escape', keyCode: 27, which: 27, bubbles: true,
});
document.dispatchEvent(esc);
window.dispatchEvent(esc);
// 先走正常关闭:点弹窗自己的 × / 关闭按钮
let closed = 0;
document.querySelectorAll(
'.el-dialog__wrapper [class*="close"],.el-message-box__wrapper [class*="close"],'
+ '.ant-modal-wrap [class*="close"],[class*="dialog"] [class*="close"],'
+ '[class*="modal"] [class*="close"],[class*="popup"] [class*="close"]'
).forEach((el) => {
if (visible(el)) { el.click(); closed += 1; }
});
let removed = 0;
const seen = new Set();
const drop = (el) => {
if (seen.has(el)) return;
const style = window.getComputedStyle(el);
if (style.position !== 'fixed' && style.position !== 'absolute') return;
if (style.display === 'none') return;
const rect = el.getBoundingClientRect();
if (rect.width === 0 || rect.height === 0) return;
if (el.closest('.aweme-record-wrapper,.list-box,.list-auth-wrap-block')) return;
seen.add(el);
el.remove();
removed += 1;
};
const selectors = [
'.el-dialog__wrapper', '.el-overlay', '.el-message-box__wrapper',
'.ant-modal-mask', '.ant-modal-wrap', '.modal-mask', '.modal-dialog',
'.v-modal', '.v-modal-inner',
'[class*="popup"]', '[class*="dialog"]', '[class*="overlay"]', '[class*="mask"]',
];
for (const selector of selectors) {
document.querySelectorAll(selector).forEach(drop);
}
if (removed === 0) {
document.querySelectorAll('body *').forEach((el) => {
const style = window.getComputedStyle(el);
const zIndex = Number.parseInt(style.zIndex, 10);
const rect = el.getBoundingClientRect();
if (Number.isFinite(zIndex) && zIndex > 500
&& rect.height > window.innerHeight * 0.5) {
drop(el);
}
});
}
// 解除滚动锁:框架加的锁 class + 内联 overflow 一起清
document.body.classList.remove(
'el-popup-parent--hidden', 'el-popup-parent--overflow-hidden', 'overflow-hidden'
);
document.documentElement.classList.remove(
'el-popup-parent--hidden', 'el-popup-parent--overflow-hidden', 'overflow-hidden'
);
document.documentElement.style.overflow = '';
document.body.style.overflow = '';
document.body.style.paddingRight = '';
return removed + closed;
}
"""
# 导出阶段弹窗状态机:'confirmable'=有可见弹窗且弹窗里有可点的
# 导出/确定/确认按钮(直接点,不能清);'present'=弹窗可见但没有可点
# 的确认按钮(空壳遮罩/公告,ESC 关不掉就强删);'none'=没有弹窗。
_EXPORT_DIALOG_STATE_SCRIPT = r"""
() => {
const visible = (el) => {
const style = window.getComputedStyle(el);
const rect = el.getBoundingClientRect();
return style.display !== 'none' && style.visibility !== 'hidden'
&& style.opacity !== '0' && rect.width > 0 && rect.height > 0;
};
const roots = [...document.querySelectorAll(
'.el-dialog__wrapper,.el-message-box__wrapper,.el-overlay,'
+ '.ant-modal-wrap,.ant-modal-mask,[role="dialog"],'
+ '[class*="dialog"],[class*="modal"],[class*="popup"]'
)].filter(visible);
for (const root of roots) {
const confirmable = [...root.querySelectorAll(
'button,a,[role="button"],[class*="confirm"],[class*="primary"]'
)].some((el) => {
if (!visible(el)) return false;
const text = (el.innerText || el.value || '').replace(/\s+/g, '');
return /导出|确定|确认/.test(text);
});
if (confirmable) return 'confirmable';
}
return roots.length ? 'present' : 'none';
}
"""
# 抓取当前可见弹窗的文字内容(每窗截前 160 字),写进日志用于定位弹窗身份。
_EXPORT_DIALOG_TEXT_SCRIPT = r"""
() => {
const visible = (el) => {
const style = window.getComputedStyle(el);
const rect = el.getBoundingClientRect();
return style.display !== 'none' && style.visibility !== 'hidden'
&& style.opacity !== '0' && rect.width > 0 && rect.height > 0;
};
return [...document.querySelectorAll(
'.el-dialog__wrapper,.el-message-box__wrapper,.el-overlay,'
+ '.ant-modal-wrap,[role="dialog"],[class*="dialog"],[class*="modal"],[class*="popup"]'
)].filter(visible).map((el) => {
const text = (el.innerText || '').replace(/\s+/g, ' ').trim();
const classes = String(el.className || '').slice(0, 80);
return `[${classes}] ${text.slice(0, 160)}`;
});
}
"""
_VIDEO_LIST_READY_SCRIPT = r""" _VIDEO_LIST_READY_SCRIPT = r"""
() => { () => {
const visible = (el) => { const visible = (el) => {
@@ -909,8 +1298,39 @@ def _max_excel_mtime():
def refresh_then_export_accounts(driver, urls, wait_seconds=REFRESH_WAIT_SECONDS): def refresh_then_export_accounts(driver, urls, wait_seconds=REFRESH_WAIT_SECONDS):
"""两阶段执行:先刷新所有账号全部分页,再等待并重新导出。""" """两阶段执行:先刷新所有账号全部分页,再等待并重新导出。
运行中途登录态失效(目标页被重定向到登录页)时,自动用账号密码重新
登录并重试当前账号一次;整套运行最多自动重登一次,避免死循环。
"""
refresh_summaries = [] refresh_summaries = []
relogin_state = {"attempted": False}
def _relogin_if_expired(exc: BaseException) -> bool:
"""登录态失效类异常时自动重登一次;返回是否已恢复登录。"""
if relogin_state["attempted"] or "登录态失效" not in str(exc):
return False
relogin_state["attempted"] = True
print("🔐 检测到登录态失效,尝试账号密码自动重新登录...")
try:
ok = login(driver)
except Exception as login_exc:
print(f"⚠️ 自动重新登录异常: {login_exc}")
return False
if ok:
print("✅ 自动重新登录成功,重试当前账号")
return bool(ok)
def _empty_summary(exc: BaseException) -> dict:
return {
"clicked": 0,
"pending": 0,
"zero_without_button": 0,
"unconfirmed": 0,
"pages": 0,
"error": str(exc),
}
for index, url in enumerate(urls, 1): for index, url in enumerate(urls, 1):
print(f"\n{'=' * 50}") print(f"\n{'=' * 50}")
print(f"🔄 刷新阶段 博主 {index}/{len(urls)}: {url}") print(f"🔄 刷新阶段 博主 {index}/{len(urls)}: {url}")
@@ -919,15 +1339,16 @@ def refresh_then_export_accounts(driver, urls, wait_seconds=REFRESH_WAIT_SECONDS
navigate_to_target(driver, url) navigate_to_target(driver, url)
summary = refresh_zero_exposure_videos(driver) summary = refresh_zero_exposure_videos(driver)
except Exception as exc: except Exception as exc:
print(f"⚠️ 博主 {index} 刷新扫描异常: {exc},继续下一个") if _relogin_if_expired(exc):
summary = { try:
"clicked": 0, navigate_to_target(driver, url)
"pending": 0, summary = refresh_zero_exposure_videos(driver)
"zero_without_button": 0, except Exception as retry_exc:
"unconfirmed": 0, print(f"⚠️ 博主 {index} 重登后刷新仍失败: {retry_exc},继续下一个")
"pages": 0, summary = _empty_summary(retry_exc)
"error": str(exc), else:
} print(f"⚠️ 博主 {index} 刷新扫描异常: {exc},继续下一个")
summary = _empty_summary(exc)
refresh_summaries.append(summary) refresh_summaries.append(summary)
should_wait = any( should_wait = any(
@@ -956,7 +1377,21 @@ def refresh_then_export_accounts(driver, urls, wait_seconds=REFRESH_WAIT_SECONDS
print(f" 解析出 {len(records)} 条视频记录") print(f" 解析出 {len(records)} 条视频记录")
all_records.extend(records) all_records.extend(records)
except Exception as exc: except Exception as exc:
print(f"⚠️ 博主 {index} 导出异常: {exc},跳过") if _relogin_if_expired(exc):
try:
navigate_to_target(driver, url)
pre_mtime = _max_excel_mtime()
excel_path = export_video_data(driver, pre_mtime)
if not excel_path:
print(f"⚠️ 博主 {index} 重登后导出失败,跳过")
continue
records, _ = parse_chanmama_export(excel_path)
print(f" 解析出 {len(records)} 条视频记录(重登后)")
all_records.extend(records)
except Exception as retry_exc:
print(f"⚠️ 博主 {index} 重登后导出仍失败: {retry_exc},跳过")
else:
print(f"⚠️ 博主 {index} 导出异常: {exc},跳过")
return all_records, refresh_summaries return all_records, refresh_summaries
@@ -978,6 +1413,10 @@ def export_video_data(driver, pre_mtime: float = 0):
except TimeoutException: except TimeoutException:
print("⚠️ 视频列表加载超时,继续尝试导出...") print("⚠️ 视频列表加载超时,继续尝试导出...")
# 刷新/翻页阶段可能延迟弹出公告/活动弹窗,吞掉导出点击或盖住确认框;
# 点导出按钮前先清一次(确认框受保护)。
_dismiss_export_stage_overlays(driver)
# 方法1: 先在可见 DOM 中找真正的交互节点。新版按钮文字常在嵌套 # 方法1: 先在可见 DOM 中找真正的交互节点。新版按钮文字常在嵌套
# span/div 内,旧的 contains(text()) 和第一个隐藏 locator 都会漏掉它。 # span/div 内,旧的 contains(text()) 和第一个隐藏 locator 都会漏掉它。
export_clicked = False export_clicked = False
@@ -989,8 +1428,11 @@ def export_video_data(driver, pre_mtime: float = 0):
except TimeoutException: except TimeoutException:
control = None control = None
if control: if control:
clicked = driver.execute_script(_CLICK_EXPORT_CONTROL_SCRIPT) # 原生点击优先(可触发内层元素的 Vue @click),JS 合成点击兜底
export_clicked = bool(clicked) export_clicked = _native_click_export_button(driver)
if not export_clicked:
clicked = driver.execute_script(_CLICK_EXPORT_CONTROL_SCRIPT)
export_clicked = bool(clicked)
if export_clicked: if export_clicked:
print(f"✅ 已点击导出按钮: {control.get('text') or control.get('className')}") print(f"✅ 已点击导出按钮: {control.get('text') or control.get('className')}")
@@ -1032,36 +1474,38 @@ def export_video_data(driver, pre_mtime: float = 0):
return scrape_table_data(driver) return scrape_table_data(driver)
# 等待导出确认弹窗出现并点击【导出】按钮 # 等待导出确认弹窗出现并点击【导出】按钮
# 蝉妈妈弹窗标题"您即将进行数据导出",按钮为 橙色"导出"(文字在 <span> 内) # 蝉妈妈弹窗标题"您即将进行数据导出",按钮为 橙色"导出"(文字在 <span> 内)。
# 逐秒探测弹窗状态:可确认→立即点导出;挂着但没有可点的确认按钮
# (空壳遮罩/公告弹窗)→先 ESC+移除正常关,持续 5 秒以上就补点导出按钮
# (第一次点击可能已被弹窗吞掉);没有弹窗→常规清理并定期补点导出。
print("⏳ 等待导出确认弹窗...") print("⏳ 等待导出确认弹窗...")
export_confirmed = False export_confirmed = False
for wait_sec in range(20): unconfirmable_since: float | None = None
for wait_sec in range(36):
time.sleep(1) time.sleep(1)
try: state = _export_dialog_state(driver)
# 关键:用 contains(.,'导出') 匹配后代文本("导出"在 <span> 内) if state == "confirmable":
confirm_selectors = [ unconfirmable_since = None
# 精确:主按钮(primary) + 含"导出"文字 if _click_export_confirm(driver):
"//div[contains(@class,'el-dialog') or @role='dialog']//button[contains(.,'导出')]", export_confirmed = True
# 弹窗内任意含"导出"的按钮
"//div[contains(@class,'el-dialog') or @role='dialog']//button[contains(@class,'el-button--primary')][contains(.,'导出')]",
# 兼容"确定"/"确认"
"//div[contains(@class,'el-dialog') or @role='dialog']//button[contains(.,'确定')]",
"//div[contains(@class,'el-dialog') or @role='dialog']//button[contains(.,'确认')]",
]
for sel in confirm_selectors:
try:
btn = driver.find_element(By.XPATH, sel)
if btn.is_displayed():
driver.execute_script("arguments[0].click();", btn)
export_confirmed = True
print(f"✅ 已在弹窗中点击导出按钮: {sel}")
break
except NoSuchElementException:
continue
if export_confirmed:
break break
except Exception:
continue continue
if state == "present":
now = time.monotonic()
if unconfirmable_since is None:
_log_export_dialog_texts(driver)
print("ℹ️ 弹窗可见但没有可点的确认按钮,尝试关闭")
unconfirmable_since = now
elif now - unconfirmable_since > 5:
if _reclick_export_control(driver):
print("🔁 清掉卡死弹窗后补点导出按钮")
unconfirmable_since = None
_dismiss_export_stage_overlays(driver)
continue
unconfirmable_since = None
_dismiss_export_stage_overlays(driver)
if wait_sec in (10, 20, 30) and _reclick_export_control(driver):
print("🔁 确认框迟迟未出现,补点导出按钮")
if not export_confirmed: if not export_confirmed:
try: try:
@@ -1093,9 +1537,20 @@ def export_video_data(driver, pre_mtime: float = 0):
"⚠️ 未找到本次下载的 Excel文件(等待 " "⚠️ 未找到本次下载的 Excel文件(等待 "
f"{CHANMAMA_EXPORT_WAIT_SECONDS + max(CHANMAMA_EXPORT_GRACE_SECONDS, 0)} 秒超时)" f"{CHANMAMA_EXPORT_WAIT_SECONDS + max(CHANMAMA_EXPORT_GRACE_SECONDS, 0)} 秒超时)"
) )
_dump_export_failure(driver)
return None return None
def _download_complete(path: str) -> bool:
"""判断下载文件是否已写完整(zip EOCD 在文件尾,截断时 is_zipfile 为假)。"""
import zipfile
try:
return zipfile.is_zipfile(path)
except OSError:
return False
def _wait_for_exported_excel( def _wait_for_exported_excel(
pre_mtime: float, pre_mtime: float,
*, *,
@@ -1125,7 +1580,13 @@ def _wait_for_exported_excel(
except Exception: except Exception:
pass pass
new_path = _find_new_excel_after(pre_mtime) new_path = _find_new_excel_after(pre_mtime)
if new_path is not None: # 文件出现 ≠ 下载完成: 当前浏览器可能以最终名边下边写,截断的
# zip 会让解析报 "File is not a zip file"。zip 的 EOCD 记录在
# 文件末尾,is_zipfile 通过即认为下载已完整;路径不存在(测试桩)
# 时保持旧行为直接返回。
if new_path is not None and (
not os.path.exists(new_path) or _download_complete(new_path)
):
return new_path return new_path
now = time.time() now = time.time()
if now >= deadline and not grace_logged and grace_deadline > deadline: if now >= deadline and not grace_logged and grace_deadline > deadline:
@@ -137,6 +137,7 @@ class ScraplingDriver:
cdp_url: str | None = None, cdp_url: str | None = None,
stealthy: bool = False, stealthy: bool = False,
real_chrome: bool = True, real_chrome: bool = True,
extra_flags: list[str] | None = None,
) -> ScraplingDriver: ) -> ScraplingDriver:
options: dict[str, Any] = { options: dict[str, Any] = {
"headless": False, "headless": False,
@@ -152,6 +153,7 @@ class ScraplingDriver:
"extra_flags": [ "extra_flags": [
"--disable-blink-features=AutomationControlled", "--disable-blink-features=AutomationControlled",
"--disable-dev-shm-usage", "--disable-dev-shm-usage",
*(extra_flags or []),
], ],
} }
reuse_existing_context = bool(cdp_url) reuse_existing_context = bool(cdp_url)
@@ -211,6 +213,9 @@ class ScraplingDriver:
[script, raw_arguments], [script, raw_arguments],
) )
def save_screenshot(self, path: str | os.PathLike[str]) -> bool:
return bool(self._page.screenshot(path=str(Path(path))))
def get_cookies(self) -> list[dict[str, Any]]: def get_cookies(self) -> list[dict[str, Any]]:
context = self._owner.context context = self._owner.context
return list(context.cookies()) if context is not None else [] return list(context.cookies()) if context is not None else []
@@ -1,4 +1,5 @@
from datetime import datetime from datetime import datetime
from pathlib import Path
from types import SimpleNamespace from types import SimpleNamespace
import pytest import pytest
@@ -723,3 +724,168 @@ def test_wait_for_exported_excel_tolerates_driver_errors(monkeypatch):
assert chanmama._wait_for_exported_excel(0.0, driver=_BrokenDriver()) == ( assert chanmama._wait_for_exported_excel(0.0, driver=_BrokenDriver()) == (
"/downloads/已有.xlsx" "/downloads/已有.xlsx"
) )
def test_refresh_then_export_relogin_on_expired_session(monkeypatch):
"""运行中途登录态失效: 自动账密重登一次并重试当前博主,不影响其他博主。"""
events = []
navigate_calls = []
def fake_navigate(_driver, url):
navigate_calls.append(url)
if len(navigate_calls) == 1:
raise RuntimeError(
"蝉妈妈登录态失效,目标页被重定向到: "
"https://www.chanmama.com/login.html"
)
monkeypatch.setattr(chanmama, "navigate_to_target", fake_navigate)
monkeypatch.setattr(
chanmama,
"refresh_zero_exposure_videos",
lambda _driver: {
"clicked": 0, "pending": 0, "zero_without_button": 0,
"unconfirmed": 0, "pages": 1,
},
)
monkeypatch.setattr(
chanmama, "login",
lambda _driver: events.append("login") or True,
)
monkeypatch.setattr(chanmama.time, "sleep", lambda _seconds: None)
monkeypatch.setattr(chanmama, "_max_excel_mtime", lambda: 0.0)
monkeypatch.setattr(
chanmama, "export_video_data", lambda _driver, mtime: "x.xlsx"
)
monkeypatch.setattr(
chanmama, "parse_chanmama_excel", lambda _path: ([{"title": "v"}], None)
)
records, summaries = chanmama.refresh_then_export_accounts(
object(), ["account-1", "account-2"], wait_seconds=0
)
assert events == ["login"] # 整套运行只自动重登一次
assert len(records) == 2 # 两个博主最终都导出成功
assert summaries[0]["pages"] == 1 # 博主 1 重登后刷新成功
def test_refresh_then_export_no_relogin_for_other_errors(monkeypatch):
"""非登录态失效的异常不触发自动重登。"""
events = []
def fake_navigate(_driver, url):
raise RuntimeError("蝉妈妈目标页加载失败: timeout")
monkeypatch.setattr(chanmama, "navigate_to_target", fake_navigate)
monkeypatch.setattr(
chanmama, "login",
lambda _driver: events.append("login") or True,
)
monkeypatch.setattr(chanmama.time, "sleep", lambda _seconds: None)
monkeypatch.setattr(
chanmama, "refresh_zero_exposure_videos", lambda _driver: {}
)
monkeypatch.setattr(chanmama, "_max_excel_mtime", lambda: 0.0)
monkeypatch.setattr(
chanmama, "export_video_data", lambda _driver, mtime: None
)
records, _summaries = chanmama.refresh_then_export_accounts(
object(), ["account-1"], wait_seconds=0
)
assert events == [] # 普通异常不触发自动重登
assert records == [] # 导出失败不产出记录
class _RenderFakeDriver:
"""navigate_to_target 渲染链路的桩:列表只在浮层清理后才就绪。"""
def __init__(self, ready_after_dismiss=1):
self.current_url = (
"https://www.chanmama.com/bloggerRank/x.html?activeTab=aweme"
)
self.visited = []
self.dismiss_runs = 0
self.ready_after_dismiss = ready_after_dismiss
self.screenshots = []
def get(self, url):
self.visited.append(url)
def find_element(self, _by, _value):
return object()
def execute_script(self, script, *_args):
if script == chanmama._OVERLAY_DISMISS_SCRIPT:
self.dismiss_runs += 1
return 2
if script == chanmama._VIDEO_LIST_READY_SCRIPT:
return self.dismiss_runs >= self.ready_after_dismiss
raise AssertionError(f"unexpected script: {script[:60]}")
def save_screenshot(self, path):
self.screenshots.append(str(path))
Path(path).write_bytes(b"png")
return True
@property
def page_source(self):
return "<html>diag</html>"
def test_navigate_to_target_clears_overlays_before_render_wait(monkeypatch):
monkeypatch.setattr(chanmama.time, "sleep", lambda _seconds: None)
driver = _RenderFakeDriver(ready_after_dismiss=1)
assert chanmama.navigate_to_target(
driver, "https://www.chanmama.com/bloggerRank/x.html"
) is True
assert driver.dismiss_runs >= 1 # 弹窗清理后才轮到渲染判定
assert driver.visited == ["https://www.chanmama.com/bloggerRank/x.html"]
def test_navigate_to_target_timeout_saves_diagnostics(monkeypatch, tmp_path):
monkeypatch.setattr(chanmama.time, "sleep", lambda _seconds: None)
monkeypatch.setattr(chanmama, "CHANMAMA_LIST_READY_TIMEOUT_SECONDS", 0.05)
monkeypatch.setattr(chanmama, "DOWNLOAD_DIR", str(tmp_path))
driver = _RenderFakeDriver(ready_after_dismiss=99)
with pytest.raises(RuntimeError, match="蝉妈妈目标页加载失败"):
chanmama.navigate_to_target(
driver, "https://www.chanmama.com/bloggerRank/x.html"
)
assert driver.dismiss_runs >= 2 # 两次尝试各自清理过浮层
assert len(driver.screenshots) == 1
diag_dir = tmp_path / "_diag"
assert list(diag_dir.glob("render_failure_*.png"))
assert list(diag_dir.glob("render_failure_*.html"))
def test_dismiss_overlays_ignores_driver_errors():
class BoomDriver:
def execute_script(self, _script, *_args):
raise RuntimeError("execution context destroyed")
assert chanmama._dismiss_overlays(BoomDriver()) == 0
def test_wait_video_list_ready_tolerates_transient_script_errors(monkeypatch):
monkeypatch.setattr(chanmama.time, "sleep", lambda _seconds: None)
class FlakyDriver:
def __init__(self):
self.ready_calls = 0
def execute_script(self, script, *_args):
if script == chanmama._OVERLAY_DISMISS_SCRIPT:
return 0
self.ready_calls += 1
if self.ready_calls == 1:
raise RuntimeError("frame detached")
return True
chanmama._wait_video_list_ready(FlakyDriver(), 2) # 不抛即通过