Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 45 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
@@ -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()
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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",
Expand Down
56 changes: 56 additions & 0 deletions tests/test_downloads.py
Original file line number Diff line number Diff line change
@@ -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"
26 changes: 26 additions & 0 deletions tests/test_formatter.py
Original file line number Diff line number Diff line change
@@ -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:
Expand Down Expand Up @@ -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",
]
16 changes: 15 additions & 1 deletion tests/test_qr_login.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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"},
Expand Down
47 changes: 46 additions & 1 deletion xhs_cli/commands/reading.py
Original file line number Diff line number Diff line change
@@ -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,
Expand Down Expand Up @@ -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"
Expand All @@ -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,
Expand Down
68 changes: 68 additions & 0 deletions xhs_cli/downloads.py
Original file line number Diff line number Diff line change
@@ -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
40 changes: 39 additions & 1 deletion xhs_cli/formatter_normalizers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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", ""),
Expand All @@ -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),
}


Expand Down
Loading