feat(search)：接入小红书 + 三平台随机限流 + 抖音改走 piaoquantv

- 小红书：search_xiaohongshu/parse_xiaohongshu，封面/标题/作者直接取自搜索回包的
  note_card（image_list[0].image_url），无需 detail（视频帖无直链）。store 拍平扩到
  三平台，run_search 加 xiaohongshu_topk。
- 限流：RateLimiter 支持 [min,max] 随机区间；三平台「搜索」各自独立闸，两次间隔随机
  10~12s；抖音详情走更轻的 3~5s 闸。微信/小红书限流照旧保持。
- 抖音改走 piaoquantv 后端（crawapi.piaoquantv.com）绕开 aiddit 强限流（code 10000）：
  config 加 douyin_base_url/account_id/cookie_batch（env CK_DOUYIN_*），search/crawler
  对抖音改用此 host + 带 account_id + cookie_batch。实测 search→detail code:0 拿视频直链。
- run_search 加 DOUYIN_ON 开关：关闭时沿用上次结果里的抖音，不发请求。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.env

@@ -80,6 +80,12 @@ CONTENTFIND_API_CRAWAPI_BASE_URL=http://crawler.aiddit.com
 CONTENTFIND_API_CRAWAPI_KEY=
 CONTENTFIND_API_CRAWAPI_TIMEOUT_SECONDS=60
 
+# 抖音单独走 piaoquantv 后端（绕开 aiddit 限流 code 10000）；search.py/crawler.py 读这三个
+# keyword + detail 都用此 host，body 需带 account_id + cookie_batch。小红书/微信仍走上面的 aiddit host。
+CK_DOUYIN_BASE_URL=http://crawapi.piaoquantv.com
+CK_DOUYIN_ACCOUNT_ID=7450041106378522636
+CK_DOUYIN_COOKIE_BATCH=default
+
 CONTENTFIND_DOUYIN_KEYWORD_PATH=/crawler/dou_yin/keyword
 CONTENTFIND_DOUYIN_BLOGGER_PATH=/crawler/dou_yin/blogger
 CONTENTFIND_DOUYIN_DETAIL_PATH=/crawler/dou_yin/detail

acquisition/crawler.py

