Create-knowledge-find-decode/acquisition/suggest.py

"""联想扩展：种子词 → 小红书 keyword_v2（返回相关帖）→ 从帖子标题/话题标签挖候选搜索词。

⚠️ 小红书没有纯「搜索联想词」接口，所以这里是退一步：拿种子打 keyword_v2，从返回的
【相关帖子】的 title + body_text 里的 #标签 + topic_list 挖候选词，不是搜索框下拉补全。
（抖音只有 keyword 搜索接口、且与搜索共享强限流，已从 demo 移除，只保留小红书。）

parse_suggest 纯函数可离线测；suggest 负责 HTTP（mirror search.py）。
"""
from __future__ import annotations

import re
from typing import Any, Optional
from urllib.parse import urljoin

import httpx

from acquisition.crawler import RateLimiter
from core.config import Settings

SUGGEST_PATH = "/crawler/xiao_hong_shu/keyword_v2"
_TAG = re.compile(r"#([^\s#＃]{2,20})")


class SuggestError(RuntimeError):
    pass


def parse_suggest(response: Any, *, limit: int = 15) -> list[str]:
    """从 keyword_v2 回包的相关帖挖候选搜索词：#标签/话题优先，标题兜底。去重截断。"""
    if not isinstance(response, dict):
        raise SuggestError("bad_response: not a dict")
    if response.get("code") not in (0, "0"):
        raise SuggestError(f"business_error: code={response.get('code')} msg={response.get('msg')}")
    posts = ((response.get("data") or {}).get("data")) or []
    tags: list[str] = []
    titles: list[str] = []
    for p in posts:
        if not isinstance(p, dict):
            continue
        tags += _TAG.findall(f"{p.get('title', '')} {p.get('body_text', '')}")
        for t in (p.get("topic_list") or []):
            name = t.get("name") if isinstance(t, dict) else t
            if name:
                tags.append(str(name))
        title = (p.get("title") or "").strip()
        if title:
            titles.append(title)
    out: list[str] = []
    seen: set[str] = set()
    for c in tags + titles:          # 标签优先、标题兜底
        c = c.strip()
        if c and c not in seen:
            seen.add(c)
            out.append(c)
            if len(out) >= limit:
                break
    return out


def suggest(
    keyword: str,
    *,
    content_type: str = "图文",
    settings: Optional[Settings] = None,
    http_client: Any = None,
    rate_limiter: Optional[RateLimiter] = None,
    env_file: str = ".env",
    limit: int = 15,
) -> list[str]:
    """种子词 → 小红书候选搜索词列表（从相关帖挖）。失败抛 SuggestError。"""
    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("xhs_suggest")
        url = urljoin(settings.aiddit_crawler_base_url, SUGGEST_PATH)
        body = {"keyword": keyword, "content_type": content_type, "sort_type": "综合", "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 SuggestError(f"http_error: {exc}") from exc
        except ValueError as exc:
            raise SuggestError("bad_json") from exc
        return parse_suggest(data, limit=limit)
    finally:
        if owns_client:
            client.close()