From 5530e7c60efca82a025837eb8bda24f5645e6788 Mon Sep 17 00:00:00 2001 From: Murphy Liu Date: Sat, 9 May 2026 00:28:51 -0700 Subject: [PATCH] feat: add video note publishing (post-video) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ports the upstream `post-video` command from jackwener/xiaohongshu-cli#15 and threads the per-request `uploadAddr` from the upload permit response through both image and video upload paths so it works on rednote too. Behavioral changes: - New `xhs post-video` command (--video required; --cover optional; topics from --topic flags + body hashtags; --private supported). - `create_image_note` now POSTs via `_creator_post` (`webapi.rednote.com` / `creator.xiaohongshu.com`) instead of the edith host. This matches the path the web client actually uses and is the upstream PR's fix for "Chinese title/body returns API error when posting." `business_binds` is now serialized with `ensure_ascii=False` for the same reason. - `get_upload_permit` returns the `uploadAddr` field from the permit response. `upload_file` / `upload_video` accept an `upload_addr` override; falls back to the static `UPLOAD_HOST` constant when empty (preserves xiaohongshu behavior). Refactor: - Note creation paths share a single `_post_note` helper instead of two near-identical 30-line builders. Test plan: - `uv run pytest` — 114 passed under both XHS_TARGET=xiaohongshu (default) and XHS_TARGET=rednote. - `ruff check` clean. - `xhs post-video --help` and validation errors (missing file, unsupported extension) rendered correctly. --- README.md | 12 ++-- xhs_cli/cli.py | 1 + xhs_cli/client_mixins.py | 89 +++++++++++++++++++++++++---- xhs_cli/commands/creator.py | 101 ++++++++++++++++++++++++++++++++- xhs_cli/formatter_renderers.py | 2 +- 5 files changed, 187 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index ca1082f..dae5524 100644 --- a/README.md +++ b/README.md @@ -25,7 +25,7 @@ A CLI for Xiaohongshu (小红书) — search, read, interact, and post via rever - 📰 **Feed** — recommendation feed, hot/trending by category - 👥 **Social** — follow/unfollow, favorites - 👍 **Interactions** — like, favorite, comment, reply, delete -- ✍️ **Creator** — post image notes, my-notes list, delete +- ✍️ **Creator** — post image / video notes, my-notes list, delete - 🔔 **Notifications** — unread count, mentions, likes, new followers - 🛡️ **Anti-detection** — consistent macOS Chrome fingerprint, `sec-ch-ua` alignment, session-stable browser identity, Gaussian jitter, captcha cooldown, exponential backoff - 📊 **Structured output** — commands support `--yaml` and `--json`; non-TTY stdout defaults to YAML @@ -133,7 +133,9 @@ xhs delete-comment # Delete own comment # ─── Creator ───────────────────────────────────── xhs my-notes # List own notes (v2 creator endpoint) xhs my-notes --page 1 # Next page -xhs post --title "标题" --body "正文" --images img.jpg # Post note +xhs post --title "标题" --body "正文" --images img.jpg # Post image note +xhs post-video --title "Trip" --body "..." --video clip.mp4 # Post video note (.mp4/.mov) +xhs post-video --title "Trip" --body "..." --video clip.mov --cover cover.jpg # With custom cover xhs delete # Delete note xhs delete -y # Skip confirmation @@ -341,7 +343,7 @@ The built-in Gaussian jitter delay (~1-1.5s between requests) is intentional to - 📰 **发现** — 推荐 Feed、按分类浏览热门 - 👥 **社交** — 关注/取关、收藏夹 - 👍 **互动** — 点赞、收藏、评论、回复、删除 -- ✍️ **创作者** — 发布图文笔记、我的笔记列表、删除 +- ✍️ **创作者** — 发布图文 / 视频笔记、我的笔记列表、删除 - 🔔 **通知** — 未读数、@、点赞、新关注 - 🛡️ **反风控** — macOS Chrome 指纹一致性、session 级浏览器身份持久化、高斯抖动延迟、验证码自动冷却、指数退避重试 - 📊 **结构化输出** — `--yaml` / `--json`，非 TTY 默认输出 YAML @@ -434,7 +436,9 @@ xhs delete-comment # 删除自己的评论 # 创作者 xhs my-notes # 我的笔记列表 -xhs post --title "标题" --body "正文" --images img.jpg # 发布笔记 +xhs post --title "标题" --body "正文" --images img.jpg # 发布图文笔记 +xhs post-video --title "标题" --body "正文" --video clip.mp4 # 发布视频笔记（.mp4/.mov） +xhs post-video --title "标题" --body "正文" --video clip.mov --cover cover.jpg # 自定义封面 xhs delete # 删除笔记 xhs delete -y # 跳过确认 diff --git a/xhs_cli/cli.py b/xhs_cli/cli.py index a33a08e..5c592b5 100644 --- a/xhs_cli/cli.py +++ b/xhs_cli/cli.py @@ -107,6 +107,7 @@ def cli(ctx, verbose: bool, cookie_source: str): # ─── Creator commands ─────────────────────────────────────────────────────── cli.add_command(creator.post) +cli.add_command(creator.post_video) cli.add_command(creator.my_notes) cli.add_command(creator.delete) diff --git a/xhs_cli/client_mixins.py b/xhs_cli/client_mixins.py index b4dada7..3b7df5b 100644 --- a/xhs_cli/client_mixins.py +++ b/xhs_cli/client_mixins.py @@ -13,7 +13,7 @@ from pathlib import Path from typing import Any -from .constants import CREATOR_HOST, HOME_URL, UPLOAD_HOST, USER_AGENT +from .constants import HOME_URL, UPLOAD_HOST, USER_AGENT from .cookies import ( cache_note_context, cookies_to_string, @@ -544,7 +544,14 @@ def get_upload_permit(self, file_type: str = "image", count: int = 1) -> dict[st "source": "web", }) permit = data["uploadTempPermits"][0] - return {"fileId": permit["fileIds"][0], "token": permit["token"]} + return { + "fileId": permit["fileIds"][0], + "token": permit["token"], + "uploadAddr": permit.get("uploadAddr", ""), + } + + def get_video_upload_permit(self, count: int = 1) -> dict[str, str]: + return self.get_upload_permit(file_type="video", count=count) def upload_file( self, @@ -552,11 +559,13 @@ def upload_file( token: str, file_path: str, content_type: str | None = None, + upload_addr: str | None = None, ) -> None: with open(file_path, "rb") as f: file_data = f.read() - url = f"{UPLOAD_HOST}/{file_id}" + host = f"https://{upload_addr}" if upload_addr else UPLOAD_HOST + url = f"{host}/{file_id}" content_type = content_type or mimetypes.guess_type(file_path)[0] or "application/octet-stream" resp = self._request_with_retry( "PUT", @@ -570,6 +579,23 @@ def upload_file( if resp.status_code >= 400: raise XhsApiError(f"Upload failed: {resp.status_code} {resp.reason_phrase}") + def upload_video( + self, + file_id: str, + token: str, + file_path: str, + content_type: str | None = None, + upload_addr: str | None = None, + ) -> None: + content_type = content_type or mimetypes.guess_type(file_path)[0] or "video/mp4" + self.upload_file( + file_id, + token, + file_path, + content_type=content_type, + upload_addr=upload_addr, + ) + def create_image_note( self, title: str, @@ -579,6 +605,50 @@ def create_image_note( is_private: bool = False, ) -> Any: images = [{"file_id": fid, "metadata": {"source": -1}} for fid in image_file_ids] + return self._post_note( + note_type="normal", + title=title, + desc=desc, + topics=topics, + is_private=is_private, + image_info={"images": images}, + video_info=None, + ) + + def create_video_note( + self, + title: str, + desc: str, + video_file_id: str, + cover_file_id: str | None = None, + topics: list[dict[str, str]] | None = None, + is_private: bool = False, + ) -> Any: + video_info = {"file_id": video_file_id, "metadata": {"source": -1}} + image_info: dict[str, Any] | None = None + if cover_file_id: + image_info = {"images": [{"file_id": cover_file_id, "metadata": {"source": -1}}]} + return self._post_note( + note_type="video", + title=title, + desc=desc, + topics=topics, + is_private=is_private, + image_info=image_info, + video_info=video_info, + ) + + def _post_note( + self, + *, + note_type: str, + title: str, + desc: str, + topics: list[dict[str, str]] | None, + is_private: bool, + image_info: dict[str, Any] | None, + video_info: dict[str, Any] | None, + ) -> Any: business_binds = { "version": 1, "noteId": 0, @@ -588,24 +658,21 @@ def create_image_note( } data = { "common": { - "type": "normal", + "type": note_type, "title": title, "note_id": "", "desc": desc, "source": '{"type":"web","ids":"","extraInfo":"{\\"subType\\":\\"official\\"}"}', - "business_binds": json.dumps(business_binds), + "business_binds": json.dumps(business_binds, ensure_ascii=False), "ats": [], "hash_tag": topics or [], "post_loc": {}, "privacy_info": {"op_type": 1, "type": 1 if is_private else 0}, }, - "image_info": {"images": images}, - "video_info": None, + "image_info": image_info, + "video_info": video_info, } - return self._main_api_post("/web_api/sns/v2/note", data, { - "origin": CREATOR_HOST, - "referer": f"{CREATOR_HOST}/", - }) + return self._creator_post("/web_api/sns/v2/note", data) def delete_note(self, note_id: str) -> dict[str, Any]: try: diff --git a/xhs_cli/commands/creator.py b/xhs_cli/commands/creator.py index 63237f5..083db99 100644 --- a/xhs_cli/commands/creator.py +++ b/xhs_cli/commands/creator.py @@ -1,5 +1,6 @@ -"""Creator commands: post, my-notes, delete.""" +"""Creator commands: post, post-video, my-notes, delete.""" +import os import re import click @@ -49,7 +50,12 @@ def _publish(client): for img_path in images: print_info(f"Uploading {img_path}...") permit = client.get_upload_permit() - client.upload_file(permit["fileId"], permit["token"], img_path) + client.upload_file( + permit["fileId"], + permit["token"], + img_path, + upload_addr=permit.get("uploadAddr"), + ) file_ids.append(permit["fileId"]) print_success(f"Uploaded: {img_path}") @@ -84,6 +90,97 @@ def _publish(client): ) +SUPPORTED_VIDEO_EXTS = {".mp4", ".mov"} + + +@click.command("post-video") +@click.option("--title", required=True, help="Video note title") +@click.option("--body", required=True, help="Video note body text") +@click.option("--video", required=True, help="Video file path (.mp4 / .mov)") +@click.option("--cover", default=None, help="Cover image file path (optional)") +@click.option("--topic", "topics_flag", multiple=True, help="Topic(s)/hashtag(s) to search and attach") +@click.option("--private", "is_private", is_flag=True, help="Publish as private note") +@structured_output_options +@click.pass_context +def post_video( + ctx, + title: str, + body: str, + video: str, + cover: str | None, + topics_flag: tuple[str, ...], + is_private: bool, + as_json: bool, + as_yaml: bool, +): + """Publish a video note.""" + if not os.path.exists(video): + raise click.BadParameter(f"Video file not found: {video}", param_hint="--video") + ext = os.path.splitext(video)[1].lower() + if ext not in SUPPORTED_VIDEO_EXTS: + raise click.BadParameter( + f"Unsupported video format {ext!r}. Supported: {', '.join(sorted(SUPPORTED_VIDEO_EXTS))}", + param_hint="--video", + ) + if cover and not os.path.exists(cover): + raise click.BadParameter(f"Cover image not found: {cover}", param_hint="--cover") + + def _publish(client): + print_info(f"Uploading video {video}...") + video_permit = client.get_video_upload_permit() + client.upload_video( + video_permit["fileId"], + video_permit["token"], + video, + upload_addr=video_permit.get("uploadAddr"), + ) + print_success(f"Uploaded video: {video}") + + cover_file_id = None + if cover: + print_info(f"Uploading cover {cover}...") + cover_permit = client.get_upload_permit() + client.upload_file( + cover_permit["fileId"], + cover_permit["token"], + cover, + upload_addr=cover_permit.get("uploadAddr"), + ) + cover_file_id = cover_permit["fileId"] + print_success(f"Uploaded cover: {cover}") + + body_hashtags = extract_hashtags(body) + all_topics = list(topics_flag) + body_hashtags + unique_topics = list(dict.fromkeys(all_topics)) + if len(unique_topics) > 10: + print_info(f"Found {len(unique_topics)} topics, using first 10") + unique_topics = unique_topics[:10] + + resolved_topics = [] + for t in unique_topics: + topic_data = client.search_topics(t) + resolved_topics.extend(select_topic_payload(topic_data, t)) + + return client.create_video_note( + title=title, + desc=body, + video_file_id=video_permit["fileId"], + cover_file_id=cover_file_id, + topics=resolved_topics, + is_private=is_private, + ) + + handle_command( + ctx, + action=_publish, + render=lambda _data: print_success( + f"Video note published: {title}" + (" (private)" if is_private else "") + ), + as_json=as_json, + as_yaml=as_yaml, + ) + + @click.command("my-notes") @click.option("--page", default=0, help="Page number (0-indexed)") @structured_output_options diff --git a/xhs_cli/formatter_renderers.py b/xhs_cli/formatter_renderers.py index 0c680fc..badf40a 100644 --- a/xhs_cli/formatter_renderers.py +++ b/xhs_cli/formatter_renderers.py @@ -7,6 +7,7 @@ from rich.panel import Panel from rich.table import Table +from .constants import HOME_URL from .formatter_normalizers import ( normalize_comments, normalize_creator_notes, @@ -19,7 +20,6 @@ normalize_user_posts, normalize_users, ) -from .constants import HOME_URL from .formatter_utils import coerce_int, console, format_count, print_error, print_info