@@ -8,6 +8,7 @@ detail path、URL→content_id 解析、平台名/id 前缀。详见 数据接
 """
 from __future__ import annotations
 
+import random
 import re
 import time
 from typing import Any, Callable, Optional
@@ -62,23 +63,31 @@ class CrawlerError(RuntimeError):
 
 
 class RateLimiter:
-    """同一 bucket 两次调用间隔 ≥ min_interval。对齐 CFA RateLimiter。"""
+    """同一 bucket 两次调用间隔 ≥ min_interval。对齐 CFA RateLimiter。
+    给 max_interval_seconds 时，每次间隔在 [min, max] 随机取（搜索接口加随机抖动躲限流）。"""
 
     def __init__(
         self,
         min_interval_seconds: float = RATE_LIMIT_SECONDS,
         now_fn: Callable[[], float] = time.monotonic,
         sleep_fn: Callable[[float], None] = time.sleep,
+        max_interval_seconds: Optional[float] = None,
     ) -> None:
         self.min_interval_seconds = min_interval_seconds
+        self.max_interval_seconds = max_interval_seconds
         self.now_fn = now_fn
         self.sleep_fn = sleep_fn
         self._last: dict[str, float] = {}
 
+    def _interval(self) -> float:
+        if self.max_interval_seconds is not None:
+            return random.uniform(self.min_interval_seconds, self.max_interval_seconds)
+        return self.min_interval_seconds
+
     def wait(self, bucket: str) -> None:
         last = self._last.get(bucket)
         if last is not None:
-            remaining = self.min_interval_seconds - (self.now_fn() - last)
+            remaining = self._interval() - (self.now_fn() - last)
             if remaining > 0:
                 self.sleep_fn(remaining)
         self._last[bucket] = self.now_fn()
@@ -171,10 +180,16 @@ def fetch_post_detail(
     owns_client = http_client is None
     client = http_client or httpx.Client()
     try:
-        url = urljoin(settings.crawler_base_url, cfg["path"])
+        is_douyin = platform == "douyin"
+        base_url = settings.douyin_base_url if is_douyin else settings.crawler_base_url
+        body = {"content_id": content_id}
+        if is_douyin:                                  # 抖音详情走 piaoquantv，需带 account_id + cookie_batch
+            body["account_id"] = settings.douyin_account_id
+            body["cookie_batch"] = settings.douyin_cookie_batch
+        url = urljoin(base_url, cfg["path"])
         resp = client.post(
             url,
-            json={"content_id": content_id},
+            json=body,
             headers={"Content-Type": "application/json"},
             timeout=settings.crawler_timeout,
         )

acquisition/search.py

@@ -33,8 +33,8 @@ PLATFORM_SEARCH = {
     "douyin": {
         "path": "/crawler/dou_yin/keyword", "id_key": "aweme_id",
         "sort_type": "综合排序", "publish_time": "不限", "cursor0": "0",
-        # 对齐 ContentFindAgentNew CONTENTFIND_DOUYIN_DEFAULT_ACCOUNT_ID；缺它抖音搜索 code 10000
-        "account_id": "771431222",
+        # account_id 改由 settings.douyin_account_id 提供（抖音走 piaoquantv 后端）；缺它搜索报错
+        "account_id": None,
     },
 }
 
@@ -82,7 +82,10 @@ def search_keyword(
     settings = settings or Settings.from_env(env_file)
     sort_type = cfg["sort_type"] if sort_type is None else sort_type
     publish_time = cfg["publish_time"] if publish_time is None else publish_time
-    account_id = cfg["account_id"] if account_id is None else account_id
+    is_douyin = platform == "douyin"
+    if account_id is None:
+        account_id = settings.douyin_account_id if is_douyin else cfg["account_id"]
+    base_url = settings.douyin_base_url if is_douyin else settings.crawler_base_url
     rate_limiter = rate_limiter or RateLimiter()
 
     owns_client = http_client is None
@@ -93,11 +96,13 @@ def search_keyword(
     try:
         for _ in range(max_pages):
             rate_limiter.wait(f"{platform}_keyword")
-            url = urljoin(settings.crawler_base_url, cfg["path"])
+            url = urljoin(base_url, cfg["path"])
             body = {"keyword": keyword, "content_type": content_type,
                     "sort_type": sort_type, "publish_time": publish_time, "cursor": cursor}
             if account_id:
                 body["account_id"] = account_id   # 抖音必带；小红书不带
+            if is_douyin:
+                body["cookie_batch"] = settings.douyin_cookie_batch   # piaoquantv 抖音必带
             try:
                 resp = client.post(url, json=body,
                                    headers={"Content-Type": "application/json"},
@@ -178,3 +183,69 @@ def search_weixin(
     finally:
         if owns_client:
             client.close()
+
+
+# ---------- 小红书搜索（封面/标题/作者已在搜索回包的 note_card 里，无需 detail）----------
+def parse_xiaohongshu(response: Any, *, limit: int = 5) -> list[dict]:
+    """小红书搜索回包 {code,data:{data:[{id, note_card:{image_list,display_title,desc,user,type}}]}}
+    → 帖子列表，直接取封面(image_list[0].image_url)+标题+作者，省去逐条 detail。无视频直链。"""
+    if not isinstance(response, dict):
+        raise SearchError("bad_response: not a dict")
+    if response.get("code") not in (0, "0"):
+        raise SearchError(f"business_error: code={response.get('code')} msg={response.get('msg')}")
+    items = ((response.get("data") or {}).get("data")) or []
+    out: list[dict] = []
+    for it in items:
+        if not isinstance(it, dict):
+            continue
+        cid = it.get("id")
+        nc = it.get("note_card") or {}
+        img_list = nc.get("image_list") or []
+        cover = (img_list[0] or {}).get("image_url") if img_list else ""
+        if not cid or not cover:
+            continue
+        title = (nc.get("display_title") or "").strip()
+        if not title:
+            title = (nc.get("desc") or "").strip().split("\n")[0][:50]
+        out.append({"id": cid, "title": title, "cover_url": cover,
+                    "nick_name": (nc.get("user") or {}).get("nickname") or "",
+                    "type": nc.get("type") or "",
+                    "url": f"https://www.xiaohongshu.com/explore/{cid}"})
+        if len(out) >= limit:
+            break
+    return out
+
+
+def search_xiaohongshu(
+    keyword: str,
+    *,
+    content_type: str = "图文",
+    limit: int = 5,
+    settings: Optional[Settings] = None,
+    http_client: Any = None,
+    rate_limiter: Optional[RateLimiter] = None,
+    env_file: str = ".env",
+) -> list[dict]:
+    """小红书搜索：query → 帖子列表（带 url/封面/标题，无需 detail；视频帖无直链）。失败抛 SearchError。"""
+    settings = settings or Settings.from_env(env_file)
+    rate_limiter = rate_limiter or RateLimiter()
+    owns_client = http_client is None
+    client = http_client or httpx.Client()
+    try:
+        rate_limiter.wait("xiaohongshu_keyword")
+        url = urljoin(settings.crawler_base_url, "/crawler/xiao_hong_shu/keyword")
+        try:
+            resp = client.post(url, json={"keyword": keyword, "content_type": content_type,
+                                          "sort_type": "综合", "publish_time": "", "cursor": ""},
+                               headers={"Content-Type": "application/json"},
+                               timeout=settings.crawler_timeout)
+            resp.raise_for_status()
+            data = resp.json()
+        except httpx.HTTPError as exc:
+            raise SearchError(f"http_error: {exc}") from exc
+        except ValueError as exc:
+            raise SearchError("bad_json") from exc
+        return parse_xiaohongshu(data, limit=limit)
+    finally:
+        if owns_client:
+            client.close()

acquisition/store.py

@@ -104,11 +104,14 @@ def insert_queries(conn: sqlite3.Connection, run_id: str, method: str,
 
 
 def _flatten_search(rec: dict) -> list[dict]:
-    """把一条 {method, query, douyin:{ok:[...],error}, weixin:{ok:[...],error}} 拍成「每个结果一行」。
-    某渠道有结果 → 每个结果一行 ok=1；空/失败 → 一行 ok=0 记 error（保留「搜过但无结果」状态）。"""
+    """把一条 {method, query, douyin/weixin/xiaohongshu:{ok:[...],error}} 拍成「每个结果一行」。
+    某渠道有结果 → 每个结果一行 ok=1；空/失败 → 一行 ok=0 记 error（保留「搜过但无结果」状态）。
+    只处理记录里实际出现的渠道（向后兼容只有抖音/微信的旧记录）。"""
     out = []
     method, query = rec.get("method", ""), rec.get("query", "")
-    for platform in ("douyin", "weixin"):
+    for platform in ("douyin", "weixin", "xiaohongshu"):
+        if platform not in rec:
+            continue
         v = rec.get(platform) or {}
         hits = v.get("ok") or []
         if hits:

core/config.py

@@ -94,6 +94,10 @@ class Settings:
     search_default_limit: int = 5
     search_content_type: str = "图文"
     search_sort_type: str = "综合"
+    # 抖音走独立后端（piaoquantv，绕开 aiddit 限流）：host / account_id / cookie_batch 单独配
+    douyin_base_url: str = "http://crawapi.piaoquantv.com"
+    douyin_account_id: str = "7450041106378522636"
+    douyin_cookie_batch: str = "default"
 
     @classmethod
     def from_env(cls, env_file: str | Path = ".env") -> "Settings":
@@ -138,4 +142,7 @@ class Settings:
             search_default_limit=int(env_value("CK_SEARCH_DEFAULT_LIMIT", file_env, "5")),
             search_content_type=env_value("CK_SEARCH_CONTENT_TYPE", file_env, "图文"),
             search_sort_type=env_value("CK_SEARCH_SORT_TYPE", file_env, "综合"),
+            douyin_base_url=env_value("CK_DOUYIN_BASE_URL", file_env, "http://crawapi.piaoquantv.com"),
+            douyin_account_id=env_value("CK_DOUYIN_ACCOUNT_ID", file_env, "7450041106378522636"),
+            douyin_cookie_batch=env_value("CK_DOUYIN_COOKIE_BATCH", file_env, "default"),
         )

scripts/run_search.py

@@ -1,10 +1,11 @@
-"""用生成的 query 真实搜帖子/视频（抖音 + 微信），媒体存本地 data/，写 search_results.json。
+"""用生成的 query 真实搜帖子/视频（抖音 + 微信公众号 + 小红书），媒体存本地 data/。
 
-每法取前 N 条 query，每条 query 每个渠道取 top-K，让三种打法的搜索结果真正拉开差距：
-  抖音：search(limit=K) → 逐个 detail → 下载视频+封面到 data/search/douyin/<id>/
-  微信：search(limit=K) → 下载每个封面到 data/search/weixin/<hash>/（正文需 detail+token，本期只存封面）
-失败/空按渠道记 error，不中断。抖音对突发调用有强冷却，所有抖音调用走一个 12s 统一闸。
-产出每条记录：{method, query, douyin:{ok:[...],error}, weixin:{ok:[...],error}}。
+每法取前 N 条 query，每条 query 每个渠道取 top-K：
+  抖音：search(top-1) → detail → 下载视频+封面（搜索/详情各自限速）
+  微信公众号：search(top-K) → 下载每个封面（封面在搜索回包里，无需 detail）
+  小红书：search(top-K) → 下载每个封面（封面在 note_card 里，无需 detail；视频帖无直链）
+限流：三个平台的「搜索」各自独立闸，两次搜索间隔随机 10~12s；抖音「详情」走更轻的 3~5s 闸。
+产出每条记录：{method, query, douyin/weixin/xiaohongshu:{ok:[...],error}}。
 用法：PYTHONPATH=. python scripts/run_search.py
 """
 from __future__ import annotations
