Create-knowledge-find-decode/creation_knowledge/integrations/search.py

"""关键词搜索：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:<content_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]