From ba2742c8c103036c00ccc8ab22ef8845f41342b5 Mon Sep 17 00:00:00 2001 From: clarkchen Date: Wed, 17 Jun 2026 23:28:56 +0800 Subject: [PATCH 1/2] Fix QR login Camoufox addon startup --- tests/test_qr_login.py | 16 +++++++++++++++- xhs_cli/qr_login.py | 13 ++++++++++++- 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/tests/test_qr_login.py b/tests/test_qr_login.py index 4f397b9..74c54d9 100644 --- a/tests/test_qr_login.py +++ b/tests/test_qr_login.py @@ -4,7 +4,12 @@ from xhs_cli.command_normalizers import normalize_xhs_user_payload from xhs_cli.exceptions import XhsApiError -from xhs_cli.qr_login import BrowserQrLoginUnavailable, _normalize_browser_cookies, qrcode_login +from xhs_cli.qr_login import ( + BrowserQrLoginUnavailable, + _camoufox_launch_options, + _normalize_browser_cookies, + qrcode_login, +) class _FakeQrClient: @@ -219,6 +224,15 @@ def test_qrcode_login_falls_back_when_browser_backend_unavailable(monkeypatch): } +def test_browser_assisted_login_excludes_default_addons(): + from camoufox import DefaultAddons + + options = _camoufox_launch_options() + + assert options["headless"] is False + assert options["exclude_addons"] == [DefaultAddons.UBO] + + def test_normalize_browser_cookies_uses_allowlist(): cookies = _normalize_browser_cookies([ {"name": "a1", "value": "a1-value", "domain": ".xiaohongshu.com"}, diff --git a/xhs_cli/qr_login.py b/xhs_cli/qr_login.py index e7b189a..e1e3f4f 100644 --- a/xhs_cli/qr_login.py +++ b/xhs_cli/qr_login.py @@ -337,6 +337,16 @@ def _ensure_camoufox_ready() -> None: ) +def _camoufox_launch_options() -> dict[str, Any]: + """Return Camoufox launch options for browser-assisted QR login.""" + from camoufox import DefaultAddons + + return { + "headless": False, + "exclude_addons": [DefaultAddons.UBO], + } + + def _browser_assisted_qrcode_login( *, on_status: callable[[str], None] | None = None, @@ -347,6 +357,7 @@ def _browser_assisted_qrcode_login( try: from camoufox.sync_api import Camoufox + launch_options = _camoufox_launch_options() except ImportError as exc: raise BrowserQrLoginUnavailable( "Camoufox sync API is unavailable in the current environment." @@ -356,7 +367,7 @@ def _browser_assisted_qrcode_login( _emit_status(on_status, "🔑 Starting browser-assisted QR login...") - with Camoufox(headless=False) as browser: + with Camoufox(**launch_options) as browser: page = browser.new_page() def _handle_response(response) -> None: From e0584b38fa1e39931728945f21030112bfa07372 Mon Sep 17 00:00:00 2001 From: clarkchen Date: Wed, 17 Jun 2026 23:54:32 +0800 Subject: [PATCH 2/2] feat: save note images from read command --- tests/test_cli.py | 45 +++++++++++++++++++++ tests/test_downloads.py | 56 ++++++++++++++++++++++++++ tests/test_formatter.py | 26 ++++++++++++ xhs_cli/commands/reading.py | 47 +++++++++++++++++++++- xhs_cli/downloads.py | 68 ++++++++++++++++++++++++++++++++ xhs_cli/formatter_normalizers.py | 40 ++++++++++++++++++- xhs_cli/formatter_renderers.py | 2 + 7 files changed, 282 insertions(+), 2 deletions(-) create mode 100644 tests/test_downloads.py create mode 100644 xhs_cli/downloads.py diff --git a/tests/test_cli.py b/tests/test_cli.py index f4bec77..3565188 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,10 +1,13 @@ """Tests for CLI commands using Click's test runner.""" +from pathlib import Path + import pytest import yaml from click.testing import CliRunner from xhs_cli.cli import cli +from xhs_cli.downloads import SavedImage from xhs_cli.exceptions import NoCookieError, SessionExpiredError, UnsupportedOperationError runner = CliRunner() @@ -266,6 +269,11 @@ def test_read_help_mentions_short_index(self): assert result.exit_code == 0 assert "index" in result.output.lower() + def test_read_help_mentions_save_images(self): + result = runner.invoke(cli, ["read", "--help"]) + assert result.exit_code == 0 + assert "--save-images" in result.output + def test_comments_help_mentions_short_index(self): result = runner.invoke(cli, ["comments", "--help"]) assert result.exit_code == 0 @@ -302,6 +310,43 @@ def fake_handle_command(ctx, action, render, as_json, as_yaml): assert called["kwargs"]["xsec_token"] == "token-abc" assert called["kwargs"]["xsec_source"] == "pc_search" + def test_read_save_images_outputs_saved_paths(self, monkeypatch, tmp_path): + class FakeClient: + def get_note_detail(self, note_id, **kwargs): + assert note_id == "note-abc" + return FAKE_NOTE_RESPONSE + + def fake_run_client_action(ctx, action): + return action(FakeClient()) + + output_dir = tmp_path / "downloads" + saved_path = output_dir / "note-abc" / "image-1.jpg" + + monkeypatch.setattr("xhs_cli.commands.reading.run_client_action", fake_run_client_action) + monkeypatch.setattr( + "xhs_cli.commands.reading.download_note_images", + lambda data, output: [ + SavedImage( + index=1, + url="https://img.example/image.jpg", + path=Path(output) / "note-abc" / "image-1.jpg", + ) + ], + ) + + result = runner.invoke(cli, ["read", "note-abc", "--save-images", "--output", str(output_dir), "--yaml"]) + + assert result.exit_code == 0 + payload = yaml.safe_load(result.output) + assert payload["ok"] is True + assert payload["data"]["saved_images"] == [ + { + "index": 1, + "url": "https://img.example/image.jpg", + "path": str(saved_path), + } + ] + def test_comments_index_resolves_note_context(self, monkeypatch): monkeypatch.setattr( "xhs_cli.note_refs.get_note_by_index", diff --git a/tests/test_downloads.py b/tests/test_downloads.py new file mode 100644 index 0000000..234c525 --- /dev/null +++ b/tests/test_downloads.py @@ -0,0 +1,56 @@ +"""Unit tests for media download helpers.""" + +from xhs_cli.downloads import download_note_images + + +class FakeResponse: + content = b"image-bytes" + + def raise_for_status(self): + return None + + +class FakeHttpClient: + requested_urls = [] + + def __init__(self, **kwargs): + self.kwargs = kwargs + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return None + + def get(self, url): + self.requested_urls.append(url) + return FakeResponse() + + +def test_download_note_images_saves_under_note_directory(monkeypatch, tmp_path): + FakeHttpClient.requested_urls = [] + monkeypatch.setattr("xhs_cli.downloads.httpx.Client", FakeHttpClient) + + data = { + "items": [ + { + "id": "note/abc", + "note_card": { + "image_list": [ + { + "url_default": "https://img.example/path/photo.webp!nd_dft_wgth_webp_3", + } + ] + }, + } + ] + } + + saved = download_note_images(data, tmp_path) + + assert FakeHttpClient.requested_urls == [ + "https://img.example/path/photo.webp!nd_dft_wgth_webp_3", + ] + assert len(saved) == 1 + assert saved[0].path == tmp_path / "note-abc" / "image-1.webp" + assert saved[0].path.read_bytes() == b"image-bytes" diff --git a/tests/test_formatter.py b/tests/test_formatter.py index c8ae6d2..d0b4b11 100644 --- a/tests/test_formatter.py +++ b/tests/test_formatter.py @@ -1,6 +1,7 @@ """Unit tests for formatter (no network required).""" from xhs_cli.formatter import coerce_int, extract_note_id, format_count, parse_note_reference +from xhs_cli.formatter_normalizers import normalize_note_detail class TestFormatCount: @@ -60,3 +61,28 @@ def test_extracts_token_and_source(self): assert note_id == "abc123" assert token == "token-1" assert source == "pc_search" + + +class TestNormalizeNoteDetail: + def test_extracts_image_urls(self): + note = normalize_note_detail({ + "items": [ + { + "note_card": { + "title": "Test", + "image_list": [ + {"url_default": "https://img.example/default.jpg"}, + {"url_list": ["https://img.example/list.jpg"]}, + {"info_list": [{"url_pre": "https://img.example/pre.jpg"}]}, + ], + } + } + ] + }) + + assert note["image_count"] == 3 + assert note["image_urls"] == [ + "https://img.example/default.jpg", + "https://img.example/list.jpg", + "https://img.example/pre.jpg", + ] diff --git a/xhs_cli/commands/reading.py b/xhs_cli/commands/reading.py index 33f5980..76087ed 100644 --- a/xhs_cli/commands/reading.py +++ b/xhs_cli/commands/reading.py @@ -1,12 +1,16 @@ """Reading commands: search, read, comments, sub-comments, user, user-posts, feed, hot, topics, search-user.""" +from pathlib import Path + import click from ..command_normalizers import normalize_paged_notes from ..cookies import cache_note_context +from ..downloads import DEFAULT_DOWNLOAD_DIR, download_note_images from ..formatter import ( maybe_print_structured, print_info, + print_success, render_comments, render_feed, render_note, @@ -82,9 +86,26 @@ def _search_action(client): @click.command() @click.argument("id_or_url") @click.option("--xsec-token", default="", help="Security token (or reuse a cached token for this note)") +@click.option("--save-images", is_flag=True, help="Download note images after reading") +@click.option( + "--output", + "output_dir", + default=DEFAULT_DOWNLOAD_DIR, + type=click.Path(file_okay=False, dir_okay=True, path_type=Path), + show_default=True, + help="Directory for --save-images downloads", +) @structured_output_options @click.pass_context -def read(ctx, id_or_url: str, xsec_token: str, as_json: bool, as_yaml: bool): +def read( + ctx, + id_or_url: str, + xsec_token: str, + save_images: bool, + output_dir: Path, + as_json: bool, + as_yaml: bool, +): """Read a note by ID, URL, or short index.""" note_id, token, url_source = resolve_note_reference(id_or_url, xsec_token=xsec_token) xsec_source = url_source or "pc_feed" @@ -97,6 +118,30 @@ def _read_action(client): kwargs["xsec_source"] = url_source return client.get_note_detail(note_id, **kwargs) + if save_images: + try: + data = run_client_action(ctx, _read_action) + saved_images = download_note_images(data, output_dir) + if not maybe_print_structured( + { + "note": data, + "saved_images": [ + {"index": image.index, "url": image.url, "path": str(image.path)} + for image in saved_images + ], + }, + as_json=as_json, + as_yaml=as_yaml, + ): + render_note(data) + if saved_images: + print_success(f"Saved {len(saved_images)} image(s) to {saved_images[0].path.parent}") + else: + print_info("No images to save") + return + except Exception as exc: + exit_for_error(exc, as_json=as_json, as_yaml=as_yaml) + handle_command( ctx, action=_read_action, diff --git a/xhs_cli/downloads.py b/xhs_cli/downloads.py new file mode 100644 index 0000000..5eb921a --- /dev/null +++ b/xhs_cli/downloads.py @@ -0,0 +1,68 @@ +"""Helpers for downloading media referenced by note payloads.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from pathlib import Path +from urllib.parse import urlparse + +import httpx + +from .formatter_normalizers import normalize_note_detail + +DEFAULT_DOWNLOAD_DIR = "xhs-downloads" + + +@dataclass(frozen=True) +class SavedImage: + index: int + url: str + path: Path + + +def _safe_path_part(value: str, fallback: str) -> str: + cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", value).strip(".-") + return cleaned or fallback + + +def _image_extension(url: str) -> str: + parsed = urlparse(url) + path = parsed.path.split("!", 1)[0] + suffix = Path(path).suffix.lower() + if suffix in {".jpg", ".jpeg", ".png", ".webp", ".gif", ".avif"}: + return suffix + return ".jpg" + + +def _note_id_from_detail(data: dict) -> str: + items = data.get("items", []) + if not items: + return "note" + note_card = items[0].get("note_card", {}) + note_id = items[0].get("id") or note_card.get("note_id") + return _safe_path_part(str(note_id or ""), "note") + + +def download_note_images(data: dict, output_dir: Path | str = DEFAULT_DOWNLOAD_DIR) -> list[SavedImage]: + """Download images referenced by a note detail payload.""" + note = normalize_note_detail(data) + if not note: + return [] + + image_urls = note.get("image_urls", []) + if not image_urls: + return [] + + note_dir = Path(output_dir) / _note_id_from_detail(data) + note_dir.mkdir(parents=True, exist_ok=True) + + saved_images = [] + with httpx.Client(follow_redirects=True, timeout=30) as client: + for index, url in enumerate(image_urls, 1): + response = client.get(url) + response.raise_for_status() + image_path = note_dir / f"image-{index}{_image_extension(url)}" + image_path.write_bytes(response.content) + saved_images.append(SavedImage(index=index, url=url, path=image_path)) + return saved_images diff --git a/xhs_cli/formatter_normalizers.py b/xhs_cli/formatter_normalizers.py index 0e1e8b9..daeeeb2 100644 --- a/xhs_cli/formatter_normalizers.py +++ b/xhs_cli/formatter_normalizers.py @@ -5,6 +5,41 @@ from typing import Any +def _extract_image_url(image: dict[str, Any]) -> str: + for key in ("url_default", "url_pre", "url"): + value = image.get(key) + if isinstance(value, str) and value: + return value + + for key in ("url_list", "info_list"): + values = image.get(key) + if not isinstance(values, list): + continue + for value in values: + if isinstance(value, str) and value: + return value + if isinstance(value, dict): + url = _extract_image_url(value) + if url: + return url + + return "" + + +def _extract_image_urls(images: Any) -> list[str]: + if not isinstance(images, list): + return [] + urls = [] + for image in images: + if isinstance(image, dict): + url = _extract_image_url(image) + if url: + urls.append(url) + elif isinstance(image, str) and image: + urls.append(image) + return urls + + def _coerce_int(value: Any, default: int = 0) -> int: if isinstance(value, bool): return int(value) @@ -47,6 +82,8 @@ def normalize_note_detail(data: dict[str, Any]) -> dict[str, Any] | None: interact = note.get("interact_info", {}) tags = note.get("tag_list", []) + image_list = note.get("image_list", []) + return { "title": note.get("title", "Untitled"), "desc": note.get("desc", ""), @@ -56,7 +93,8 @@ def normalize_note_detail(data: dict[str, Any]) -> dict[str, Any] | None: "comment_count": interact.get("comment_count", "0"), "share_count": interact.get("share_count", "0"), "tags": [tag.get("name", "") for tag in tags if tag.get("name")], - "image_count": len(note.get("image_list", [])), + "image_count": len(image_list) if isinstance(image_list, list) else 0, + "image_urls": _extract_image_urls(image_list), } diff --git a/xhs_cli/formatter_renderers.py b/xhs_cli/formatter_renderers.py index c29669a..7be8757 100644 --- a/xhs_cli/formatter_renderers.py +++ b/xhs_cli/formatter_renderers.py @@ -125,6 +125,8 @@ def render_note(data: dict[str, Any]) -> None: if note["image_count"]: table.add_row("图片", f"{note['image_count']} 张") + for index, image_url in enumerate(note.get("image_urls", []), 1): + table.add_row(f"图片 {index}", f"[link={image_url}]{image_url}[/link]") console.print(Panel(table, title=f"📝 {title}", border_style="green"))