feat(news): research 时讯加可信过滤、去重与国内配额

候选池按独立事件拉取，白名单过滤低质源，同事件与 tech 主题去重后按展示上限打包。 Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-31 15:47:50 +08:00
parent 0384e2c7a9
commit e2f609a027
6 changed files with 848 additions and 23 deletions
--- a/.env.example
+++ b/.env.example
@@ -38,6 +38,11 @@ DAILY_WECOM_GITHUB_TOPIC=5
 DAILY_WECOM_AI_NEWS=10
 # research 模式额外技术类时讯条数（叠加在 AI 时讯精选之上）
 DAILY_WECOM_AI_NEWS_TECH=5
+# research 主列表国内最少条数；0=按约 30% 推算（10→3）
+# DAILY_WECOM_AI_NEWS_CN_MIN=3
+# research 去重后候选池（独立事件数；默认展示×2；0=自动）
+# DAILY_AI_NEWS_RESEARCH_POOL=20
+# DAILY_AI_NEWS_RESEARCH_TECH_POOL=10
 DAILY_WECOM_CN_AI_NEWS=10
 # 企微新闻摘要字数（句读/词边界截断，不加省略号）
 # DAILY_WECOM_NEWS_DESC_LIMIT=72
--- a/daily/news/research.py
+++ b/daily/news/research.py
@@ -13,6 +13,7 @@ from daily.config import OUTPUT_DIR, ROOT, env, env_int, wecom_ai_news_tech_limi
 from daily.llm_client import cursor_agent_prompt, extract_json_object, has_cursor_configured
 from daily.news.fetch import brief_news_summary, _normalize_link
 from daily.news.pushed_links import filter_unpushed_items
+from daily.news.research_quality import post_process_research_news, research_cn_min

 logger = logging.getLogger(__name__)

@@ -40,6 +41,25 @@ def research_limit() -> int:
    return max(1, env_int("DAILY_WECOM_AI_NEWS", 10))


+def research_pool_limit(display_limit: int | None = None) -> int:
+    """Agent 原始候选条数（展示上限之上多拉，供可信/去重筛）。"""
+    lim = display_limit if display_limit is not None else research_limit()
+    explicit = env_int("DAILY_AI_NEWS_RESEARCH_POOL", 0)
+    if explicit > 0:
+        return max(lim, explicit)
+    return max(lim * 2, lim + 8)
+
+
+def research_tech_pool_limit(display_limit: int | None = None) -> int:
+    tech = display_limit if display_limit is not None else research_tech_limit()
+    if tech <= 0:
+        return 0
+    explicit = env_int("DAILY_AI_NEWS_RESEARCH_TECH_POOL", 0)
+    if explicit > 0:
+        return max(tech, explicit)
+    return max(tech * 2, tech + 4)
+
+
 def research_json_path(date_str: str) -> Path:
    return OUTPUT_DIR / f"{date_str}.ai-news-research.json"

@@ -90,7 +110,7 @@ def _normalize_research_item(raw: dict[str, Any]) -> dict[str, Any] | None:
    if not title or not link or not link.startswith("http"):
        return None
    desc = brief_news_summary(str(raw.get("desc_short") or raw.get("summary") or ""))
