fix(content): 蝉妈妈导出标签页崩溃自愈与 cookie 文件格式对齐

导出点击后 tab 死亡(渲染进程被清理)时:页面级死亡先同 context 重开
新页(facade 新增 reopen_page),context 也死亡才整套重建驱动,按
profile → cookie 文件 → 账密顺序恢复会话,两种自愈各限一次防死循环。
cookie 文件改存裸列表,与 vault 同步及适配器 BrowserCookieStore 三方
格式对齐(字典格式会被适配器拒绝加载);adapters 暴露
is_browser_closed_error 供模块识别目标已关闭类异常。
This commit is contained in:
wangyunlong committed 2026-10-14 18:35:44 +08:00
1 parent 6cd563eab0
commit 7b305b0b12
4 files changed
+325 -42

No files matched your search

+3 -3
View File
@@ -37,8 +37,8 @@ def is_browser_timeout_error(error: BaseException) -> bool:
) )
def _is_browser_closed_error(error: BaseException) -> bool: def is_browser_closed_error(error: BaseException) -> bool:
"""Recognize a browser target that disappeared during cleanup.""" """Recognize a browser target that disappeared (page/context/browser)."""
if any( if any(
error_type.__name__ == "TargetClosedError" error_type.__name__ == "TargetClosedError"
@@ -155,7 +155,7 @@ class ScraplingBrowser:
# and context-manager cleanup. The current state cannot # and context-manager cleanup. The current state cannot
# be exported then, but that must not turn a successful # be exported then, but that must not turn a successful
# download into a false failure or mask the action error. # download into a false failure or mask the action error.
if not _is_browser_closed_error(exc): if not is_browser_closed_error(exc):
raise raise
finally: finally:
try: try:
@@ -16,6 +16,7 @@ import time
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from gyxx_flow.adapters.scrapling import is_browser_closed_error
from gyxx_flow.core.exit_codes import COOKIE_SKIP_EXIT_CODE from gyxx_flow.core.exit_codes import COOKIE_SKIP_EXIT_CODE
from gyxx_flow.modules.content_marketing import chanmama_export from gyxx_flow.modules.content_marketing import chanmama_export
from gyxx_flow.modules.content_marketing.runtime_paths import PATHS from gyxx_flow.modules.content_marketing.runtime_paths import PATHS
@@ -141,17 +142,18 @@ def create_driver():
def save_cookies(driver): def save_cookies(driver):
"""保存 cookie 到文件""" """保存 cookie 到文件。
写裸列表格式,与账号 vault 同步(sync_binding_state)和适配器
BrowserCookieStore 的绑定 cookie 文件格式保持一致 —— 同一文件
被三方读写,字典格式会让适配器下次启动/重建时直接拒绝加载。
"""
try: try:
cookies = driver.get_cookies() cookies = driver.get_cookies()
data = {
"cookies": cookies,
"timestamp": time.time(),
}
cookie_path = Path(COOKIE_FILE) cookie_path = Path(COOKIE_FILE)
cookie_path.parent.mkdir(parents=True, exist_ok=True) cookie_path.parent.mkdir(parents=True, exist_ok=True)
temp_path = cookie_path.with_name(f".{cookie_path.name}.tmp") temp_path = cookie_path.with_name(f".{cookie_path.name}.tmp")
temp_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") temp_path.write_text(json.dumps(cookies, ensure_ascii=False, indent=2), encoding="utf-8")
os.replace(temp_path, cookie_path) os.replace(temp_path, cookie_path)
print(f"💾 已保存 cookie ({len(cookies)} 条) 到: {COOKIE_FILE}") print(f"💾 已保存 cookie ({len(cookies)} 条) 到: {COOKIE_FILE}")
return True return True
@@ -163,6 +165,9 @@ def save_cookies(driver):
def load_cookies(driver): def load_cookies(driver):
"""从文件加载 cookie 到浏览器 """从文件加载 cookie 到浏览器
兼容两种格式: 裸列表(vault/适配器格式,按文件 mtime 判断时效)与
旧字典格式({"cookies": [...], "timestamp": ...})。
Returns: Returns:
bool: cookie 是否有效(有效期内) bool: cookie 是否有效(有效期内)
""" """
@@ -171,16 +176,20 @@ def load_cookies(driver):
return False return False
try: try:
data = json.loads(Path(COOKIE_FILE).read_text(encoding="utf-8")) payload = json.loads(Path(COOKIE_FILE).read_text(encoding="utf-8"))
if isinstance(payload, list):
cookies = payload
timestamp = os.path.getmtime(COOKIE_FILE)
else:
cookies = payload.get("cookies", [])
timestamp = payload.get("timestamp", 0)
timestamp = data.get("timestamp", 0)
age = time.time() - timestamp age = time.time() - timestamp
if age > COOKIE_MAX_AGE: if age > COOKIE_MAX_AGE:
print(f"⏰ cookie 已过期 ({int(age // 86400)} 天前),需要重新登录") print(f"⏰ cookie 已过期 ({int(age // 86400)} 天前),需要重新登录")
return False return False
cookies = data.get("cookies", [])
if not cookies: if not cookies:
print("📭 cookie 为空,需要重新登录") print("📭 cookie 为空,需要重新登录")
return False return False
@@ -1301,10 +1310,15 @@ def refresh_then_export_accounts(driver, urls, wait_seconds=REFRESH_WAIT_SECONDS
"""两阶段执行:先刷新所有账号全部分页,再等待并重新导出。 """两阶段执行:先刷新所有账号全部分页,再等待并重新导出。
运行中途登录态失效(目标页被重定向到登录页)时,自动用账号密码重新 运行中途登录态失效(目标页被重定向到登录页)时,自动用账号密码重新
登录并重试当前账号一次;整套运行最多自动重登一次,避免死循环。 登录并重试当前账号一次;浏览器/页面被外部关闭(如进程树被清理)时,
自动重建浏览器并恢复会话后重试当前账号一次。两种自愈整套运行各最多
一次,避免死循环。返回 (记录列表, 刷新摘要, 最终生效的驱动) —— 驱动
可能在重建后已不是调用方传入的那个,调用方需用返回值收尾关闭。
""" """
refresh_summaries = [] refresh_summaries = []
relogin_state = {"attempted": False} relogin_state = {"attempted": False}
rebuild_state = {"attempted": False}
driver_box = {"driver": driver}
def _relogin_if_expired(exc: BaseException) -> bool: def _relogin_if_expired(exc: BaseException) -> bool:
"""登录态失效类异常时自动重登一次;返回是否已恢复登录。""" """登录态失效类异常时自动重登一次;返回是否已恢复登录。"""
@@ -1313,7 +1327,7 @@ def refresh_then_export_accounts(driver, urls, wait_seconds=REFRESH_WAIT_SECONDS
relogin_state["attempted"] = True relogin_state["attempted"] = True
print("🔐 检测到登录态失效,尝试账号密码自动重新登录...") print("🔐 检测到登录态失效,尝试账号密码自动重新登录...")
try: try:
ok = login(driver) ok = login(driver_box["driver"])
except Exception as login_exc: except Exception as login_exc:
print(f"⚠️ 自动重新登录异常: {login_exc}") print(f"⚠️ 自动重新登录异常: {login_exc}")
return False return False
@@ -1321,6 +1335,51 @@ def refresh_then_export_accounts(driver, urls, wait_seconds=REFRESH_WAIT_SECONDS
print("✅ 自动重新登录成功,重试当前账号") print("✅ 自动重新登录成功,重试当前账号")
return bool(ok) return bool(ok)
def _restore_session(current) -> bool:
"""重建浏览器后按 profile → cookie 文件 → 账密的顺序恢复登录态。"""
try:
if verify_login(current):
return True
if load_cookies(current) and verify_login(current):
return True
return bool(login(current))
except Exception as exc:
print(f"⚠️ 重建后会话恢复异常: {exc}")
return False
def _rebuild_if_browser_closed(exc: BaseException) -> bool:
"""浏览器/页面被外部关闭类异常时自愈一次;返回是否已恢复。
先尝试同一 context 重开新页(页面级死亡,登录态保留、代价最小);
context 也死了才整套重建驱动并按 profile → cookie → 账密恢复会话。
"""
if rebuild_state["attempted"] or not is_browser_closed_error(exc):
return False
rebuild_state["attempted"] = True
print("🌐 检测到浏览器页面被关闭,尝试自愈...")
current = driver_box["driver"]
try:
current.reopen_page()
print("✅ 浏览器 context 存活,已重开新页面,重试当前账号")
return True
except Exception as reopen_exc:
print(f"ℹ️ 页面重开失败({reopen_exc}),改为整套重建浏览器...")
try:
current.quit()
except Exception:
# 旧驱动随浏览器一起死亡,清理失败不影响重建。
pass
try:
driver_box["driver"] = create_driver()
except Exception as create_exc:
print(f"⚠️ 浏览器重建失败: {create_exc}")
return False
if not _restore_session(driver_box["driver"]):
print("⚠️ 重建后会话恢复失败,放弃自动重试")
return False
print("✅ 浏览器重建完成,重试当前账号")
return True
def _empty_summary(exc: BaseException) -> dict: def _empty_summary(exc: BaseException) -> dict:
return { return {
"clicked": 0, "clicked": 0,
@@ -1336,15 +1395,15 @@ def refresh_then_export_accounts(driver, urls, wait_seconds=REFRESH_WAIT_SECONDS
print(f"🔄 刷新阶段 博主 {index}/{len(urls)}: {url}") print(f"🔄 刷新阶段 博主 {index}/{len(urls)}: {url}")
print(f"{'=' * 50}") print(f"{'=' * 50}")
try: try:
navigate_to_target(driver, url) navigate_to_target(driver_box["driver"], url)
summary = refresh_zero_exposure_videos(driver) summary = refresh_zero_exposure_videos(driver_box["driver"])
except Exception as exc: except Exception as exc:
if _relogin_if_expired(exc): if _relogin_if_expired(exc) or _rebuild_if_browser_closed(exc):
try: try:
navigate_to_target(driver, url) navigate_to_target(driver_box["driver"], url)
summary = refresh_zero_exposure_videos(driver) summary = refresh_zero_exposure_videos(driver_box["driver"])
except Exception as retry_exc: except Exception as retry_exc:
print(f"⚠️ 博主 {index} 重登后刷新仍失败: {retry_exc},继续下一个") print(f"⚠️ 博主 {index} 自愈后刷新仍失败: {retry_exc},继续下一个")
summary = _empty_summary(retry_exc) summary = _empty_summary(retry_exc)
else: else:
print(f"⚠️ 博主 {index} 刷新扫描异常: {exc},继续下一个") print(f"⚠️ 博主 {index} 刷新扫描异常: {exc},继续下一个")
@@ -1367,9 +1426,9 @@ def refresh_then_export_accounts(driver, urls, wait_seconds=REFRESH_WAIT_SECONDS
print(f"📌 导出阶段 博主 {index}/{len(urls)}: {url}") print(f"📌 导出阶段 博主 {index}/{len(urls)}: {url}")
print(f"{'=' * 50}") print(f"{'=' * 50}")
try: try:
navigate_to_target(driver, url) navigate_to_target(driver_box["driver"], url)
pre_mtime = _max_excel_mtime() pre_mtime = _max_excel_mtime()
excel_path = export_video_data(driver, pre_mtime) excel_path = export_video_data(driver_box["driver"], pre_mtime)
if not excel_path: if not excel_path:
print(f"⚠️ 博主 {index} 导出失败,跳过") print(f"⚠️ 博主 {index} 导出失败,跳过")
continue continue
@@ -1377,23 +1436,23 @@ 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:
if _relogin_if_expired(exc): if _relogin_if_expired(exc) or _rebuild_if_browser_closed(exc):
try: try:
navigate_to_target(driver, url) navigate_to_target(driver_box["driver"], url)
pre_mtime = _max_excel_mtime() pre_mtime = _max_excel_mtime()
excel_path = export_video_data(driver, pre_mtime) excel_path = export_video_data(driver_box["driver"], pre_mtime)
if not excel_path: if not excel_path:
print(f"⚠️ 博主 {index} 重登后导出失败,跳过") print(f"⚠️ 博主 {index} 自愈后导出失败,跳过")
continue continue
records, _ = parse_chanmama_export(excel_path) records, _ = parse_chanmama_export(excel_path)
print(f" 解析出 {len(records)} 条视频记录(重登后)") print(f" 解析出 {len(records)} 条视频记录(自愈后)")
all_records.extend(records) all_records.extend(records)
except Exception as retry_exc: except Exception as retry_exc:
print(f"⚠️ 博主 {index} 重登后导出仍失败: {retry_exc},跳过") print(f"⚠️ 博主 {index} 自愈后导出仍失败: {retry_exc},跳过")
else: else:
print(f"⚠️ 博主 {index} 导出异常: {exc},跳过") print(f"⚠️ 博主 {index} 导出异常: {exc},跳过")
return all_records, refresh_summaries return all_records, refresh_summaries, driver_box["driver"]
def export_video_data(driver, pre_mtime: float = 0): def export_video_data(driver, pre_mtime: float = 0):
@@ -1569,6 +1628,8 @@ def _wait_for_exported_excel(
deadline 之后落盘的竞态;两个窗口内都未出现则返回 None。 deadline 之后落盘的竞态;两个窗口内都未出现则返回 None。
传入 driver 时每轮执行一次无害的 evaluate 以泵 Playwright 事件循环, 传入 driver 时每轮执行一次无害的 evaluate 以泵 Playwright 事件循环,
确保 download 事件处理器(_save_download)在 sleep 轮询期间也能执行。 确保 download 事件处理器(_save_download)在 sleep 轮询期间也能执行。
泵调用发现浏览器/页面已被外部关闭时,若文件已完整落盘则照常返回,
否则抛出原始关闭异常交由上层重建浏览器重试,不再空等满超时。
""" """
deadline = time.time() + max(timeout_seconds, 0) deadline = time.time() + max(timeout_seconds, 0)
grace_deadline = deadline + max(grace_seconds, 0) grace_deadline = deadline + max(grace_seconds, 0)
@@ -1577,8 +1638,17 @@ def _wait_for_exported_excel(
if driver is not None: if driver is not None:
try: try:
driver.execute_script("1") driver.execute_script("1")
except Exception: except Exception as pump_exc:
pass if is_browser_closed_error(pump_exc):
# 浏览器/页面被外部关闭(进程树清理等): 先看文件是否
# 已在关闭前完整落盘,再决定放弃,避免误杀已完成下载。
settled = _find_new_excel_after(pre_mtime)
if settled is not None and (
not os.path.exists(settled) or _download_complete(settled)
):
return settled
raise
# 其余瞬态异常(页面导航中等)不致命,继续轮询文件系统。
new_path = _find_new_excel_after(pre_mtime) new_path = _find_new_excel_after(pre_mtime)
# 文件出现 ≠ 下载完成: 当前浏览器可能以最终名边下边写,截断的 # 文件出现 ≠ 下载完成: 当前浏览器可能以最终名边下边写,截断的
# zip 会让解析报 "File is not a zip file"。zip 的 EOCD 记录在 # zip 会让解析报 "File is not a zip file"。zip 的 EOCD 记录在
@@ -2290,7 +2360,8 @@ def main():
print("⚠️ 登录可能未成功,但继续尝试后续步骤...") print("⚠️ 登录可能未成功,但继续尝试后续步骤...")
# 4. 先刷新两个账号的全部分页,再统一等待并重新进入页面导出。 # 4. 先刷新两个账号的全部分页,再统一等待并重新进入页面导出。
all_records, _refresh_summaries = refresh_then_export_accounts( # 返回的驱动可能已在浏览器被外部关闭后重建,后续收尾必须用它。
all_records, _refresh_summaries, driver = refresh_then_export_accounts(
driver, TARGET_URLS, wait_seconds=REFRESH_WAIT_SECONDS driver, TARGET_URLS, wait_seconds=REFRESH_WAIT_SECONDS
) )
@@ -229,6 +229,24 @@ class ScraplingDriver:
def quit(self) -> None: def quit(self) -> None:
self._owner.close() self._owner.close()
def reopen_page(self) -> Any:
"""Rebind to a fresh page inside the still-live context.
A page/tab can die (site closed it, renderer gone) while the browser
context survives with its login state. Returns the new page, or
raises the underlying error when the context/browser is gone too —
the caller should rebuild the whole driver then.
"""
context = self._owner.context
if context is None:
raise RuntimeError("Scrapling browser context is closed")
# 永远开新页: context.pages 里可能还挂着半死的目标页。
page = context.new_page()
self._page = page
self._owner.page = page
page.on("download", self._save_download)
return page
def _save_download(self, download: Any) -> None: def _save_download(self, download: Any) -> None:
target = self._download_dir / download.suggested_filename target = self._download_dir / download.suggested_filename
download.save_as(str(target)) download.save_as(str(target))
@@ -1,3 +1,5 @@
import json
import time
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from types import SimpleNamespace from types import SimpleNamespace
@@ -33,7 +35,7 @@ def test_refresh_then_export_waits_once_after_all_accounts(monkeypatch):
monkeypatch.setattr(chanmama, "export_video_data", lambda _driver, mtime: events.append(("export", mtime)) or "x.xlsx") monkeypatch.setattr(chanmama, "export_video_data", lambda _driver, mtime: events.append(("export", mtime)) or "x.xlsx")
monkeypatch.setattr(chanmama, "parse_chanmama_excel", lambda _path: ([{"title": "video"}], None)) monkeypatch.setattr(chanmama, "parse_chanmama_excel", lambda _path: ([{"title": "video"}], None))
records, summaries = chanmama.refresh_then_export_accounts(object(), urls, wait_seconds=600) records, summaries, _driver = chanmama.refresh_then_export_accounts(object(), urls, wait_seconds=600)
assert len(records) == 2 assert len(records) == 2
assert [item["clicked"] for item in summaries] == [2, 0] assert [item["clicked"] for item in summaries] == [2, 0]
@@ -54,7 +56,7 @@ def test_refresh_then_export_skips_wait_when_nothing_needs_refresh(monkeypatch):
monkeypatch.setattr(chanmama, "_max_excel_mtime", lambda: 0.0) monkeypatch.setattr(chanmama, "_max_excel_mtime", lambda: 0.0)
monkeypatch.setattr(chanmama, "export_video_data", lambda *_args: None) monkeypatch.setattr(chanmama, "export_video_data", lambda *_args: None)
records, _ = chanmama.refresh_then_export_accounts(object(), ["account-1"], wait_seconds=600) records, _, _driver = chanmama.refresh_then_export_accounts(object(), ["account-1"], wait_seconds=600)
assert records == [] assert records == []
assert sleeps == [] assert sleeps == []
@@ -159,7 +161,6 @@ def test_missing_get_data_button_does_not_trigger_refresh_wait(monkeypatch):
chanmama.refresh_then_export_accounts(object(), ["account-1"], wait_seconds=600) chanmama.refresh_then_export_accounts(object(), ["account-1"], wait_seconds=600)
assert sleeps == [] assert sleeps == []
def test_create_driver_uses_runtime_cdp_profile_and_download_bindings(monkeypatch, tmp_path): def test_create_driver_uses_runtime_cdp_profile_and_download_bindings(monkeypatch, tmp_path):
calls = [] calls = []
expected = object() expected = object()
@@ -183,7 +184,9 @@ def test_create_driver_uses_runtime_cdp_profile_and_download_bindings(monkeypatc
"profile_dir": str(tmp_path / "profile"), "profile_dir": str(tmp_path / "profile"),
"download_dir": str(tmp_path / "downloads"), "download_dir": str(tmp_path / "downloads"),
"cdp_url": "http://127.0.0.1:22002", "cdp_url": "http://127.0.0.1:22002",
"stealthy": True, "stealthy": False,
"real_chrome": True,
"extra_flags": ["--no-proxy-server"],
}] }]
@@ -206,8 +209,10 @@ def test_create_driver_starts_scrapling_without_cdp(monkeypatch, tmp_path):
assert chanmama.create_driver() is expected assert chanmama.create_driver() is expected
assert calls[0]["cdp_url"] is None assert calls[0]["cdp_url"] is None
assert calls[0]["stealthy"] is True # 蝉妈妈 WAF 拉黑代理出口且压制 stealthy 补丁,必须真 Chrome + 直连。
assert calls[0]["real_chrome"] is False assert calls[0]["stealthy"] is False
assert calls[0]["real_chrome"] is True
assert calls[0]["extra_flags"] == ["--no-proxy-server"]
def test_create_driver_falls_back_when_bound_cdp_is_not_listening( def test_create_driver_falls_back_when_bound_cdp_is_not_listening(
@@ -241,8 +246,8 @@ def test_create_driver_falls_back_when_bound_cdp_is_not_listening(
"http://127.0.0.1:22002", "http://127.0.0.1:22002",
None, None,
] ]
assert all(call["stealthy"] is True for call in calls) assert all(call["stealthy"] is False for call in calls)
assert calls[1]["real_chrome"] is False assert all(call["real_chrome"] is True for call in calls)
def test_create_driver_keeps_explicit_remote_cdp_fail_closed( def test_create_driver_keeps_explicit_remote_cdp_fail_closed(
@@ -761,7 +766,7 @@ def test_refresh_then_export_relogin_on_expired_session(monkeypatch):
chanmama, "parse_chanmama_excel", lambda _path: ([{"title": "v"}], None) chanmama, "parse_chanmama_excel", lambda _path: ([{"title": "v"}], None)
) )
records, summaries = chanmama.refresh_then_export_accounts( records, summaries, _driver = chanmama.refresh_then_export_accounts(
object(), ["account-1", "account-2"], wait_seconds=0 object(), ["account-1", "account-2"], wait_seconds=0
) )
@@ -791,7 +796,7 @@ def test_refresh_then_export_no_relogin_for_other_errors(monkeypatch):
chanmama, "export_video_data", lambda _driver, mtime: None chanmama, "export_video_data", lambda _driver, mtime: None
) )
records, _summaries = chanmama.refresh_then_export_accounts( records, _summaries, _driver = chanmama.refresh_then_export_accounts(
object(), ["account-1"], wait_seconds=0 object(), ["account-1"], wait_seconds=0
) )
@@ -889,3 +894,192 @@ def test_wait_video_list_ready_tolerates_transient_script_errors(monkeypatch):
return True return True
chanmama._wait_video_list_ready(FlakyDriver(), 2) # 不抛即通过 chanmama._wait_video_list_ready(FlakyDriver(), 2) # 不抛即通过
def test_wait_for_exported_excel_raises_when_browser_closed_without_file(monkeypatch):
"""浏览器被外部关闭且文件未落盘: 立刻抛出关闭异常,不再空等满超时。"""
class _DeadDriver:
def execute_script(self, _script, *_args):
raise RuntimeError(
"Page.goto: Target page, context or browser has been closed"
)
monkeypatch.setattr(chanmama.time, "time", lambda: 1000.0)
monkeypatch.setattr(
chanmama.time,
"sleep",
lambda _seconds: pytest.fail("浏览器已死不应继续轮询等待"),
)
monkeypatch.setattr(chanmama, "_find_new_excel_after", lambda _pre_mtime: None)
with pytest.raises(RuntimeError, match="closed"):
chanmama._wait_for_exported_excel(0.0, driver=_DeadDriver())
def test_wait_for_exported_excel_returns_file_if_landed_before_close(monkeypatch):
"""关闭前下载已完整落盘: 仍返回文件路径,不因浏览器死亡误报失败。"""
class _DeadDriver:
def execute_script(self, _script, *_args):
raise RuntimeError(
"Download.save_as: Target page, context or browser has been closed"
)
landed = None
def fake_find(_pre_mtime):
return landed
def fake_complete(path):
return True
monkeypatch.setattr(chanmama.time, "time", lambda: 1000.0)
monkeypatch.setattr(chanmama, "_find_new_excel_after", fake_find)
monkeypatch.setattr(chanmama, "_download_complete", fake_complete)
landed = "/downloads/关闭前已落盘.xlsx"
assert chanmama._wait_for_exported_excel(0.0, driver=_DeadDriver()) == landed
def test_refresh_then_export_rebuilds_browser_when_closed_mid_run(monkeypatch):
"""导出中途浏览器被外部关闭: 重建驱动+恢复会话后重试当前博主一次。"""
events = []
navigate_calls = []
dead_driver = object()
live_driver = object()
def fake_navigate(current, url):
navigate_calls.append((current, url))
if current is dead_driver and len(navigate_calls) == 3:
# 刷新阶段两个博主都成功后,导出阶段第一次导航即发现浏览器死亡
raise RuntimeError(
"蝉妈妈目标页加载失败: x.html; Page.goto: "
"Target page, context or browser has been closed"
)
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.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)
)
monkeypatch.setattr(chanmama, "verify_login", lambda _driver: True)
monkeypatch.setattr(
chanmama, "create_driver", lambda: events.append("rebuild") or live_driver
)
monkeypatch.setattr(
chanmama, "login",
lambda _driver: events.append("login") or True,
)
monkeypatch.setattr(
chanmama, "load_cookies",
lambda _driver: (_ for _ in ()).throw(
AssertionError("profile 直接恢复,不应走 cookie 文件")
),
)
class _DeadDriver:
def quit(self):
events.append("quit-dead")
dead_driver = _DeadDriver()
records, _summaries, final_driver = chanmama.refresh_then_export_accounts(
dead_driver, ["account-1", "account-2"], wait_seconds=0
)
assert events == ["quit-dead", "rebuild"] # 旧驱动清理+重建一次
assert final_driver is live_driver # 返回重建后的驱动供收尾关闭
assert len(records) == 2 # 两个博主最终都导出成功
# 第 3 次导航(导出博主 1)发现死亡,其后所有导航都在重建后的驱动上
assert all(current is live_driver for current, _ in navigate_calls[3:])
def test_refresh_then_export_rebuilds_only_once(monkeypatch):
"""环境持续杀浏览器时只重建一次,后续博主直接失败不再死循环。"""
rebuilds = []
def always_closed_navigate(_driver, _url):
raise RuntimeError(
"Page.goto: Target page, context or browser has been closed"
)
class _Driver:
def quit(self):
pass
monkeypatch.setattr(chanmama, "navigate_to_target", always_closed_navigate)
monkeypatch.setattr(
chanmama, "create_driver", lambda: rebuilds.append(1) or _Driver()
)
monkeypatch.setattr(chanmama, "verify_login", lambda _driver: True)
monkeypatch.setattr(chanmama.time, "sleep", lambda _seconds: None)
records, _summaries, _driver = chanmama.refresh_then_export_accounts(
_Driver(), ["account-1", "account-2"], wait_seconds=0
)
assert len(rebuilds) == 1
assert records == []
def test_save_cookies_writes_adapter_list_format(monkeypatch, tmp_path):
"""cookie 文件与 vault/适配器共用,必须写裸列表而不是字典。"""
cookie_file = tmp_path / "cookies.json"
monkeypatch.setattr(chanmama, "COOKIE_FILE", str(cookie_file))
class _Driver:
@staticmethod
def get_cookies():
return [{"name": "session", "domain": ".chanmama.com"}]
assert chanmama.save_cookies(_Driver()) is True
payload = json.loads(cookie_file.read_text(encoding="utf-8"))
assert payload == [{"name": "session", "domain": ".chanmama.com"}]
def test_load_cookies_accepts_list_and_legacy_dict(monkeypatch, tmp_path):
cookie_file = tmp_path / "cookies.json"
injected = []
class _Driver:
@staticmethod
def get(_url):
return None
@staticmethod
def add_cookie(cookie):
injected.append(cookie)
monkeypatch.setattr(chanmama, "COOKIE_FILE", str(cookie_file))
monkeypatch.setattr(chanmama.time, "sleep", lambda _seconds: None)
cookie_file.write_text(
json.dumps([{"name": "a", "domain": ".chanmama.com"}]),
encoding="utf-8",
)
assert chanmama.load_cookies(_Driver()) is True
assert [c["name"] for c in injected] == ["a"]
injected.clear()
cookie_file.write_text(
json.dumps(
{"cookies": [{"name": "b", "domain": ".chanmama.com"}], "timestamp": time.time()}
),
encoding="utf-8",
)
assert chanmama.load_cookies(_Driver()) is True
assert [c["name"] for c in injected] == ["b"]