fix: fix favorites command not working

tests/test_client.py

@@ -6,11 +6,248 @@
 import pytest
 
 from xhs_cli.client import XhsClient
+from xhs_cli.constants import USER_AGENT
 from xhs_cli.cookies import cache_note_context, get_cached_note_context
-from xhs_cli.exceptions import UnsupportedOperationError, XhsApiError
+from xhs_cli.exceptions import SignatureError, UnsupportedOperationError, XhsApiError
 
 
 class TestFavorites:
+    def test_get_user_favorites_uses_current_web_params(self, monkeypatch):
+        captured = {}
+
+        def fake_get(self, uri, params=None):
+            captured["uri"] = uri
+            captured["params"] = params
+            return {"notes": [], "has_more": False, "cursor": ""}
+
+        monkeypatch.setattr(XhsClient, "_main_api_get", fake_get)
+
+        client = XhsClient({"a1": "cookie"})
+        try:
+            result = client.get_user_favorites("user-123", cursor="cursor-1")
+        finally:
+            client.close()
+
+        assert result["notes"] == []
+        assert captured["uri"] == "/api/sns/web/v2/note/collect/page"
+        assert captured["params"] == {
+            "num": 30,
+            "cursor": "cursor-1",
+            "user_id": "user-123",
+            "image_formats": "jpg,webp,avif",
+            "xsec_token": "",
+            "xsec_source": "",
+        }
+
+    def test_get_user_favorites_falls_back_to_browser_on_common_api_error(self, monkeypatch):
+        captured = {}
+
+        def fake_get(self, uri, params=None):
+            raise XhsApiError(
+                "API error: {\"code\": -1, \"success\": false}",
+                code=-1,
+                response={"code": -1, "success": False},
+            )
+
+        def fake_browser_fallback(self, user_id, cursor=""):
+            captured["user_id"] = user_id
+            captured["cursor"] = cursor
+            return {"notes": [{"note_id": "note-1"}], "has_more": False, "cursor": ""}
+
+        monkeypatch.setattr(XhsClient, "_main_api_get", fake_get)
+        monkeypatch.setattr(XhsClient, "_get_user_favorites_with_browser", fake_browser_fallback)
+
+        client = XhsClient({"a1": "cookie"})
+        try:
+            result = client.get_user_favorites("user-123", cursor="cursor-1")
+        finally:
+            client.close()
+
+        assert result["notes"] == [{"note_id": "note-1"}]
+        assert captured == {"user_id": "user-123", "cursor": "cursor-1"}
+
+    def test_get_user_favorites_falls_back_to_browser_on_signature_error(self, monkeypatch):
+        captured = {}
+
+        def fake_get(self, uri, params=None):
+            raise SignatureError()
+
+        def fake_browser_fallback(self, user_id, cursor=""):
+            captured["user_id"] = user_id
+            captured["cursor"] = cursor
+            return {"notes": [{"note_id": "note-1"}], "has_more": False, "cursor": ""}
+
+        monkeypatch.setattr(XhsClient, "_main_api_get", fake_get)
+        monkeypatch.setattr(XhsClient, "_get_user_favorites_with_browser", fake_browser_fallback)
+
+        client = XhsClient({"a1": "cookie"})
+        try:
+            result = client.get_user_favorites("user-123", cursor="cursor-1")
+        finally:
+            client.close()
+
+        assert result["notes"] == [{"note_id": "note-1"}]
+        assert captured == {"user_id": "user-123", "cursor": "cursor-1"}
+
+    def test_get_user_favorites_browser_fallback_uses_camoufox_backend(self, monkeypatch):
+        captured = {}
+
+        class FakeResponse:
+            url = "https://edith.xiaohongshu.com/api/sns/web/v2/note/collect/page?cursor=cursor-1"
+
+            def json(self):
+                return {
+                    "success": True,
+                    "data": {"notes": [{"note_id": "note-1"}], "has_more": False, "cursor": ""},
+                }
+
+        class FakeMouse:
+            def wheel(self, x, y):
+                captured["wheel"] = (x, y)
+
+        class FakeLocator:
+            def __init__(self, page):
+                self.page = page
+
+            def click(self, timeout):
+                captured["click_timeout"] = timeout
+                self.page.handlers["response"](FakeResponse())
+
+        class FakePage:
+            def __init__(self):
+                self.handlers = {}
+                self.mouse = FakeMouse()
+
+            def on(self, event, handler):
+                self.handlers[event] = handler
+
+            def goto(self, url, wait_until, timeout):
+                captured["goto"] = (url, wait_until, timeout)
+
+            def wait_for_timeout(self, timeout):
+                captured.setdefault("waits", []).append(timeout)
+
+            def get_by_text(self, text, exact):
+                captured["tab_locator"] = (text, exact)
+                return FakeLocator(self)
+
+        class FakeContext:
+            def add_cookies(self, cookies):
+                captured["cookies"] = cookies
+
+            def new_page(self):
+                captured["new_page"] = True
+                return FakePage()
+
+        class FakeBrowser:
+            def new_context(self, **kwargs):
+                captured["context_kwargs"] = kwargs
+                return FakeContext()
+
+        class FakeCamoufox:
+            def __init__(self, **kwargs):
+                captured["launch_kwargs"] = kwargs
+
+            def __enter__(self):
+                return FakeBrowser()
+
+            def __exit__(self, *args):
+                captured["closed"] = True
+
+        monkeypatch.setattr("xhs_cli.qr_login._ensure_camoufox_ready", lambda: None)
+        monkeypatch.setattr("camoufox.sync_api.Camoufox", FakeCamoufox)
+
+        client = XhsClient({"a1": "cookie", "saved_at": "ignored"})
+        try:
+            result = client._get_user_favorites_with_browser("user-123", cursor="cursor-1")
+        finally:
+            client.close()
+
+        assert result["notes"] == [{"note_id": "note-1"}]
+        assert captured["launch_kwargs"] == {"headless": True}
+        assert captured["context_kwargs"] == {"user_agent": USER_AGENT, "locale": "zh-CN"}
+        assert captured["cookies"] == [{
+            "name": "a1",
+            "value": "cookie",
+            "domain": ".xiaohongshu.com",
+            "path": "/",
+            "secure": True,
+            "sameSite": "Lax",
+        }]
+        assert captured["goto"] == (
+            "https://www.xiaohongshu.com/user/profile/user-123",
+            "domcontentloaded",
+            45_000,
+        )
+        assert captured["tab_locator"] == ("收藏", True)
+        assert captured["click_timeout"] == 5_000
+        assert captured["closed"] is True
+
+    def test_get_user_favorites_browser_fallback_wraps_start_failure(self, monkeypatch):
+        class FakeCamoufox:
+            def __init__(self, **kwargs):
+                pass
+
+            def __enter__(self):
+                raise RuntimeError("browser failed to launch")
+
+            def __exit__(self, *args):
+                return None
+
+        monkeypatch.setattr("xhs_cli.qr_login._ensure_camoufox_ready", lambda: None)
+        monkeypatch.setattr("camoufox.sync_api.Camoufox", FakeCamoufox)
+
+        client = XhsClient({"a1": "cookie"})
+        try:
+            with pytest.raises(UnsupportedOperationError, match="could not start Camoufox"):
+                client._get_user_favorites_with_browser("user-123")
+        finally:
+            client.close()
+
+    def test_get_user_favorites_browser_fallback_wraps_navigation_failure(self, monkeypatch):
+        class FakePage:
+            def __init__(self):
+                self.mouse = object()
+
+            def on(self, event, handler):
+                return None
+
+            def goto(self, url, wait_until, timeout):
+                raise RuntimeError("navigation failed")
+
+        class FakeContext:
+            def add_cookies(self, cookies):
+                return None
+
+            def new_page(self):
+                return FakePage()
+
+        class FakeBrowser:
+            def new_context(self, **kwargs):
+                return FakeContext()
+
+        class FakeCamoufox:
+            def __init__(self, **kwargs):
+                pass
+
+            def __enter__(self):
+                return FakeBrowser()
+
+            def __exit__(self, *args):
+                return None
+
+        monkeypatch.setattr("xhs_cli.qr_login._ensure_camoufox_ready", lambda: None)
+        monkeypatch.setattr("camoufox.sync_api.Camoufox", FakeCamoufox)
+
+        client = XhsClient({"a1": "cookie"})
+        try:
+            with pytest.raises(XhsApiError, match="failed while driving Camoufox") as exc_info:
+                client._get_user_favorites_with_browser("user-123")
+        finally:
+            client.close()
+
+        assert not isinstance(exc_info.value, UnsupportedOperationError)
+
     def test_unfavorite_uses_note_ids_payload(self, monkeypatch):
         captured = {}
 

