| 1 | """Xiaohongshu HTTP API search client for last30days. |
| 2 | |
| 3 | Uses xpzouying/xiaohongshu-mcp REST endpoints: |
| 4 | - GET/POST /api/v1/feeds/search |
| 5 | - GET /api/v1/login/status |
| 6 | """ |
| 7 | |
| 8 | from datetime import datetime, timezone |
| 9 | from typing import Any, Dict, List, Optional |
| 10 | |
| 11 | from . import http |
| 12 | |
| 13 | |
| 14 | def _to_int(value: Any) -> int: |
| 15 | """Convert Xiaohongshu count strings to int. |
| 16 | |
| 17 | Supports plain ints and Chinese suffixes like 1.2万 / 3亿. |
| 18 | """ |
| 19 | if value is None: |
| 20 | return 0 |
| 21 | if isinstance(value, (int, float)): |
| 22 | return int(value) |
| 23 | |
| 24 | text = str(value).strip().lower().replace(",", "") |
| 25 | if not text: |
| 26 | return 0 |
| 27 | |
| 28 | try: |
| 29 | if text.endswith("万"): |
| 30 | return int(float(text[:-1]) * 10000) |
| 31 | if text.endswith("亿"): |
| 32 | return int(float(text[:-1]) * 100000000) |
| 33 | return int(float(text)) |
| 34 | except (TypeError, ValueError): |
| 35 | return 0 |
| 36 | |
| 37 | |
| 38 | def _timestamp_to_date_ms(ts: Any) -> Optional[str]: |
| 39 | """Convert millisecond timestamp to YYYY-MM-DD.""" |
| 40 | try: |
| 41 | iv = int(ts) |
| 42 | if iv <= 0: |
| 43 | return None |
| 44 | # API examples use milliseconds. |
| 45 | dt = datetime.fromtimestamp(iv / 1000.0, tz=timezone.utc) |
| 46 | return dt.strftime("%Y-%m-%d") |
| 47 | except (TypeError, ValueError, OSError): |
| 48 | return None |
| 49 | |
| 50 | |
| 51 | def _relevance_from_interactions(likes: int, comments: int, favorites: int) -> float: |
| 52 | """Heuristic relevance score from engagement metrics.""" |
| 53 | # Weighted engagement with soft caps to [0, 1]. |
| 54 | weighted = (likes * 1.0) + (comments * 2.5) + (favorites * 1.5) |
| 55 | # 5000 weighted engagement ~= strong relevance. |
| 56 | score = min(1.0, max(0.05, weighted / 5000.0)) |
| 57 | return round(score, 3) |
| 58 | |
| 59 | |
| 60 | def _build_note_url(feed_id: str, xsec_token: str) -> str: |
| 61 | """Build a stable Xiaohongshu note URL.""" |
| 62 | if xsec_token: |
| 63 | return f"https://www.xiaohongshu.com/explore/{feed_id}?xsec_token={xsec_token}" |
| 64 | return f"https://www.xiaohongshu.com/explore/{feed_id}" |
| 65 | |
| 66 | |
| 67 | def search_feeds( |
| 68 | topic: str, |
| 69 | from_date: str, |
| 70 | to_date: str, |
| 71 | base_url: str, |
| 72 | depth: str = "default", |
| 73 | ) -> List[Dict[str, Any]]: |
| 74 | """Search Xiaohongshu feeds and normalize to web-item shape.""" |
| 75 | base = (base_url or "").rstrip("/") |
| 76 | if not base: |
| 77 | raise ValueError("Missing Xiaohongshu API base URL") |
| 78 | |
| 79 | # Quick login sanity check. |
| 80 | login = http.get(f"{base}/api/v1/login/status", timeout=8, retries=1) |
| 81 | is_logged_in = ( |
| 82 | login.get("data", {}).get("is_logged_in") |
| 83 | if isinstance(login, dict) else False |
| 84 | ) |
| 85 | if not is_logged_in: |
| 86 | raise http.HTTPError("Xiaohongshu API reachable but not logged in") |
| 87 | |
| 88 | # API supports filters; use recency-oriented defaults. |
| 89 | publish_time = "一天内" if depth == "quick" else "一周内" if depth == "default" else "半年内" |
| 90 | payload = { |
| 91 | "keyword": topic, |
| 92 | "filters": { |
| 93 | "sort_by": "综合", |
| 94 | "note_type": "不限", |
| 95 | "publish_time": publish_time, |
| 96 | "search_scope": "不限", |
| 97 | "location": "不限", |
| 98 | }, |
| 99 | } |
| 100 | |
| 101 | resp = http.post(f"{base}/api/v1/feeds/search", payload, timeout=20, retries=1) |
| 102 | feeds = resp.get("data", {}).get("feeds", []) if isinstance(resp, dict) else [] |
| 103 | if not isinstance(feeds, list): |
| 104 | feeds = [] |
| 105 | |
| 106 | # Cap source volume similarly to other web sources. |
| 107 | limit = {"quick": 8, "default": 15, "deep": 25}.get(depth, 15) |
| 108 | items: List[Dict[str, Any]] = [] |
| 109 | |
| 110 | for i, feed in enumerate(feeds[:limit]): |
| 111 | if not isinstance(feed, dict): |
| 112 | continue |
| 113 | note = feed.get("noteCard") or {} |
| 114 | if not isinstance(note, dict): |
| 115 | note = {} |
| 116 | interact = note.get("interactInfo") or {} |
| 117 | if not isinstance(interact, dict): |
| 118 | interact = {} |
| 119 | |
| 120 | feed_id = str(feed.get("id") or note.get("noteId") or "").strip() |
| 121 | if not feed_id: |
| 122 | continue |
| 123 | |
| 124 | xsec_token = str(feed.get("xsecToken") or note.get("xsecToken") or "").strip() |
| 125 | title = str( |
| 126 | note.get("displayTitle") |
| 127 | or note.get("title") |
| 128 | or "" |
| 129 | ).strip() |
| 130 | snippet = str( |
| 131 | note.get("desc") |
| 132 | or note.get("displayDesc") |
| 133 | or title |
| 134 | or "" |
| 135 | ).strip() |
| 136 | |
| 137 | likes = _to_int(interact.get("likedCount")) |
| 138 | comments = _to_int(interact.get("commentCount")) |
| 139 | favorites = _to_int(interact.get("collectedCount")) |
| 140 | |
| 141 | date_value = _timestamp_to_date_ms(note.get("time")) |
| 142 | why = f"Xiaohongshu engagement: likes={likes}, comments={comments}, favorites={favorites}" |
| 143 | |
| 144 | items.append({ |
| 145 | "id": f"XHS{i+1}", |
| 146 | "title": title[:200] if title else f"Xiaohongshu note {feed_id}", |
| 147 | "url": _build_note_url(feed_id, xsec_token), |
| 148 | "source_domain": "xiaohongshu.com", |
| 149 | "snippet": snippet[:500], |
| 150 | "date": date_value, |
| 151 | "date_confidence": "high" if date_value else "low", |
| 152 | "relevance": _relevance_from_interactions(likes, comments, favorites), |
| 153 | "why_relevant": why, |
| 154 | # Keep raw engagement for debugging/possible future rendering. |
| 155 | "engagement": { |
| 156 | "likes": likes, |
| 157 | "comments": comments, |
| 158 | "favorites": favorites, |
| 159 | }, |
| 160 | }) |
| 161 | |
| 162 | return items |
| 163 |