a

my_xiaohongshu_cli/README.md

@@ -0,0 +1,71 @@
+# My Xiaohongshu CLI
+
+基于自定义 Cookie 的小红书命令行工具。
+
+## 安装
+
+```bash
+cd /home/ubuntu/nanocode/my_iaohongshu_cli
+uv pip install -e .
+```
+
+或使用 pip:
+```bash
+pip install -e .
+```
+
+## 使用方法
+
+```bash
+# 检查登录状态
+myxhs status
+
+# 搜索笔记
+myxhs search "旅行"
+
+# 查看笔记详情
+myxhs read 69ddda7a000000002100786c
+
+# 获取首页推荐
+myxhs feed
+
+# 获取热搜榜
+myxhs hot
+
+# 查看用户信息
+myxhs user 5f3d2e1a000000000101c1f2
+
+# 查看用户笔记
+myxhs user_notes 5f3d2e1a000000000101c1f2
+
+# 点赞/取消点赞
+myxhs like 69ddda7a000000002100786c
+myxhs unlike 69ddda7a000000002100786c
+```
+
+## Cookie 文件
+
+默认从 `/home/ubuntu/xiaohongshu_cookies.txt` 读取 Cookie。
+
+可通过 `--cookie-file` 指定其他路径:
+```bash
+myxhs --cookie-file /path/to/cookies.json status
+```
+
+Cookie 文件格式支持:
+- Chrome 扩展导出的 JSON 数组格式
+- 简单的 key-value JSON 对象
+
+## 输出格式
+
+所有命令支持 `--json-output` 输出 JSON 格式:
+```bash
+myxhs search "美食" --json-output
+```
+
+## 调试
+
+使用 `-v` 开启调试日志:
+```bash
+myxhs -v search "编程"
+```

my_xiaohongshu_cli/pyproject.toml

@@ -0,0 +1,20 @@
+[project]
+name = "my-xiaohongshu-cli"
+version = "0.1.0"
+description = "My custom Xiaohongshu CLI with custom cookies"
+requires-python = ">=3.9"
+dependencies = [
+    "httpx>=0.27.0",
+    "click>=8.0.0",
+    "xhshow>=0.1.0",
+]
+
+[project.scripts]
+myxhs = "my_xhs_cli.cli:main"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/my_xhs_cli"]

my_xiaohongshu_cli/src/my_xhs_cli/__init__.py

@@ -0,0 +1,3 @@
+"""My Xiaohongshu CLI - Custom cookie-based XHS client."""
+
+__version__ = "0.1.0"

my_xiaohongshu_cli/src/my_xhs_cli/cli.py

