"""点赞/收藏操作,对应 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}已执行")