From 2b5663a313b3b5a67410bfc2770aad72dd8d1d1e Mon Sep 17 00:00:00 2001 From: bee Date: Mon, 6 Jul 2026 12:12:21 +0800 Subject: [PATCH] feat: add text2img command for text-to-image generation Call text2imgv3 API to generate multi-style cover images from text. - Add client.text2img() method, reuse _creator_post and _search_request_id - Add xhs text2img command with --text/--page/--download options - Slugify download filenames + path validation, non-zero exit on download failure Closes #69 --- xhs_cli/cli.py | 2 + xhs_cli/client_mixins.py | 44 +++++++++++++ xhs_cli/commands/creator.py | 123 +++++++++++++++++++++++++++++++++++- 3 files changed, 168 insertions(+), 1 deletion(-) diff --git a/xhs_cli/cli.py b/xhs_cli/cli.py index a33a08e..32307d9 100644 --- a/xhs_cli/cli.py +++ b/xhs_cli/cli.py @@ -21,6 +21,7 @@ xhs unread xhs post --title "..." --body "..." --images img.png xhs delete [-y] + xhs text2img --text "内容" [--page 1] [--download ./out/] """ from __future__ import annotations @@ -109,6 +110,7 @@ def cli(ctx, verbose: bool, cookie_source: str): cli.add_command(creator.post) cli.add_command(creator.my_notes) cli.add_command(creator.delete) +cli.add_command(creator.text2img) # ─── Notification commands ────────────────────────────────────────────────── diff --git a/xhs_cli/client_mixins.py b/xhs_cli/client_mixins.py index b4dada7..1d7af8e 100644 --- a/xhs_cli/client_mixins.py +++ b/xhs_cli/client_mixins.py @@ -627,6 +627,50 @@ def get_creator_note_list(self, tab: int = 0, page: int = 0) -> dict[str, Any]: "page": page, }) + def text2img( + self, + text: str, + page_num: int = 1, + style_json: str = "", + ) -> dict[str, Any]: + """调用小红书"文字配图"接口,根据文字生成多风格封面图片。 + + 接口一次返回多个风格(基础/弥散/插图/简约/涂写等)的图片 URL。 + + Args: + text: 要生成图片的文字内容 + page_num: 风格分页,1 表示第一批风格(接口按 has_next 翻页) + style_json: 指定样式的 JSON 字符串,空字符串表示用推荐样式 + + Returns: + 接口 data 字段,含 has_next 和 text2_img_type_results(各风格图片列表) + """ + request_id = self._search_request_id() + data = { + "page_type": "cover", + "text_config_list": [ + { + "index": 0, + "text": text, + "text_attribute": json.dumps({ + "edit_content": text, + "text_list": [text], + "text_size": 0, + "line_space": 0, + "word_space": 0, + "use_system_fonts": 0, + }, separators=(",", ":"), ensure_ascii=False), + } + ], + "style_json": style_json, + "page_num": page_num, + "request_id": request_id, + "extra_info": "", + } + return self._creator_post( + "/api/galaxy/v2/creator/post/inspiration/text2imgv3", data + ) + class SocialEndpointsMixin: """Social graph and saved-content endpoints.""" diff --git a/xhs_cli/commands/creator.py b/xhs_cli/commands/creator.py index 63237f5..da291a8 100644 --- a/xhs_cli/commands/creator.py +++ b/xhs_cli/commands/creator.py @@ -1,10 +1,12 @@ -"""Creator commands: post, my-notes, delete.""" +"""Creator commands: post, my-notes, delete, text2img.""" import re +from pathlib import Path import click from ..command_normalizers import select_topic_payload +from ..exceptions import XhsApiError from ..formatter import ( extract_note_id, maybe_print_structured, @@ -123,3 +125,122 @@ def delete(ctx, id_or_url: str, as_json: bool, as_yaml: bool, yes: bool): print_success(f"Deleted note {note_id}") except Exception as exc: exit_for_error(exc, as_json=as_json, as_yaml=as_yaml) + + +@click.command("text2img") +@click.option("--text", required=True, help="Text content to render as image") +@click.option( + "--page", "page_num", default=1, show_default=True, + help="Style page number (1 = first batch of styles)", +) +@click.option( + "--download", "download_dir", default=None, + help="Download all style images to this directory (default: print URLs only)", +) +@structured_output_options +@click.pass_context +def text2img( + ctx, + text: str, + page_num: int, + download_dir: str | None, + as_json: bool, + as_yaml: bool, +): + """Generate cover images from text via Xiaohongshu text2imgv3 API. + + Returns multiple style variants (基础/弥散/插图/简约/涂写 etc.) per request. + By default prints each style's image URL; use --download to save them locally. + """ + + def _text2img_action(client): + data = client.text2img(text=text, page_num=page_num) + if download_dir: + results = data.get("text2_img_type_results", []) if isinstance(data, dict) else [] + failures = _download_images(results, download_dir) + if failures: + failed_names = ", ".join(a for a, _ in failures) + raise XhsApiError( + f"{len(failures)} image(s) failed to download: {failed_names}" + ) + return data + + def _render(data): + results = data.get("text2_img_type_results", []) if isinstance(data, dict) else [] + if not results: + print_info("No image styles returned") + return + + print_success(f"Generated {len(results)} style(s):") + for i, item in enumerate(results): + album = item.get("album_name", "unknown") + img_url = item.get("img_url", "") + config_id = item.get("config_id", "") + print(f" [{i}] {album} (config_id={config_id})") + print(f" {img_url}") + + has_next = data.get("has_next", False) if isinstance(data, dict) else False + if has_next: + print_info("More styles available, use --page 2 for next batch") + + handle_command( + ctx, + action=_text2img_action, + render=_render, + as_json=as_json, + as_yaml=as_yaml, + ) + + +def _download_images( + results: list[dict], download_dir: str +) -> list[tuple[str, str]]: + """下载生成的各风格图片到指定目录。 + + 返回失败列表 [(album, error_message), ...],调用方据此决定是否非零退出。 + 文件名做 slug 化处理并校验解析后路径仍在 download_dir 内,防止接口返回值越权写入。 + """ + import httpx + + out_dir = Path(download_dir).resolve() + out_dir.mkdir(parents=True, exist_ok=True) + failures: list[tuple[str, str]] = [] + used_names: set[str] = set() + + for i, item in enumerate(results): + album = item.get("album_name") or f"style_{i}" + img_url = item.get("img_url", "") + if not img_url: + continue + + # slug 化文件名:只保留非路径分隔符的字符,防止 ../ 或绝对路径注入 + safe_stem = re.sub(r"[^\w\u4e00-\u9fff.()-]+", "_", album).strip("._") or f"style_{i}" + ext = ".jpg" + for e in (".png", ".jpg", ".jpeg", ".webp"): + if e in img_url: + ext = e + break + # 同名去重 + filename = f"{safe_stem}{ext}" + n = 1 + while filename in used_names: + filename = f"{safe_stem}_{n}{ext}" + n += 1 + used_names.add(filename) + + # 校验解析后路径仍在 download_dir 内 + filepath = (out_dir / filename).resolve() + if not filepath.is_relative_to(out_dir): + failures.append((album, f"resolved path escapes download dir: {filepath}")) + continue + + try: + resp = httpx.get(img_url, timeout=30.0, follow_redirects=True) + resp.raise_for_status() + filepath.write_bytes(resp.content) + print_success(f"Downloaded: {filepath}") + except Exception as exc: + failures.append((album, str(exc))) + print_info(f"Failed to download {album}: {exc}") + + return failures