"""Research 时讯质量：可信源、同事件去重、tech 主题过滤、国内配额。""" from __future__ import annotations import json import unittest from pathlib import Path from unittest import mock from daily.news.research_quality import ( build_deduped_candidate_pool, dedupe_same_event, filter_tech_against_items, is_cn_item, is_trusted_item, pack_with_cn_quota, post_process_research_news, research_cn_min, ) def _item( title: str, link: str, *, source_name: str = "", desc_short: str = "", region: str | None = None, ) -> dict: row = { "title": title, "link": link, "source_name": source_name, "desc_short": desc_short, "summary_plain": desc_short, "published_fmt": "", } if region is not None: row["region"] = region return row class TestTrustedAndCn(unittest.TestCase): def test_trusted_official_and_authority(self): self.assertTrue( is_trusted_item(_item("x", "https://openai.com/blog/x", source_name="OpenAI")) ) self.assertTrue( is_trusted_item(_item("x", "https://techcrunch.com/a", source_name="TechCrunch")) ) self.assertTrue( is_trusted_item(_item("x", "https://www.qbitai.com/a", source_name="量子位")) ) def test_rejects_low_quality(self): self.assertFalse( is_trusted_item( _item("x", "https://wap.stockstar.com/detail/IG1", source_name="证券之星") ) ) self.assertFalse( is_trusted_item(_item("x", "https://random-blog.xyz/a", source_name="Unknown")) ) self.assertFalse( is_trusted_item( _item("x", "https://www.techtimes.com/articles/1.htm", source_name="TechTimes") ) ) def test_trusted_cn_majors_and_engadget(self): self.assertTrue( is_trusted_item(_item("x", "https://www.yicai.com/news/1.html", source_name="第一财经")) ) self.assertTrue( is_trusted_item( _item("x", "https://www.news.cn/world/20260729/a/c.html", source_name="新华网") ) ) self.assertTrue( is_trusted_item( _item("x", "https://www.engadget.com/2225849/google/", source_name="Engadget") ) ) def test_cn_by_whitelist_region_and_cn_tld(self): self.assertTrue(is_cn_item(_item("x", "https://www.qbitai.com/a", source_name="量子位"))) self.assertTrue( is_cn_item(_item("x", "https://techcrunch.com/a", source_name="TechCrunch", region="cn")) ) self.assertTrue(is_cn_item(_item("x", "https://news.example.cn/a", source_name="X"))) self.assertFalse(is_cn_item(_item("x", "https://techcrunch.com/a", source_name="TechCrunch"))) class TestSameEventAndTech(unittest.TestCase): def test_petition_cluster_keeps_one(self): items = [ _item( "OpenAI, Anthropic scientists ask U.S. for tools to pace AI development", "https://www.nbcnews.com/tech/a", source_name="NBC News", desc_short="超千名前沿实验室员工联名，吁美政府支持控制 AI 研发节奏", ), _item( "Sam Altman is ready to decelerate", "https://techcrunch.com/2026/07/28/sam-altman-is-ready-to-decelerate/", source_name="TechCrunch", desc_short="奥特曼称或需控制 AI 发展速度，并支持员工联名请愿", ), ] out = dedupe_same_event(items) self.assertEqual(len(out), 1) def test_distinct_clusters_kept(self): items = [ _item( "Sam Altman is ready to decelerate", "https://techcrunch.com/a", source_name="TechCrunch", desc_short="奥特曼称或需控制 AI 发展速度", ), _item( "OpenAI’s agent siege forced rebuild at Hugging Face", "https://www.theregister.com/ai/a", source_name="The Register", desc_short="Hugging Face 因 OpenAI 智能体入侵重建基础设施", ), ] out = dedupe_same_event(items) self.assertEqual(len(out), 2) def test_tech_drops_kimi_adapt_when_items_have_kimi_open_source(self): items = [ _item( "Moonshot Open-Sources Kimi K3", "https://www.caixinglobal.com/a", source_name="Caixin", desc_short="月之暗面开放 Kimi K3 权重与技术报告", ) ] tech = [ _item( "moonshotai/Kimi-K3 · Hugging Face", "https://huggingface.co/moonshotai/Kimi-K3", source_name="Hugging Face", desc_short="Kimi K3 开源权重上线", ), _item( "华为官宣昇腾 Day0 支持 Kimi K3", "https://www.ithome.com/0/982/615.htm", source_name="IT之家", desc_short="昇腾宣布适配 Kimi K3 训练与推理", ), _item( "MCP Specification 2026-07-28", "https://blog.modelcontextprotocol.io/posts/2026-07-28/", source_name="MCP Blog", desc_short="MCP 正式发布新规范", ), ] kept = filter_tech_against_items(tech, items) self.assertEqual(len(kept), 1) self.assertIn("MCP", kept[0]["title"]) class TestDedupedCandidatePool(unittest.TestCase): def test_pool_is_unique_events_after_fetch(self): items = [ _item( "OpenAI, Anthropic scientists ask U.S. for tools to pace AI development", "https://www.nbcnews.com/tech/a", source_name="NBC News", desc_short="超千名前沿实验室员工联名，吁美政府支持控制 AI 研发节奏", ), _item( "Sam Altman is ready to decelerate", "https://techcrunch.com/2026/07/28/sam-altman-is-ready-to-decelerate/", source_name="TechCrunch", desc_short="奥特曼称或需控制 AI 发展速度，并支持员工联名请愿", ), _item( "junk", "https://wap.stockstar.com/detail/1", source_name="证券之星", desc_short="营销稿", ), ] pool, tech = build_deduped_candidate_pool(items, []) self.assertEqual(tech, []) self.assertEqual(len(pool), 1) self.assertTrue(all(is_trusted_item(i) for i in pool)) class TestCnQuota(unittest.TestCase): def test_cn_min_default_30_percent(self): with mock.patch.dict("os.environ", {"DAILY_WECOM_AI_NEWS_CN_MIN": "0"}, clear=False): self.assertEqual(research_cn_min(10), 3) with mock.patch.dict("os.environ", {"DAILY_WECOM_AI_NEWS_CN_MIN": "4"}, clear=False): self.assertEqual(research_cn_min(10), 4) def test_pack_reserves_cn_slots(self): items = [ _item("I1", "https://techcrunch.com/1", source_name="TechCrunch", desc_short="国际1"), _item("I2", "https://techcrunch.com/2", source_name="TechCrunch", desc_short="国际2"), _item("I3", "https://techcrunch.com/3", source_name="TechCrunch", desc_short="国际3"), _item("C1", "https://www.qbitai.com/1", source_name="量子位", desc_short="国内1"), _item("C2", "https://www.36kr.com/1", source_name="36氪", desc_short="国内2"), _item("C3", "https://www.jiqizhixin.com/1", source_name="机器之心", desc_short="国内3"), ] out = pack_with_cn_quota(items, limit=5, min_cn=3) self.assertEqual(len(out), 5) self.assertGreaterEqual(sum(1 for i in out if is_cn_item(i)), 3) class TestPostProcessIntegration(unittest.TestCase): def test_sample_day_filters_stockstar_and_dedupes(self): sample = Path(__file__).resolve().parents[1] / "output" / "2026-07-29.ai-news-research.json" if not sample.exists(): self.skipTest("sample research json missing") raw = json.loads(sample.read_text(encoding="utf-8")) items = [ _item( r["title"], r["link"], source_name=r.get("source_name", ""), desc_short=r.get("desc_short", ""), ) for r in raw["items"] ] tech = [ _item( r["title"], r["link"], source_name=r.get("source_name", ""), desc_short=r.get("desc_short", ""), ) for r in raw["tech_items"] ] out_items, out_tech = post_process_research_news( items, tech, limit=10, tech_limit=5, min_cn=3 ) links = {i["link"] for i in out_items + out_tech} self.assertTrue(all("stockstar" not in link for link in links)) # 联名/减速同簇只留一条 petitionish = [ i for i in out_items if "decelerat" in i["title"].lower() or "联名" in (i.get("desc_short") or "") or "pace AI" in i["title"] ] self.assertLessEqual(len(petitionish), 1) # Kimi 适配不应再堆在 tech kimi_tech = [t for t in out_tech if "kimi" in (t["title"] + t.get("desc_short", "")).lower()] self.assertEqual(kimi_tech, []) def test_sample_2026_07_30_keeps_cn_majors(self): sample = Path(__file__).resolve().parents[1] / "output" / "2026-07-30.ai-news-research.json" if not sample.exists(): self.skipTest("sample research json missing") raw = json.loads(sample.read_text(encoding="utf-8")) items = [ _item( r["title"], r["link"], source_name=r.get("source_name", ""), desc_short=r.get("desc_short", ""), region=r.get("region"), ) for r in raw["items"] ] tech = [ _item( r["title"], r["link"], source_name=r.get("source_name", ""), desc_short=r.get("desc_short", ""), region=r.get("region"), ) for r in raw["tech_items"] ] out_items, out_tech = post_process_research_news( items, tech, limit=10, tech_limit=5, min_cn=3 ) # 不应再被白名单误杀成只剩 1 条 self.assertGreaterEqual(len(out_items) + len(out_tech), 4) self.assertGreaterEqual(sum(1 for i in out_items if is_cn_item(i)), 2) hosts = " ".join(i["link"] for i in out_items + out_tech) self.assertNotIn("techtimes.com", hosts) self.assertTrue("yicai.com" in hosts or "news.cn" in hosts) if __name__ == "__main__": unittest.main()