"""关键词搜索:query → content_id 列表(翻页 + 去重 + 截断)。 接口(实测 2026-06-26,host = settings.crawler_base_url): POST /crawler/xiao_hong_shu/keyword body{keyword, content_type:"图文"|"视频", sort_type:"综合", publish_time:"", cursor:""} → {code:0, data:{has_more, next_cursor, data:[{id:, note_card:{...预览...}}]}} 搜索只给预览;全文/全图/视频仍需 crawler.fetch_post_detail 逐 id 取详情。 parse_search_response 纯函数可离线测;search_keyword 负责 HTTP + 翻页。 抖音 /keyword 实测 code:10000(疑需 cookie/鉴权)→ 抛 SearchError,由调用方跳过。 """ from __future__ import annotations from typing import Any, Optional from urllib.parse import urljoin import httpx from creation_knowledge.config import Settings from creation_knowledge.integrations.crawler import RateLimiter SEARCH_PATHS = { "xiaohongshu": "/crawler/xiao_hong_shu/keyword", "douyin": "/crawler/dou_yin/keyword", # TODO: 实测 code 10000,需 cookie/鉴权 } class SearchError(RuntimeError): pass def parse_search_response(response: Any, *, platform: str = "xiaohongshu") -> tuple[list[str], bool, str]: """信封 {code,msg,data:{has_more,next_cursor,data:[...]}} → (content_ids, has_more, next_cursor)。 code!=0 抛 SearchError(如抖音 10000)。""" if not isinstance(response, dict): raise SearchError("bad_response: not a dict") code = response.get("code") if code not in (0, "0"): raise SearchError(f"business_error: code={code} msg={response.get('msg')}") data = response.get("data") or {} items = data.get("data") or [] ids = [it.get("id") for it in items if isinstance(it, dict) and it.get("id")] return ids, bool(data.get("has_more")), str(data.get("next_cursor") or "") def search_keyword( keyword: str, *, platform: str = "xiaohongshu", content_type: str = "图文", sort_type: str = "综合", limit: int = 5, publish_time: str = "", settings: Optional[Settings] = None, http_client: Any = None, rate_limiter: Optional[RateLimiter] = None, env_file: str = ".env", max_pages: int = 10, ) -> list[str]: """搜索 → content_id 列表(翻页直到够 limit 或 has_more=False;按 id 去重、截断 limit)。 复用 crawler.RateLimiter('{platform}_keyword' bucket)。失败抛 SearchError。""" settings = settings or Settings.from_env(env_file) path = SEARCH_PATHS.get(platform) if not path: raise SearchError(f"unsupported platform: {platform}") rate_limiter = rate_limiter or RateLimiter() owns_client = http_client is None client = http_client or httpx.Client() out: list[str] = [] seen: set[str] = set() cursor = "" try: for _ in range(max_pages): rate_limiter.wait(f"{platform}_keyword") url = urljoin(settings.crawler_base_url, path) body = {"keyword": keyword, "content_type": content_type, "sort_type": sort_type, "publish_time": publish_time, "cursor": cursor} try: resp = client.post(url, json=body, 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 ids, has_more, next_cursor = parse_search_response(data, platform=platform) for cid in ids: if cid not in seen: seen.add(cid) out.append(cid) if len(out) >= limit: return out if not has_more or not next_cursor or next_cursor == cursor: break cursor = next_cursor finally: if owns_client: client.close() return out[:limit]