zzh
5a26ae0048
- 新增小红书前端模块（启动服务、内容分析、爆款创作、发布记录、技能卡片、拟人设置、AI 助手） - 新增 xhs-agent-desktop 独立桌面 Agent，对接 OMS 任务调度 - 登录页重构，整合微信登录与 OMS JWT 认证 - 订单、渠道、用户中心等页面与 API 增强 Co-authored-by: Cursor <cursoragent@cursor.com>
81 lines
2.2 KiB
Python
81 lines
2.2 KiB
Python
"""通过 XHS Bridge 扩展从 Chrome 读取小红书 Cookie。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import time
|
||
from typing import Any
|
||
|
||
import websockets.sync.client as ws_client
|
||
|
||
_CACHE: dict[str, tuple[float, str]] = {}
|
||
_CACHE_TTL_SEC = 60.0
|
||
|
||
|
||
def clear_cookie_cache() -> None:
|
||
_CACHE.clear()
|
||
|
||
|
||
def cookies_to_header(cookies: list[dict[str, Any]]) -> str:
|
||
parts: list[str] = []
|
||
seen: set[str] = set()
|
||
for item in cookies:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
name = item.get("name")
|
||
value = item.get("value")
|
||
if not name or value is None:
|
||
continue
|
||
if name in seen:
|
||
continue
|
||
seen.add(name)
|
||
parts.append(f"{name}={value}")
|
||
return "; ".join(parts)
|
||
|
||
|
||
def fetch_cookies_from_extension(
|
||
bridge_url: str,
|
||
*,
|
||
domain: str = "xiaohongshu.com",
|
||
timeout: float = 10.0,
|
||
) -> list[dict[str, Any]]:
|
||
"""调用扩展 get_cookies，返回 chrome.cookies 对象列表。"""
|
||
msg = {
|
||
"role": "cli",
|
||
"method": "get_cookies",
|
||
"params": {"domain": domain},
|
||
}
|
||
try:
|
||
with ws_client.connect(bridge_url, max_size=4 * 1024 * 1024, open_timeout=3) as ws:
|
||
ws.send(json.dumps(msg, ensure_ascii=False))
|
||
raw = ws.recv(timeout=timeout)
|
||
except OSError as exc:
|
||
raise RuntimeError(f"无法连接 Bridge（{bridge_url}）：{exc}") from exc
|
||
|
||
data = json.loads(raw)
|
||
if data.get("error"):
|
||
raise RuntimeError(str(data["error"]))
|
||
result = data.get("result")
|
||
if not isinstance(result, list):
|
||
return []
|
||
return [x for x in result if isinstance(x, dict)]
|
||
|
||
|
||
def get_cookie_header_from_extension(
|
||
bridge_url: str,
|
||
*,
|
||
force_refresh: bool = False,
|
||
) -> str:
|
||
"""读取 Cookie 头；带短时缓存，避免频繁打扩展。"""
|
||
now = time.time()
|
||
if not force_refresh and bridge_url in _CACHE:
|
||
ts, header = _CACHE[bridge_url]
|
||
if (now - ts) < _CACHE_TTL_SEC and header:
|
||
return header
|
||
|
||
cookies = fetch_cookies_from_extension(bridge_url)
|
||
header = cookies_to_header(cookies)
|
||
if header:
|
||
_CACHE[bridge_url] = (now, header)
|
||
return header
|