"""用生成的 query 真实搜帖子/视频（抖音 + 微信公众号 + 小红书），媒体存本地 data/。 每法取前 N 条 query，每条 query 每个渠道取 top-K： 抖音：search(top-1) → detail → 下载视频+封面（搜索/详情各自限速） 微信公众号：search(top-K) → 下载每个封面（封面在搜索回包里，无需 detail） 小红书：search(top-K) → 下载每个封面（封面在 note_card 里，无需 detail；视频帖无直链） 限流：三个平台的「搜索」各自独立闸，两次搜索间隔随机 10~12s；抖音「详情」走更轻的 3~5s 闸。 产出每条记录：{method, query, douyin/weixin/xiaohongshu:{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, search_xiaohongshu 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" DOUYIN_ON = True # 抖音走 piaoquantv，真跑 N = 6 # 每法取前 N 条 query K_DOUYIN = 3 # 抖音每条 query 取前 3 个视频（每个要 1 详情 + 下 mp4，最重，故取少） K_WEIXIN = 10 # 微信公众号每条 query 取前 10 个封面 K_XHS = 10 # 小红书每条 query 取前 10 帖（逐帖下完整图+正文） SEARCH_MIN, SEARCH_MAX = 10.0, 12.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, search_rl: RateLimiter, detail_rl: RateLimiter) -> dict: """抖音搜 top-K 视频：返回 {ok:[{title,url,cover,video}], error}。搜索/详情各走各的闸。""" try: ids = search_keyword(query, platform="douyin", content_type="视频", limit=K_DOUYIN, settings=settings, rate_limiter=search_rl) 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: post = fetch_post_detail(vid, settings=settings, rate_limiter=detail_rl) 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 _dedup(urls): seen, out = set(), [] for u in urls: if u and u not in seen: seen.add(u) out.append(u) return out def xiaohongshu_topk(query: str, settings: Settings, search_rl: RateLimiter, detail_rl: RateLimiter) -> dict: """小红书搜 top-K → 逐条调详情拉【完整帖子】：下载全部图片 + 存正文到本地。 小红书链接会被反爬封，必须把整帖存档，不能只留封面+链接。返回 {ok:[{title,url,nick,cover,images,body_text}], error}。""" try: posts = search_xiaohongshu(query, content_type="图文", limit=K_XHS, settings=settings, rate_limiter=search_rl) except Exception as exc: return {"ok": [], "error": f"搜索失败: {str(exc)[:50]}"} if not posts: return {"ok": [], "error": "未搜到"} recs = [] for p in posts[:K_XHS]: cid = p["id"] try: post = fetch_post_detail(cid, settings=settings, rate_limiter=detail_rl) except Exception: post = None base, pub = DATA / "search" / "xiaohongshu" / cid, f"/data/search/xiaohongshu/{cid}" srcs = _dedup((post.image_urls if post else None) or ([p["cover_url"]] if p.get("cover_url") else [])) imgs = [] for i, u in enumerate(srcs): local = _dl(u, "xiaohongshu", base / f"img_{i}.jpg", f"{pub}/img_{i}.jpg") if local: imgs.append(local) if not imgs: continue # 一张都没下来就丢（链接全失效） recs.append({"title": (post.title if post and post.title else p["title"]), "url": p["url"], "nick": p["nick_name"], "body_text": (post.body_text if post else ""), "cover": imgs[0], "images": imgs}) return {"ok": recs, "error": None if recs else "详情/下载均失败"} def _prev_douyin() -> dict: """抖音关闭时，从上次的 search_results.json 沿用抖音结果（按 query 文本取），保住已抓到的视频。""" if not OUT.exists(): return {} try: prev = json.loads(OUT.read_text("utf-8")) except Exception: return {} return {r["query"]: r.get("douyin") for r in prev if r.get("query")} def main() -> None: settings = Settings.from_env() demo = json.loads(DEMO.read_text("utf-8")) # 三平台搜索各自独立闸（随机 10~12s）；抖音详情更轻（3~5s） dy_search = RateLimiter(min_interval_seconds=SEARCH_MIN, max_interval_seconds=SEARCH_MAX) dy_detail = RateLimiter(min_interval_seconds=3.0, max_interval_seconds=5.0) wx_search = RateLimiter(min_interval_seconds=SEARCH_MIN, max_interval_seconds=SEARCH_MAX) xhs_search = RateLimiter(min_interval_seconds=SEARCH_MIN, max_interval_seconds=SEARCH_MAX) xhs_detail = RateLimiter(min_interval_seconds=3.0, max_interval_seconds=5.0) # 小红书详情（拉完整帖） prev_dy = {} if DOUYIN_ON else _prev_douyin() if not DOUYIN_ON: print(f"（抖音已关闭：沿用上次 {len(prev_dy)} 条 query 的抖音结果，本次只跑微信+小红书）") 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}") if DOUYIN_ON: dy = douyin_topk(q, settings, dy_search, dy_detail) else: dy = prev_dy.get(q) or {"ok": [], "error": "抖音限流暂停，未跑"} wx = weixin_topk(q, settings, wx_search) xhs = xiaohongshu_topk(q, settings, xhs_search, xhs_detail) xhs_imgs = sum(len(r.get("images") or []) for r in xhs["ok"]) print(f" 抖音 {len(dy['ok'])} 视频{_err(dy)}" f" 微信 {len(wx['ok'])} 封面{_err(wx)}" f" 小红书 {len(xhs['ok'])} 帖/{xhs_imgs} 图{_err(xhs)}") results.append({"method": name, "query": q, "douyin": dy, "weixin": wx, "xiaohongshu": xhs}) OUT.parent.mkdir(parents=True, exist_ok=True) OUT.write_text(json.dumps(results, ensure_ascii=False, indent=1), encoding="utf-8") tot = lambda k: sum(len(r[k]["ok"]) for r in results) print(f"\nwrote {len(results)} 条 query（抖音 {tot('douyin')} / 微信 {tot('weixin')} / " f"小红书 {tot('xiaohongshu')}）→ {OUT}") def _err(d: dict) -> str: return " / " + d["error"] if d.get("error") else "" if __name__ == "__main__": main()