diff --git a/README.md b/README.md index 69a6168..654cd2a 100644 --- a/README.md +++ b/README.md @@ -89,7 +89,8 @@ xhs comments "" --all --json # All comments as JSON xhs comments --xsec-token T # Use note_id + explicit xsec_token xhs comments # Reuse cached token if available xhs sub-comments # View replies to a comment -xhs user # User profile +xhs user # User profile; profile URL may include xsec_token/xsec_source +xhs user --xsec-token T --xsec-source pc_search # User profile with explicit xsec context xhs user-posts # User's published notes xhs user-posts --cursor X # Paginate with cursor @@ -382,7 +383,8 @@ xhs comments "" --all --json # 全部评论，JSON 格式 xhs comments --xsec-token T # 用 note_id + 显式 xsec_token xhs comments # 如果之前访问过 URL，会复用缓存 token xhs sub-comments # 查看评论的回复 -xhs user # 用户主页 +xhs user # 用户主页；主页 URL 可携带 xsec_token/xsec_source +xhs user --xsec-token T --xsec-source pc_search # 显式传 xsec 上下文 xhs user-posts # 用户发布的笔记 # 发现 diff --git a/tests/test_cli.py b/tests/test_cli.py index f4bec77..28eec46 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -351,6 +351,37 @@ def test_search_empty_results_clear_previous_index(self, monkeypatch): assert saved == [[]] + def test_user_profile_url_passes_xsec_context(self, monkeypatch): + called = {} + + class FakeClient: + def get_user_info(self, user_id, xsec_token="", xsec_source=""): + called["user_id"] = user_id + called["xsec_token"] = xsec_token + called["xsec_source"] = xsec_source + return {} + + def fake_handle_command(ctx, action, render, as_json, as_yaml): + action(FakeClient()) + return None + + monkeypatch.setattr("xhs_cli.commands.reading.handle_command", fake_handle_command) + + result = runner.invoke( + cli, + [ + "user", + "https://www.xiaohongshu.com/user/profile/user-1?xsec_token=token-abc&xsec_source=pc_search", + ], + ) + + assert result.exit_code == 0 + assert called == { + "user_id": "user-1", + "xsec_token": "token-abc", + "xsec_source": "pc_search", + } + def test_user_posts_saves_index_entries(self, monkeypatch): saved = [] monkeypatch.setattr("xhs_cli.note_refs.save_note_index", lambda items: saved.append(items)) diff --git a/tests/test_client.py b/tests/test_client.py index eb55af7..d53529d 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -69,6 +69,44 @@ def fake_post(self, uri, data): client.close() +class TestUserProfileEndpoint: + def test_get_user_info_fetches_profile_html_with_xsec_context(self, monkeypatch): + captured = {} + html = ( + '" + ) + + def fake_request(self, method, url, **kwargs): + captured["method"] = method + captured["url"] = url + captured["headers"] = kwargs.get("headers", {}) + return httpx.Response(200, text=html) + + monkeypatch.setattr(XhsClient, "_request_with_retry", fake_request) + + client = XhsClient({"a1": "cookie"}) + try: + result = client.get_user_info( + "user-1", + xsec_token="token-abc", + xsec_source="pc_search", + ) + finally: + client.close() + + assert captured["method"] == "GET" + assert captured["url"] == ( + "https://www.xiaohongshu.com/user/profile/user-1" + "?xsec_token=token-abc&xsec_source=pc_search" + ) + assert "a1=cookie" in captured["headers"]["cookie"] + assert result["basicInfo"]["nickname"] == "Alice" + + class TestTransportCookies: def test_request_with_retry_merges_response_cookies(self, monkeypatch): request = httpx.Request("POST", "https://edith.xiaohongshu.com/api/test") diff --git a/xhs_cli/cli.py b/xhs_cli/cli.py index a33a08e..82b942c 100644 --- a/xhs_cli/cli.py +++ b/xhs_cli/cli.py @@ -5,7 +5,7 @@ xhs search [--sort popular|latest] [--type video|image] [--page N] xhs read [--xsec-token TOKEN] xhs comments - xhs user + xhs user [--xsec-token TOKEN] [--xsec-source SOURCE] xhs user-posts [--cursor CURSOR] xhs feed xhs hot [--category CATEGORY] diff --git a/xhs_cli/client_mixins.py b/xhs_cli/client_mixins.py index b4dada7..54fee47 100644 --- a/xhs_cli/client_mixins.py +++ b/xhs_cli/client_mixins.py @@ -12,6 +12,7 @@ from collections import OrderedDict from pathlib import Path from typing import Any +from urllib.parse import urlencode from .constants import CREATOR_HOST, HOME_URL, UPLOAD_HOST, USER_AGENT from .cookies import ( @@ -22,7 +23,7 @@ invalidate_note_context, ) from .exceptions import NeedVerifyError, UnsupportedOperationError, XhsApiError -from .html_parser import extract_note_from_html +from .html_parser import extract_note_from_html, extract_user_from_html logger = logging.getLogger(__name__) @@ -219,6 +220,27 @@ def _fetch_note_html( ) return resp.text + def _fetch_user_html( + self, + user_id: str, + xsec_token: str = "", + xsec_source: str = "", + ) -> str: + url = f"{HOME_URL}/user/profile/{user_id}" + if xsec_token and xsec_source: + url = f"{url}?{urlencode({'xsec_token': xsec_token, 'xsec_source': xsec_source})}" + + resp = self._request_with_retry( + "GET", + url, + headers={ + "user-agent": USER_AGENT, + "referer": f"{HOME_URL}/", + "cookie": cookies_to_string(self.cookies), + }, + ) + return resp.text + def resolve_xsec_context( self, note_id: str, @@ -258,10 +280,14 @@ def resolve_xsec_token(self, note_id: str, preferred_token: str = "") -> str: def get_self_info(self) -> dict[str, Any]: return self._main_api_get("/api/sns/web/v2/user/me") - def get_user_info(self, user_id: str) -> dict[str, Any]: - return self._main_api_get("/api/sns/web/v1/user/otherinfo", { - "target_user_id": user_id, - }) + def get_user_info( + self, + user_id: str, + xsec_token: str = "", + xsec_source: str = "", + ) -> dict[str, Any]: + html = self._fetch_user_html(user_id, xsec_token=xsec_token, xsec_source=xsec_source) + return extract_user_from_html(html) def get_user_notes(self, user_id: str, cursor: str = "") -> dict[str, Any]: return self._main_api_get("/api/sns/web/v1/user_posted", { diff --git a/xhs_cli/commands/reading.py b/xhs_cli/commands/reading.py index 33f5980..3b21175 100644 --- a/xhs_cli/commands/reading.py +++ b/xhs_cli/commands/reading.py @@ -6,6 +6,7 @@ from ..cookies import cache_note_context from ..formatter import ( maybe_print_structured, + parse_user_reference, print_info, render_comments, render_feed, @@ -149,13 +150,20 @@ def _render_comments(data): @click.command() @click.argument("user_id") +@click.option("--xsec-token", default="", help="Security token from a profile URL") +@click.option("--xsec-source", default="", help="Security source from a profile URL") @structured_output_options @click.pass_context -def user(ctx, user_id: str, as_json: bool, as_yaml: bool): +def user(ctx, user_id: str, xsec_token: str, xsec_source: str, as_json: bool, as_yaml: bool): """View user profile info.""" + parsed_user_id, url_token, url_source = parse_user_reference(user_id) handle_command( ctx, - action=lambda client: client.get_user_info(user_id), + action=lambda client: client.get_user_info( + parsed_user_id, + xsec_token=xsec_token or url_token, + xsec_source=xsec_source or url_source, + ), render=render_user_info, as_json=as_json, as_yaml=as_yaml, diff --git a/xhs_cli/formatter.py b/xhs_cli/formatter.py index d7d8766..4f19b6d 100644 --- a/xhs_cli/formatter.py +++ b/xhs_cli/formatter.py @@ -65,3 +65,18 @@ def extract_note_id(id_or_url: str) -> str: """Extract note ID from URL or return as-is (drops query params).""" note_id, _ = parse_note_url(id_or_url) return note_id + + +def parse_user_reference(id_or_url: str) -> tuple[str, str, str]: + """Extract user ID, xsec_token, and xsec_source from a profile URL or plain ID.""" + if "xiaohongshu.com" in id_or_url: + from urllib.parse import parse_qs, urlparse + + parsed = urlparse(id_or_url) + parts = parsed.path.rstrip("/").split("/") + user_id = parts[-1] + qs = parse_qs(parsed.query) + xsec_token = qs.get("xsec_token", [""])[0] + xsec_source = qs.get("xsec_source", [""])[0] + return user_id, xsec_token, xsec_source + return id_or_url, "", "" diff --git a/xhs_cli/formatter_normalizers.py b/xhs_cli/formatter_normalizers.py index 0e1e8b9..e708214 100644 --- a/xhs_cli/formatter_normalizers.py +++ b/xhs_cli/formatter_normalizers.py @@ -19,7 +19,7 @@ def _coerce_int(value: Any, default: int = 0) -> int: def normalize_user_info(data: dict[str, Any]) -> dict[str, Any]: - basic = data.get("basic_info", data) + basic = data.get("basic_info", data.get("basicInfo", data)) interactions = data.get("interactions", []) stats = {} @@ -28,10 +28,13 @@ def normalize_user_info(data: dict[str, Any]) -> dict[str, Any]: return { "nickname": basic.get("nickname", basic.get("nick_name", "Unknown")), - "red_id": basic.get("red_id", ""), + "red_id": basic.get("red_id", basic.get("redId", "")), "desc": basic.get("desc", ""), - "ip_location": basic.get("ip_location", ""), - "user_id": basic.get("user_id", data.get("user_id", "")), + "ip_location": basic.get("ip_location", basic.get("ipLocation", "")), + "user_id": basic.get( + "user_id", + basic.get("userId", data.get("user_id", data.get("userId", ""))), + ), "gender": basic.get("gender"), "stats": stats, } diff --git a/xhs_cli/html_parser.py b/xhs_cli/html_parser.py index 9a22cae..0c4fa16 100644 --- a/xhs_cli/html_parser.py +++ b/xhs_cli/html_parser.py @@ -71,3 +71,18 @@ def extract_note_from_html(html: str, note_id: str) -> dict[str, Any]: """High-level: parse HTML → extract note in one step.""" state = parse_initial_state(html) return extract_note_from_state(state, note_id) + + +def extract_user_from_state(state: dict[str, Any]) -> dict[str, Any]: + """Extract profile data from a parsed XHS user homepage state.""" + user_page_data = state.get("user", {}).get("userPageData") + if isinstance(user_page_data, dict): + return user_page_data + + raise XhsApiError("User profile not found in HTML state") + + +def extract_user_from_html(html: str) -> dict[str, Any]: + """High-level: parse HTML → extract user profile in one step.""" + state = parse_initial_state(html) + return extract_user_from_state(state)