diff --git a/apps/api/routers/cs_feeds.py b/apps/api/routers/cs_feeds.py index de33b1a..e3fe409 100644 --- a/apps/api/routers/cs_feeds.py +++ b/apps/api/routers/cs_feeds.py @@ -6,6 +6,7 @@ from fastapi import APIRouter, Depends, Query, Request +from packages.ai.cs_feed_orchestrator import CSFeedOrchestrator from packages.domain.task_tracker import global_tracker from packages.storage.db import SessionLocal from packages.storage.repositories import CSFeedRepository @@ -165,19 +166,28 @@ def _fetch_fn(progress_callback=None): progress_callback(f"开始入库 ({total_papers} 篇)...", 50, 100) count = 0 + paper_ids: list[str] = [] with session_scope() as session: paper_repo = PaperRepository(session) for i, p in enumerate(papers): - paper_repo.upsert_paper(p) + saved = paper_repo.upsert_paper(p) count += 1 + paper_ids.append(saved.id) if progress_callback: progress_callback( f"入库中 ({i + 1}/{total_papers})...", 50 + int((i + 1) / total_papers * 40), 100, ) + # 关联到 cs_feed 分类 topic + 触发 auto_link（与 orchestrator 行为一致） + if paper_ids: + CSFeedOrchestrator._link_cs_papers_to_topic(session, category_code, paper_ids) repo.update_run_status(category_code, count) + # 抓取后触发引用自动关联（与 orchestrator 对齐） + if paper_ids: + CSFeedOrchestrator._trigger_auto_link(paper_ids) + if progress_callback: progress_callback("抓取完成", 95, 100) return {"fetched": count} diff --git a/packages/ai/cs_feed_orchestrator.py b/packages/ai/cs_feed_orchestrator.py index fcd6cd8..c8020f9 100644 --- a/packages/ai/cs_feed_orchestrator.py +++ b/packages/ai/cs_feed_orchestrator.py @@ -149,6 +149,12 @@ def run(self): if title: titles.append(title) + # 关联论文到 cs_feed 分类对应的 topic（每分类一个 disabled topic） + # 使 cs_feed 论文接入主题侧边栏/图谱/统计。enabled=False 防 + # topic_dispatch 重复抓取同一分类（仅 cs_feed_dispatch 负责） + if paper_ids: + self._link_cs_papers_to_topic(sub_session, category_code, paper_ids) + sub_repo.update_run_status(category_code, count) sub_session.commit() logger.info("[CSFeed] %s: ingested %d papers", category_code, count) @@ -158,6 +164,8 @@ def run(self): # 此前 cs_feed 论文处于 unread 无 embedding 无 topic，只能靠 # idle_processor 事后补——改为抓取即处理 self._process_cs_papers(paper_ids) + # 抓取后触发引用自动关联（与 ingest_arxiv 对齐，此前 cs_feed 缺这步） + self._trigger_auto_link(paper_ids) except Exception as e: sub_session.rollback() err_str = str(e) @@ -217,6 +225,43 @@ def _process_cs_papers(self, paper_ids: list[str]) -> None: finally: limiter.end_task() + @staticmethod + def _link_cs_papers_to_topic(session, category_code: str, paper_ids: list[str]) -> None: + """把 cs_feed 抓取的论文关联到对应分类的 topic（每分类一个 disabled topic）。 + + 命名约定：topic name = `csfeed:{category_code}`，query = `cat:{category_code}`， + enabled=False（防 topic_dispatch 重复抓取同一分类，仅 cs_feed_dispatch 负责）。 + upsert_topic 按 name 幂等，link_to_topic 受 uq_paper_topic 约束兜底幂等。 + 关联后 cs_feed 论文即接入主题侧边栏/图谱视图/统计/推荐 boost。 + """ + from packages.storage.repositories import PaperRepository, TopicRepository + + topic_repo = TopicRepository(session) + topic = topic_repo.upsert_topic( + name=f"csfeed:{category_code}", + query=f"cat:{category_code}", + enabled=False, + ) + paper_repo = PaperRepository(session) + for pid in paper_ids: + paper_repo.link_to_topic(pid, topic.id) + + @staticmethod + def _trigger_auto_link(paper_ids: list[str]) -> None: + """抓取后触发引用自动关联（复用 paper_pipelines 的有界线程池）。 + + 此前 cs_feed 只入库不 auto_link，论文间引用关系靠后续 sync_incremental 补。 + 与 ingest_arxiv 对齐，抓取即提交后台关联。 + """ + if not paper_ids: + return + try: + from packages.ai.pipelines.paper_pipelines import _auto_link_pool, _bg_auto_link + + _auto_link_pool.submit(_bg_auto_link, paper_ids) + except Exception as exc: + logger.warning("[CSFeed] 触发 auto_link 失败: %s", exc) + def _notify_digest(self, digest: list[tuple[str, int, list[str]]]) -> None: """抓取入库后发送邮件摘要；SMTP 未配置或无收件人时静默跳过""" from packages.config import get_settings diff --git a/tests/test_repositories.py b/tests/test_repositories.py index 1a04b86..ae00fd4 100644 --- a/tests/test_repositories.py +++ b/tests/test_repositories.py @@ -355,3 +355,72 @@ def test_skim_report_roundtrip(self, db_session): assert "一句话总结" in report.summary_md assert report.skim_score == 0.85 assert report.key_insights.get("skim_one_liner") == "一句话总结" + + +class TestCSFeedTopicLink: + def test_link_creates_disabled_topic_and_links_papers(self, db_session): + """cs_feed 论文关联到每分类的 disabled topic（接入主题侧边栏/图谱/统计）""" + from packages.ai.cs_feed_orchestrator import CSFeedOrchestrator + + repo = PaperRepository(db_session) + p1 = repo.upsert_paper( + PaperCreate(arxiv_id="2401.00201", title="t1", abstract="a", metadata={}) + ) + p2 = repo.upsert_paper( + PaperCreate(arxiv_id="2401.00202", title="t2", abstract="a", metadata={}) + ) + db_session.flush() + + CSFeedOrchestrator._link_cs_papers_to_topic(db_session, "cs.AI", [p1.id, p2.id]) + db_session.commit() + + topic = TopicRepository(db_session).get_by_name("csfeed:cs.AI") + assert topic is not None + # enabled=False 防 topic_dispatch 重复抓取同一分类 + assert topic.enabled is False + assert topic.query == "cat:cs.AI" + linked = PaperRepository(db_session).list_by_topic(topic.id) + assert len(linked) == 2 + + def test_link_idempotent_no_duplicate_rows(self, db_session): + """重复关联同一 (paper, topic) 不产生重复行（uq_paper_topic 兜底）""" + from packages.ai.cs_feed_orchestrator import CSFeedOrchestrator + + repo = PaperRepository(db_session) + p = repo.upsert_paper( + PaperCreate(arxiv_id="2401.00203", title="t", abstract="a", metadata={}) + ) + db_session.flush() + + CSFeedOrchestrator._link_cs_papers_to_topic(db_session, "cs.LG", [p.id]) + db_session.commit() + # 再关联一次 + CSFeedOrchestrator._link_cs_papers_to_topic(db_session, "cs.LG", [p.id]) + db_session.commit() + + topic = TopicRepository(db_session).get_by_name("csfeed:cs.LG") + linked = PaperRepository(db_session).list_by_topic(topic.id) + assert len(linked) == 1 # 幂等，不重复 + + def test_link_category_isolation(self, db_session): + """不同分类建独立 topic，论文不串""" + from packages.ai.cs_feed_orchestrator import CSFeedOrchestrator + + repo = PaperRepository(db_session) + p_ai = repo.upsert_paper( + PaperCreate(arxiv_id="2401.00204", title="ai", abstract="a", metadata={}) + ) + p_lg = repo.upsert_paper( + PaperCreate(arxiv_id="2401.00205", title="lg", abstract="a", metadata={}) + ) + db_session.flush() + + CSFeedOrchestrator._link_cs_papers_to_topic(db_session, "cs.AI", [p_ai.id]) + CSFeedOrchestrator._link_cs_papers_to_topic(db_session, "cs.LG", [p_lg.id]) + db_session.commit() + + t_ai = TopicRepository(db_session).get_by_name("csfeed:cs.AI") + t_lg = TopicRepository(db_session).get_by_name("csfeed:cs.LG") + assert len(PaperRepository(db_session).list_by_topic(t_ai.id)) == 1 + assert len(PaperRepository(db_session).list_by_topic(t_lg.id)) == 1 + assert t_ai.id != t_lg.id