zzh
5a26ae0048
- 新增小红书前端模块（启动服务、内容分析、爆款创作、发布记录、技能卡片、拟人设置、AI 助手） - 新增 xhs-agent-desktop 独立桌面 Agent，对接 OMS 任务调度 - 登录页重构，整合微信登录与 OMS JWT 认证 - 订单、渠道、用户中心等页面与 API 增强 Co-authored-by: Cursor <cursoragent@cursor.com>
134 lines
3.8 KiB
Python
134 lines
3.8 KiB
Python
"""点赞/收藏操作，对应 Go xiaohongshu/like_favorite.go。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import time
|
|
|
|
from .cdp import Page
|
|
from .errors import NoFeedDetailError
|
|
from .feed_detail import navigate_to_feed_detail
|
|
from .selectors import COLLECT_BUTTON, LIKE_BUTTON
|
|
from .types import ActionResult
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_GET_INTERACT_STATE_JS = """
|
|
(() => {
|
|
if (window.__INITIAL_STATE__ &&
|
|
window.__INITIAL_STATE__.note &&
|
|
window.__INITIAL_STATE__.note.noteDetailMap) {
|
|
return JSON.stringify(window.__INITIAL_STATE__.note.noteDetailMap);
|
|
}
|
|
return "";
|
|
})()
|
|
"""
|
|
|
|
|
|
def _get_interact_state(page: Page, feed_id: str) -> tuple[bool, bool]:
|
|
result = page.evaluate(_GET_INTERACT_STATE_JS)
|
|
if not result:
|
|
raise NoFeedDetailError()
|
|
|
|
note_detail_map = json.loads(result)
|
|
detail = note_detail_map.get(feed_id)
|
|
if not detail and len(note_detail_map) == 1:
|
|
detail = next(iter(note_detail_map.values()))
|
|
|
|
if not detail:
|
|
raise NoFeedDetailError()
|
|
|
|
interact = detail.get("note", {}).get("interactInfo", {})
|
|
return interact.get("liked", False), interact.get("collected", False)
|
|
|
|
|
|
def _prepare_page(
|
|
page: Page,
|
|
feed_id: str,
|
|
xsec_token: str,
|
|
*,
|
|
on_progress=None,
|
|
) -> None:
|
|
navigate_to_feed_detail(page, feed_id, xsec_token)
|
|
time.sleep(1)
|
|
|
|
|
|
def like_feed(
|
|
page: Page,
|
|
feed_id: str,
|
|
xsec_token: str,
|
|
*,
|
|
on_progress=None,
|
|
) -> ActionResult:
|
|
_prepare_page(page, feed_id, xsec_token, on_progress=on_progress)
|
|
return _toggle_like(page, feed_id, target_liked=True)
|
|
|
|
|
|
def unlike_feed(
|
|
page: Page,
|
|
feed_id: str,
|
|
xsec_token: str,
|
|
*,
|
|
on_progress=None,
|
|
) -> ActionResult:
|
|
_prepare_page(page, feed_id, xsec_token, on_progress=on_progress)
|
|
return _toggle_like(page, feed_id, target_liked=False)
|
|
|
|
|
|
def favorite_feed(
|
|
page: Page,
|
|
feed_id: str,
|
|
xsec_token: str,
|
|
*,
|
|
on_progress=None,
|
|
) -> ActionResult:
|
|
_prepare_page(page, feed_id, xsec_token, on_progress=on_progress)
|
|
return _toggle_favorite(page, feed_id, target_collected=True)
|
|
|
|
|
|
def unfavorite_feed(
|
|
page: Page,
|
|
feed_id: str,
|
|
xsec_token: str,
|
|
*,
|
|
on_progress=None,
|
|
) -> ActionResult:
|
|
_prepare_page(page, feed_id, xsec_token, on_progress=on_progress)
|
|
return _toggle_favorite(page, feed_id, target_collected=False)
|
|
|
|
|
|
def _toggle_like(page: Page, feed_id: str, *, target_liked: bool) -> ActionResult:
|
|
action_name = "点赞" if target_liked else "取消点赞"
|
|
try:
|
|
liked, _ = _get_interact_state(page, feed_id)
|
|
except NoFeedDetailError:
|
|
liked = False
|
|
if liked == target_liked:
|
|
return ActionResult(feed_id=feed_id, success=True, message=f"已是{action_name}状态")
|
|
if not page.has_element(LIKE_BUTTON):
|
|
raise RuntimeError("未找到点赞按钮")
|
|
page.click_element(LIKE_BUTTON)
|
|
time.sleep(1.2)
|
|
return ActionResult(feed_id=feed_id, success=True, message=f"{action_name}已执行")
|
|
|
|
|
|
def _toggle_favorite(page: Page, feed_id: str, *, target_collected: bool) -> ActionResult:
|
|
action_name = "收藏" if target_collected else "取消收藏"
|
|
try:
|
|
_, collected = _get_interact_state(page, feed_id)
|
|
except NoFeedDetailError:
|
|
collected = False
|
|
if collected == target_collected:
|
|
return ActionResult(feed_id=feed_id, success=True, message=f"已是{action_name}状态")
|
|
deadline = time.monotonic() + 5.0
|
|
while time.monotonic() < deadline:
|
|
if page.has_element(COLLECT_BUTTON):
|
|
break
|
|
time.sleep(0.2)
|
|
else:
|
|
raise RuntimeError("未找到收藏按钮")
|
|
page.click_element(COLLECT_BUTTON)
|
|
time.sleep(1.2)
|
|
return ActionResult(feed_id=feed_id, success=True, message=f"{action_name}已执行")
|