diff --git a/README.md b/README.md
index a5327c240..074bcdafc 100644
--- a/README.md
+++ b/README.md
@@ -166,6 +166,9 @@ uv run main.py --platform xhs --lt qrcode --type detail
# 打开对应APP扫二维码登录
+# 开始爬取前，先检查已保存的 cookie / 代理是否还能用（只发一次请求，不会启动爬虫）
+uv run main.py --platform xhs --check_session
+
# 其他平台爬虫使用示例，执行下面的命令查看
uv run main.py --help
```
diff --git a/README_en.md b/README_en.md
index 559e4326d..2fadcdf95 100644
--- a/README_en.md
+++ b/README_en.md
@@ -146,6 +146,9 @@ uv run main.py --platform xhs --lt qrcode --type detail
# Open corresponding APP to scan QR code for login
+# Before crawling, check whether the saved cookies/proxy still work (one request, no crawling)
+uv run main.py --platform xhs --check_session
+
# For other platform crawler usage examples, execute the following command to view
uv run main.py --help
```
diff --git a/README_es.md b/README_es.md
index db76c86eb..0d179bb30 100644
--- a/README_es.md
+++ b/README_es.md
@@ -146,6 +146,9 @@ uv run main.py --platform xhs --lt qrcode --type detail
# Abrir la APP correspondiente para escanear código QR para login
+# Antes de rastrear, compruebe si las cookies/proxy guardados siguen funcionando (una sola petición, sin rastreo)
+uv run main.py --platform xhs --check_session
+
# Para ejemplos de uso de rastreador de otras plataformas, ejecute el siguiente comando para ver
uv run main.py --help
```
diff --git a/cmd_arg/arg.py b/cmd_arg/arg.py
index 20d1d3976..1267b28f1 100644
--- a/cmd_arg/arg.py
+++ b/cmd_arg/arg.py
@@ -251,6 +251,14 @@ def main(
rich_help_panel="Account Configuration",
),
] = config.COOKIES,
+ check_session: Annotated[
+ bool,
+ typer.Option(
+ "--check_session",
+ help="Check whether the saved cookies/proxy still work, then exit without crawling",
+ rich_help_panel="Runtime Configuration",
+ ),
+ ] = False,
specified_id: Annotated[
str,
typer.Option(
@@ -412,6 +420,7 @@ def main(
headless=config.HEADLESS,
save_data_option=config.SAVE_DATA_OPTION,
init_db=init_db_value,
+ check_session=check_session,
cookies=config.COOKIES,
specified_id=specified_id,
creator_id=creator_id,
diff --git "a/docs/\345\270\270\350\247\201\351\227\256\351\242\230.md" "b/docs/\345\270\270\350\247\201\351\227\256\351\242\230.md"
index cfc03192d..b248419a2 100644
--- "a/docs/\345\270\270\350\247\201\351\227\256\351\242\230.md"
+++ "b/docs/\345\270\270\350\247\201\351\227\256\351\242\230.md"
@@ -24,6 +24,11 @@ A：在config/base_config.py 中 XHS_SPECIFIED_ID_LIST 参数用于控制需要
Q: 刚开始能爬取数据，过一段时间就是失效了？
A：出现这种情况多半是由于你的账号触发了平台风控机制了，❗️❗️请勿大规模对平台进行爬虫，影响平台。
+## 如何在开始爬取前确认登录态还在
+Q: 每次都要跑起来才发现 cookie 过期 / 代理不通，有没有更快的办法？
+A：执行 `uv run main.py --platform xhs --check_session`。它只发一次登录态请求就退出，并告诉你失败的可能原因（cookie 失效、代理不可用、访问受限等）。全部通过时退出码为 0，有失败时为 1，方便定时任务判断。
+注意：抖音(dy)与百度贴吧(tieba)的登录态只能在浏览器里确认，预检对这两个平台只能给出 cookie 层面的线索，结果为“未知”。
+
## 如何更换另一个账号
Q: 如何更换登录账号？
A：删除项目根目录下的 brower_data/ 文件夹即可
diff --git a/main.py b/main.py
index 2823a5f5e..1515aa9af 100644
--- a/main.py
+++ b/main.py
@@ -44,6 +44,7 @@
from media_platform.xhs import XiaoHongShuCrawler
from media_platform.zhihu import ZhihuCrawler
from tools.async_file_writer import AsyncFileWriter
+from tools.session_check import run_session_check
from var import crawler_type_var
@@ -101,6 +102,13 @@ async def main() -> None:
global crawler
args = await cmd_arg.parse_cmd()
+
+ # 预检只做一次登录态请求就退出，不建表、不启动爬虫。
+ # 放在 --init_db 之前：两个都传时，--init_db 的 return 会让预检被静默跳过。
+ if args.check_session:
+ exit_code = await run_session_check([config.PLATFORM])
+ raise SystemExit(exit_code)
+
if args.init_db:
await db.init_db(args.init_db)
print(f"Database {args.init_db} initialized successfully.")
diff --git a/tests/test_session_check.py b/tests/test_session_check.py
new file mode 100644
index 000000000..767230f5e
--- /dev/null
+++ b/tests/test_session_check.py
@@ -0,0 +1,438 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) 2025 relakkes@gmail.com
+#
+# This file is part of MediaCrawler project.
+# Repository: https://github.com/NanmiCoder/MediaCrawler/blob/main/tests/test_session_check.py
+# GitHub: https://github.com/NanmiCoder
+# Licensed under NON-COMMERCIAL LEARNING LICENSE 1.1
+#
+# 声明：本代码仅供学习和研究目的使用。使用者应遵守以下原则：
+# 1. 不得用于任何商业用途。
+# 2. 使用时应遵守目标平台的使用条款和robots.txt规则。
+# 3. 不得进行大规模爬取或对平台造成运营干扰。
+# 4. 应合理控制请求频率，避免给目标平台带来不必要的负担。
+# 5. 不得用于任何非法或不当的用途。
+#
+# 详细许可条款请参阅项目根目录下的LICENSE文件。
+# 使用本代码即表示您同意遵守上述原则和LICENSE中的所有条款。
+
+"""
+会话预检 (--check_session) 的单元测试。
+
+全部离线：探测那一步 (_probe) 被 monkeypatch 掉，不会构造真实 client，也不会
+发出任何请求。测试关注两件事：判定结果，以及失败原因的归类。
+"""
+
+import asyncio
+
+import httpx
+import pytest
+from tenacity import AsyncRetrying, RetryError, stop_after_attempt
+
+import config
+from tools import session_check
+from tools.session_check import (
+ CAUSE_BLOCKED,
+ CAUSE_BROWSER_REQUIRED,
+ CAUSE_EXPIRED,
+ CAUSE_MISSING_COOKIE_KEYS,
+ CAUSE_NETWORK,
+ CAUSE_NO_SESSION,
+ CAUSE_PROXY,
+ CAUSE_TIMEOUT,
+ CAUSE_UNEXPECTED,
+ STATUS_FAILED,
+ STATUS_OK,
+ STATUS_UNKNOWN,
+ check_platform_session,
+ run_session_check,
+)
+
+# 满足各平台 REQUIRED_COOKIE_KEYS 的最小 cookie
+COOKIES = {
+ "xhs": "a1=abc;web_session=abc",
+ "bili": "SESSDATA=abc",
+ "wb": "SUB=abc",
+ "zhihu": "d_c0=abc;z_c0=abc",
+ "ks": "did=web_abc",
+}
+
+
+class PlatformAccessError(Exception):
+ """名字与各平台 exception.py 中的一致，_classify_exception 按类名归类。"""
+
+
+class DataFetchError(Exception):
+ pass
+
+
+@pytest.fixture(autouse=True)
+def _offline(monkeypatch):
+ """默认关闭代理，并让构造 client 这一步不做任何事。"""
+
+ monkeypatch.setattr(config, "ENABLE_IP_PROXY", False)
+ monkeypatch.setattr(
+ session_check, "_build_client", lambda platform, cookie_str, httpx_proxy: object()
+ )
+
+
+def _probe_returns(value, delay=0.0):
+ async def _inner(platform, client):
+ if delay:
+ await asyncio.sleep(delay)
+ return value
+
+ return _inner
+
+
+def _probe_raises(exc, delay=0.0):
+ async def _inner(platform, client):
+ if delay:
+ await asyncio.sleep(delay)
+ raise exc
+
+ return _inner
+
+
+def _install_probe(monkeypatch, probe):
+ calls = []
+
+ async def _counting(platform, client):
+ calls.append(platform)
+ return await probe(platform, client)
+
+ monkeypatch.setattr(session_check, "_probe", _counting)
+ return calls
+
+
+@pytest.mark.asyncio
+async def test_missing_session_fails_without_touching_the_network(monkeypatch):
+ def _boom(*args, **kwargs):
+ raise AssertionError("没有登录态时不应该构造 client")
+
+ monkeypatch.setattr(session_check, "_build_client", _boom)
+ monkeypatch.setattr(session_check, "_probe", _probe_raises(AssertionError("不应该发请求")))
+
+ result = await check_platform_session("bili", cookie_str=" ")
+
+ assert result.status == STATUS_FAILED
+ assert result.cause == CAUSE_NO_SESSION
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+ "platform,cookie_str,missing",
+ [
+ ("zhihu", "z_c0=abc", "d_c0"),
+ ("xhs", "web_session=abc", "a1"),
+ ],
+)
+async def test_cookies_missing_a_required_field_fail_early(monkeypatch, platform, cookie_str, missing):
+ """xhs 的签名缺 a1、zhihu 的 _pre_headers 缺 d_c0 都会在发请求前就抛异常。"""
+
+ monkeypatch.setattr(session_check, "_probe", _probe_raises(AssertionError("不应该发请求")))
+
+ result = await check_platform_session(platform, cookie_str=cookie_str)
+
+ assert result.status == STATUS_FAILED
+ assert result.cause == CAUSE_MISSING_COOKIE_KEYS
+ assert missing in result.detail
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("platform", sorted(COOKIES))
+async def test_live_session_passes_with_exactly_one_request(monkeypatch, platform):
+ calls = _install_probe(monkeypatch, _probe_returns(True))
+
+ result = await check_platform_session(platform, cookie_str=COOKIES[platform])
+
+ assert result.status == STATUS_OK
+ assert result.cause == ""
+ assert calls == [platform], "预检只能发一次请求"
+
+
+@pytest.mark.asyncio
+async def test_expired_session_is_reported_as_expired(monkeypatch):
+ _install_probe(monkeypatch, _probe_returns(False))
+
+ result = await check_platform_session("xhs", cookie_str=COOKIES["xhs"])
+
+ assert result.status == STATUS_FAILED
+ assert result.cause == CAUSE_EXPIRED
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+ "exc,enable_proxy,expected_cause",
+ [
+ (httpx.ProxyError("proxy refused"), True, CAUSE_PROXY),
+ (httpx.ConnectError("no route"), False, CAUSE_NETWORK),
+ (httpx.ConnectError("no route"), True, CAUSE_PROXY),
+ (PlatformAccessError("HTTP 403"), False, CAUSE_BLOCKED),
+ (DataFetchError("bad json"), False, CAUSE_UNEXPECTED),
+ (ValueError("something else"), False, CAUSE_UNEXPECTED),
+ ],
+)
+async def test_failures_are_attributed_to_a_cause(monkeypatch, exc, enable_proxy, expected_cause):
+ """这些异常只有绕开 pong() 直接探测才会冒出来——pong() 会把它们吞掉。"""
+
+ monkeypatch.setattr(config, "ENABLE_IP_PROXY", enable_proxy)
+ if enable_proxy:
+ monkeypatch.setattr(session_check, "_resolve_proxy", _async_return("http://127.0.0.1:7890"))
+ _install_probe(monkeypatch, _probe_raises(exc))
+
+ result = await check_platform_session("bili", cookie_str=COOKIES["bili"])
+
+ assert result.status == STATUS_FAILED
+ assert result.cause == expected_cause
+
+
+@pytest.mark.asyncio
+async def test_retry_error_is_attributed_to_the_wrapped_exception(monkeypatch):
+ """weibo/zhihu 的 client 带 tenacity 重试，真实原因藏在 RetryError 里。"""
+
+ retry_error = await _make_retry_error(httpx.ProxyError("proxy refused"))
+ monkeypatch.setattr(config, "ENABLE_IP_PROXY", False)
+ _install_probe(monkeypatch, _probe_raises(retry_error))
+
+ result = await check_platform_session("wb", cookie_str=COOKIES["wb"])
+
+ assert result.status == STATUS_FAILED
+ assert result.cause == CAUSE_PROXY, "不应该退化成泛泛的网络错误"
+
+
+@pytest.mark.asyncio
+async def test_slow_platform_times_out_instead_of_hanging(monkeypatch):
+ monkeypatch.setattr(session_check, "CHECK_TIMEOUT_SECONDS", 0.01)
+ _install_probe(monkeypatch, _probe_returns(True, delay=0.2))
+
+ result = await check_platform_session("wb", cookie_str=COOKIES["wb"])
+
+ assert result.status == STATUS_FAILED
+ assert result.cause == CAUSE_TIMEOUT
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+ "platform,cookie_str,expected_hint",
+ [
+ ("dy", "LOGIN_STATUS=1", "LOGIN_STATUS=1 in cookies"),
+ ("dy", "LOGIN_STATUS=0", "no LOGIN_STATUS=1 in cookies"),
+ ("tieba", "BDUSS=abc", "BDUSS/STOKEN/PTOKEN present"),
+ ("tieba", "BAIDUID=abc", "no BDUSS/STOKEN/PTOKEN in cookies"),
+ ],
+)
+async def test_browser_only_platforms_are_unknown_not_a_verdict(
+ monkeypatch, platform, cookie_str, expected_hint
+):
+ """dy 的 pong() 先看 localStorage，cookie 只是兜底，所以这里不能断言未登录。"""
+
+ monkeypatch.setattr(session_check, "_probe", _probe_raises(AssertionError("dy/tieba 不该联网探测")))
+
+ result = await check_platform_session(platform, cookie_str=cookie_str)
+
+ assert result.status == STATUS_UNKNOWN
+ assert result.cause == CAUSE_BROWSER_REQUIRED
+ assert expected_hint in result.detail
+
+
+@pytest.mark.asyncio
+async def test_exit_code_is_zero_when_nothing_failed(monkeypatch, capsys):
+ _install_probe(monkeypatch, _probe_returns(True))
+
+ assert await run_session_check(["xhs"], cookie_str=COOKIES["xhs"]) == 0
+
+ output = capsys.readouterr().out
+ assert "Session preflight" in output
+ assert "ok 1 / failed 0 / unknown 0" in output
+
+
+@pytest.mark.asyncio
+async def test_exit_code_is_one_when_any_platform_failed(monkeypatch):
+ _install_probe(monkeypatch, _probe_returns(False))
+
+ assert await run_session_check(["xhs"], cookie_str=COOKIES["xhs"]) == 1
+
+
+@pytest.mark.asyncio
+async def test_unknown_alone_does_not_fail_the_run(monkeypatch):
+ """否则 dy / tieba 的定时任务会永远返回非零。"""
+
+ assert await run_session_check(["dy"], cookie_str="LOGIN_STATUS=1") == 0
+
+
+@pytest.mark.asyncio
+async def test_a_failure_after_an_unknown_still_sets_the_exit_code(monkeypatch, capsys):
+ _install_probe(monkeypatch, _probe_returns(False))
+
+ exit_code = await run_session_check(["dy", "xhs"], cookie_str=COOKIES["xhs"])
+
+ assert exit_code == 1
+ assert "ok 0 / failed 1 / unknown 1" in capsys.readouterr().out
+
+
+@pytest.mark.asyncio
+async def test_explicit_cookies_win_and_skip_the_browser_profile(monkeypatch, tmp_path):
+ """传了 --cookies / 配了 config.COOKIES 时不该去开浏览器。"""
+
+ monkeypatch.chdir(tmp_path)
+ monkeypatch.setattr(config, "COOKIES", COOKIES["bili"])
+
+ async def _boom(platform):
+ raise AssertionError("有 cookie 时不应该读浏览器 profile")
+
+ monkeypatch.setattr(session_check, "_cookies_from_browser_profile", _boom)
+ _install_probe(monkeypatch, _probe_returns(True))
+
+ result = await check_platform_session("bili")
+
+ assert result.status == STATUS_OK
+ assert "config.COOKIES" in result.detail
+
+
+@pytest.mark.asyncio
+async def test_saved_browser_profile_is_used_when_cookies_are_empty(monkeypatch, tmp_path):
+ """默认的 qrcode 流程不写 config.COOKIES，登录态只在 browser_data/ 里。"""
+
+ monkeypatch.chdir(tmp_path)
+ monkeypatch.setattr(config, "COOKIES", "")
+ monkeypatch.setattr(config, "ENABLE_CDP_MODE", False)
+ (tmp_path / "browser_data" / (config.USER_DATA_DIR % "bili")).mkdir(parents=True)
+
+ read_from = []
+
+ async def _fake_read(platform):
+ read_from.append(platform)
+ return COOKIES["bili"]
+
+ monkeypatch.setattr(session_check, "_cookies_from_browser_profile", _fake_read)
+ _install_probe(monkeypatch, _probe_returns(True))
+
+ result = await check_platform_session("bili")
+
+ assert result.status == STATUS_OK
+ assert read_from == ["bili"]
+ assert "browser_data/" in result.detail
+
+
+@pytest.mark.asyncio
+async def test_unreadable_browser_profile_is_reported_not_raised(monkeypatch, tmp_path):
+ monkeypatch.chdir(tmp_path)
+ monkeypatch.setattr(config, "COOKIES", "")
+ monkeypatch.setattr(config, "ENABLE_CDP_MODE", False)
+ (tmp_path / "browser_data" / (config.USER_DATA_DIR % "bili")).mkdir(parents=True)
+
+ async def _fail(platform):
+ raise RuntimeError("browser is not installed")
+
+ monkeypatch.setattr(session_check, "_cookies_from_browser_profile", _fail)
+
+ result = await check_platform_session("bili")
+
+ assert result.status == STATUS_FAILED
+ assert result.cause == CAUSE_NO_SESSION
+ assert "browser is not installed" in result.detail
+
+
+@pytest.mark.asyncio
+async def test_main_turns_the_exit_code_into_systemexit(monkeypatch):
+ """--check_session 的对外契约：通过 0，失败 1，且不建表、不启动爬虫。"""
+
+ import main
+
+ async def _fake_parse_cmd():
+ from types import SimpleNamespace
+
+ return SimpleNamespace(init_db=None, check_session=True)
+
+ async def _fake_run_session_check(platforms, cookie_str=None):
+ assert platforms == [config.PLATFORM]
+ return 1
+
+ def _no_crawler(*args, **kwargs):
+ raise AssertionError("预检不应该创建爬虫")
+
+ async def _no_init_db(*args, **kwargs):
+ raise AssertionError("预检不应该建表")
+
+ monkeypatch.setattr(main.cmd_arg, "parse_cmd", _fake_parse_cmd)
+ monkeypatch.setattr(main, "run_session_check", _fake_run_session_check)
+ monkeypatch.setattr(main.CrawlerFactory, "create_crawler", staticmethod(_no_crawler))
+ monkeypatch.setattr(main.db, "init_db", _no_init_db)
+
+ with pytest.raises(SystemExit) as excinfo:
+ await main.main()
+
+ assert excinfo.value.code == 1
+
+
+@pytest.mark.asyncio
+async def test_check_session_runs_even_when_init_db_is_requested(monkeypatch):
+ """两个都传时预检要先跑，否则写在 cron 里的 --init_db --check_session 会静默跳过检查。"""
+
+ import main
+
+ ran = []
+
+ async def _fake_parse_cmd():
+ from types import SimpleNamespace
+
+ return SimpleNamespace(init_db="sqlite", check_session=True)
+
+ async def _fake_run_session_check(platforms, cookie_str=None):
+ ran.append(platforms)
+ return 0
+
+ monkeypatch.setattr(main.cmd_arg, "parse_cmd", _fake_parse_cmd)
+ monkeypatch.setattr(main, "run_session_check", _fake_run_session_check)
+
+ with pytest.raises(SystemExit) as excinfo:
+ await main.main()
+
+ assert ran, "预检没有执行"
+ assert excinfo.value.code == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("argv", [["--check_session"], ["--check_session", "--platform", "xhs"]])
+async def test_cli_flag_sets_check_session(monkeypatch, argv):
+ from cmd_arg import parse_cmd
+
+ # parse_cmd 会就地改写 config 的全局变量，用 monkeypatch 兜住，避免污染其他测试
+ for name in ("PLATFORM", "LOGIN_TYPE", "CRAWLER_TYPE", "COOKIES", "SAVE_DATA_OPTION"):
+ monkeypatch.setattr(config, name, getattr(config, name))
+
+ args = await parse_cmd(argv)
+
+ assert args.check_session is True
+
+
+@pytest.mark.asyncio
+async def test_cli_flag_defaults_to_off(monkeypatch):
+ from cmd_arg import parse_cmd
+
+ for name in ("PLATFORM", "LOGIN_TYPE", "CRAWLER_TYPE", "COOKIES", "SAVE_DATA_OPTION"):
+ monkeypatch.setattr(config, name, getattr(config, name))
+
+ args = await parse_cmd(["--platform", "xhs"])
+
+ assert args.check_session is False
+
+
+async def _make_retry_error(exc: Exception) -> RetryError:
+ """构造一个真实的 tenacity RetryError，内部包着 exc。"""
+
+ try:
+ async for attempt in AsyncRetrying(stop=stop_after_attempt(1), reraise=False):
+ with attempt:
+ raise exc
+ except RetryError as retry_error:
+ return retry_error
+ raise AssertionError("expected a RetryError")
+
+
+def _async_return(value):
+ async def _inner(*args, **kwargs):
+ return value
+
+ return _inner
diff --git a/tools/session_check.py b/tools/session_check.py
new file mode 100644
index 000000000..c882c2b36
--- /dev/null
+++ b/tools/session_check.py
@@ -0,0 +1,493 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) 2025 relakkes@gmail.com
+#
+# This file is part of MediaCrawler project.
+# Repository: https://github.com/NanmiCoder/MediaCrawler/blob/main/tools/session_check.py
+# GitHub: https://github.com/NanmiCoder
+# Licensed under NON-COMMERCIAL LEARNING LICENSE 1.1
+#
+
+# 声明：本代码仅供学习和研究目的使用。使用者应遵守以下原则：
+# 1. 不得用于任何商业用途。
+# 2. 使用时应遵守目标平台的使用条款和robots.txt规则。
+# 3. 不得进行大规模爬取或对平台造成运营干扰。
+# 4. 应合理控制请求频率，避免给目标平台带来不必要的负担。
+# 5. 不得用于任何非法或不当的用途。
+#
+# 详细许可条款请参阅项目根目录下的LICENSE文件。
+# 使用本代码即表示您同意遵守上述原则和LICENSE中的所有条款。
+
+"""
+会话预检 (session preflight)：开爬之前先确认登录态与代理是否可用。
+
+三条设计约束：
+
+1. 只发一次平台请求。预检不该给平台增加压力，也不做自动登录或 token 刷新。
+2. 要能说出原因。各平台 client 的 pong() 把异常吞掉只返回 bool，所以这里直接调用
+ pong() 内部的那一个请求，让异常冒出来再归类（cookie 失效 / 代理不通 / 被限流
+ / 响应异常）。
+3. 登录态从用户实际使用的地方读：优先 config.COOKIES（--cookies 传入），否则读
+ qrcode 登录留下的浏览器 profile（browser_data/），两条路径都只发一次请求。
+
+抖音与百度贴吧的 pong() 读的是浏览器 localStorage / cookie，没有真正的登录态接口，
+所以这两个平台只给 cookie 层面的线索，结果标记为 unknown 而不是伪装成功。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from dataclasses import dataclass
+from typing import Dict, List, Optional, Tuple
+
+import httpx
+
+import config
+from tools import utils
+
+# 单平台预检的最长等待时间，避免代理不通时卡住。
+CHECK_TIMEOUT_SECONDS = 20
+
+# 无需浏览器即可联网验证登录态的平台。
+NETWORK_CHECKABLE_PLATFORMS = ("xhs", "bili", "wb", "zhihu", "ks")
+# pong() 只读浏览器状态、无法在预检里联网验证的平台。
+BROWSER_ONLY_PLATFORMS = ("dy", "tieba")
+
+PLATFORM_LABELS: Dict[str, str] = {
+ "xhs": "xiaohongshu",
+ "dy": "douyin",
+ "ks": "kuaishou",
+ "bili": "bilibili",
+ "wb": "weibo",
+ "tieba": "tieba",
+ "zhihu": "zhihu",
+}
+
+# 缺少这些 cookie 时，请求还没发出去就会失败，先拦下来给出更准确的提示：
+# xhs 的 sign_with_xhshow 缺 a1 会抛 ValueError；zhihu 的 _pre_headers 缺 d_c0 会抛异常。
+REQUIRED_COOKIE_KEYS: Dict[str, Tuple[str, ...]] = {
+ "xhs": ("a1",),
+ "zhihu": ("d_c0",),
+}
+
+STATUS_OK = "ok"
+STATUS_FAILED = "failed"
+STATUS_UNKNOWN = "unknown"
+
+CAUSE_NO_SESSION = "no saved session"
+CAUSE_MISSING_COOKIE_KEYS = "cookie is missing required fields"
+CAUSE_EXPIRED = "session expired or cookie is invalid"
+CAUSE_PROXY = "proxy is not usable"
+CAUSE_NETWORK = "network unreachable"
+CAUSE_BLOCKED = "blocked or rate limited by the platform"
+CAUSE_TIMEOUT = "request timed out"
+CAUSE_UNEXPECTED = "unexpected response"
+CAUSE_BROWSER_REQUIRED = "needs a browser to verify"
+
+
+@dataclass
+class SessionCheckResult:
+ """单个平台的预检结果。"""
+
+ platform: str
+ status: str
+ cause: str = ""
+ detail: str = ""
+
+ @property
+ def label(self) -> str:
+ return PLATFORM_LABELS.get(self.platform, self.platform)
+
+ def render(self) -> str:
+ icon = {STATUS_OK: "[ OK ]", STATUS_FAILED: "[FAIL]", STATUS_UNKNOWN: "[ ?? ]"}.get(self.status, "[ ?? ]")
+ line = f"{icon} {self.platform:<6} {self.label:<12} {self.status}"
+ if self.cause:
+ line = f"{line} - {self.cause}"
+ if self.detail:
+ line = f"{line} ({self.detail})"
+ return line
+
+
+def _classify_exception(exc: BaseException) -> Tuple[str, str]:
+ """把探测过程中的异常归类成 (原因, 细节)。"""
+
+ if isinstance(exc, asyncio.TimeoutError):
+ return CAUSE_TIMEOUT, f"no answer within {CHECK_TIMEOUT_SECONDS}s"
+
+ if isinstance(exc, httpx.ProxyError):
+ return CAUSE_PROXY, str(exc) or exc.__class__.__name__
+
+ if isinstance(exc, (httpx.ConnectError, httpx.ConnectTimeout, httpx.ReadTimeout)):
+ cause = CAUSE_PROXY if config.ENABLE_IP_PROXY else CAUSE_NETWORK
+ return cause, exc.__class__.__name__
+
+ # 各平台异常类名相同但定义在不同模块，按名字归类比逐个 import 更省事。
+ exc_name = exc.__class__.__name__
+ if exc_name in ("IPBlockError", "PlatformAccessError", "ForbiddenError"):
+ return CAUSE_BLOCKED, exc_name
+
+ if exc_name == "RetryError":
+ # tenacity 重试耗尽后包了一层，取最后一次的真实异常再归类。
+ last_exc = _unwrap_retry_error(exc)
+ if last_exc is not None and last_exc is not exc:
+ return _classify_exception(last_exc)
+ return (CAUSE_PROXY if config.ENABLE_IP_PROXY else CAUSE_NETWORK), exc_name
+
+ if exc_name == "DataFetchError":
+ return CAUSE_UNEXPECTED, str(exc) or exc_name
+
+ return CAUSE_UNEXPECTED, f"{exc_name}: {exc}"
+
+
+def _unwrap_retry_error(exc: BaseException) -> Optional[BaseException]:
+ """取出 tenacity RetryError 内部最后一次失败的异常。"""
+
+ last_attempt = getattr(exc, "last_attempt", None)
+ if last_attempt is None:
+ return None
+ try:
+ if last_attempt.failed:
+ return last_attempt.exception()
+ except Exception: # noqa: BLE001 - 拿不到就按 RetryError 本身处理
+ return None
+ return None
+
+
+async def _resolve_proxy() -> Optional[str]:
+ """按当前配置取一个可用代理，返回 httpx 可用的代理地址。
+
+ 只取一个 IP，不预热整个代理池——预检要保持轻量。
+ """
+
+ if not config.ENABLE_IP_PROXY:
+ return None
+
+ from proxy.proxy_ip_pool import create_ip_pool
+
+ ip_pool = await create_ip_pool(1, enable_validate_ip=True)
+ ip_proxy_info = await ip_pool.get_proxy()
+ _, httpx_proxy = utils.format_proxy_info(ip_proxy_info)
+ return httpx_proxy
+
+
+def _user_data_dir(platform: str) -> str:
+ """qrcode 登录后保存登录态的浏览器 profile 目录。
+
+ 与 media_platform/*/core.py 的 launch_browser 和 tools/cdp_browser.py:255-263
+ 保持一致：CDP 模式多一个 cdp_ 前缀。
+ """
+
+ dir_name = config.USER_DATA_DIR % platform
+ if config.ENABLE_CDP_MODE:
+ dir_name = f"cdp_{dir_name}"
+ return os.path.join(os.getcwd(), "browser_data", dir_name)
+
+
+async def _cookies_from_browser_profile(platform: str) -> str:
+ """从已保存的浏览器 profile 里读出 cookie，读完立刻关闭浏览器。
+
+ 默认的 `--lt qrcode` 流程不会往 config.COOKIES 写任何东西，登录态只存在于
+ browser_data/ 里，所以不读这里的话预检对大多数用户都只会说“没有 cookie”。
+ """
+
+ from playwright.async_api import async_playwright
+
+ user_data_dir = _user_data_dir(platform)
+ async with async_playwright() as playwright:
+ browser_context = await playwright.chromium.launch_persistent_context(
+ user_data_dir=user_data_dir,
+ accept_downloads=True,
+ headless=True,
+ viewport={"width": 1920, "height": 1080},
+ user_agent=utils.get_user_agent(),
+ )
+ try:
+ cookie_str, _ = await utils.convert_browser_context_cookies(browser_context)
+ finally:
+ await browser_context.close()
+
+ return cookie_str
+
+
+async def _load_cookie_str(platform: str) -> Tuple[str, str]:
+ """返回 (cookie 字符串, 来源描述)。取不到就返回空串。"""
+
+ if config.COOKIES.strip():
+ return config.COOKIES, "config.COOKIES"
+
+ user_data_dir = _user_data_dir(platform)
+ if not os.path.isdir(user_data_dir):
+ return "", ""
+
+ cookie_str = await _cookies_from_browser_profile(platform)
+ return cookie_str, f"browser_data/{os.path.basename(user_data_dir)}"
+
+
+def _build_client(platform: str, cookie_str: str, httpx_proxy: Optional[str]):
+ """按平台构造一个仅用于预检的 API client。
+
+ header 与各平台 create_*_client 的关键字段保持一致，尤其是 Cookie 的大小写：
+ xhs 读 headers["Cookie"]，zhihu 读 default_headers["cookie"]。
+ """
+
+ cookie_dict = utils.convert_str_cookie_to_dict(cookie_str)
+
+ if platform == "xhs":
+ from media_platform.xhs.client import XiaoHongShuClient
+
+ index_url = "https://www.rednote.com" if config.XHS_INTERNATIONAL else "https://www.xiaohongshu.com"
+ return XiaoHongShuClient(
+ proxy=httpx_proxy,
+ headers={
+ "accept": "application/json, text/plain, */*",
+ "accept-language": "zh-CN,zh;q=0.9",
+ "content-type": "application/json;charset=UTF-8",
+ "origin": index_url,
+ "referer": f"{index_url}/",
+ "user-agent": utils.get_user_agent(),
+ "Cookie": cookie_str,
+ },
+ playwright_page=None,
+ cookie_dict=cookie_dict,
+ )
+
+ if platform == "bili":
+ from media_platform.bilibili.client import BilibiliClient
+
+ return BilibiliClient(
+ proxy=httpx_proxy,
+ headers={
+ "User-Agent": utils.get_user_agent(),
+ "Cookie": cookie_str,
+ "Origin": "https://www.bilibili.com",
+ "Referer": "https://www.bilibili.com",
+ "Content-Type": "application/json;charset=UTF-8",
+ },
+ playwright_page=None,
+ cookie_dict=cookie_dict,
+ )
+
+ if platform == "wb":
+ from media_platform.weibo.client import WeiboClient
+
+ return WeiboClient(
+ proxy=httpx_proxy,
+ headers={
+ "User-Agent": utils.get_mobile_user_agent(),
+ "Cookie": cookie_str,
+ "Origin": "https://m.weibo.cn",
+ "Referer": "https://m.weibo.cn",
+ "Content-Type": "application/json;charset=UTF-8",
+ },
+ playwright_page=None,
+ cookie_dict=cookie_dict,
+ )
+
+ if platform == "zhihu":
+ from media_platform.zhihu.client import ZhiHuClient
+
+ return ZhiHuClient(
+ proxy=httpx_proxy,
+ headers={
+ "accept": "*/*",
+ "accept-language": "zh-CN,zh;q=0.9",
+ "cookie": cookie_str,
+ "priority": "u=1, i",
+ "referer": "https://www.zhihu.com/",
+ "user-agent": utils.get_user_agent(),
+ "x-api-version": "3.0.91",
+ "x-app-za": "OS=Web",
+ "x-requested-with": "fetch",
+ "x-zse-93": "101_3_3.0",
+ },
+ playwright_page=None,
+ cookie_dict=cookie_dict,
+ )
+
+ if platform == "ks":
+ from media_platform.kuaishou.client import KuaiShouClient
+
+ return KuaiShouClient(
+ proxy=httpx_proxy,
+ headers={
+ "User-Agent": utils.get_user_agent(),
+ "Cookie": cookie_str,
+ "Origin": "https://www.kuaishou.com",
+ "Referer": "https://www.kuaishou.com",
+ "Content-Type": "application/json;charset=UTF-8",
+ },
+ playwright_page=None,
+ cookie_dict=cookie_dict,
+ )
+
+ raise ValueError(f"platform {platform!r} does not support a browserless session check")
+
+
+async def _probe(platform: str, client) -> bool:
+ """发出各平台 pong() 内部的那一个请求，异常交给调用方归类。
+
+ 刻意不复用 pong()：它内部 `except Exception: return False`，一旦复用，代理不通、
+ 被限流、响应异常全都会被报成“cookie 失效”。
+ """
+
+ if platform == "xhs":
+ # 走 client.get()（而不是 query_self()）：前者经过 request()，
+ # 403/429 会抛 PlatformAccessError，风控码会抛 IPBlockError。
+ data = await client.get("/api/sns/web/v1/user/selfinfo", {})
+ return bool((data or {}).get("result", {}).get("success"))
+
+ if platform == "bili":
+ data = await client.get("/x/web-interface/nav")
+ return bool((data or {}).get("isLogin"))
+
+ if platform == "wb":
+ # 绕开 WeiboClient.request()：它带 5 次重试（一次预检要 12s 以上），
+ # 而且响应不是 JSON 时会去解引用 playwright_page（这里是 None）。
+ from tools.httpx_util import make_async_client
+
+ async with make_async_client(proxy=client.proxy) as http_client:
+ response = await http_client.get(
+ f"{client._host}/api/config",
+ headers=client.headers,
+ timeout=client.timeout,
+ )
+ payload = response.json()
+ return bool(payload.get("data", {}).get("login"))
+
+ if platform == "zhihu":
+ data = await client.get_current_user_info()
+ return bool(data.get("uid") and data.get("name"))
+
+ if platform == "ks":
+ data = await client.post(
+ "",
+ {
+ "operationName": "visionProfileUserList",
+ "variables": {"ftype": 1},
+ "query": client.graphql.get("vision_profile_user_list"),
+ },
+ )
+ return (data or {}).get("visionProfileUserList", {}).get("result") == 1
+
+ raise ValueError(f"platform {platform!r} does not support a browserless session check")
+
+
+def _browser_only_result(platform: str, cookie_str: str) -> SessionCheckResult:
+ """dy / tieba：pong() 读的是浏览器状态，预检只能给 cookie 层面的线索。"""
+
+ cookie_dict = utils.convert_str_cookie_to_dict(cookie_str)
+
+ if platform == "dy":
+ # DouYinClient.pong 先看 localStorage.HasUserLogin，cookie 只是兜底，
+ # 所以 cookie 里没有这个标记并不能证明未登录。
+ signal = cookie_dict.get("LOGIN_STATUS") == "1"
+ hint = "LOGIN_STATUS=1 in cookies" if signal else "no LOGIN_STATUS=1 in cookies"
+ else: # tieba
+ signal = any(key in cookie_dict for key in ("BDUSS", "STOKEN", "PTOKEN"))
+ hint = "BDUSS/STOKEN/PTOKEN present" if signal else "no BDUSS/STOKEN/PTOKEN in cookies"
+
+ return SessionCheckResult(
+ platform=platform,
+ status=STATUS_UNKNOWN,
+ cause=CAUSE_BROWSER_REQUIRED,
+ detail=f"{hint}; this platform has no login-state endpoint to probe",
+ )
+
+
+async def check_platform_session(platform: str, cookie_str: Optional[str] = None) -> SessionCheckResult:
+ """对单个平台做一次登录态预检。"""
+
+ source = "explicit cookie"
+ if cookie_str is None:
+ try:
+ cookie_str, source = await _load_cookie_str(platform)
+ except Exception as exc: # noqa: BLE001 - 读不到登录态本身就是一种预检结果
+ return SessionCheckResult(
+ platform=platform,
+ status=STATUS_FAILED,
+ cause=CAUSE_NO_SESSION,
+ detail=f"could not read the saved browser session: {exc.__class__.__name__}: {exc}",
+ )
+
+ if not cookie_str.strip():
+ return SessionCheckResult(
+ platform=platform,
+ status=STATUS_FAILED,
+ cause=CAUSE_NO_SESSION,
+ detail="config.COOKIES is empty and no saved browser profile was found; pass --cookies or log in once",
+ )
+
+ cookie_dict = utils.convert_str_cookie_to_dict(cookie_str)
+ missing = [key for key in REQUIRED_COOKIE_KEYS.get(platform, ()) if key not in cookie_dict]
+ if missing:
+ return SessionCheckResult(
+ platform=platform,
+ status=STATUS_FAILED,
+ cause=CAUSE_MISSING_COOKIE_KEYS,
+ detail=f"missing {', '.join(missing)} (source: {source}); the cookie may belong to another platform",
+ )
+
+ if platform in BROWSER_ONLY_PLATFORMS:
+ return _browser_only_result(platform, cookie_str)
+
+ try:
+ httpx_proxy = await _resolve_proxy()
+ except Exception as exc: # noqa: BLE001 - 代理拿不到也是一种预检结果
+ return SessionCheckResult(
+ platform=platform,
+ status=STATUS_FAILED,
+ cause=CAUSE_PROXY,
+ detail=f"could not obtain a proxy: {exc.__class__.__name__}: {exc}",
+ )
+
+ try:
+ client = _build_client(platform, cookie_str, httpx_proxy)
+ except ValueError as exc:
+ return SessionCheckResult(
+ platform=platform,
+ status=STATUS_UNKNOWN,
+ cause=CAUSE_BROWSER_REQUIRED,
+ detail=str(exc),
+ )
+
+ try:
+ alive = await asyncio.wait_for(_probe(platform, client), timeout=CHECK_TIMEOUT_SECONDS)
+ except Exception as exc: # noqa: BLE001 - 归类后上报，预检不把异常抛给调用方
+ cause, detail = _classify_exception(exc)
+ return SessionCheckResult(platform=platform, status=STATUS_FAILED, cause=cause, detail=detail)
+
+ if alive:
+ return SessionCheckResult(platform=platform, status=STATUS_OK, detail=f"source: {source}")
+
+ return SessionCheckResult(
+ platform=platform,
+ status=STATUS_FAILED,
+ cause=CAUSE_EXPIRED,
+ detail=f"the platform answered 'not logged in' (source: {source})",
+ )
+
+
+async def run_session_check(platforms: List[str], cookie_str: Optional[str] = None) -> int:
+ """预检若干平台，打印结果并返回进程退出码。
+
+ 返回 0 表示没有失败；只要有一个 failed 就返回 1。unknown 不算失败，否则
+ dy / tieba 的定时任务会永远是红的。
+ """
+
+ print("=" * 72)
+ print("Session preflight")
+ print(f"proxy: {'on' if config.ENABLE_IP_PROXY else 'off'}")
+ print("=" * 72)
+
+ results: List[SessionCheckResult] = []
+ for platform in platforms:
+ result = await check_platform_session(platform, cookie_str=cookie_str)
+ results.append(result)
+ print(result.render())
+
+ failed = [r for r in results if r.status == STATUS_FAILED]
+ unknown = [r for r in results if r.status == STATUS_UNKNOWN]
+ print("=" * 72)
+ print(f"ok {len(results) - len(failed) - len(unknown)} / failed {len(failed)} / unknown {len(unknown)}")
+
+ return 1 if failed else 0