feat(search)：抖音 + 微信公众号真实搜索与抓取

acquisition/search.py：search_keyword 支持小红书/抖音（抖音补 account_id + 综合排序 +
aweme_id 解析），新增 search_weixin（/crawler/wei_xin/keyword，清洗标题保留 url/封面）。
scripts/run_search.py：用生成的 query 真搜，每法前 N 条、每渠道 top-K，下载视频/封面到
data/search/，写 search_results.json；抖音走 12s 统一闸躲限流（K_DOUYIN=1 稳）。

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

creation_knowledge/integrations/search.py → acquisition/search.py

@@ -15,13 +15,14 @@ parse_search_response 纯函数可离线测；search_keyword 负责 HTTP + 翻
 """
 from __future__ import annotations
 
+import re
 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
+from core.config import Settings
+from acquisition.crawler import RateLimiter
 
 # 每平台的搜索参数差异（path / 条目 id 字段 / 排序 / 发布时间 / 起始 cursor / account_id）
 PLATFORM_SEARCH = {
@@ -121,3 +122,59 @@ def search_keyword(
         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 去 <em> 标记）。"""
+    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()
+    owns_client = http_client is None
+    client = http_client or httpx.Client()
+    try:
+        rate_limiter.wait("weixin_keyword")
+        url = urljoin(settings.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()

scripts/run_search.py

@@ -0,0 +1,115 @@
+"""用生成的 query 真实搜帖子/视频（抖音 + 微信），媒体存本地 data/，写 search_results.json。
+
+每法取前 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}}。
+用法：PYTHONPATH=. python scripts/run_search.py
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+
+from acquisition.crawler import RateLimiter, fetch_post_detail
+from acquisition.search import search_keyword, search_weixin
+from core.config import Settings
+from creation_knowledge.integrations.video_extract import _default_download
+
+ROOT = Path(__file__).resolve().parent.parent
+DATA = ROOT / "data"
+DEMO = DATA / "queries" / "demo.json"
+OUT = DATA / "queries" / "search_results.json"
+N = 6                                    # 每法取前 N 条 query
+K_DOUYIN = 1                             # 抖音每条 query 只取第 1 个视频（1 搜索+1 详情，躲限流）
+K_WEIXIN = 5                             # 微信每条 query 取前 K 个封面
+_NOOP = RateLimiter(min_interval_seconds=0.0)   # 让搜索/详情内部限流让位，统一走外层闸
+
+
+def _dl(url: str, platform: str, dst: Path, public: str):
+    """下载到 dst，返回公开路径；失败返回 None。"""
+    try:
+        dst.parent.mkdir(parents=True, exist_ok=True)
+        dst.write_bytes(_default_download(url, platform))
+        return public
+    except Exception:
+        return None
+
+
+def douyin_topk(query: str, settings: Settings, gate: 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)
+    except Exception as exc:
+        return {"ok": [], "error": f"搜索失败: {str(exc)[:50]}"}
+    if not ids:
+        return {"ok": [], "error": "未搜到"}
+    recs = []
+    for vid in ids[:K_DOUYIN]:
+        try:
+            gate.wait("douyin")
+            post = fetch_post_detail(vid, settings=settings, rate_limiter=_NOOP)
+        except Exception:
+            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:
+            rec["cover"] = _dl(post.image_urls[0], "douyin", base / "cover.jpg", pub + "/cover.jpg")
+        if post.video_urls:
+            rec["video"] = _dl(post.video_urls[0], "douyin", base / "video.mp4", pub + "/video.mp4")
+        if rec["video"] or rec["cover"]:
+            recs.append(rec)
+    return {"ok": recs, "error": None if recs else "详情/下载均失败"}
+
+
+def weixin_topk(query: str, settings: Settings, rl: RateLimiter) -> dict:
+    """微信搜 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:
+        return {"ok": [], "error": f"搜索失败: {str(exc)[:50]}"}
+    if not arts:
+        return {"ok": [], "error": "未搜到"}
+    recs = []
+    for a in arts[:K_WEIXIN]:
+        h = hashlib.md5(a["url"].encode()).hexdigest()[:16]
+        base, pub = DATA / "search" / "weixin" / h, f"/data/search/weixin/{h}"
+        rec = {"title": a["title"], "url": a["url"], "nick": a["nick_name"], "cover": None}
+        if a["cover_url"]:
+            rec["cover"] = _dl(a["cover_url"], "weixin", base / "cover.jpg", pub + "/cover.jpg")
+        recs.append(rec)
+    return {"ok": recs, "error": None}
+
+
+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)
+    results = []
+    for tac in ("tactic1", "tactic2", "tactic4"):
+        t = demo.get(tac) or {}
+        name = t.get("name", tac)
+        for it in (t.get("items") or [])[:N]:
+            q = it.get("query")
+            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})
+    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}")
+
+
+if __name__ == "__main__":
+    main()

tests/test_search.py

@@ -6,10 +6,10 @@ from pathlib import Path
 
 import pytest
 
-from creation_knowledge.config import PgConfig, Settings
-from creation_knowledge.integrations.crawler import RateLimiter
-from creation_knowledge.integrations.search import (
-    SearchError, parse_search_response, search_keyword,
+from core.config import PgConfig, Settings
+from acquisition.crawler import RateLimiter
+from acquisition.search import (
+    SearchError, parse_search_response, parse_weixin, search_keyword,
 )
 
 FIX = Path(__file__).parent / "fixtures"
@@ -112,6 +112,24 @@ def test_douyin_body_has_account_id_and_params():
     assert body["cursor"] == "0"                       # 抖音起始 cursor
 
 
+def test_parse_weixin_cleans_title_and_keeps_url_cover():
+    resp = {"code": 0, "data": {"data": [
+        {"title": "复旦教授<em class='highlight'>历史科普</em>", "url": "http://mp.weixin.qq.com/s?x=1",
+         "cover_url": "https://mmbiz.qpic.cn/a.jpg", "nick_name": "某号", "time": "1天前"},
+        {"title": "无链接的应被跳过"},   # 无 url → 跳过
+    ]}}
+    out = parse_weixin(resp, limit=5)
+    assert len(out) == 1
+    assert out[0]["title"] == "复旦教授历史科普"          # <em> 标记被清掉
+    assert out[0]["url"].startswith("http://mp.weixin.qq.com")
+    assert out[0]["cover_url"].endswith("a.jpg") and out[0]["nick_name"] == "某号"
+
+
+def test_parse_weixin_business_error():
+    with pytest.raises(SearchError):
+        parse_weixin({"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])