@@ -14,7 +15,7 @@ import json
 from pathlib import Path
 
 from acquisition.crawler import RateLimiter, fetch_post_detail
-from acquisition.search import search_keyword, search_weixin
+from acquisition.search import search_keyword, search_weixin, search_xiaohongshu
 from core.config import Settings
 from creation_knowledge.integrations.video_extract import _default_download
 
@@ -22,10 +23,12 @@ ROOT = Path(__file__).resolve().parent.parent
 DATA = ROOT / "data"
 DEMO = DATA / "queries" / "demo.json"
 OUT = DATA / "queries" / "search_results.json"
+DOUYIN_ON = False                       # 抖音限流暂停：False 时不发抖音请求，沿用上次结果里的抖音
 N = 6                                    # 每法取前 N 条 query
-K_DOUYIN = 1                             # 抖音每条 query 只取第 1 个视频（1 搜索+1 详情，躲限流）
-K_WEIXIN = 5                             # 微信每条 query 取前 K 个封面
-_NOOP = RateLimiter(min_interval_seconds=0.0)   # 让搜索/详情内部限流让位，统一走外层闸
+K_DOUYIN = 1                             # 抖音每条 query 只取第 1 个视频（躲限流）
+K_WEIXIN = 5                             # 微信公众号每条 query 取前 K 个封面
+K_XHS = 5                               # 小红书每条 query 取前 K 个封面
+SEARCH_MIN, SEARCH_MAX = 10.0, 12.0     # 每平台两次搜索之间随机间隔（秒）
 
 
 def _dl(url: str, platform: str, dst: Path, public: str):
