Create-knowledge-find-decode/scripts/run_search.py

"""用生成的 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()