@@ -0,0 +1,330 @@
+"""CLI entry point for my-xiaohongshu-cli."""
+
+from __future__ import annotations
+
+import json
+import logging
+import sys
+from pathlib import Path
+
+import click
+
+from . import __version__
+from .client import XhsCookieClient
+
+# Default cookie file path
+DEFAULT_COOKIE_FILE = "/home/ubuntu/xiaohongshu_cookies.txt"
+
+
+def get_client(cookie_file: str) -> XhsCookieClient:
+    """Create client with error handling."""
+    try:
+        return XhsCookieClient(cookie_file)
+    except FileNotFoundError as e:
+        click.echo(f"Error: {e}", err=True)
+        click.echo(f"Please ensure cookie file exists: {cookie_file}", err=True)
+        sys.exit(1)
+    except Exception as e:
+        click.echo(f"Error loading cookies: {e}", err=True)
+        sys.exit(1)
+
+
+@click.group()
+@click.version_option(version=__version__, prog_name="myxhs")
+@click.option("-v", "--verbose", is_flag=True, help="Enable debug logging")
+@click.option(
+    "--cookie-file",
+    type=click.Path(exists=False),
+    default=DEFAULT_COOKIE_FILE,
+    help="Path to cookie file (JSON format)",
+)
+@click.pass_context
+def cli(ctx: click.Context, verbose: bool, cookie_file: str):
+    """myxhs — My custom Xiaohongshu CLI 📕
+    
+    Uses custom cookies from /home/ubuntu/xiaohongshu_cookies.txt
+    """
+    ctx.ensure_object(dict)
+    ctx.obj["cookie_file"] = cookie_file
+
+    if verbose:
+        logging.basicConfig(level=logging.DEBUG, format="%(name)s %(message)s")
+    else:
+        logging.basicConfig(level=logging.WARNING)
+
+
+# ============ Auth Commands ============
+
+@cli.command()
+@click.pass_context
+def status(ctx: click.Context):
+    """Check login status with current cookies."""
+    cookie_file = ctx.obj["cookie_file"]
+    click.echo(f"Cookie file: {cookie_file}")
+    
+    if not Path(cookie_file).exists():
+        click.echo("Status: ❌ Cookie file not found")
+        sys.exit(1)
+    
+    try:
+        with get_client(cookie_file) as client:
+            cookies = client.cookies
+            click.echo(f"Status: ✅ Loaded {len(cookies)} cookies")
+            click.echo(f"Key cookies: a1={cookies.get('a1', 'N/A')[:20]}...")
+            click.echo(f"             web_session={cookies.get('web_session', 'N/A')[:20]}...")
+    except Exception as e:
+        click.echo(f"Status: ❌ Error - {e}")
+        sys.exit(1)
+
+
+# ============ Reading Commands ============
+
+@cli.command()
+@click.argument("keyword")
+@click.option("--page", default=1, help="Page number")
+@click.option("--limit", default=10, help="Results per page")
+@click.option("--json-output", is_flag=True, help="Output as JSON")
+@click.pass_context
+def search(ctx: click.Context, keyword: str, page: int, limit: int, json_output: bool):
+    """Search notes by keyword."""
+    with get_client(ctx.obj["cookie_file"]) as client:
+        results = client.search_notes(keyword, page=page, page_size=limit)
+        
+        if json_output:
+            click.echo(json.dumps(results, ensure_ascii=False, indent=2))
+            return
+        
+        if not results:
+            click.echo("No results found.")
+            return
+        
+        click.echo(f"\n🔍 Search results for '{keyword}':\n")
+        for item in results:
+            click.echo(f"📌 {item['title'] or '(无标题)'}")
+            click.echo(f"   ID: {item['id']}")
+            click.echo(f"   Author: {item['user']} | ❤️ {item['likes']}")
+            if item['desc']:
+                click.echo(f"   {item['desc'][:80]}...")
+            click.echo()
+
+
+@cli.command()
+@click.argument("note_id")
+@click.option("--xsec-token", help="X-Sec-Token for the note")
+@click.option("--json-output", is_flag=True, help="Output as JSON")
+@click.pass_context
+def read(ctx: click.Context, note_id: str, xsec_token: str | None, json_output: bool):
+    """Read note details by ID."""
+    # Strip URL if full URL provided
+    if "/" in note_id:
+        # Extract ID from URL like https://www.xiaohongshu.com/explore/xxx
+        parts = note_id.split("/")
+        note_id = parts[-1].split("?")[0]
+    
+    with get_client(ctx.obj["cookie_file"]) as client:
+        note = client.get_note_detail(note_id, xsec_token)
+        
+        if json_output:
+            click.echo(json.dumps(note, ensure_ascii=False, indent=2))
+            return
+        
+        click.echo(f"\n📖 {note['title'] or '(无标题)'}")
+        click.echo(f"   Author: {note['user']}")
+        click.echo(f"   ❤️ {note['likes']} | ⭐ {note['collects']} | 💬 {note['comments']} | ↗️ {note['shares']}")
+        click.echo(f"\n{note['desc'][:500]}...")
+        
+        if note['tags']:
+            click.echo(f"\nTags: {', '.join(note['tags'])}")
+        
+        if note['images']:
+            click.echo(f"\n🖼️ {len(note['images'])} images")
+
+
+@cli.command()
+@click.option("--num", default=10, help="Number of items")
+@click.option("--json-output", is_flag=True, help="Output as JSON")
+@click.pass_context
+def feed(ctx: click.Context, num: int, json_output: bool):
+    """Get home feed recommendations."""
+    with get_client(ctx.obj["cookie_file"]) as client:
+        items = client.get_feed(num=num)
+        
+        if json_output:
+            click.echo(json.dumps(items, ensure_ascii=False, indent=2))
+            return
+        
+        click.echo(f"\n🏠 Home Feed:\n")
+        for item in items:
+            click.echo(f"📌 {item['title'] or '(无标题)'}")
+            click.echo(f"   ID: {item['id']} | Author: {item.get('author', 'Unknown')} | ❤️ {item['likes']}")
+            click.echo()
+
+
+@cli.command()
+@click.option("--category", default="homefeed.fashion_v3", help="Feed category")
+@click.option("--num", default=10, help="Number of items")
+@click.option("--json-output", is_flag=True, help="Output as JSON")
+@click.pass_context
+def hot(ctx: click.Context, category: str, num: int, json_output: bool):
+    """Get hot/trending feed."""
+    with get_client(ctx.obj["cookie_file"]) as client:
+        items = client.get_hot_feed(category=category, num=num)
+        
+        if json_output:
+            click.echo(json.dumps(items, ensure_ascii=False, indent=2))
+            return
+        
+        click.echo(f"\n🔥 Hot Feed ({category}):\n")
+        for item in items:
+            click.echo(f"  📌 {item['title'] or '(无标题)'} | {item['user']} | ❤️ {item['likes']}")
+
+
+# ============ User Commands ============
+
+@cli.command()
+@click.argument("user_id")
+@click.option("--json-output", is_flag=True, help="Output as JSON")
+@click.pass_context
+def user(ctx: click.Context, user_id: str, json_output: bool):
+    """Get user profile info."""
+    with get_client(ctx.obj["cookie_file"]) as client:
+        info = client.get_user_info(user_id)
+        
+        if json_output:
+            click.echo(json.dumps(info, ensure_ascii=False, indent=2))
+            return
+        
+        click.echo(f"\n👤 {info['nickname']}")
+        click.echo(f"   ID: {info['id']}")
+        click.echo(f"   Bio: {info['desc']}")
+        click.echo(f"   Followers: {info['followers']} | Following: {info['following']}")
+        click.echo(f"   Notes: {info['notes']} | Collected: {info['collected']}")
+
+
+@cli.command()
+@click.argument("user_id")
+@click.option("--num", default=10, help="Number of notes")
+@click.option("--json-output", is_flag=True, help="Output as JSON")
+@click.pass_context
+def user_notes(ctx: click.Context, user_id: str, num: int, json_output: bool):
+    """Get user's published notes."""
+    with get_client(ctx.obj["cookie_file"]) as client:
+        notes = client.get_user_notes(user_id, num=num)
+        
+        if json_output:
+            click.echo(json.dumps(notes, ensure_ascii=False, indent=2))
+            return
+        
+        click.echo(f"\n📝 User Notes ({len(notes)}):\n")
+        for note in notes:
+            click.echo(f"  📌 {note['title'] or '(无标题)'}")
+            click.echo(f"     ID: {note['id']} | ❤️ {note['likes']} | {note['time']}")
+
+
+# ============ Interaction Commands ============
+
+@cli.command()
+@click.argument("note_id")
+@click.pass_context
+def like(ctx: click.Context, note_id: str):
+    """Like a note."""
+    if "/" in note_id:
+        parts = note_id.split("/")
+        note_id = parts[-1].split("?")[0]
+    
+    with get_client(ctx.obj["cookie_file"]) as client:
+        success = client.like_note(note_id)
+        if success:
+            click.echo(f"❤️ Liked note {note_id}")
+        else:
+            click.echo("Failed to like note")
+
+
+@cli.command()
+@click.argument("note_id")
+@click.pass_context
+def unlike(ctx: click.Context, note_id: str):
+    """Unlike a note."""
+    if "/" in note_id:
+        parts = note_id.split("/")
+        note_id = parts[-1].split("?")[0]
+    
+    with get_client(ctx.obj["cookie_file"]) as client:
+        success = client.unlike_note(note_id)
+        if success:
+            click.echo(f"💔 Unliked note {note_id}")
+        else:
+            click.echo("Failed to unlike note")
+
+
+@cli.command()
+@click.option("--title", required=True, help="Note title")
+@click.option("--body", required=True, help="Note body text")
+@click.option("--images", multiple=True, help="Image file paths (can use multiple)")
+@click.option("--topic", help="Topic/hashtag to search and attach")
+@click.option("--private", "is_private", is_flag=True, help="Publish as private note")
+@click.pass_context
+def post(ctx: click.Context, title: str, body: str, images: tuple, topic: str | None, is_private: bool):
+    """Publish a new image note."""
+    with get_client(ctx.obj["cookie_file"]) as client:
+        # Upload images
+        file_ids = []
+        for img_path in images:
+            click.echo(f"Uploading {img_path}...")
+            permit = client.get_upload_permit()
+            client.upload_file(permit["fileId"], permit["token"], img_path)
+            file_ids.append(permit["fileId"])
+            click.echo(f"✅ Uploaded: {img_path}")
+        
+        # Search topic if provided
+        topics = []
+        if topic:
+            topic_data = client.search_topics(topic)
+            # Simple selection: take first matching topic
+            if topic_data.get("topics"):
+                t = topic_data["topics"][0]
+                topics.append({"id": t.get("id", ""), "name": t.get("name", topic), "type": "topic"})
+        
+        # Publish
+        click.echo("Publishing note...")
+        result = client.create_image_note(
+            title=title,
+            desc=body,
+            image_file_ids=file_ids,
+            topics=topics,
+            is_private=is_private,
+        )
+        if result.get("success"):
+            click.echo(f"✅ Published: {title}" + (" (private)" if is_private else ""))
+        else:
+            click.echo(f"❌ Failed: {result.get('data', 'Unknown error')}")
+
+
+@cli.command()
+@click.option("--num", default=10, help="Number of notes")
+@click.option("--json-output", is_flag=True, help="Output as JSON")
+@click.pass_context
+def my_notes(ctx: click.Context, num: int, json_output: bool):
+    """Get my published notes."""
+    with get_client(ctx.obj["cookie_file"]) as client:
+        notes = client.get_my_notes(num=num)
+        
+        if json_output:
+            click.echo(json.dumps(notes, ensure_ascii=False, indent=2))
+            return
+        
+        click.echo(f"\n📝 My Notes ({len(notes)}):\n")
+        for note in notes:
+            click.echo(f"  📌 {note['title'] or '(无标题)'}")
+            click.echo(f"     ID: {note['id']} | ❤️ {note['likes']} | {note['time']}")
+
+
+# ============ Main Entry ============
+
+def main():
+    """Entry point."""
+    cli()
+
+
+if __name__ == "__main__":
+    main()