@@ -38,12 +41,11 @@ def _dl(url: str, platform: str, dst: Path, public: str):
         return None
 
 
-def douyin_topk(query: str, settings: Settings, gate: RateLimiter) -> dict:
-    """抖音搜 top-K 视频：返回 {ok:[{title,url,cover,video}], error}。"""
+def douyin_topk(query: str, settings: Settings, search_rl: RateLimiter, detail_rl: RateLimiter) -> dict:
+    """抖音搜 top-K 视频：返回 {ok:[{title,url,cover,video}], error}。搜索/详情各走各的闸。"""
     try:
-        gate.wait("douyin")
         ids = search_keyword(query, platform="douyin", content_type="视频", limit=K_DOUYIN,
-                             settings=settings, rate_limiter=_NOOP)
+                             settings=settings, rate_limiter=search_rl)
     except Exception as exc:
         return {"ok": [], "error": f"搜索失败: {str(exc)[:50]}"}
     if not ids:
@@ -51,10 +53,9 @@ def douyin_topk(query: str, settings: Settings, gate: RateLimiter) -> dict:
     recs = []
     for vid in ids[:K_DOUYIN]:
         try:
-            gate.wait("douyin")
-            post = fetch_post_detail(vid, settings=settings, rate_limiter=_NOOP)
+            post = fetch_post_detail(vid, settings=settings, rate_limiter=detail_rl)
         except Exception:
