| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051 |
- """decompose 入口二的纯逻辑离线测：slugify + expand_queries 跨 query 去重。"""
- from __future__ import annotations
- import scripts.decompose as d
- from creation_knowledge.config import PgConfig, Settings
- def _settings() -> Settings:
- return Settings(
- pg=PgConfig(host="h", port=5432, user="u", password="p", database="d"),
- crawler_base_url="x", crawler_key="", crawler_timeout=30,
- video_model="m", gemini_api_key="",
- openrouter_base_url="b", openrouter_api_key="k",
- llm_model="m", max_cards=12, frames_dir="f", douyin_ratio="540p", data_dir="")
- def test_slugify():
- assert d.slugify("分镜脚本") == "分镜脚本" # 中文保留
- assert d.slugify("A B!c") == "a-b-c" # 空格/标点→- 且小写
- assert d.slugify("") == "run" # 空→run
- assert d.slugify("!!!") == "run" # 全标点→run
- def test_expand_queries_dedup(monkeypatch):
- ret = {"q1": ["a", "b"], "q2": ["b", "c"]} # q1/q2 共享 id "b"
- def fake_search(keyword, **kw):
- return ret[keyword]
- monkeypatch.setattr(d, "search_keyword", fake_search)
- srcs = d.expand_queries(
- [{"query": "q1", "platform": "xiaohongshu"},
- {"query": "q2", "platform": "xiaohongshu"}], _settings())
- assert [s["cid"] for s in srcs] == ["a", "b", "c"] # b 去重，只出现一次
- assert all(s["from"] == "live" for s in srcs)
- assert srcs[0]["query"] == "q1" and srcs[2]["query"] == "q2" # b 归属首次命中的 q1
- def test_expand_queries_skips_failed(monkeypatch):
- def fake_search(keyword, **kw):
- if keyword == "bad":
- raise RuntimeError("search down")
- return ["x"]
- monkeypatch.setattr(d, "search_keyword", fake_search)
- srcs = d.expand_queries(
- [{"query": "bad", "platform": "xiaohongshu"},
- {"query": "ok", "platform": "xiaohongshu"}], _settings())
- assert [s["cid"] for s in srcs] == ["x"] # bad 整条跳过，不阻断 ok
|