-    return {
+    item: dict[str, Any] = {
        "title": title,
        "link": link,
        "source_name": _guess_source_name(link, str(raw.get("source_name") or "")),
@@ -98,6 +118,10 @@ def _normalize_research_item(raw: dict[str, Any]) -> dict[str, Any] | None:
        "desc_short": desc,
        "summary_plain": desc,
    }
+    region = str(raw.get("region") or "").strip().lower()
+    if region:
+        item["region"] = region
+    return item


 def research_tech_limit() -> int:
@@ -133,11 +157,17 @@ def parse_research_response(
    *,
    limit: int,
    tech_limit: int = 0,
+    pool_limit: int | None = None,
+    tech_pool_limit: int | None = None,
 ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    parsed = extract_json_object(raw)
+    item_cap = pool_limit if pool_limit is not None else limit
+    tech_cap = tech_pool_limit if tech_pool_limit is not None else tech_limit
    seen: set[str] = set()
-    items = _parse_items_array(parsed.get("items"), limit=limit, seen=seen)
-    tech_items = _parse_items_array(parsed.get("tech_items"), limit=tech_limit, seen=seen) if tech_limit else []
+    items = _parse_items_array(parsed.get("items"), limit=item_cap, seen=seen)
+    tech_items = (
+        _parse_items_array(parsed.get("tech_items"), limit=tech_cap, seen=seen) if tech_cap else []
+    )
    return items, tech_items


@@ -189,25 +219,38 @@ def fetch_ai_news_research(

    skill = _load_skill()
    now_cst = datetime.now(timezone(timedelta(hours=8)))
+    cn_min = research_cn_min(lim)
+    pool = research_pool_limit(lim)
+    tech_pool = research_tech_pool_limit(tech_lim)
+    # 候选池内国内目标略高于展示配额，避免筛完国内不足
+    cn_pool_target = max(cn_min * 2, cn_min + 2)
+    # 解析多收一点原始行，输出前/后处理再压成「去重后候选池」
+    raw_cap = max(pool + 10, (pool * 3) // 2)
+    tech_raw_cap = max(tech_pool + 4, (tech_pool * 3) // 2) if tech_pool else 0
    tech_clause = ""
-    if tech_lim:
+    if tech_pool:
        tech_clause = (
-            f"\n另输出 **tech_items 恰好 {tech_lim} 条**，聚焦工程技术："
+            f"\n另输出 **去重后** 约 **{tech_pool} 条** tech_items 候选（最终展示约 {tech_lim} 条），聚焦工程技术："
            "模型/框架发布、开源项目、芯片算力、开发者工具、推理与工程实践。"
-            "与 items 不得重复 link。"
+            "不得与 items 重复 link/同事件；输出前自行去重，候选池内每条应为独立事件。"
        )
    system = (
        f"{skill}\n\n"
        "当前执行 **早报 AI 时讯调研**。\n"
        f"时间窗口：近 **{h}** 小时（截至 {now_cst.strftime('%Y-%m-%d %H:%M')} UTC+8）。\n"
-        f"输出 **恰好 {lim} 条** items，按重要性排序。{tech_clause}\n"
-        "使用 WebSearch 检索；不要读取本项目文档或 RSS 配置。"
+        f"输出 **同事件去重后** 约 **{pool} 条** items 候选（按重要性排序；最终展示约 {lim} 条）。\n"
+        "候选池条数 = 独立事件数：同一事件多源报道只留一条最权威源，禁止用换源重复充数。\n"
+        f"去重后的候选中国内可信源尽量不少于 **{cn_pool_target}** 条（展示侧至少 {cn_min} 条）。\n"
+        f"禁止用低质源凑数；可信独立事件不足才少返回。{tech_clause}\n"
+        "只采用官方博客/新闻稿、政府监管原文、一线权威媒体、学术官方；"
+        "禁止二手搬运、标题党、营销号。使用 WebSearch 检索；不要读取本项目文档或 RSS 配置。"
    )
    user = (
        f"/deep-research 获取近 {h} 小时的 AI 人工智能新闻资讯，"
-        "不区分国内国外，合并精选。"
-        f"只输出 JSON，items 长度={lim}"
-        + (f"，tech_items 长度={tech_lim}" if tech_lim else "")
+        f"国内与国际合并；items 去重后约 {pool} 条独立事件（国内可信尽量 ≥{cn_pool_target}）；"
+        "输出前完成同事件去重；可信度不足则不写。"
+        f"只输出 JSON，items 去重后目标约 {pool} 条"
+        + (f"，tech_items 去重后目标约 {tech_pool} 条" if tech_pool else "")
        + "。"
    )

@@ -234,7 +277,13 @@ def fetch_ai_news_research(
            "stats": {"error": "empty_response"},
        }

-    items, tech_items = parse_research_response(raw, limit=lim, tech_limit=tech_lim)
+    items, tech_items = parse_research_response(
+        raw,
+        limit=lim,
+        tech_limit=tech_lim,
+        pool_limit=raw_cap,
+        tech_pool_limit=tech_raw_cap,
+    )
    payload = extract_json_object(raw)
    if payload:
        _save_research_json(research_json_path(date_str), payload)
@@ -250,10 +299,23 @@ def fetch_ai_news_research(
            "stats": {"error": "invalid_json"},
        }

+    items, tech_items = post_process_research_news(
+        items,
+        tech_items,
+        limit=lim,
+        tech_limit=tech_lim,
+        min_cn=cn_min,
+    )
    items = _apply_pushed_dedup(items, date_str=date_str, limit=lim)
    if tech_items:
        tech_items = _apply_pushed_dedup(tech_items, date_str=date_str, limit=tech_lim)
-    logger.info("AI 时讯 research 完成：%d 条 + %d 技术", len(items), len(tech_items))
+    logger.info(
+        "AI 时讯 research 完成：%d 条 + %d 技术（候选池 %d/%d）",
+        len(items),
+        len(tech_items),
+        pool,
+        tech_pool,
+    )

    flat = [
        {
--- a/daily/news/research_quality.py
+++ b/daily/news/research_quality.py
@@ -0,0 +1,442 @@
+"""Research 时讯后处理：可信源、同事件去重、tech 主题过滤、国内配额。"""
+
+from __future__ import annotations
+
+import logging
+import re
+from typing import Any
+from urllib.parse import urlparse
+
+from daily.config import env_int
+
+logger = logging.getLogger(__name__)
+
+# 官方域（同事件去重时优先保留）
+_OFFICIAL_HOST_SUFFIXES: tuple[str, ...] = (
+    "openai.com",
+    "anthropic.com",
+    "deepmind.google",
+    "blog.google",
+    "ai.googleblog.com",
+    "microsoft.com",
+    "meta.com",
+    "engineering.fb.com",
+    "nvidia.com",
+    "huggingface.co",
+    "arxiv.org",
+    "github.com",
+    "github.blog",
+    "modelcontextprotocol.io",
+    "cursor.com",
+    "vercel.com",
+    "langchain.dev",
+    "cohere.com",
+    "moonshot.cn",
+    "moonshot.ai",
+    "sktelecom.com",
+    "tether.io",
+)
+
+# 权威媒体 / 可信站（国际 + 国内）
+_TRUSTED_HOST_SUFFIXES: tuple[str, ...] = _OFFICIAL_HOST_SUFFIXES + (
+    "techcrunch.com",
+    "theverge.com",
+    "wired.com",
+    "arstechnica.com",
+    "venturebeat.com",
+    "technologyreview.com",
+    "engadget.com",
+    "cnet.com",
+    "zdnet.com",
+    "axios.com",
+    "reuters.com",
+    "bloomberg.com",
+    "bloomberglaw.com",
+    "bbc.com",
+    "bbc.co.uk",
+    "nytimes.com",
+    "wsj.com",
+    "ft.com",
+    "nbcnews.com",
+    "time.com",
+    "theguardian.com",
+    "washingtonpost.com",
+    "theregister.com",
+    "nature.com",
+    "science.org",
+    "scmp.com",
+    "caixinglobal.com",
+    "caixin.com",
+    "qbitai.com",
+    "36kr.com",
+    "leiphone.com",
+    "jiqizhixin.com",
+    "ithome.com",
+    "tmtpost.com",
+    "huxiu.com",
+    "solidot.org",
+    "synched.cn",
+    "infoq.cn",
+    "yicai.com",
+    "news.cn",
+    "xinhuanet.com",
+    "people.com.cn",
+    "cls.cn",
+    "geekpark.net",
+    "standard.com",
+    "business-standard.com",
+    "siliconvalley.com",
+)
+
+_TRUSTED_SOURCE_NAMES: frozenset[str] = frozenset(
+    {
+        "techcrunch",
+        "the verge",
+        "wired",
+        "ars technica",
+        "engadget",
+        "reuters",
+        "bloomberg",
+        "bloomberg law",
+        "nbc news",
+        "time",
+        "the register",
+        "openai",
+        "anthropic",
+        "arxiv",
+        "hugging face",
+        "mcp blog",
+        "github",
+        "sk telecom",
+        "tether",
+        "量子位",
+        "36氪",
+        "36kr",
+        "雷锋网",
+        "机器之心",
+        "it之家",
+        "财新",
+        "caixin",
+        "钛媒体",
+        "虎嗅",
+        "第一财经",
+        "新华网",
+        "新华社",
+        "财联社",
+        "极客公园",
+        "人民日报",
+    }
+)
+
+_CN_HOST_SUFFIXES: tuple[str, ...] = (
+    "qbitai.com",
+    "36kr.com",
+    "leiphone.com",
+    "jiqizhixin.com",
+    "ithome.com",
+    "caixin.com",
+    "caixinglobal.com",
+    "tmtpost.com",
+    "huxiu.com",
+    "solidot.org",
+    "synched.cn",
+    "infoq.cn",
+    "moonshot.cn",
+    "yicai.com",
+    "news.cn",
+    "xinhuanet.com",
+    "people.com.cn",
+    "cls.cn",
+    "geekpark.net",
+    "zhihu.com",
+    "sina.com.cn",
+    "qq.com",
+    "163.com",
+)
+
+_CN_SOURCE_NAMES: frozenset[str] = frozenset(
+    {
+        "量子位",
+        "36氪",
+        "36kr",
+        "雷锋网",
+        "机器之心",
+        "it之家",
+        "财新",
+        "caixin",
+        "钛媒体",
+        "虎嗅",
+        "月之暗面",
+        "第一财经",
+        "新华网",
+        "新华社",
+        "财联社",
+        "极客公园",
+        "人民日报",
+    }
+)
+
+_ENTITIES: tuple[tuple[str, tuple[str, ...]], ...] = (
+    ("openai", ("openai", "altman", "chatgpt", "奥特曼")),
+    ("anthropic", ("anthropic", "amodei", "claude")),
+    ("kimi", ("kimi", "moonshot", "月之暗面")),
+    ("mcp", ("mcp", "model context protocol", "modelcontextprotocol")),
+    ("nvidia", ("nvidia", "英伟达")),
+    ("amd", ("amd",)),
+    ("hugging_face", ("hugging face", "huggingface")),
+    ("google", ("google", "deepmind", "gemini")),
+    ("meta", ("meta", "llama")),
+    ("microsoft", ("microsoft", "copilot")),
+    ("huawei", ("huawei", "华为", "昇腾", "ascend")),
+    ("moore", ("摩尔线程", "moore threads", "musa")),
+)
+
+_EVENT_CLUSTERS: tuple[tuple[str, tuple[str, ...]], ...] = (
+    ("petition", ("petition", "联名", "decelerat", "pace ai", "控制", "减速")),
+    ("hack", ("hack", "入侵", "siege", "breach", "攻击", "逃逸")),
+    ("open_source", ("open-source", "opensource", "open sources", "开源", "open-sources")),
+    ("adapt", ("适配", "adapt", "day-0", "day0", "day 0", "推理部署", "训练适配")),
+    ("release", ("release", "发布", "specification", "规范", "v2.0", "changelog")),
+    ("chip", ("chip", "芯片", "data center", "数据中心", "mi455")),
+    ("regulate", ("framework", "监管", "voluntary", "审核", "ban", "禁止")),
+)
+
+_RELEASE_FAMILY = frozenset({"open_source", "adapt", "release"})
+# 分发平台，不参与 tech↔items 主题冲突（避免 HF 上架与 HF 被黑误杀）
+_PLATFORM_ENTITIES = frozenset({"hugging_face"})
+
+
+def _host(link: str) -> str:
+    return urlparse(link).netloc.lower().removeprefix("www.")
+
+
+def _ends_with_any(host: str, suffixes: tuple[str, ...]) -> bool:
+    return any(host == s or host.endswith("." + s) for s in suffixes)
+
+
+def _norm_text(*parts: str) -> str:
+    text = " ".join(p for p in parts if p).lower()
+    text = re.sub(r"[\s\-_/|·，。、：:（）()【】\[\]]+", " ", text)
+    return text.strip()
+
+
+def _item_text(item: dict[str, Any]) -> str:
+    return _norm_text(
+        str(item.get("title") or ""),
+        str(item.get("desc_short") or item.get("summary_plain") or ""),
+    )
+
+
+def _match_labels(text: str, table: tuple[tuple[str, tuple[str, ...]], ...]) -> frozenset[str]:
+    hit: set[str] = set()
+    for label, kws in table:
+        if any(kw in text for kw in kws):
+            hit.add(label)
+    return frozenset(hit)
+
+
+def entities_of(item: dict[str, Any]) -> frozenset[str]:
+    return _match_labels(_item_text(item), _ENTITIES)
+
+
+def events_of(item: dict[str, Any]) -> frozenset[str]:
+    return _match_labels(_item_text(item), _EVENT_CLUSTERS)
+
+
+def event_key(item: dict[str, Any]) -> tuple[frozenset[str], frozenset[str]] | None:
+    ents = entities_of(item)
+    evs = events_of(item)
+    if not ents or not evs:
+        return None
+    return (ents, evs)
+
+
+def same_event(a: dict[str, Any], b: dict[str, Any]) -> bool:
+    """共享至少一实体且共享至少一事件簇 → 同事件（保守合并）。"""
+    ea, eva = entities_of(a), events_of(a)
+    eb, evb = entities_of(b), events_of(b)
+    if not ea or not eb or not eva or not evb:
+        return False
+    return bool(ea & eb) and bool(eva & evb)
+
+
+def is_official_item(item: dict[str, Any]) -> bool:
+    return _ends_with_any(_host(str(item.get("link") or "")), _OFFICIAL_HOST_SUFFIXES)
+
+
+def _is_institutional_cn_host(host: str) -> bool:
+    """新华社 / 政府站等机构域，默认可信。"""
+    if host.endswith(".gov.cn") or host.endswith(".gov.cn."):
+        return True
+    if host == "news.cn" or host.endswith(".news.cn"):
+        return True
+    if host.endswith("xinhuanet.com") or host.endswith("people.com.cn"):
+        return True
+    return False
+
+
+def is_trusted_item(item: dict[str, Any]) -> bool:
+    host = _host(str(item.get("link") or ""))
+    if _is_institutional_cn_host(host):
+        return True
+    if _ends_with_any(host, _TRUSTED_HOST_SUFFIXES):
+        return True
+    name = str(item.get("source_name") or "").strip().lower()
+    return name in _TRUSTED_SOURCE_NAMES
+
+
+def is_cn_item(item: dict[str, Any]) -> bool:
+    region = str(item.get("region") or "").strip().lower()
+    if region in {"cn", "china", "zh", "zh-cn"}:
+        return True
+    if region in {"intl", "international", "global", "en"}:
+        return False
+    host = _host(str(item.get("link") or ""))
+    if host.endswith(".cn") or _ends_with_any(host, _CN_HOST_SUFFIXES):
+        return True
+    name = str(item.get("source_name") or "").strip().lower()
+    return name in {n.lower() for n in _CN_SOURCE_NAMES}
+
+
+def research_cn_min(limit: int) -> int:
+    explicit = env_int("DAILY_WECOM_AI_NEWS_CN_MIN", 0)
+    if explicit > 0:
+        return min(explicit, max(1, limit))
+    return max(1, limit * 3 // 10)
+
+
+def _trust_rank(item: dict[str, Any]) -> int:
+    if is_official_item(item):
+        return 0
+    if is_trusted_item(item):
+        return 1
+    return 2
+
+
+def filter_trusted(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    kept = [i for i in items if is_trusted_item(i)]
+    dropped = len(items) - len(kept)
+    if dropped:
+        logger.info("news_dedup_drop:trusted=%s", dropped)
+    return kept
+
+
+def dedupe_same_event(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """同事件只留一条；官方源优先，否则保留更靠前的。"""
+    kept: list[dict[str, Any]] = []
+    for item in items:
+        replaced = False
+        for idx, prev in enumerate(kept):
+            if not same_event(item, prev):
+                continue
+            if _trust_rank(item) < _trust_rank(prev):
+                kept[idx] = item
+            replaced = True
+            break
+        if not replaced:
+            kept.append(item)
+
+    dropped = len(items) - len(kept)
+    if dropped:
+        logger.info("news_dedup_drop:same_event=%s", dropped)
+    return kept
+
+
+def filter_tech_against_items(
+    tech_items: list[dict[str, Any]],
+    items: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    kept: list[dict[str, Any]] = []
+    for tech in tech_items:
+        t_ents = entities_of(tech)
+        t_evs = events_of(tech)
+        conflict = False
+        for item in items:
+            if same_event(tech, item):
+                conflict = True
+                break
+            shared = (t_ents & entities_of(item)) - _PLATFORM_ENTITIES
+            if shared and ((t_evs | events_of(item)) & _RELEASE_FAMILY):
+                conflict = True
+                break
+        if not conflict:
+            kept.append(tech)
+    dropped = len(tech_items) - len(kept)
+    if dropped:
+        logger.info("news_dedup_drop:tech_topic=%s", dropped)
+    return kept
+
+
+def pack_with_cn_quota(
+    items: list[dict[str, Any]],
+    *,
+    limit: int,
+    min_cn: int,
+) -> list[dict[str, Any]]:
+    if limit <= 0:
+        return []
+    min_cn = max(0, min(min_cn, limit))
+    out: list[dict[str, Any]] = []
+    used: set[str] = set()
+    cn_got = 0
+
+    def _take(item: dict[str, Any]) -> None:
+        nonlocal cn_got
+        link = str(item.get("link") or "")
+        if not link or link in used:
+            return
+        out.append(item)
+        used.add(link)
+        if is_cn_item(item):
+            cn_got += 1
+
+    for item in items:
+        if len(out) >= limit:
+            break
+        link = str(item.get("link") or "")
+        if not link or link in used:
+            continue
+        slots_left = limit - len(out)
+        need_cn = max(0, min_cn - cn_got)
+        if not is_cn_item(item) and slots_left <= need_cn:
+            continue
+        _take(item)
+
+    if len(out) < limit:
+        for item in items:
+            if len(out) >= limit:
+                break
+            _take(item)
+
+    final_cn = sum(1 for i in out if is_cn_item(i))
+    if final_cn < min_cn:
+        logger.info("news_cn_short:%s", final_cn)
+    return out[:limit]
+
+
+def build_deduped_candidate_pool(
+    items: list[dict[str, Any]],
+    tech_items: list[dict[str, Any]],
+) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    """拉取后立即得到去重候选池：可信过滤 + 同事件去重 + tech 相对 items 主题过滤。"""
+    items = dedupe_same_event(filter_trusted(items))
+    tech_items = dedupe_same_event(filter_trusted(tech_items))
+    tech_items = filter_tech_against_items(tech_items, items)
+    logger.info("research_pool_deduped:items=%s tech=%s", len(items), len(tech_items))
+    return items, tech_items
+
+
+def post_process_research_news(
+    items: list[dict[str, Any]],
+    tech_items: list[dict[str, Any]],
+    *,
+    limit: int,
+    tech_limit: int,
+    min_cn: int | None = None,
+) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    """去重候选池 → 国内配额打包 → 截到展示上限。"""
+    cn_min = research_cn_min(limit) if min_cn is None else max(0, min_cn)
+    items, tech_items = build_deduped_candidate_pool(items, tech_items)
+    items = pack_with_cn_quota(items, limit=limit, min_cn=cn_min)
+    tech_items = filter_tech_against_items(tech_items, items)
+    return items, tech_items[: max(0, tech_limit)]
--- a/skills/daily-ai-news-research/SKILL.md
+++ b/skills/daily-ai-news-research/SKILL.md
@@ -1,15 +1,16 @@
 # AI 时讯 Deep Research（早报专用）

-你是 **AI 时讯调研员**。使用 **WebSearch** 与网页抓取工具，收集近 N 小时全球 AI 新闻（不区分国内/国外），输出供企微早报使用的结构化 JSON。
+你是 **AI 时讯调研员**。使用 **WebSearch** 与网页抓取工具，收集近 N 小时 AI 新闻（国内 + 国际合并展示），输出供企微早报使用的结构化 JSON。

 ## 工作流

-1. 将任务拆成 3–5 个子问题（模型发布、监管政策、大厂动态、芯片算力、研究突破等）
+1. 将任务拆成 3–5 个子问题（模型发布、监管政策、大厂动态、芯片算力、研究突破等）；**中英文检索都要做**
 2. 每个子问题用 WebSearch 检索 2–3 组关键词（中英文混合）
-3. 交叉验证：优先权威媒体 / 官方博客 / 学术来源
-4. 精选最多 **10 条**最重要、可核实的新闻（`items`）；窗口内不足则少返回，勿凑数
-5. 另精选最多 **5 条**工程技术向新闻（`tech_items`）：模型/框架发布、开源、芯片算力、开发者工具、推理与工程实践；不得与 `items` 重复 link；不足则少返回
-6. **只输出 JSON**，不要 Markdown 报告，不要代码块
+3. 交叉验证：只采用 **官方博客 / 新闻稿、政府或监管原文、一线权威媒体、学术官方**
+4. 按请求输出 **已去重候选池**（通常多于最终展示，如展示 10 → 去重后约 20）：**条数 = 独立事件数**。输出前必须完成同事件/同 link 去重；多源只留最权威一条。国内可信独立事件也要明显多于展示配额；禁止换源重复充数或用低质源灌满
+5. 另输出已去重的 `tech_items` 候选；不得与 `items` 重复 link/同事件；技术区不强制国内
+6. **候选池内同事件只允许一条**（官方 > 一线媒体）
+7. **只输出 JSON**，不要 Markdown 报告，不要代码块

 ## 质量规则

@@ -19,6 +20,8 @@
 4. `desc_short` 用中文一句话摘要（≤72 字）
 5. `title` 保留原文标题；中文源可用中文标题
 6. `source_name` 为媒体/站点简称（如 TechCrunch、量子位、OpenAI Blog）
+7. 可选 `region`: `"cn"` 或 `"intl"`（国内源标 `cn`）
+8. **禁止**二手搬运、标题党、不明自媒体、证券营销号；无权威源交叉验证则 **不写**

 ## 输出格式（严格 JSON）

@@ -29,6 +32,7 @@
      "title": "Apple sues OpenAI over trade secret theft",
      "link": "https://techcrunch.com/...",
      "source_name": "TechCrunch",
+      "region": "intl",
      "desc_short": "苹果起诉 OpenAI 涉嫌窃取硬件商业机密",
      "published_fmt": "07-11 05:00"
    }
@@ -36,8 +40,9 @@
  "tech_items": [
    {
      "title": "Meta Iris AI chip enters production",
-      "link": "https://example.com/...",
+      "link": "https://techcrunch.com/...",
      "source_name": "TechCrunch",
+      "region": "intl",
      "desc_short": "Meta 自研 Iris 芯片 9 月量产",
      "published_fmt": ""
    }
@@ -46,8 +51,8 @@
 }
 ```

- `items` 数组长度 **必须等于** 请求的 limit（默认 10）
- `tech_items` 数组长度 **必须等于** 请求的 tech limit（默认 5）；聚焦工程技术，可与 `items` 主题重叠但 link 不得重复
+- `items` / `tech_items`：条数以请求的**去重后候选目标**为准（独立事件数；可略少，不可灌重复或低质源）
+- `tech_items` 聚焦工程技术；可与 `items` 领域相近，但 **事件与产品不得重复**
 - `published_fmt` 格式 `MM-DD HH:MM`（UTC+8），无法确定则留空字符串
 - 不要输出 `items` 以外的长文；`methodology` 可选，一行即可

@@ -56,3 +61,4 @@
 - 不要输出 ```json 代码块包裹（直接输出 JSON 对象）
 - 不要输出 Executive Summary / Key Takeaways 等报告章节
 - 不要使用本项目 RSS 或本地文档作为来源
+- 不要为凑国内配额或条数而写入低可信来源
--- a/tests/test_ai_news_research.py
+++ b/tests/test_ai_news_research.py
@@ -5,7 +5,24 @@ from __future__ import annotations
 import unittest

 from daily.format_wecom import _ai_news_lines, replace_wecom_news_sections
-from daily.news.research import parse_research_response
+from daily.news.research import (
+    parse_research_response,
+    research_pool_limit,
+    research_tech_pool_limit,
+)
+
+
+class TestAiNewsResearchPool(unittest.TestCase):
+    def test_pool_defaults_above_display(self):
+        self.assertEqual(research_pool_limit(10), 20)
+        self.assertEqual(research_tech_pool_limit(5), 10)
+
+    def test_pool_env_override(self):
+        import os
+        from unittest import mock
+
+        with mock.patch.dict(os.environ, {"DAILY_AI_NEWS_RESEARCH_POOL": "24"}, clear=False):
+            self.assertEqual(research_pool_limit(10), 24)


 class TestAiNewsResearchParse(unittest.TestCase):
--- a/tests/test_ai_news_research_quality.py
+++ b/tests/test_ai_news_research_quality.py
@@ -0,0 +1,293 @@
+"""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()