-            continue                                    # 单条详情失败跳过，不毁整条
+            continue
         base, pub = DATA / "search" / "douyin" / post.id, f"/data/search/douyin/{post.id}"
         rec = {"title": post.title, "url": post.url, "cover": None, "video": None}
         if post.image_urls:
@@ -67,7 +68,7 @@ def douyin_topk(query: str, settings: Settings, gate: RateLimiter) -> dict:
 
 
 def weixin_topk(query: str, settings: Settings, rl: RateLimiter) -> dict:
-    """微信搜 top-K 文章封面：返回 {ok:[{title,url,nick,cover}], error}。"""
+    """微信公众号搜 top-K 文章封面：返回 {ok:[{title,url,nick,cover}], error}。"""
     try:
         arts = search_weixin(query, limit=K_WEIXIN, settings=settings, rate_limiter=rl)
     except Exception as exc:
@@ -85,11 +86,47 @@ def weixin_topk(query: str, settings: Settings, rl: RateLimiter) -> dict:
     return {"ok": recs, "error": None}
 
 
+def xiaohongshu_topk(query: str, settings: Settings, rl: RateLimiter) -> dict:
+    """小红书搜 top-K 帖子封面：返回 {ok:[{title,url,nick,cover}], error}。封面来自搜索回包，无需 detail。"""
+    try:
+        posts = search_xiaohongshu(query, content_type="图文", limit=K_XHS,
+                                   settings=settings, rate_limiter=rl)
+    except Exception as exc:
+        return {"ok": [], "error": f"搜索失败: {str(exc)[:50]}"}
+    if not posts:
+        return {"ok": [], "error": "未搜到"}
+    recs = []
+    for p in posts[:K_XHS]:
+        base, pub = DATA / "search" / "xiaohongshu" / p["id"], f"/data/search/xiaohongshu/{p['id']}"
+        rec = {"title": p["title"], "url": p["url"], "nick": p["nick_name"], "cover": None}
+        if p["cover_url"]:
+            rec["cover"] = _dl(p["cover_url"], "xiaohongshu", base / "cover.jpg", pub + "/cover.jpg")
+        recs.append(rec)
+    return {"ok": recs, "error": None}
+
+
+def _prev_douyin() -> dict:
+    """抖音关闭时，从上次的 search_results.json 沿用抖音结果（按 query 文本取），保住已抓到的视频。"""
+    if not OUT.exists():
+        return {}
+    try:
+        prev = json.loads(OUT.read_text("utf-8"))
+    except Exception:
+        return {}
+    return {r["query"]: r.get("douyin") for r in prev if r.get("query")}
+
+
 def main() -> None:
     settings = Settings.from_env()
     demo = json.loads(DEMO.read_text("utf-8"))
-    dy_gate = RateLimiter(min_interval_seconds=12.0)   # 抖音统一闸（躲冷却）
-    wx_rl = RateLimiter(min_interval_seconds=1.0)
+    # 三平台搜索各自独立闸（随机 10~12s）；抖音详情更轻（3~5s）
+    dy_search = RateLimiter(min_interval_seconds=SEARCH_MIN, max_interval_seconds=SEARCH_MAX)
+    dy_detail = RateLimiter(min_interval_seconds=3.0, max_interval_seconds=5.0)
+    wx_search = RateLimiter(min_interval_seconds=SEARCH_MIN, max_interval_seconds=SEARCH_MAX)
+    xhs_search = RateLimiter(min_interval_seconds=SEARCH_MIN, max_interval_seconds=SEARCH_MAX)
+    prev_dy = {} if DOUYIN_ON else _prev_douyin()
+    if not DOUYIN_ON:
+        print(f"（抖音已关闭：沿用上次 {len(prev_dy)} 条 query 的抖音结果，本次只跑微信+小红书）")
     results = []
     for tac in ("tactic1", "tactic2", "tactic4"):
         t = demo.get(tac) or {}
