zzh
5a26ae0048
- 新增小红书前端模块（启动服务、内容分析、爆款创作、发布记录、技能卡片、拟人设置、AI 助手） - 新增 xhs-agent-desktop 独立桌面 Agent，对接 OMS 任务调度 - 登录页重构，整合微信登录与 OMS JWT 认证 - 订单、渠道、用户中心等页面与 API 增强 Co-authored-by: Cursor <cursoragent@cursor.com>
50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
from __future__ import annotations
|
||
|
||
import requests
|
||
|
||
from xhs_bridge.config import load_config
|
||
from xhs_bridge.cookies_loader import image_request_headers, normalize_image_url
|
||
from xhs_bridge.extension_cookies import get_cookie_header_from_extension
|
||
|
||
_TIMEOUT = 30
|
||
|
||
|
||
def _fetch_once(url: str, cookie_header: str | None) -> bytes:
|
||
headers = image_request_headers(cookie_header=cookie_header)
|
||
resp = requests.get(url, headers=headers, timeout=_TIMEOUT)
|
||
resp.raise_for_status()
|
||
|
||
content_type = (resp.headers.get("Content-Type") or "").lower()
|
||
data = resp.content
|
||
if len(data) < 128:
|
||
raise ValueError("图片数据过短")
|
||
if "text/html" in content_type or "application/json" in content_type:
|
||
raise ValueError("CDN 返回了非图片内容")
|
||
return data
|
||
|
||
|
||
def fetch_image_bytes(url: str) -> bytes:
|
||
"""下载小红书 CDN 图片；失败时从 Chrome 扩展读取 Cookie 重试。"""
|
||
url = normalize_image_url(url)
|
||
if not url:
|
||
raise ValueError("图片地址为空")
|
||
|
||
cfg = load_config()
|
||
bridge_url = cfg.skills.bridge_ws_url
|
||
last_error: Exception | None = None
|
||
|
||
try:
|
||
return _fetch_once(url, None)
|
||
except Exception as exc:
|
||
last_error = exc
|
||
|
||
try:
|
||
cookie = get_cookie_header_from_extension(bridge_url, force_refresh=True)
|
||
if not cookie:
|
||
raise ValueError("扩展未返回 Cookie，请确认 Chrome 已登录小红书")
|
||
return _fetch_once(url, cookie)
|
||
except Exception as exc:
|
||
last_error = exc
|
||
|
||
raise ValueError(f"图片加载失败：{last_error}") from last_error
|