diff --git a/tests/test_client.py b/tests/test_client.py index eb55af7..953bd0c 100644 --- a/tests/test_client.py +++ b/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 = {} diff --git a/xhs_cli/client_mixins.py b/xhs_cli/client_mixins.py index b4dada7..7dcb6ff 100644 --- a/xhs_cli/client_mixins.py +++ b/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", {