@@ -99,16 +136,25 @@ def main() -> None:
             if not q:
                 continue
             print(f"[{name}] {q}")
-            dy = douyin_topk(q, settings, dy_gate)
-            wx = weixin_topk(q, settings, wx_rl)
-            print(f"   抖音 {len(dy['ok'])} 视频{' / ' + dy['error'] if dy['error'] else ''}"
-                  f"   微信 {len(wx['ok'])} 封面{' / ' + wx['error'] if wx['error'] else ''}")
-            results.append({"method": name, "query": q, "douyin": dy, "weixin": wx})
+            if DOUYIN_ON:
+                dy = douyin_topk(q, settings, dy_search, dy_detail)
+            else:
+                dy = prev_dy.get(q) or {"ok": [], "error": "抖音限流暂停，未跑"}
+            wx = weixin_topk(q, settings, wx_search)
+            xhs = xiaohongshu_topk(q, settings, xhs_search)
+            print(f"   抖音 {len(dy['ok'])} 视频{_err(dy)}"
+                  f"   微信 {len(wx['ok'])} 封面{_err(wx)}"
+                  f"   小红书 {len(xhs['ok'])} 封面{_err(xhs)}")
+            results.append({"method": name, "query": q, "douyin": dy, "weixin": wx, "xiaohongshu": xhs})
     OUT.parent.mkdir(parents=True, exist_ok=True)
     OUT.write_text(json.dumps(results, ensure_ascii=False, indent=1), encoding="utf-8")
-    dy_total = sum(len(r["douyin"]["ok"]) for r in results)
-    wx_total = sum(len(r["weixin"]["ok"]) for r in results)
-    print(f"\nwrote {len(results)} 条 query（抖音 {dy_total} 视频 / 微信 {wx_total} 封面）→ {OUT}")
+    tot = lambda k: sum(len(r[k]["ok"]) for r in results)
+    print(f"\nwrote {len(results)} 条 query（抖音 {tot('douyin')} / 微信 {tot('weixin')} / "
+          f"小红书 {tot('xiaohongshu')}）→ {OUT}")
+
+
+def _err(d: dict) -> str:
+    return " / " + d["error"] if d.get("error") else ""
 
 
 if __name__ == "__main__":

tests/test_search.py

@@ -9,7 +9,7 @@ import pytest
 from core.config import PgConfig, Settings
 from acquisition.crawler import RateLimiter
 from acquisition.search import (
-    SearchError, parse_search_response, parse_weixin, search_keyword,
+    SearchError, parse_search_response, parse_weixin, parse_xiaohongshu, search_keyword,
 )
 
 FIX = Path(__file__).parent / "fixtures"
@@ -106,7 +106,8 @@ def test_douyin_body_has_account_id_and_params():
                          settings=_settings(), http_client=client, rate_limiter=_no_wait(), limit=10)
     assert out == ["a1", "a2"]                         # 用 aweme_id 解析
     body = client.calls[0]
-    assert body["account_id"] == "771431222"           # 抖音必带
+    assert body["account_id"] == "7450041106378522636"  # piaoquantv 抖音账号（settings.douyin_account_id）
+    assert body["cookie_batch"] == "default"            # piaoquantv 抖音必带
     assert body["sort_type"] == "综合排序"             # 平台默认（非小红书的"综合"）
     assert body["publish_time"] == "不限"
     assert body["cursor"] == "0"                       # 抖音起始 cursor
@@ -130,6 +131,35 @@ def test_parse_weixin_business_error():
         parse_weixin({"code": 10000, "msg": "x"})
 
 
