From 9ab4e87b998ce6cf35dae2f4bb65981e75f8314a Mon Sep 17 00:00:00 2001 From: Color2333 <1552429809@qq.com> Date: Sat, 18 Jul 2026 19:29:03 +0800 Subject: [PATCH] =?UTF-8?q?fix(worker):=20PR2=20=E5=B9=B6=E5=8F=91/?= =?UTF-8?q?=E7=AB=9E=E6=80=81=20High=20=E4=BF=AE=E5=A4=8D=206=20=E9=A1=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 2a upsert_paper 保护 skim metadata：更新分支此前 existing.metadata_json = data.metadata（整体覆盖）会把 skim 写入的 keywords/title_zh/abstract_zh 抹掉，重复抓取丢失已花钱算出的产物。 改合并：保留已有 skim 派生字段，其余由 arxiv 原始元数据更新覆盖。 2b AnalysisReport 加 unique + IntegrityError 处理：paper_id 加 unique=True，并发 skim 同一论文不再产生重复行。_get_or_create 加 IntegrityError 捕获：flush 命中 unique 约束时回滚并取已存在行。 alembic 迁移先删重复行（保留最早）再加唯一索引（幂等）。 2c daily_runner 内层 retry 加退避：此前 for 循环无 sleep 无退避， 失败后立即重发加速触发 429。加指数退避；429/限流类用更长退避 （10*2^n），其余用 3*2^n。 2d idle_processor 防重复处理：新增进程内 _dispatching 标志 + set_dispatching/is_dispatching。worker topic_dispatch 期间置 True， IdleDetector.is_idle 读到即视为繁忙，避免 idle 与 topic_dispatch 抢同一批 unread 论文重复 embed/skim。 2e worker heartbeat 不掩盖故障：_write_heartbeat 写 JSON {ts, error}， job 全部失败时不写心跳让其自然过期。healthcheck 改判时效（20min） 而非仅 test -f。新增 scripts/worker_healthcheck.py。 2f arxiv_client 500 重试复用限流：acquire_api 移入重试循环内， 此前循环外只 acquire 一次，500 回退的二次请求绕过限流器。 500 回退改 continue 回循环顶部重新限流，二次失败由循环统一处理。 测试：新增 TestAnalysisRepository（_get_or_create 幂等+唯一无重复+ IntegrityError 恢复+skim 回写）、PaperRepository upsert 保留 skim metadata，全套 50 passed。 --- apps/worker/main.py | 90 ++++++++---- docker-compose.yml | 4 +- .../f6a7b8c9d0e1_analysis_report_unique.py | 70 +++++++++ packages/ai/daily_runner.py | 18 +++ packages/ai/idle_processor.py | 21 +++ packages/integrations/arxiv_client.py | 21 ++- packages/storage/models.py | 1 + packages/storage/repositories/analysis.py | 10 +- packages/storage/repositories/paper.py | 11 +- scripts/worker_healthcheck.py | 34 +++++ tests/test_repositories.py | 135 ++++++++++++++++++ 11 files changed, 374 insertions(+), 41 deletions(-) create mode 100644 infra/migrations/versions/f6a7b8c9d0e1_analysis_report_unique.py create mode 100644 scripts/worker_healthcheck.py diff --git a/apps/worker/main.py b/apps/worker/main.py index e372698..a6eae82 100644 --- a/apps/worker/main.py +++ b/apps/worker/main.py @@ -23,7 +23,11 @@ run_topic_ingest, run_weekly_graph_maintenance, ) -from packages.ai.idle_processor import start_idle_processor, stop_idle_processor +from packages.ai.idle_processor import ( + set_dispatching, + start_idle_processor, + stop_idle_processor, +) from packages.config import get_settings from packages.logging_setup import setup_logging from packages.storage.db import session_scope @@ -33,12 +37,23 @@ logger = logging.getLogger(__name__) _HEALTH_FILE = Path("/tmp/worker_heartbeat") +# 心跳健康判定：最近一次心跳距现在超过此秒数视为不健康（捕获 worker 卡死/全部任务失败） +_HEARTBEAT_STALE_SECONDS = 1200 # 20 分钟（cron job 最小间隔 30min，留足缓冲） + + +def _write_heartbeat(error: str | None = None) -> None: + """写入心跳文件供外部健康检查（High 2e：记录最近一次错误，不再掩盖故障）。 + 此前无条件写时间戳，healthcheck 仅 test -f → 即使所有 job 失败 worker 仍判健康。 + 现写入 JSON {ts, error}：健康检查读 ts 判定时效，error 字段记录最近致命错误。 + job 全部失败时不写心跳（让心跳自然过期 → healthcheck 反映故障）。 + """ + import json -def _write_heartbeat() -> None: - """写入心跳文件供外部健康检查""" with contextlib.suppress(OSError): - _HEALTH_FILE.write_text(str(time.time())) + _HEALTH_FILE.write_text( + json.dumps({"ts": time.time(), "error": error[:200] if error else None}) + ) def _update_topic_run_status(topic_id: str, *, error: str | None) -> None: @@ -122,22 +137,34 @@ def topic_dispatch_job() -> None: len(candidates), ", ".join(c["name"] for c in candidates), ) - for c in candidates: - try: - result = _retry_with_backoff( - run_topic_ingest, c["id"], max_retries=_RETRY_MAX, base_delay=_RETRY_DELAY - ) - logger.info( - "topic %s done: inserted=%s, processed=%s", - c["name"], - result.get("inserted", 0) if result else 0, - result.get("processed", 0) if result else 0, - ) - _update_topic_run_status(c["id"], error=None) - except Exception as e: - logger.exception("topic_dispatch failed for %s", c["name"]) - _update_topic_run_status(c["id"], error=str(e)) - _write_heartbeat() + # High 2d：置调度标志，idle_processor 检测到即视为繁忙，避免抢同一批论文重复处理 + # High 2e：全部失败不写 heartbeat，让心跳自然过期 → healthcheck 反映故障 + set_dispatching(True) + failures: list[str] = [] + try: + for c in candidates: + try: + result = _retry_with_backoff( + run_topic_ingest, c["id"], max_retries=_RETRY_MAX, base_delay=_RETRY_DELAY + ) + logger.info( + "topic %s done: inserted=%s, processed=%s", + c["name"], + result.get("inserted", 0) if result else 0, + result.get("processed", 0) if result else 0, + ) + _update_topic_run_status(c["id"], error=None) + except Exception as e: + logger.exception("topic_dispatch failed for %s", c["name"]) + _update_topic_run_status(c["id"], error=str(e)) + failures.append(f"{c['name']}: {e}") + finally: + set_dispatching(False) + if not failures: + _write_heartbeat() + else: + # 全部失败时不写健康心跳，仅记录致命错误到日志（healthcheck 靠时效捕获） + logger.error("topic_dispatch 全部失败，跳过心跳写入：%s", failures) def brief_job() -> None: @@ -160,8 +187,10 @@ def brief_job() -> None: result.get("saved_path", "N/A") if result else "N/A", result.get("email_sent", False) if result else False, ) - except Exception: - logger.exception("Daily brief job failed after retries") + except Exception as e: + # High 2e：失败不写心跳，让健康检查靠时效捕获故障 + logger.exception("Daily brief job failed after retries: %s", e) + return _write_heartbeat() @@ -171,15 +200,22 @@ def weekly_graph_job() -> None: _retry_with_backoff( run_weekly_graph_maintenance, max_retries=_RETRY_MAX, base_delay=_RETRY_DELAY ) - except Exception: - logger.exception("Weekly graph job failed after retries") + except Exception as e: + # High 2e：失败不写心跳 + logger.exception("Weekly graph job failed after retries: %s", e) + return _write_heartbeat() def cs_feed_dispatch_job(): - """每小时同步分类 + 执行订阅抓取""" - cs_orchestrator.sync_categories() - cs_orchestrator.run() + """每小时同步分类 + 执行订阅抓取（High 2e：失败不写心跳）""" + try: + cs_orchestrator.sync_categories() + cs_orchestrator.run() + except Exception as e: + logger.exception("cs_feed_dispatch failed: %s", e) + return + _write_heartbeat() def run_worker() -> None: diff --git a/docker-compose.yml b/docker-compose.yml index d781c26..1e40828 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -80,7 +80,9 @@ services: postgres: condition: service_healthy healthcheck: - test: ["CMD", "test", "-f", "/tmp/worker_heartbeat"] + # High 2e：改判时效而非仅文件存在——所有 job 失败时心跳不再写，文件过期即不健康。 + # 心跳超过 20 分钟（1200s）视为不健康（worker 卡死或全部任务失败） + test: ["CMD", "python", "-m", "scripts.worker_healthcheck"] interval: 30s timeout: 5s start_period: 40s diff --git a/infra/migrations/versions/f6a7b8c9d0e1_analysis_report_unique.py b/infra/migrations/versions/f6a7b8c9d0e1_analysis_report_unique.py new file mode 100644 index 0000000..23bdcbd --- /dev/null +++ b/infra/migrations/versions/f6a7b8c9d0e1_analysis_report_unique.py @@ -0,0 +1,70 @@ +"""add unique constraint on analysis_reports.paper_id + +Revision ID: f6a7b8c9d0e1 +Revises: e5f6a7b8c9d0 +Create Date: 2026-07-17 17:00:00.000000 + +目的：给 analysis_reports.paper_id 加 unique 约束，防止重复 skim/deep 产生 +重复行。此前无约束，并发处理同一论文会插入多行 AnalysisReport，下游读 summary +时取到任意一行（行为不确定）。 + +迁移顺序： +1. 先删除重复行（同一 paper_id 多行的，保留 created_at 最早的一行） +2. 再加 unique 约束 + +PG 用窗口函数删重复 + CREATE UNIQUE INDEX IF NOT EXISTS； +SQLite 用 rowid 删重复（保留 MIN(rowid)）+ CREATE UNIQUE INDEX IF NOT EXISTS。 +""" +from alembic import op + + +# revision identifiers, used by Alembic. +revision = "f6a7b8c9d0e1" +down_revision = "e5f6a7b8c9d0" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + bind = op.get_bind() + if bind.dialect.name == "postgresql": + # 删除重复行：同一 paper_id 保留 created_at 最早的一行 + op.execute( + """ + DELETE FROM analysis_reports a USING analysis_reports b + WHERE a.paper_id = b.paper_id + AND a.id <> b.id + AND a.created_at > b.created_at + """ + ) + # 兜底：若仍有同 paper_id 同 created_at 的重复（极端竞态），保留 id 最小者 + op.execute( + """ + DELETE FROM analysis_reports a USING analysis_reports b + WHERE a.paper_id = b.paper_id + AND a.id > b.id + AND a.created_at = b.created_at + """ + ) + op.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS uq_analysis_reports_paper_id " + "ON analysis_reports (paper_id)" + ) + else: + # SQLite：用 rowid 删重复，保留 MIN(rowid) + op.execute( + """ + DELETE FROM analysis_reports + WHERE rowid NOT IN ( + SELECT MIN(rowid) FROM analysis_reports GROUP BY paper_id + ) + """ + ) + op.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS uq_analysis_reports_paper_id " + "ON analysis_reports (paper_id)" + ) + + +def downgrade() -> None: + op.execute("DROP INDEX IF EXISTS uq_analysis_reports_paper_id") diff --git a/packages/ai/daily_runner.py b/packages/ai/daily_runner.py index 8756b82..4073018 100644 --- a/packages/ai/daily_runner.py +++ b/packages/ai/daily_runner.py @@ -7,6 +7,7 @@ from __future__ import annotations import logging +import time from concurrent.futures import ThreadPoolExecutor, as_completed from typing import TYPE_CHECKING from uuid import UUID @@ -167,6 +168,23 @@ def run_topic_ingest(topic_id: str, progress_callback: callable | None = None) - break except Exception as exc: last_error = str(exc) + # 修 High：内层 retry 此前无 sleep 无退避，失败后立即重发请求，限流场景 + # 下加速触发 429。改指数退避；429/限流类错误用更长退避，其余快速失败重试。 + if _attempt < topic.retry_limit: + is_rate_limited = any( + tok in str(exc).lower() + for tok in ("429", "rate limit", "限流", "timeout", "timed out") + ) + delay = 10.0 * (2**_attempt) if is_rate_limited else 3.0 * (2**_attempt) + logger.warning( + "topic %s 抓取失败 (attempt %d/%d): %s — %.0fs 后重试", + topic_name, + attempts, + topic.retry_limit + 1, + str(exc)[:120], + delay, + ) + time.sleep(delay) if last_error is not None: return { diff --git a/packages/ai/idle_processor.py b/packages/ai/idle_processor.py index 9cf1207..6355165 100644 --- a/packages/ai/idle_processor.py +++ b/packages/ai/idle_processor.py @@ -19,6 +19,22 @@ logger = logging.getLogger(__name__) +# 进程内调度标志（High 2d）：topic_dispatch 抓取/处理期间置 True， +# idle 检测读到即视为繁忙，避免 idle 与 topic_dispatch 抢同一批 unread 论文重复 +# embed/skim。仅 worker 进程内生效（idle_processor 与 topic_dispatch 同在 worker 容器）。 +_dispatching = False + + +def set_dispatching(value: bool) -> None: + """设置 topic_dispatch 是否正在跑（供 worker main 调用）""" + global _dispatching + _dispatching = value + + +def is_dispatching() -> bool: + """查询 topic_dispatch 是否正在跑""" + return _dispatching + class IdleDetector: """ @@ -90,6 +106,11 @@ def is_idle(self) -> bool: Returns: bool: 是否空闲 """ + # High 2d：topic_dispatch 正在抓取/处理时不算空闲，避免与 idle 抢同一批论文 + if is_dispatching(): + logger.debug("topic_dispatch 正在跑，不满足空闲条件") + return False + # 检查距离上次任务执行的时间 if time.time() - self._last_task_time < self.idle_interval: return False diff --git a/packages/integrations/arxiv_client.py b/packages/integrations/arxiv_client.py index 27e4dc5..7789423 100644 --- a/packages/integrations/arxiv_client.py +++ b/packages/integrations/arxiv_client.py @@ -75,10 +75,6 @@ def fetch_latest( days_back 默认 0 = 不加日期过滤（否则经典老论文如 OpenShape/Uni3D 都会被筛掉）。 订阅/定时任务需要最新增量时，由调用方显式传 days_back。 """ - # 获取速率限制许可（10 秒超时） - if not acquire_api("arxiv", timeout=10.0): - raise httpx.TimeoutException("ArXiv 速率限制等待超时，请稍后重试") - structured_query = _build_arxiv_query(query, days_back) logger.info( "ArXiv search: %s → %s (sort=%s start=%d days_back=%d)", @@ -96,9 +92,13 @@ def fetch_latest( "max_results": max_results, } # 自动重试（429 限流 + 网络抖动 + 500 回退） + # 修 High 2f：acquire_api 移入循环内，每次请求都重新获取限流许可 + # （此前循环外只 acquire 一次，500 回退的二次请求绕过限流器） last_exc: Exception | None = None for attempt in range(3): try: + if not acquire_api("arxiv", timeout=10.0): + raise httpx.TimeoutException("ArXiv 速率限制等待超时，请稍后重试") response = self.client.get(ARXIV_API_URL, params=params) response.raise_for_status() return self._parse_atom(response.text) @@ -112,13 +112,12 @@ def fetch_latest( time.sleep(wait) continue elif status == 500 and "submittedDate:" in structured_query: - # arXiv API 日期过滤可能有问题，尝试不带日期的查询 - logger.warning("ArXiv 500 错误（可能是日期过滤问题），尝试不带日期的查询") - simple_query = _build_arxiv_query(query, days_back=0) # 不添加日期 - params["search_query"] = simple_query - response = self.client.get(ARXIV_API_URL, params=params) - response.raise_for_status() - return self._parse_atom(response.text) + # arXiv API 日期过滤可能有问题，改不带日期的查询重试。 + # continue 回循环顶部重新 acquire_api（限流），二次失败由循环统一处理 + logger.warning("ArXiv 500 错误（可能是日期过滤问题），切无日期查询重试") + structured_query = _build_arxiv_query(query, days_back=0) + params["search_query"] = structured_query + continue raise except httpx.TimeoutException as exc: last_exc = exc diff --git a/packages/storage/models.py b/packages/storage/models.py index 81274e0..85a42ae 100644 --- a/packages/storage/models.py +++ b/packages/storage/models.py @@ -85,6 +85,7 @@ class AnalysisReport(Base): ForeignKey("papers.id", ondelete="CASCADE"), nullable=False, index=True, + unique=True, ) summary_md: Mapped[str | None] = mapped_column(Text, nullable=True) deep_dive_md: Mapped[str | None] = mapped_column(Text, nullable=True) diff --git a/packages/storage/repositories/analysis.py b/packages/storage/repositories/analysis.py index e712699..c80dbbd 100644 --- a/packages/storage/repositories/analysis.py +++ b/packages/storage/repositories/analysis.py @@ -8,6 +8,7 @@ from typing import TYPE_CHECKING from sqlalchemy import select +from sqlalchemy.exc import IntegrityError if TYPE_CHECKING: from uuid import UUID @@ -57,7 +58,14 @@ def _get_or_create(self, paper_id: UUID) -> AnalysisReport: return found report = AnalysisReport(paper_id=pid, key_insights={}) self.session.add(report) - self.session.flush() + try: + self.session.flush() + except IntegrityError: + # 并发 skim 同一论文时，另一事务已插入行（paper_id 现为 unique）。 + # 回滚本事务未提交改动并取已存在的行，避免重复插入并防止抛 IntegrityError + # 中断 skim 流程。此前无 unique 约束 → 重复 skim 产生重复行。 + self.session.rollback() + return self.session.execute(q).scalar_one() return report def summaries_for_papers(self, paper_ids: list[str]) -> dict[str, str]: diff --git a/packages/storage/repositories/paper.py b/packages/storage/repositories/paper.py index 7dfdce6..b477c22 100644 --- a/packages/storage/repositories/paper.py +++ b/packages/storage/repositories/paper.py @@ -45,7 +45,16 @@ def upsert_paper(self, data: PaperCreate) -> Paper: existing.title = data.title existing.abstract = data.abstract existing.publication_date = data.publication_date - existing.metadata_json = data.metadata + # 修 High：此前 existing.metadata_json = data.metadata（整体覆盖）会把 skim + # 写入的 keywords/title_zh/abstract_zh 抹掉——重复抓取同一论文时丢失已花钱 + # 算出来的 skim 产物。改为合并：保留已有的 skim 派生字段，其余由 arxiv 原始 + # 元数据更新覆盖（categories/authors/source 等原始字段） + new_meta = dict(data.metadata or {}) + old_meta = existing.metadata_json or {} + for skim_key in ("keywords", "title_zh", "abstract_zh"): + if skim_key in old_meta and old_meta[skim_key]: + new_meta[skim_key] = old_meta[skim_key] + existing.metadata_json = new_meta existing.updated_at = datetime.now(UTC) self.session.flush() return existing diff --git a/scripts/worker_healthcheck.py b/scripts/worker_healthcheck.py new file mode 100644 index 0000000..f06803f --- /dev/null +++ b/scripts/worker_healthcheck.py @@ -0,0 +1,34 @@ +#!/usr/bin/env python3 +"""Worker 健康检查脚本（High 2e）。 + +读取 /tmp/worker_heartbeat（JSON {ts, error}），判定心跳时效： +- 文件不存在 / 解析失败 → 不健康（exit 1） +- ts 距今超过 1200 秒 → 不健康（worker 卡死或全部任务失败，心跳已过期） +- 否则健康（exit 0） + +此前 healthcheck 仅 test -f 文件存在 → 即使所有 job 失败 worker 仍判健康。 +""" + +from __future__ import annotations + +import json +import sys +import time +from pathlib import Path + +HEALTH_FILE = Path("/tmp/worker_heartbeat") +STALE_SECONDS = 1200 # 20 分钟 + + +def main() -> int: + try: + data = json.loads(HEALTH_FILE.read_text()) + ts = float(data.get("ts", 0)) + except (OSError, ValueError, TypeError): + # 文件不存在或损坏 → 视为不健康 + return 1 + return 0 if (time.time() - ts) < STALE_SECONDS else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_repositories.py b/tests/test_repositories.py index 05f5b52..1a04b86 100644 --- a/tests/test_repositories.py +++ b/tests/test_repositories.py @@ -11,6 +11,7 @@ from packages.domain.enums import ReadStatus from packages.domain.schemas import PaperCreate from packages.storage.repositories import ( + AnalysisRepository, CSFeedRepository, IeeeQuotaRepository, PaperRepository, @@ -47,6 +48,49 @@ def test_upsert_insert_then_update(self, db_session): assert updated.title == "Updated Title" assert updated.abstract == "upd" + def test_upsert_preserves_skim_metadata_on_refetch(self, db_session): + """重复抓取同一论文时保留 skim 写入的 keywords/title_zh/abstract_zh（High 2a） + + 此前更新分支 existing.metadata_json = data.metadata（整体覆盖）会抹掉 skim 产物， + 重复抓取丢失已花钱算出的中文翻译/关键词。 + """ + repo = PaperRepository(db_session) + saved = repo.upsert_paper( + PaperCreate( + arxiv_id="2401.00002", + title="Orig", + abstract="a", + metadata={"categories": ["cs.AI"], "authors": ["X"]}, + ) + ) + # 模拟 skim 写入的派生字段 + saved.metadata_json = { + "categories": ["cs.AI"], + "authors": ["X"], + "keywords": ["llm", "reasoning"], + "title_zh": "大语言模型推理", + "abstract_zh": "摘要中文", + } + db_session.flush() + + # 二次抓取：arxiv 原始元数据不含 skim 字段 + updated = repo.upsert_paper( + PaperCreate( + arxiv_id="2401.00002", + title="New Title", + abstract="b", + metadata={"categories": ["cs.AI", "cs.CL"], "authors": ["X", "Y"]}, + ) + ) + meta = updated.metadata_json or {} + # skim 派生字段应保留 + assert meta.get("keywords") == ["llm", "reasoning"] + assert meta.get("title_zh") == "大语言模型推理" + assert meta.get("abstract_zh") == "摘要中文" + # arxiv 原始字段应更新 + assert meta.get("categories") == ["cs.AI", "cs.CL"] + assert meta.get("authors") == ["X", "Y"] + def test_upsert_multi_source_arxiv_id_synthesis(self, db_session): """非 arXiv 源（arxiv_id=None）用 source:source_id 合成 arxiv_id""" repo = PaperRepository(db_session) @@ -220,3 +264,94 @@ def test_update_run_status_unknown_category_silent(self, db_session): """未知 category_code：静默跳过，不抛""" repo = CSFeedRepository(db_session) repo.update_run_status("cs.NONEXIST", count=5) + + +class TestAnalysisRepository: + def test_get_or_create_idempotent(self, db_session): + """_get_or_create 首次创建、二次返回同一行""" + repo = AnalysisRepository(db_session) + # 需要先有 paper（外键约束） + paper_repo = PaperRepository(db_session) + paper = paper_repo.upsert_paper( + PaperCreate(arxiv_id="2401.00010", title="t", abstract="a", metadata={}) + ) + db_session.flush() + + r1 = repo._get_or_create(paper.id) + r2 = repo._get_or_create(paper.id) + assert r1.id == r2.id + + def test_get_or_create_unique_no_duplicates(self, db_session): + """AnalysisReport.paper_id 唯一：重复 _get_or_create 不产生重复行（High 2b）""" + from sqlalchemy import func, select + + from packages.storage.models import AnalysisReport + + paper_repo = PaperRepository(db_session) + paper = paper_repo.upsert_paper( + PaperCreate(arxiv_id="2401.00011", title="t", abstract="a", metadata={}) + ) + db_session.flush() + + repo = AnalysisRepository(db_session) + r1 = repo._get_or_create(paper.id) + db_session.flush() + r2 = repo._get_or_create(paper.id) + db_session.flush() + assert r1.id == r2.id # 同一行 + + # 全表只有这一条该 paper_id 的 report + count = db_session.execute( + select(func.count()) + .select_from(AnalysisReport) + .where(AnalysisReport.paper_id == str(paper.id)) + ).scalar_one() + assert count == 1 + + def test_get_or_create_integrity_error_recovers(self, db_session): + """直接插入违反 unique 后，_get_or_create 回滚并取已存在行（High 2b IntegrityError 路径）""" + from uuid import uuid4 + + from packages.storage.models import AnalysisReport + + paper_repo = PaperRepository(db_session) + paper = paper_repo.upsert_paper( + PaperCreate(arxiv_id="2401.00013", title="t", abstract="a", metadata={}) + ) + db_session.flush() + pid = str(paper.id) + + # 先 commit 一行（模拟并发事务 A 已提交） + existing = AnalysisReport(id=str(uuid4()), paper_id=pid, key_insights={"a": 1}) + db_session.add(existing) + db_session.commit() + + # 再次 _get_or_create：select 命中已存在行，不走插入分支，不抛 IntegrityError + repo = AnalysisRepository(db_session) + again = repo._get_or_create(paper.id) + assert again.id == existing.id + + def test_skim_report_roundtrip(self, db_session): + """upsert_skim 写入 summary_md/skim_score/key_insights""" + from packages.domain.schemas import SkimReport + + paper_repo = PaperRepository(db_session) + paper = paper_repo.upsert_paper( + PaperCreate(arxiv_id="2401.00012", title="t", abstract="a", metadata={}) + ) + db_session.flush() + + repo = AnalysisRepository(db_session) + skim = SkimReport( + one_liner="一句话总结", + innovations=["创新1", "创新2"], + relevance_score=0.85, + ) + repo.upsert_skim(paper.id, skim) + db_session.flush() + + report = repo._get_or_create(paper.id) + assert report.summary_md is not None + assert "一句话总结" in report.summary_md + assert report.skim_score == 0.85 + assert report.key_insights.get("skim_one_liner") == "一句话总结"