"""关键词搜索：query → content_id 列表（翻页 + 去重 + 截断）。 接口（实测 2026-06-26，host = settings.aiddit_crawler_base_url）： 小红书 POST /crawler/xiao_hong_shu/keyword body{keyword, sort_type:"综合", publish_time:"", cursor:""} → {code:0, data:{has_more, next_cursor, data:[{id:, note_card:{...}}]}} 抖音 POST /crawler/dou_yin/keyword body{keyword, content_type:"视频", sort_type:"综合排序", publish_time:"不限", cursor:"0", account_id:"771431222"} # ← 缺 account_id 会 code 10000 → {code:0, data:{has_more, next_cursor, data:[{aweme_id:, ...}]}} 各平台请求参数 / id 字段不同，集中在 PLATFORM_SEARCH 表里。 搜索只给预览；全文/全图/视频仍需 crawler.fetch_post_detail 逐 id 取详情（detail 只需 content_id）。 parse_search_response 纯函数可离线测；search_keyword 负责 HTTP + 翻页。 """ from __future__ import annotations from dataclasses import dataclass, field import re from typing import Any, Optional from urllib.parse import urljoin import httpx from core.config import Settings from core.text_limits import TITLE_FALLBACK_MAX_CHARS, clip_text from acquisition.crawler import ( CRAWLER_MAX_INTERVAL_SECONDS, CRAWLER_MIN_INTERVAL_SECONDS, RateLimiter, ) # 每平台的搜索参数差异（path / 条目 id 字段 / 排序 / 发布时间 / 起始 cursor / account_id） PLATFORM_SEARCH = { "xiaohongshu": { "path": "/crawler/xiao_hong_shu/keyword", "id_key": "id", "sort_type": "综合", "publish_time": "", "cursor0": "", "account_id": None, }, "douyin": { "path": "/crawler/dou_yin/keyword", "id_key": "aweme_id", "sort_type": "综合排序", "publish_time": "不限", "cursor0": "0", # account_id 改由 settings.piaoquantv_douyin_account_id 提供（抖音走 piaoquantv 后端）；缺它搜索报错 "account_id": None, }, } class SearchError(RuntimeError): pass @dataclass(frozen=True) class SearchHit: content_id: str provider: str = "" raw: dict[str, Any] = field(default_factory=dict) request_payload: dict[str, Any] = field(default_factory=dict) @dataclass(frozen=True) class SearchPage: rows: list[Any] raw_count: int cursor: str = "" next_cursor: str = "" has_more: bool = False provider: str = "" request_payload: dict[str, Any] = field(default_factory=dict) raw_metadata: dict[str, Any] = field(default_factory=dict) DOUYIN_PROVIDERS = ("piaoquantv", "aiddit") def _douyin_cursor0(provider: str) -> str: return "" if provider == "aiddit" else "0" def _douyin_base_url(settings: Settings, provider: str) -> str: if provider == "piaoquantv": return settings.piaoquantv_douyin_base_url if provider == "aiddit": return settings.aiddit_crawler_base_url raise SearchError(f"unsupported douyin provider: {provider}") def _douyin_search_body( keyword: str, *, provider: str, cursor: str, settings: Settings, content_type: str | None, sort_type: str | None, publish_time: str | None, account_id: str | None, ) -> dict[str, Any]: if provider == "piaoquantv": body: dict[str, Any] = { "keyword": keyword, "sort_type": sort_type or "综合排序", "publish_time": publish_time or "不限", "cursor": cursor, } if content_type: body["content_type"] = content_type body["account_id"] = account_id or settings.piaoquantv_douyin_account_id body["cookie_batch"] = settings.piaoquantv_douyin_cookie_batch return body if provider == "aiddit": body = { "keyword": keyword, "cursor": cursor, "sort_type": sort_type or "最多点赞", } if content_type: body["content_type"] = content_type return body raise SearchError(f"unsupported douyin provider: {provider}") def _parse_search_items(response: Any, *, platform: str = "xiaohongshu") -> tuple[list[dict[str, Any]], int, bool, str]: 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 [] if not isinstance(items, list): raise SearchError("bad_response: data.data is not a list") id_key = (PLATFORM_SEARCH.get(platform) or {}).get("id_key", "id") valid = [it for it in items if isinstance(it, dict) and it.get(id_key)] return valid, len(items), bool(data.get("has_more")), str(data.get("next_cursor") or "") 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)。 条目 id 字段按平台取（小红书 id / 抖音 aweme_id）。code!=0 抛 SearchError。""" id_key = (PLATFORM_SEARCH.get(platform) or {}).get("id_key", "id") items, _, has_more, next_cursor = _parse_search_items(response, platform=platform) ids = [it[id_key] for it in items] return ids, has_more, next_cursor def search_douyin_page( keyword: str, *, provider: str = "piaoquantv", cursors: dict[str, str] | None = None, content_type: str | None = "视频", sort_type: Optional[str] = None, publish_time: Optional[str] = None, account_id: Optional[str] = None, limit: int = 10, settings: Optional[Settings] = None, http_client: Any = None, rate_limiter: Optional[RateLimiter] = None, env_file: str = ".env", ) -> SearchPage: """Fetch one Douyin page, using piaoquantv first and AIDDIT as fallback.""" settings = settings or Settings.from_env(env_file) rate_limiter = rate_limiter or RateLimiter( min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS, max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS, ) owns_client = http_client is None client = http_client or httpx.Client() cursors = dict(cursors or {p: _douyin_cursor0(p) for p in DOUYIN_PROVIDERS}) provider_order = [provider] + [p for p in DOUYIN_PROVIDERS if p != provider] errors: list[str] = [] try: for attempt_provider in provider_order: cursor = cursors.get(attempt_provider, _douyin_cursor0(attempt_provider)) body = _douyin_search_body( keyword, provider=attempt_provider, cursor=cursor, settings=settings, content_type=content_type, sort_type=sort_type, publish_time=publish_time, account_id=account_id, ) rate_limiter.wait("douyin") url = urljoin(_douyin_base_url(settings, attempt_provider), PLATFORM_SEARCH["douyin"]["path"]) try: resp = client.post( url, json=body, headers={"Content-Type": "application/json"}, timeout=settings.crawler_timeout, ) resp.raise_for_status() data = resp.json() items, raw_count, has_more, next_cursor = _parse_search_items(data, platform="douyin") except httpx.HTTPError as exc: errors.append(f"{attempt_provider}: http_error: {exc}") continue except ValueError: errors.append(f"{attempt_provider}: bad_json") continue except SearchError as exc: errors.append(f"{attempt_provider}: {exc}") continue if not items: errors.append(f"{attempt_provider}: empty_aweme_id") continue id_key = PLATFORM_SEARCH["douyin"]["id_key"] hits = [ SearchHit( content_id=str(item[id_key]), provider=attempt_provider, raw=item, request_payload=dict(body), ) for item in items[:limit] ] next_cursors = dict(cursors) if has_more and next_cursor and next_cursor != cursor: next_cursors[attempt_provider] = next_cursor return SearchPage( rows=hits, raw_count=raw_count, cursor=cursor, next_cursor=next_cursor, has_more=has_more, provider=attempt_provider, request_payload=dict(body), raw_metadata={"cursors": next_cursors, "errors": errors}, ) finally: if owns_client: client.close() raise SearchError("; ".join(errors) or "douyin_search_failed") def search_douyin_hits( keyword: str, *, content_type: str | None = "视频", sort_type: Optional[str] = None, publish_time: Optional[str] = None, account_id: Optional[str] = None, limit: int = 5, settings: Optional[Settings] = None, http_client: Any = None, rate_limiter: Optional[RateLimiter] = None, env_file: str = ".env", max_pages: int = 10, ) -> list[SearchHit]: """Douyin search with piaoquantv primary and AIDDIT fallback per cursor.""" settings = settings or Settings.from_env(env_file) rate_limiter = rate_limiter or RateLimiter( min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS, max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS, ) owns_client = http_client is None client = http_client or httpx.Client() out: list[SearchHit] = [] seen: set[str] = set() provider = "piaoquantv" cursors = {p: _douyin_cursor0(p) for p in DOUYIN_PROVIDERS} try: for _ in range(max_pages): page = search_douyin_page( keyword, provider=provider, cursors=cursors, content_type=content_type, sort_type=sort_type, publish_time=publish_time, account_id=account_id, limit=limit, settings=settings, http_client=client, rate_limiter=rate_limiter, env_file=env_file, ) page_hits = list(page.rows) provider = page.provider current_cursor = cursors.get(provider, _douyin_cursor0(provider)) for hit in page_hits: if hit.content_id not in seen: seen.add(hit.content_id) out.append(hit) if len(out) >= limit: return out if not page.has_more or not page.next_cursor or page.next_cursor == current_cursor: break cursors = dict(page.raw_metadata.get("cursors") or cursors) finally: if owns_client: client.close() return out[:limit] def search_keyword( keyword: str, *, platform: str = "xiaohongshu", content_type: str | None = "图文", sort_type: Optional[str] = None, # None → 平台默认（小红书:综合 / 抖音:综合排序） publish_time: Optional[str] = None, # None → 平台默认（抖音:不限） account_id: Optional[str] = None, # None → 平台默认（抖音:771431222；小红书无） limit: int = 5, 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）。 平台专属参数（path/sort_type/publish_time/起始cursor/account_id）取自 PLATFORM_SEARCH， 显式入参可覆盖。复用 crawler.RateLimiter（'{platform}_keyword' bucket）。失败抛 SearchError。""" cfg = PLATFORM_SEARCH.get(platform) if not cfg: raise SearchError(f"unsupported platform: {platform}") settings = settings or Settings.from_env(env_file) is_douyin = platform == "douyin" if account_id is None: account_id = settings.piaoquantv_douyin_account_id if is_douyin else cfg["account_id"] if is_douyin: return [ hit.content_id for hit in search_douyin_hits( keyword, content_type=content_type, sort_type=sort_type, publish_time=publish_time, account_id=account_id, limit=limit, settings=settings, http_client=http_client, rate_limiter=rate_limiter, env_file=env_file, max_pages=max_pages, ) ] 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 base_url = settings.piaoquantv_douyin_base_url if is_douyin else settings.aiddit_crawler_base_url rate_limiter = rate_limiter or RateLimiter( min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS, max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS, ) owns_client = http_client is None client = http_client or httpx.Client() out: list[str] = [] seen: set[str] = set() cursor = cfg["cursor0"] try: for _ in range(max_pages): rate_limiter.wait(f"{platform}_keyword") url = urljoin(base_url, cfg["path"]) body = {"keyword": keyword, "sort_type": sort_type, "publish_time": publish_time, "cursor": cursor} if content_type: body["content_type"] = content_type if account_id: body["account_id"] = account_id # 抖音必带；小红书不带 if is_douyin: body["cookie_batch"] = settings.piaoquantv_douyin_cookie_batch # piaoquantv 抖音必带 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] # ---------- 微信公众号搜索（回的是带 url/封面的文章，不是纯 id，故单列）---------- _EM = re.compile(r"<[^>]+>") def parse_weixin(response: Any, *, limit: int = 5) -> list[dict]: """微信搜索回包 {code,data:{data:[{title,url,cover_url,nick_name,time}]}} → 文章列表（title 去 标记）。""" 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) or not it.get("url"): continue out.append({"title": _EM.sub("", it.get("title", "")).strip(), "url": it.get("url", ""), "cover_url": it.get("cover_url") or "", "nick_name": it.get("nick_name") or "", "time": it.get("time") or ""}) if len(out) >= limit: break return out def search_weixin( keyword: 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( min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS, max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS, ) owns_client = http_client is None client = http_client or httpx.Client() try: rate_limiter.wait("weixin_keyword") url = urljoin(settings.aiddit_crawler_base_url, "/crawler/wei_xin/keyword") try: resp = client.post(url, json={"keyword": keyword, "cursor": "0"}, 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_weixin(data, limit=limit) finally: if owns_client: client.close() def search_weixin_page( keyword: str, *, cursor: str = "0", limit: int = 10, settings: Optional[Settings] = None, http_client: Any = None, rate_limiter: Optional[RateLimiter] = None, env_file: str = ".env", ) -> SearchPage: """微信公众号搜索单页：query + cursor → SearchPage。""" settings = settings or Settings.from_env(env_file) rate_limiter = rate_limiter or RateLimiter( min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS, max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS, ) owns_client = http_client is None client = http_client or httpx.Client() body = {"keyword": keyword, "cursor": cursor} try: rate_limiter.wait("weixin_keyword") url = urljoin(settings.aiddit_crawler_base_url, "/crawler/wei_xin/keyword") 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 if not isinstance(data, dict): raise SearchError("bad_response: not a dict") if data.get("code") not in (0, "0"): raise SearchError(f"business_error: code={data.get('code')} msg={data.get('msg')}") envelope = data.get("data") or {} raw_rows = envelope.get("data") or [] if not isinstance(raw_rows, list): raise SearchError("bad_response: data.data is not a list") return SearchPage( rows=parse_weixin(data, limit=limit), raw_count=len(raw_rows), cursor=cursor, next_cursor=str(envelope.get("next_cursor") or ""), has_more=bool(envelope.get("has_more")), provider="aiddit", request_payload=body, ) finally: if owns_client: client.close() def fetch_weixin_detail( url: str, *, settings: Optional[Settings] = None, http_client: Any = None, rate_limiter: Optional[RateLimiter] = None, env_file: str = ".env", ) -> tuple[str, list[str]]: """微信公众号文章详情：content_link → (body_text, image_urls)。实测不需 token。失败抛 SearchError。""" settings = settings or Settings.from_env(env_file) rate_limiter = rate_limiter or RateLimiter( min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS, max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS, ) owns_client = http_client is None client = http_client or httpx.Client() try: rate_limiter.wait("weixin_detail") api = urljoin(settings.aiddit_crawler_base_url, "/crawler/wei_xin/detail") try: resp = client.post(api, json={"content_link": url}, 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 if data.get("code") not in (0, "0"): raise SearchError(f"business_error: code={data.get('code')} msg={data.get('msg')}") inner = ((data.get("data") or {}).get("data")) or {} imgs = [] for im in inner.get("image_url_list") or []: u = im.get("image_url") if isinstance(im, dict) else im if u: imgs.append(u) return inner.get("body_text") or "", imgs 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 = clip_text((nc.get("desc") or "").strip().split("\n")[0], TITLE_FALLBACK_MAX_CHARS) 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 | None = None, 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( min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS, max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS, ) owns_client = http_client is None client = http_client or httpx.Client() try: rate_limiter.wait("xiaohongshu_keyword") url = urljoin(settings.aiddit_crawler_base_url, "/crawler/xiao_hong_shu/keyword") try: body = {"keyword": keyword, "sort_type": "综合", "publish_time": "", "cursor": ""} if content_type: body["content_type"] = content_type 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 return parse_xiaohongshu(data, limit=limit) finally: if owns_client: client.close() def search_xiaohongshu_page( keyword: str, *, cursor: str = "", content_type: str | None = None, limit: int = 10, settings: Optional[Settings] = None, http_client: Any = None, rate_limiter: Optional[RateLimiter] = None, env_file: str = ".env", ) -> SearchPage: """小红书搜索单页：query + cursor → SearchPage。""" settings = settings or Settings.from_env(env_file) rate_limiter = rate_limiter or RateLimiter( min_interval_seconds=CRAWLER_MIN_INTERVAL_SECONDS, max_interval_seconds=CRAWLER_MAX_INTERVAL_SECONDS, ) owns_client = http_client is None client = http_client or httpx.Client() body = {"keyword": keyword, "sort_type": "综合", "publish_time": "", "cursor": cursor} if content_type: body["content_type"] = content_type try: rate_limiter.wait("xiaohongshu_keyword") url = urljoin(settings.aiddit_crawler_base_url, "/crawler/xiao_hong_shu/keyword") 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 if not isinstance(data, dict): raise SearchError("bad_response: not a dict") if data.get("code") not in (0, "0"): raise SearchError(f"business_error: code={data.get('code')} msg={data.get('msg')}") envelope = data.get("data") or {} raw_rows = envelope.get("data") or [] if not isinstance(raw_rows, list): raise SearchError("bad_response: data.data is not a list") return SearchPage( rows=parse_xiaohongshu(data, limit=limit), raw_count=len(raw_rows), cursor=cursor, next_cursor=str(envelope.get("next_cursor") or ""), has_more=bool(envelope.get("has_more")), provider="aiddit", request_payload=body, ) finally: if owns_client: client.close()