xhs_cli/client_mixins.py

@@ -21,7 +21,7 @@
     get_config_dir,
     invalidate_note_context,
 )
-from .exceptions import NeedVerifyError, UnsupportedOperationError, XhsApiError
+from .exceptions import NeedVerifyError, SignatureError, UnsupportedOperationError, XhsApiError
 from .html_parser import extract_note_from_html
 
 logger = logging.getLogger(__name__)
@@ -638,11 +638,132 @@ def unfollow_user(self, user_id: str) -> dict[str, Any]:
         return self._main_api_post("/api/sns/web/v1/user/unfollow", {"target_user_id": user_id})
 
     def get_user_favorites(self, user_id: str, cursor: str = "") -> dict[str, Any]:
-        return self._main_api_get("/api/sns/web/v2/note/collect/page", {
-            "user_id": user_id,
-            "cursor": cursor,
+        params = {
             "num": 30,
-        })
+            "cursor": cursor,
+            "user_id": user_id,
+            "image_formats": "jpg,webp,avif",
+            "xsec_token": "",
+            "xsec_source": "",
+        }
+        try:
+            return self._main_api_get("/api/sns/web/v2/note/collect/page", params)
+        except XhsApiError as exc:
+            if not isinstance(exc, SignatureError) and exc.code not in {-1, "-1", 300015, "300015"}:
+                raise
+            logger.debug("Signed favorites API failed; falling back to browser request")
+            return self._get_user_favorites_with_browser(user_id, cursor=cursor)
+
+    def _get_user_favorites_with_browser(self, user_id: str, cursor: str = "") -> dict[str, Any]:
+        try:
+            from camoufox.sync_api import Camoufox
+            from playwright.sync_api import TimeoutError as PlaywrightTimeoutError
+        except ImportError as exc:
+            raise UnsupportedOperationError(
+                "Favorites API requires the current web signing algorithm. "
+                "Install Camoufox/Playwright or update xiaohongshu-cli/xhshow to use the browser fallback."
+            ) from exc
+
+        from urllib.parse import parse_qs, urlparse
+
+        from .qr_login import BrowserQrLoginUnavailable, _ensure_camoufox_ready
+
+        target_path = "/api/sns/web/v2/note/collect/page"
+        responses: dict[str, dict[str, Any]] = {}
+        errors: list[dict[str, Any]] = []
+
+        def _response_cursor(url: str) -> str:
+            return parse_qs(urlparse(url).query, keep_blank_values=True).get("cursor", [""])[0]
+
+        try:
+            _ensure_camoufox_ready()
+        except BrowserQrLoginUnavailable as exc:
+            raise UnsupportedOperationError(str(exc)) from exc
+
+        try:
+            browser_context = Camoufox(headless=True)
+        except Exception as exc:
+            raise UnsupportedOperationError(
+                "Favorites browser fallback could not start Camoufox. "
+                "Run `python -m camoufox fetch` first."
+            ) from exc
+
+        try:
+            with browser_context as browser:
+                try:
+                    context = browser.new_context(user_agent=USER_AGENT, locale="zh-CN")
+                    context.add_cookies([
+                        {
+                            "name": name,
+                            "value": str(value),
+                            "domain": ".xiaohongshu.com",
+                            "path": "/",
+                            "secure": True,
+                            "sameSite": "Lax",
+                        }
+                        for name, value in self.cookies.items()
+                        if name != "saved_at" and value is not None
+                    ])
+                    page = context.new_page()
+
+                    def _capture_response(resp) -> None:
+                        if target_path not in resp.url:
+                            return
+                        try:
+                            payload = resp.json()
+                        except Exception:
+                            return
+                        if (
+                            isinstance(payload, dict)
+                            and payload.get("success")
+                            and isinstance(payload.get("data"), dict)
+                        ):
+                            responses[_response_cursor(resp.url)] = payload["data"]
+                        elif isinstance(payload, dict):
+                            errors.append(payload)
+
+                    page.on("response", _capture_response)
+                    page.goto(f"{HOME_URL}/user/profile/{user_id}", wait_until="domcontentloaded", timeout=45_000)
+                    page.wait_for_timeout(2_500)
+
+                    try:
+                        page.get_by_text("收藏", exact=True).click(timeout=5_000)
+                    except PlaywrightTimeoutError as exc:
+                        raise XhsApiError("Browser fallback could not find the favorites tab") from exc
+
+                    deadline = time.time() + 12
+                    while time.time() < deadline and cursor not in responses:
+                        page.wait_for_timeout(500)
+
+                    scroll_attempts = 0
+                    while cursor not in responses and scroll_attempts < 8:
+                        page.mouse.wheel(0, 3_000)
+                        scroll_attempts += 1
+                        deadline = time.time() + 4
+                        while time.time() < deadline and cursor not in responses:
+                            page.wait_for_timeout(500)
+
+                    if cursor in responses:
+                        return responses[cursor]
+
+                    if errors:
+                        raise XhsApiError(
+                            f"Browser fallback API error: {json.dumps(errors[-1], ensure_ascii=False)[:300]}",
+                            code=errors[-1].get("code"),
+                            response=errors[-1],
+                        )
+                    raise XhsApiError("Browser fallback did not receive favorites data")
+                except XhsApiError:
+                    raise
+                except Exception as exc:
+                    raise XhsApiError("Favorites browser fallback failed while driving Camoufox") from exc
+        except XhsApiError:
+            raise
+        except Exception as exc:
+            raise UnsupportedOperationError(
+                "Favorites browser fallback could not start Camoufox. "
+                "Run `python -m camoufox fetch` first."
+            ) from exc
 
     def get_user_likes(self, user_id: str, cursor: str = "") -> dict[str, Any]:
         return self._main_api_get("/api/sns/web/v1/note/like/page", {