my_xiaohongshu_cli/src/my_xhs_cli/client.py

@@ -0,0 +1,549 @@
+"""Xiaohongshu API client with proper signing for both Main and Creator APIs."""
+
+from __future__ import annotations
+
+import json
+import logging
+import mimetypes
+import random
+import time
+from typing import Any
+from pathlib import Path
+
+import httpx
+
+from .signing import MainApiSigner, sign_creator, XHSHOW_AVAILABLE, CRYPTO_AVAILABLE
+
+logger = logging.getLogger(__name__)
+
+# API constants
+USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36"
+CHROME_VERSION = "135"
+HOME_URL = "https://www.xiaohongshu.com"
+EDITH_HOST = "https://edith.xiaohongshu.com"
+CREATOR_HOST = "https://creator.xiaohongshu.com"
+UPLOAD_HOST = "https://ros-upload.xiaohongshu.com"
+
+
+class XhsCookieClient:
+    """Xiaohongshu API client with proper signing."""
+
+    def __init__(
+        self,
+        cookie_file: str | Path = "/home/ubuntu/xiaohongshu_cookies.txt",
+        timeout: float = 30.0,
+        request_delay: float = 1.0,
+        max_retries: int = 3,
+    ):
+        self.cookie_file = Path(cookie_file)
+        self.cookies = self._load_cookies()
+        self._http = httpx.Client(timeout=timeout, follow_redirects=True)
+        self._request_delay = request_delay
+        self._base_request_delay = request_delay
+        self._max_retries = max_retries
+        self._last_request_time = 0.0
+        self._request_count = 0
+        self._verify_count = 0
+
+        # Initialize main API signer
+        if XHSHOW_AVAILABLE:
+            try:
+                self._main_signer = MainApiSigner()
+                logger.info("Using xhshow library for main API signing")
+            except Exception as e:
+                logger.warning("Failed to initialize xhshow: %s", e)
+                self._main_signer = None
+        else:
+            logger.warning("xhshow not available, main API will fail")
+            self._main_signer = None
+
+        # Check creator signing
+        if not CRYPTO_AVAILABLE:
+            logger.warning("pycryptodome not available, creator API will fail")
+
+    def _load_cookies(self) -> dict[str, str]:
+        """Load cookies from Chrome extension export format."""
+        if not self.cookie_file.exists():
+            raise FileNotFoundError(f"Cookie file not found: {self.cookie_file}")
+
+        data = json.loads(self.cookie_file.read_text())
+
+        # Chrome extension format: list of cookie objects
+        if isinstance(data, list):
+            cookies = {}
+            for item in data:
+                if isinstance(item, dict) and "name" in item and "value" in item:
+                    cookies[item["name"]] = item["value"]
+            logger.info("Loaded %d cookies from %s", len(cookies), self.cookie_file)
+            return cookies
+
+        # Simple dict format
+        if isinstance(data, dict):
+            return data
+
+        raise ValueError(f"Unknown cookie format in {self.cookie_file}")
+
+    def _cookies_to_string(self) -> str:
+        """Convert cookies dict to header string."""
+        return "; ".join(f"{k}={v}" for k, v in self.cookies.items())
+
+    def _rate_limit_delay(self) -> None:
+        """Rate limiting with jitter."""
+        if self._request_delay <= 0:
+            return
+        elapsed = time.time() - self._last_request_time
+        if elapsed < self._request_delay:
+            jitter = max(0, random.gauss(0.3, 0.15))
+            if random.random() < 0.05:
+                jitter += random.uniform(2.0, 5.0)
+            sleep_time = self._request_delay - elapsed + jitter
+            logger.debug("Rate-limit delay: %.2fs", sleep_time)
+            time.sleep(sleep_time)
+
+    def _mark_request(self) -> None:
+        self._last_request_time = time.time()
+        self._request_count += 1
+
+    def _base_headers(self) -> dict[str, str]:
+        """Build base request headers."""
+        return {
+            "user-agent": USER_AGENT,
+            "content-type": "application/json;charset=UTF-8",
+            "cookie": self._cookies_to_string(),
+            "origin": HOME_URL,
+            "referer": f"{HOME_URL}/",
+            "sec-ch-ua": f'"Google Chrome";v="{CHROME_VERSION}", "Chromium";v="{CHROME_VERSION}"',
+            "sec-ch-ua-mobile": "?0",
+            "sec-ch-ua-platform": '"macOS"',
+            "sec-fetch-dest": "empty",
+            "sec-fetch-mode": "cors",
+            "sec-fetch-site": "same-site",
+            "accept": "application/json, text/plain, */*",
+            "accept-language": "zh-CN,zh;q=0.9,en;q=0.8",
+            "dnt": "1",
+            "priority": "u=1, i",
+        }
+
+    def _merge_response_cookies(self, resp: httpx.Response) -> None:
+        """Persist response cookies back into session."""
+        for name, value in resp.cookies.items():
+            if value:
+                self.cookies[name] = value
+
+    def _request_with_retry(self, method: str, url: str, **kwargs) -> httpx.Response:
+        """Make request with retry logic."""
+        self._rate_limit_delay()
+        last_exc: Exception | None = None
+
+        for attempt in range(self._max_retries):
+            try:
+                resp = self._http.request(method, url, **kwargs)
+                self._merge_response_cookies(resp)
+                self._mark_request()
+
+                # Handle rate limiting
+                if resp.status_code in (429, 500, 502, 503, 504):
+                    wait = (2 ** attempt) + random.uniform(0, 1)
+                    logger.warning(
+                        "HTTP %d from %s, retrying in %.1fs (attempt %d/%d)",
+                        resp.status_code, url[:80], wait, attempt + 1, self._max_retries
+                    )
+                    time.sleep(wait)
+                    continue
+
+                return resp
+
+            except (httpx.TimeoutException, httpx.NetworkError) as exc:
+                last_exc = exc
+                wait = (2 ** attempt) + random.uniform(0, 1)
+                logger.warning(
+                    "Network error: %s, retrying in %.1fs (attempt %d/%d)",
+                    exc, wait, attempt + 1, self._max_retries
+                )
+                time.sleep(wait)
+
+        if last_exc:
+            raise Exception(f"Request failed after {self._max_retries} retries: {last_exc}")
+        raise Exception(f"Request failed after {self._max_retries} retries")
+
+    def _handle_response(self, resp: httpx.Response) -> Any:
+        """Parse and validate API response."""
+        # Handle verification/captcha
+        if resp.status_code in (461, 471):
+            self._verify_count += 1
+            cooldown = min(30, 5 * (2 ** (self._verify_count - 1)))
+            logger.warning("Captcha triggered, cooling down %.0fs", cooldown)
+            self._request_delay = max(self._request_delay, self._base_request_delay * 2)
+            time.sleep(cooldown)
+            raise Exception(f"Verification required (status {resp.status_code})")
+
+        self._verify_count = 0
+        text = resp.text
+        if not text:
+            return None
+
+        try:
+            data = json.loads(text)
+        except json.JSONDecodeError:
+            raise Exception(f"Non-JSON response: {text[:200]}")
+
+        if data.get("success"):
+            return data.get("data", data)
+
+        code = data.get("code")
+        if code == 300012:
+            raise Exception("IP blocked")
+        if code == 300015:
+            raise Exception(f"Signature error: {data.get('msg', 'unknown')}")
+        if code == 300011:
+            raise Exception(f"Account error: {data.get('msg', 'account exception')}")
+        if code == -100:
+            raise Exception("Session expired - need fresh cookies")
+
+        raise Exception(f"API error: {json.dumps(data)[:300]}")
+
+    # ============ Main API Methods ============
+
+    def _main_api_get(self, uri: str, params: dict[str, Any] | None = None) -> Any:
+        """Make signed GET request to Main API."""
+        if not self._main_signer:
+            raise Exception("Main API signer not available")
+
+        sign_headers = self._main_signer.sign_get(uri, self.cookies, params=params)
+        full_uri = self._main_signer.build_uri(uri, params)
+        url = f"{EDITH_HOST}{full_uri}"
+
+        resp = self._request_with_retry("GET", url, headers={**self._base_headers(), **sign_headers})
+        return self._handle_response(resp)
+
+    def _main_api_post(
+        self,
+        uri: str,
+        data: dict[str, Any],
+        header_overrides: dict[str, str] | None = None,
+    ) -> Any:
+        """Make signed POST request to Main API."""
+        if not self._main_signer:
+            raise Exception("Main API signer not available")
+
+        sign_headers = self._main_signer.sign_post(uri, self.cookies, payload=data)
+        url = f"{EDITH_HOST}{uri}"
+
+        headers = {**self._base_headers(), **sign_headers}
+        if header_overrides:
+            headers.update(header_overrides)
+
+        body = json.dumps(data, separators=(",", ":"), ensure_ascii=False)
+        resp = self._request_with_retry("POST", url, headers=headers, content=body)
+        return self._handle_response(resp)
+
+    # ============ Creator API Methods ============
+
+    def _creator_host(self, uri: str) -> str:
+        """Determine host for creator API."""
+        return CREATOR_HOST if uri.startswith("/api/galaxy/") else EDITH_HOST
+
+    def _creator_get(self, uri: str, params: dict[str, Any] | None = None) -> Any:
+        """Make signed GET request to Creator API."""
+        if not CRYPTO_AVAILABLE:
+            raise Exception("pycryptodome not available for creator signing")
+
+        if "a1" not in self.cookies:
+            raise Exception("a1 cookie required for creator API")
+
+        # Build full URI with params
+        if params:
+            query = "&".join(f"{k}={v}" for k, v in params.items())
+            full_uri = f"{uri}?{query}"
+        else:
+            full_uri = uri
+
+        sign = sign_creator(f"url={full_uri}", None, self.cookies["a1"])
+        host = self._creator_host(uri)
+        url = f"{host}{full_uri}"
+
+        headers = {
+            **self._base_headers(),
+            "x-s": sign["x-s"],
+            "x-t": sign["x-t"],
+            "origin": CREATOR_HOST,
+            "referer": f"{CREATOR_HOST}/",
+        }
+
+        resp = self._request_with_retry("GET", url, headers=headers)
+        return self._handle_response(resp)
+
+    def _creator_post(self, uri: str, data: dict[str, Any]) -> Any:
+        """Make signed POST request to Creator API."""
+        if not CRYPTO_AVAILABLE:
+            raise Exception("pycryptodome not available for creator signing")
+
+        if "a1" not in self.cookies:
+            raise Exception("a1 cookie required for creator API")
+
+        sign = sign_creator(f"url={uri}", data, self.cookies["a1"])
+        host = self._creator_host(uri)
+        url = f"{host}{uri}"
+
+        headers = {
+            **self._base_headers(),
+            "x-s": sign["x-s"],
+            "x-t": sign["x-t"],
+            "origin": CREATOR_HOST,
+            "referer": f"{CREATOR_HOST}/",
+        }
+
+        body = json.dumps(data, separators=(",", ":"), ensure_ascii=False)
+        resp = self._request_with_retry("POST", url, headers=headers, content=body)
+        return self._handle_response(resp)
+
+    def close(self) -> None:
+        self._http.close()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        self.close()
+
+    # ============ API Methods ============
+
+    def search_notes(
+        self,
+        keyword: str,
+        page: int = 1,
+        page_size: int = 20,
+        sort: str = "general",
+    ) -> list[dict]:
+        """Search notes by keyword."""
+        search_id = self._generate_search_id()
+
+        result = self._main_api_post("/api/sns/web/v1/search/notes", {
+            "keyword": keyword,
+            "page": page,
+            "page_size": page_size,
+            "search_id": search_id,
+            "sort": sort,
+            "note_type": 0,
+        })
+
+        items = result.get("items", [])
+        return [
+            {
+                "id": item.get("id"),
+                "title": item.get("note_card", {}).get("title", ""),
+                "desc": item.get("note_card", {}).get("desc", "")[:100],
+                "likes": item.get("note_card", {}).get("interact_info", {}).get("liked_count", 0),
+            }
+            for item in items
+            if item.get("model_type") == "note"
+        ]
+
+    def get_note_detail(self, note_id: str) -> dict[str, Any]:
+        """Get note detail by ID."""
+        return self._main_api_post("/api/sns/web/v1/feed", {
+            "source_note_id": note_id,
+            "image_formats": ["jpg", "webp", "avif"],
+            "extra": {"need_body_topic": 1},
+        })
+
+    def get_feed(self, num: int = 20, category: str = "homefeed.recommend") -> list[dict]:
+        """Get home feed."""
+        result = self._main_api_post("/api/sns/web/v1/homefeed", {
+            "cursor_score": "",
+            "num": num,
+            "refresh_type": 1,
+            "note_index": 0,
+            "unread_begin_note_id": "",
+            "unread_end_note_id": "",
+            "unread_note_count": 0,
+            "category": category,
+        })
+
+        items = result.get("items", [])
+        return [
+            {
+                "id": item.get("id"),
+                "title": item.get("note_card", {}).get("title", ""),
+                "author": item.get("note_card", {}).get("user", {}).get("nickname", ""),
+                "likes": item.get("note_card", {}).get("interact_info", {}).get("liked_count", 0),
+            }
+            for item in items
+            if item.get("model_type") == "note"
+        ]
+
+    def like_note(self, note_id: str) -> bool:
+        """Like a note."""
+        result = self._main_api_post("/api/sns/web/v1/note/like", {"note_id": note_id})
+        return result.get("success", False)
+
+    def unlike_note(self, note_id: str) -> bool:
+        """Unlike a note."""
+        result = self._main_api_post("/api/sns/web/v1/note/unlike", {"note_id": note_id})
+        return result.get("success", False)
+
+    def get_user_info(self, user_id: str) -> dict[str, Any]:
+        """Get user info by ID."""
+        return self._main_api_get("/api/sns/web/v1/user/otherinfo", {"target_user_id": user_id})
+
+    def get_user_notes(self, user_id: str, num: int = 20) -> list[dict]:
+        """Get notes by user ID."""
+        result = self._main_api_post("/api/sns/web/v1/user_posted", {
+            "user_id": user_id,
+            "num": num,
+            "cursor": "",
+        })
+
+        notes = result.get("notes", [])
+        return [
+            {
+                "id": note.get("note_id"),
+                "title": note.get("title", ""),
+                "likes": note.get("interact_info", {}).get("liked_count", 0),
+                "time": note.get("time"),
+            }
+            for note in notes
+        ]
+
+    def get_comments(self, note_id: str, num: int = 20) -> list[dict]:
+        """Get comments for a note."""
+        result = self._main_api_post("/api/sns/web/v2/comment/page", {
+            "note_id": note_id,
+            "cursor": "",
+            "top_comment_id": "",
+            "image_formats": ["jpg", "webp", "avif"],
+            "num": num,
+        })
+
+        return result.get("comments", [])
+
+    # ============ Creator API Methods ============
+
+    def search_topics(self, keyword: str) -> dict[str, Any]:
+        """Search topics by keyword."""
+        return self._creator_post("/web_api/sns/v1/search/topic", {
+            "keyword": keyword,
+            "suggest_topic_request": {"title": "", "desc": ""},
+            "page": {"page_size": 20, "page": 1},
+        })
+
+    def get_upload_permit(self, file_type: str = "image", count: int = 1) -> dict[str, str]:
+        """Get upload permit for images/videos."""
+        result = self._creator_get("/api/media/v1/upload/web/permit", {
+            "biz_name": "spectrum",
+            "scene": file_type,
+            "file_count": count,
+            "version": 1,
+            "source": "web",
+        })
+
+        permit = result.get("uploadTempPermits", [{}])[0]
+        return {
+            "fileId": permit.get("fileIds", [""])[0],
+            "token": permit.get("token", ""),
+            "uploadUrl": permit.get("uploadUrl", ""),
+        }
+
+    def upload_file(
+        self,
+        file_id: str,
+        token: str,
+        file_path: str,
+        content_type: str | None = None,
+    ) -> None:
+        """Upload a file to cloud storage."""
+        url = f"{UPLOAD_HOST}/{file_id}"
+        content_type = content_type or mimetypes.guess_type(file_path)[0] or "application/octet-stream"
+
+        with open(file_path, "rb") as f:
+            resp = self._http.put(
+                url,
+                content=f.read(),
+                headers={
+                    "X-Cos-Security-Token": token,
+                    "Content-Type": content_type,
+                },
+            )
+        resp.raise_for_status()
+
+    def create_image_note(
+        self,
+        title: str,
+        desc: str,
+        image_file_ids: list[str],
+        topics: list[dict[str, str]] | None = None,
+        is_private: bool = False,
+    ) -> dict:
+        """Create and publish an image note."""
+        images = [{"file_id": fid, "metadata": {"source": -1}} for fid in image_file_ids]
+        business_binds = {
+            "version": 1,
+            "noteId": 0,
+            "noteOrderBind": {},
+            "notePostTiming": {"postTime": None},
+            "noteCollectionBind": {"id": ""},
+        }
+
+        payload = {
+            "common": {
+                "type": "normal",
+                "title": title,
+                "note_id": "",
+                "desc": desc,
+                "source": '{"type":"web","ids":"","extraInfo":"{\\"subType\\":\\"official\\"}"}',
+                "business_binds": json.dumps(business_binds),
+                "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,
+        }
+
+        result = self._main_api_post(
+            "/web_api/sns/v2/note",
+            payload,
+            header_overrides={
+                "origin": CREATOR_HOST,
+                "referer": f"{CREATOR_HOST}/",
+            },
+        )
+        return {"success": True, "data": result}
+
+    def get_creator_note_list(self, tab: int = 0, page: int = 0) -> dict[str, Any]:
+        """Get my published notes from creator platform."""
+        return self._creator_get("/api/galaxy/v2/creator/note/user/posted", {
+            "tab": tab,
+            "page": page,
+        })
+
+    def get_my_notes(self, num: int = 20) -> list[dict]:
+        """Get my published notes."""
+        result = self.get_creator_note_list(tab=0, page=0)
+        notes = result.get("notes", [])
+        return [
+            {
+                "id": note.get("note_id"),
+                "title": note.get("title", ""),
+                "likes": note.get("interact_info", {}).get("liked_count", 0),
+                "time": note.get("time"),
+            }
+            for note in notes[:num]
+        ]
+
+    @staticmethod
+    def _generate_search_id() -> str:
+        """Generate a unique search ID (base36 of timestamp << 64 + random)."""
+        e = int(time.time() * 1000) << 64
+        t = random.randint(0, 2147483646)
+        num = e + t
+
+        alphabet = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+        if num == 0:
+            return "0"
+        result = ""
+        while num > 0:
+            result = alphabet[num % 36] + result
+            num //= 36
+        return result

my_xiaohongshu_cli/src/my_xhs_cli/signing.py

@@ -0,0 +1,168 @@
+"""
+Signing utilities for Xiaohongshu API.
+
+Includes:
+- Main API signing (edith.xiaohongshu.com) using xhshow
+- Creator API signing (creator.xiaohongshu.com) using AES encryption
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import time
+from typing import Any
+
+# Try to import xhshow for main API signing
+try:
+    from xhshow import CryptoConfig, SessionManager, Xhshow
+    from xhshow.utils.url_utils import extract_uri as _extract_uri
+    from xhshow.utils.url_utils import build_url as _build_url
+
+    XHSHOW_AVAILABLE = True
+except ImportError:
+    XHSHOW_AVAILABLE = False
+
+# Creator API signing constants (from xiaohongshu-cli)
+try:
+    from Crypto.Cipher import AES
+    from Crypto.Util.Padding import pad
+
+    CRYPTO_AVAILABLE = True
+except ImportError:
+    CRYPTO_AVAILABLE = False
+
+# Constants
+USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36"
+APP_ID = "xhs-pc-web"
+PLATFORM = "xhs-pc"
+SDK_VERSION = "3.8.6"
+
+# Creator API AES key/iv
+CREATOR_AES_KEY = b"7cc4adla5ay0701v"
+CREATOR_AES_IV = b"4uzjr7mbsibcaldp"
+
+
+class MainApiSigner:
+    """Main API (edith.xiaohongshu.com) signer using xhshow."""
+
+    def __init__(self):
+        if not XHSHOW_AVAILABLE:
+            raise ImportError("xhshow library not available. Install with: pip install xhshow")
+
+        self._config = CryptoConfig().with_overrides(
+            PUBLIC_USERAGENT=USER_AGENT,
+            SIGNATURE_DATA_TEMPLATE={
+                "x0": SDK_VERSION,
+                "x1": APP_ID,
+                "x2": PLATFORM,
+                "x3": "",
+                "x4": "",
+            },
+            SIGNATURE_XSCOMMON_TEMPLATE={
+                "s0": 5,
+                "s1": "",
+                "x0": "1",
+                "x1": SDK_VERSION,
+                "x2": PLATFORM,
+                "x3": APP_ID,
+                "x4": "4.86.0",
+                "x5": "",
+                "x6": "",
+                "x7": "",
+                "x8": "",
+                "x9": -596800761,
+                "x10": 0,
+                "x11": "normal",
+            },
+        )
+        self._xhshow = Xhshow(self._config)
+        self._session = SessionManager(self._config)
+
+    def sign_get(
+        self,
+        uri: str,
+        cookies: dict[str, str],
+        params: dict[str, Any] | None = None,
+    ) -> dict[str, str]:
+        """Sign GET request."""
+        return self._xhshow.sign_headers_get(
+            uri, cookies, params=params, session=self._session
+        )
+
+    def sign_post(
+        self,
+        uri: str,
+        cookies: dict[str, str],
+        payload: dict[str, Any] | None = None,
+    ) -> dict[str, str]:
+        """Sign POST request."""
+        return self._xhshow.sign_headers_post(
+            uri, cookies, payload=payload, session=self._session
+        )
+
+    def build_uri(self, uri: str, params: dict[str, Any] | None = None) -> str:
+        """Build full URI with query parameters."""
+        if not params:
+            return uri
+        return _build_url(uri, params)
+
+    def extract_uri(self, url: str) -> str:
+        """Extract URI from full URL."""
+        return _extract_uri(url)
+
+
+def sign_creator(api: str, data: dict[str, Any] | None, a1: str) -> dict[str, str]:
+    """
+    Generate creator platform signature headers.
+
+    Algorithm: MD5(api + JSON(data)) → AES-128-CBC encrypt → XYW_ prefix
+
+    Args:
+        api: The API path, prefixed with "url=" (e.g., "url=/web_api/sns/v2/note")
+        data: Request body data (None for GET requests)
+        a1: The a1 cookie value
+
+    Returns:
+        Dict with x-s and x-t headers
+    """
+    if not CRYPTO_AVAILABLE:
+        raise ImportError("pycryptodome not available. Install with: pip install pycryptodome")
+
+    content = api
+    if data is not None:
+        content += json.dumps(data, separators=(",", ":"), ensure_ascii=False)
+
+    x1 = hashlib.md5(content.encode("utf-8")).hexdigest()
+    x2 = "0|0|0|1|0|0|1|0|0|0|1|0|0|0|0|1|0|0|0"
+    x3 = a1
+    x4 = int(time.time() * 1000)
+
+    plaintext = f"x1={x1};x2={x2};x3={x3};x4={x4};"
+    payload = _aes_encrypt(base64.b64encode(plaintext.encode("utf-8")).decode("utf-8"))
+
+    envelope = {
+        "signSvn": "56",
+        "signType": "x2",
+        "appId": "ugc",
+        "signVersion": "1",
+        "payload": payload,
+    }
+
+    xs = "XYW_" + base64.b64encode(
+        json.dumps(envelope, separators=(",", ":")).encode("utf-8")
+    ).decode("utf-8")
+
+    return {
+        "x-s": xs,
+        "x-t": str(x4),
+    }
+
+
+def _aes_encrypt(data: str) -> str:
+    """AES-128-CBC encrypt and return hex string."""
+    cipher = AES.new(CREATOR_AES_KEY, AES.MODE_CBC, CREATOR_AES_IV)
+    padded = pad(data.encode("utf-8"), AES.block_size)
+    encrypted = cipher.encrypt(padded)
+    return encrypted.hex()