+def test_parse_xiaohongshu_takes_cover_title_from_note_card():
+    resp = {"code": 0, "data": {"data": [
+        {"id": "abc123", "note_card": {
+            "type": "video", "display_title": "社会运行规则",
+            "image_list": [{"image_url": "https://ci.xiaohongshu.com/cover.jpg"}],
+            "user": {"nickname": "某号"}}},
+        {"id": "noimg", "note_card": {"display_title": "无封面应跳过", "image_list": []}},
+    ]}}
+    out = parse_xiaohongshu(resp, limit=5)
+    assert len(out) == 1                                  # 无封面的被跳过
+    assert out[0]["id"] == "abc123" and out[0]["title"] == "社会运行规则"
+    assert out[0]["cover_url"].endswith("cover.jpg") and out[0]["nick_name"] == "某号"
+    assert out[0]["url"] == "https://www.xiaohongshu.com/explore/abc123"
+
+
+def test_parse_xiaohongshu_title_falls_back_to_desc():
+    resp = {"code": 0, "data": {"data": [
+        {"id": "x1", "note_card": {"display_title": "", "desc": "那些赤裸裸的社会真相！\n第二行",
+                                   "image_list": [{"image_url": "u.jpg"}]}},
+    ]}}
+    out = parse_xiaohongshu(resp, limit=5)
+    assert out[0]["title"] == "那些赤裸裸的社会真相！"      # display_title 空 → 取 desc 首行
+
+
+def test_parse_xiaohongshu_business_error():
+    with pytest.raises(SearchError):
+        parse_xiaohongshu({"code": 10000, "msg": "x"})
+
+
 def test_xiaohongshu_body_no_account_id():
     page = {"code": 0, "data": {"has_more": False, "next_cursor": "", "data": [{"id": "x1"}]}}
     client = _FakeClient([page])

tests/test_store.py

@@ -36,10 +36,13 @@ def test_search_flatten_topk_and_ok(tmp_path):
          "weixin": {"ok": [{"title": "w", "url": "wu", "cover": "/wc.jpg", "nick": "号"}], "error": None}},
         {"method": "M", "query": "q2",
          "douyin": {"ok": [], "error": "未搜到"},
-         "weixin": {"ok": [], "error": "搜索失败: x"}},
+         "weixin": {"ok": [], "error": "搜索失败: x"},
+         "xiaohongshu": {"ok": [{"title": "x", "url": "xu", "cover": "/xc.jpg", "nick": "薯"}], "error": None}},
     ]
-    # q1: 2 抖音 + 1 微信 = 3 个 ok 行；q2: 抖音空 + 微信空 = 2 个 ok=0 行 → 共 5
-    assert store.insert_search_results(c, "s1", recs, ts=5) == 5
+    # q1(无 xhs 键，向后兼容跳过): 2 抖音 + 1 微信 = 3；q2: 抖音空 + 微信空 + 1 小红书 = 3 → 共 6
+    assert store.insert_search_results(c, "s1", recs, ts=5) == 6
+    xhs = store.list_search(c, platform="xiaohongshu", ok=True)
+    assert xhs["total"] == 1 and xhs["items"][0]["extra"]["nick"] == "薯"
     ok_dy = store.list_search(c, platform="douyin", ok=True)
     assert ok_dy["total"] == 2                                   # q1 的两个视频
     assert {i["video"] for i in ok_dy["items"]} == {"/v.mp4", "/v2.mp4"}
@@ -48,9 +51,9 @@ def test_search_flatten_topk_and_ok(tmp_path):
     assert bad["total"] == 1 and "失败" in bad["items"][0]["extra"]["error"]
     good_wx = store.list_search(c, platform="weixin", ok=True)
     assert good_wx["items"][0]["extra"]["nick"] == "号"
-    # summary：q1 有 3 个结果，q2 有 0 个但被记为搜过（total=2 行）
+    # summary：q1 有 3 个结果(抖音2+微信1)；q2 抖音/微信空 + 小红书 1 → ok=1, total=3 行
     s = store.search_summary(c)
-    assert s["q1"]["ok"] == 3 and s["q2"]["ok"] == 0 and s["q2"]["total"] == 2
+    assert s["q1"]["ok"] == 3 and s["q2"]["ok"] == 1 and s["q2"]["total"] == 3
 
 
 def test_pagination(tmp_path):