From d43d30ee3f9b59c424f34d024c79e9283f57968f Mon Sep 17 00:00:00 2001 From: potaser Date: Fri, 9 Oct 2026 16:54:03 +0800 Subject: [PATCH] feat: add review_service with plant genomics presubmit, journals endpoint, and BFF proxy --- .gitignore | 8 + review_service/.env.example | 27 ++ review_service/__init__.py | 7 + review_service/app/__init__.py | 1 + review_service/app/config.py | 78 ++++++ review_service/app/db.py | 191 +++++++++++++ review_service/app/main.py | 71 +++++ review_service/app/routers/__init__.py | 1 + review_service/app/routers/journals.py | 292 ++++++++++++++++++++ review_service/app/routers/presubmit.py | 87 ++++++ review_service/app/services/__init__.py | 1 + review_service/app/services/journal_kb.py | 223 +++++++++++++++ review_service/app/services/journal_rank.py | 210 ++++++++++++++ review_service/app/services/llm.py | 203 ++++++++++++++ review_service/app/services/paper_parse.py | 139 ++++++++++ review_service/app/services/self_review.py | 155 +++++++++++ review_service/data/journals/journals.json | 62 +++++ review_service/pytest.ini | 6 + review_service/restart.sh | 33 +++ review_service/tests/test_presubmit.py | 163 +++++++++++ review_service/tests/test_recommend.py | 271 ++++++++++++++++++ sdk/app/config.py | 5 + sdk/app/main.py | 3 + sdk/app/routers/review_proxy.py | 85 ++++++ 24 files changed, 2322 insertions(+) create mode 100644 review_service/.env.example create mode 100644 review_service/__init__.py create mode 100644 review_service/app/__init__.py create mode 100644 review_service/app/config.py create mode 100644 review_service/app/db.py create mode 100644 review_service/app/main.py create mode 100644 review_service/app/routers/__init__.py create mode 100644 review_service/app/routers/journals.py create mode 100644 review_service/app/routers/presubmit.py create mode 100644 review_service/app/services/__init__.py create mode 100644 review_service/app/services/journal_kb.py create mode 100644 review_service/app/services/journal_rank.py create mode 100644 review_service/app/services/llm.py create mode 100644 review_service/app/services/paper_parse.py create mode 100644 review_service/app/services/self_review.py create mode 100644 review_service/data/journals/journals.json create mode 100644 review_service/pytest.ini create mode 100755 review_service/restart.sh create mode 100644 review_service/tests/test_presubmit.py create mode 100644 review_service/tests/test_recommend.py create mode 100644 sdk/app/routers/review_proxy.py diff --git a/.gitignore b/.gitignore index 3fc8ecb..b36139e 100644 --- a/.gitignore +++ b/.gitignore @@ -42,3 +42,11 @@ paper_writer_service/paper.db paper_writer_service/paper.db-shm paper_writer_service/paper.db-wal paper_writer_service/paper.log + +# review_service 运行时产物（不入库） +review_service/review.db +review_service/review.db-wal +review_service/review.db-shm +review_service/review.log +# 期刊种子 journals.json 是源码：顶层 **/data/ 会误伤 review_service/data/，显式放行 +!review_service/data/ diff --git a/review_service/.env.example b/review_service/.env.example new file mode 100644 index 0000000..197247e --- /dev/null +++ b/review_service/.env.example @@ -0,0 +1,27 @@ +# review_service/.env.example —— 运行时配置模板 +# 用法: cp .env.example .env 后填写（.env 已被仓库 .gitignore 忽略，勿提交） +# restart.sh 会 `set -a; source .env` 导出给 uvicorn 进程。 + +# ---- LLM（DeepSeek；REVIEW_LLM_* 优先 → DOCGEN_LLM_* 回退 → 内置默认，见 services/llm.py）---- +# 最简做法：从部署树抄一行即可复用同一把 key： +# grep '^DOCGEN_LLM_API_KEY=' /home/xzq/5090/teamWorkspace/zh/PlantAI-rag-new/sdk/.env >> .env +REVIEW_LLM_BASE_URL=https://api.deepseek.com/v1 +REVIEW_LLM_API_KEY= +REVIEW_LLM_MODEL=deepseek-chat +#REVIEW_LLM_TIMEOUT=300 + +# ---- 上游（只读消费）---- +#REVIEW_VSTORE_BASE_URL=http://127.0.0.1:8094 +#REVIEW_METHODOLOGY_BASE_URL=http://127.0.0.1:8092 # 预留（MinerU 重链路，当前未用） + +# ---- PDF 解析（经 vstore）---- +#REVIEW_PARSE_TIMEOUT=180 # ingest→取文→清理 总读超时（秒） +#REVIEW_PARSE_DELETE_AFTER=1 # 1=解析后删除 vstore 临时文档（0=保留便于排查） + +# ---- 自审稿 ---- +#REVIEW_PRESUBMIT_READ_TIMEOUT=360 # LLM 长调用读超时（秒） +#REVIEW_MAX_TEXT_CHARS=60000 # 送 LLM 的正文截断上限 +#REVIEW_MAX_PDF_BYTES=52428800 # 上传 PDF 上限 50MB + +# ---- 服务（restart.sh 读取）---- +#REVIEW_BIND_HOST=127.0.0.1 diff --git a/review_service/__init__.py b/review_service/__init__.py new file mode 100644 index 0000000..4a7865b --- /dev/null +++ b/review_service/__init__.py @@ -0,0 +1,7 @@ +"""review_service —— 科研投稿与智能审稿助手（8100）。 + +独立部署树（模式同 paper_writer_service 8097 / bioflow_service 8098）： +- BFF(sdk 8090) 经 routers/review_proxy.py 条件反代 /api/review/*（REVIEW_ENABLED）。 +- 上游只读消费：vstore 8094（PDF 解析）、DeepSeek（LLM，REVIEW_LLM_*/DOCGEN_LLM_*）。 +- 约束见根目录 AGENTS.md：审稿相关新代码只允许落在本目录。 +""" diff --git a/review_service/app/__init__.py b/review_service/app/__init__.py new file mode 100644 index 0000000..e701edd --- /dev/null +++ b/review_service/app/__init__.py @@ -0,0 +1 @@ +"""review 服务（8100）应用包。""" diff --git a/review_service/app/config.py b/review_service/app/config.py new file mode 100644 index 0000000..cf9533a --- /dev/null +++ b/review_service/app/config.py @@ -0,0 +1,78 @@ +"""config.py —— review 服务（8100）环境变量驱动配置（pydantic-settings）。 + +全部键可用环境变量覆盖（前缀 REVIEW_）。LLM 沿用 docgen 的裸键模式： +优先 REVIEW_LLM_*，回退 DOCGEN_LLM_*（与 paper_writer_service 共享同一套 +DeepSeek 凭据，由 services/llm.py 直接读 os.environ）。零硬编码宿主路径。 +""" +from __future__ import annotations + +from functools import lru_cache +from pathlib import Path + +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict( + env_file=".env", + env_file_encoding="utf-8", + extra="ignore", + env_prefix="REVIEW_", + ) + + app_name: str = "plantai-review-service" + version: str = "0.1.0" + + # ---- 上游（只读消费；不修改 vstore/methodology 任何代码）---- + # vstore 8094：POST /api/vstore/ingest + GET/DELETE /api/vstore/documents/{id} + # （pymupdf 段落解析，同步秒级；methodology 8092 的 MinerU 链路是 + # 分钟级异步 job，不适配同步自审稿，仅预留地址供后续扩展） + vstore_base_url: str = "http://127.0.0.1:8094" + methodology_base_url: str = "http://127.0.0.1:8092" + + # ---- PDF 解析 ---- + parse_timeout: float = 180.0 # vstore ingest→fetch→delete 读超时（秒） + parse_delete_after: bool = True # 解析完成后删除 vstore 临时文档（不污染私有库） + + # ---- 自审稿（presubmit）---- + presubmit_read_timeout: float = 360.0 # LLM 长调用读超时（秒） + max_text_chars: int = 60000 # 送 LLM 的正文截断上限（约 3 万汉字/6 万字符） + max_pdf_bytes: int = 50 * 1024 * 1024 # 上传 PDF 上限 50MB（对齐 extraction FR-EXTRACT-004） + + # ---- 期刊情报与投稿推荐（recommend）---- + db_path: str = "review.db" # 独立 SQLite（相对 review_service/；REVIEW_DB_PATH 覆盖） + journals_seed_path: str = "data/journals/journals.json" # 种子（相对 review_service/） + vstore_match_timeout: float = 30.0 # scope 入库/向量检索读超时（秒） + recommend_read_timeout: float = 60.0 # LLM 重排读超时（秒） + recommend_query_chars: int = 3000 # 送 vstore 的检索 query 截断（防嵌入模型超长） + recommend_scope_top_k: int = 10 # search-chunks top_k（vstore 上限 20） + + @property + def _service_root(self) -> "Path": + # review_service/app/config.py → review_service/（相对本文件定位，不依赖 cwd） + return Path(__file__).resolve().parent.parent + + @property + def db_path_abs(self) -> "Path": + # 绝对 db_path 会被 Path 除法原样返回；相对路径落在 review_service/ 下 + return self._service_root / self.db_path + + @property + def journals_seed_abs(self) -> "Path": + return self._service_root / self.journals_seed_path + + @property + def vstore_base(self) -> str: + return self.vstore_base_url.rstrip("/") + + @property + def methodology_base(self) -> str: + return self.methodology_base_url.rstrip("/") + + +@lru_cache(maxsize=1) +def get_settings() -> Settings: + return Settings() + + +settings = get_settings() diff --git a/review_service/app/db.py b/review_service/app/db.py new file mode 100644 index 0000000..1c312a8 --- /dev/null +++ b/review_service/app/db.py @@ -0,0 +1,191 @@ +"""app/db.py —— review 服务独立 SQLite（aiosqlite + WAL，无 ORM）：journals 表。 + +- 单连接 + asyncio 写锁（模式对齐 paper_writer_service/bioflow，但表结构与种子 + 逻辑为本模块自研）；路径默认 review_service/review.db（REVIEW_DB_PATH 覆盖）。 +- 种子：data/journals/journals.json → journals 表（幂等 upsert，vstore_doc_id 列 + 永不被种子覆盖——该列记录 scope_desc 在 vstore 8094 的 document_id）。 +- 只在 review_service/ 内使用，不触碰 BFF/生产服务数据库。 +""" +from __future__ import annotations + +import asyncio +import json +from pathlib import Path +from typing import Any, Iterable + +import aiosqlite + +from .config import get_settings + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS journals ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + scope_desc TEXT NOT NULL, + impact_factor REAL, + review_time_weeks REAL, + acceptance_rate REAL, + submission_deadline TEXT, + official_url TEXT, + last_updated TEXT NOT NULL, + vstore_doc_id TEXT +); +""" + +# 种子 upsert 列（刻意不含 vstore_doc_id：种子刷新不得清掉向量库映射） +_SEED_COLS = ( + "id", + "name", + "scope_desc", + "impact_factor", + "review_time_weeks", + "acceptance_rate", + "submission_deadline", + "official_url", + "last_updated", +) +_UPSERT_SQL = ( + "INSERT INTO journals (" + ", ".join(_SEED_COLS) + ") VALUES (" + + ", ".join("?" for _ in _SEED_COLS) + ") " + + "ON CONFLICT(id) DO UPDATE SET " + + ", ".join(f"{c}=excluded.{c}" for c in _SEED_COLS if c != "id") +) + + +class SeedError(Exception): + """种子文件缺失/损坏（journals.json 不可读）。""" + + +class Database: + """aiosqlite 单连接封装（WAL；写串行化）。""" + + def __init__(self, path: str | Path | None = None) -> None: + self._path: Path | None = Path(path) if path is not None else None + self._conn: aiosqlite.Connection | None = None + self._lock = asyncio.Lock() + + @property + def path(self) -> Path: + if self._path is None: + self._path = get_settings().db_path_abs + return self._path + + def set_path(self, path: str | Path) -> None: + """测试注入：在进入 TestClient lifespan 之前预置库路径（须尚未 connect）。""" + if self._conn is not None: + raise RuntimeError("Database 已连接，无法再改路径（先 close_db()）") + self._path = Path(path) + + @property + def connected(self) -> bool: + return self._conn is not None + + async def connect(self) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + self._conn = await aiosqlite.connect(self.path) + self._conn.row_factory = aiosqlite.Row + await self._conn.execute("PRAGMA journal_mode=WAL") + await self._conn.execute("PRAGMA busy_timeout=5000") + await self._conn.executescript(_SCHEMA) + await self._conn.commit() + + async def close(self) -> None: + if self._conn is not None: + await self._conn.close() + self._conn = None + + @property + def conn(self) -> aiosqlite.Connection: + if self._conn is None: + raise RuntimeError("Database 未初始化：先 await init_db()") + return self._conn + + async def fetchall(self, sql: str, params: Iterable[Any] = ()) -> list[dict[str, Any]]: + async with self.conn.execute(sql, tuple(params)) as cur: + rows = await cur.fetchall() + return [dict(r) for r in rows] + + async def fetchone(self, sql: str, params: Iterable[Any] = ()) -> dict[str, Any] | None: + async with self.conn.execute(sql, tuple(params)) as cur: + row = await cur.fetchone() + return dict(row) if row is not None else None + + async def execute(self, sql: str, params: Iterable[Any] = ()) -> int: + async with self._lock: + cur = await self.conn.execute(sql, tuple(params)) + await self._conn.commit() # type: ignore[union-attr] + return cur.lastrowid or 0 + + async def executemany(self, sql: str, seq: list[tuple[Any, ...]]) -> None: + async with self._lock: + await self.conn.executemany(sql, seq) + await self._conn.commit() # type: ignore[union-attr] + + +db = Database() + + +async def init_db(path: str | Path | None = None) -> None: + """幂等初始化：已连接则跳过（测试可在 lifespan 前 set_path 注入路径）。""" + if path is not None: + db.set_path(path) + if db.connected: + return + await db.connect() + + +async def close_db() -> None: + await db.close() + + +def set_path(path: str | Path) -> None: + """同步注入库路径（测试 fixture 用；须在 init_db 之前调用）。""" + db.set_path(path) + + +async def ensure_journals_seeded() -> int: + """装载种子期刊（幂等）。返回 journals 表行数。""" + row = await db.fetchone("SELECT COUNT(*) AS c FROM journals") + if row and int(row["c"]) > 0: + return int(row["c"]) + + seed_file = get_settings().journals_seed_abs + try: + payload = json.loads(Path(seed_file).read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise SeedError(f"期刊种子文件不存在：{seed_file}") from exc + except (OSError, ValueError) as exc: + raise SeedError(f"期刊种子文件不可读：{exc}") from exc + + journals = payload.get("journals") if isinstance(payload, dict) else None + if not isinstance(journals, list) or not journals: + raise SeedError(f"期刊种子文件缺少 journals 数组：{seed_file}") + + rows: list[tuple[Any, ...]] = [] + for item in journals: + if not isinstance(item, dict) or not item.get("id") or not item.get("name"): + continue + rows.append(tuple(item.get(c) for c in _SEED_COLS)) + if not rows: + raise SeedError(f"期刊种子文件无有效条目：{seed_file}") + + await db.executemany(_UPSERT_SQL, rows) + return len(rows) + + +async def list_journals() -> list[dict[str, Any]]: + """全部期刊（impact_factor 降序，NULL 靠后）。""" + return await db.fetchall( + "SELECT * FROM journals ORDER BY impact_factor DESC, name ASC" + ) + + +async def get_journal(journal_id: str) -> dict[str, Any] | None: + return await db.fetchone("SELECT * FROM journals WHERE id = ?", (journal_id,)) + + +async def set_vstore_doc_id(journal_id: str, doc_id: str) -> None: + """记录 scope_desc 在 vstore 的 document_id（向量白名单检索用）。""" + await db.execute( + "UPDATE journals SET vstore_doc_id = ? WHERE id = ?", (doc_id, journal_id) + ) diff --git a/review_service/app/main.py b/review_service/app/main.py new file mode 100644 index 0000000..1611e90 --- /dev/null +++ b/review_service/app/main.py @@ -0,0 +1,71 @@ +"""main.py —— review 服务应用入口（uvicorn 入口）。 + +- 独立部署树 review_service/（科研投稿与智能审稿助手，端口 8100）。 +- 端点：/health（GET，BFF 探活反代）+ /api/review/*（自审稿等业务路由）。 +- 不依赖 BFF/生产服务（8090 等绝不探测/连接）；上游仅消费 vstore 8094。 +""" +from __future__ import annotations + +import logging +from contextlib import asynccontextmanager + +import httpx +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware + +from .config import get_settings +from .db import close_db, init_db +from .routers import journals, presubmit + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + +def create_app() -> FastAPI: + settings = get_settings() + app = FastAPI(title=settings.app_name, version=settings.version) + + app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=False, + allow_methods=["*"], + allow_headers=["*"], + ) + + @asynccontextmanager + async def _lifespan(_app: FastAPI): + # journals 推荐依赖独立 SQLite：启动即建表（幂等；测试可先 set_path 注入 + # tmp 路径，本函数沿用注入路径） + await init_db() + try: + yield + finally: + await close_db() + + app.router.lifespan_context = _lifespan + + @app.get("/health") + async def health() -> dict[str, object]: + vstore_ok = False + vstore_error: str | None = None + try: + async with httpx.AsyncClient(timeout=2.0) as client: + resp = await client.get(f"{settings.vstore_base}/api/vstore/health") + resp.raise_for_status() + vstore_ok = bool(resp.json().get("ok")) + except Exception as exc: # noqa: BLE001 + vstore_error = str(exc) + return { + "ok": True, + "status": "ok" if vstore_ok else "degraded", + "upstreams": {"vstore": {"connected": vstore_ok, "error": vstore_error}}, + "version": settings.version, + } + + app.include_router(presubmit.router) + app.include_router(journals.router) + return app + + +app = create_app() diff --git a/review_service/app/routers/__init__.py b/review_service/app/routers/__init__.py new file mode 100644 index 0000000..f026b28 --- /dev/null +++ b/review_service/app/routers/__init__.py @@ -0,0 +1 @@ +"""review 服务路由包。""" diff --git a/review_service/app/routers/journals.py b/review_service/app/routers/journals.py new file mode 100644 index 0000000..ad294fe --- /dev/null +++ b/review_service/app/routers/journals.py @@ -0,0 +1,292 @@ +"""routers/journals.py —— 期刊情报与智能投稿推荐。 + +- GET /api/review/journals 只读：journals 表全量列表（?field=/&order= 排序）， + meta 含 count / last_updated / data_source_note。无任何写接口；空表返回 + 空数组 + meta.count=0（HTTP 200，绝不 500）。 +- POST /api/review/recommend 智能投稿推荐：vstore 向量匹配（摘要 vs + scope_desc，白名单检索）→ LLM 按 id 白名单重排（幻觉期刊丢弃、缺漏规则 + 补位）→ 三档输出：挑战型 / 适配型 / 稳妥型，附匹配理由与风险提示。 + +降级哲学（同 presubmit）：向量失败 → 全量候选 + vector_ok=false； +LLM 失败 → 规则分档 + llm_ok=false；推荐永不 5xx。仅入参非法返回 +VALIDATION 错误信封（HTTP 200，与本服务其余端点一致）。 +""" +from __future__ import annotations + +import logging +from typing import Any, Literal + +from fastapi import APIRouter +from pydantic import BaseModel, Field + +from ..config import get_settings +from ..db import SeedError, ensure_journals_seeded, list_journals +from ..services import journal_kb, journal_rank, llm + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/review", tags=["review-journals"]) + + +# ---- 请求/响应模型 ---- + +class Preferences(BaseModel): + max_review_weeks: float | None = Field(default=None, ge=1, le=104) + min_impact_factor: float | None = Field(default=None, ge=0, le=200) + tier_focus: Literal["all", "challenge", "fit", "safe"] = "all" + + +class RecommendRequest(BaseModel): + abstract: str + preferences: Preferences = Preferences() + + +class JournalItem(BaseModel): + id: str + name: str + impact_factor: float | None = None + review_time_weeks: float | None = None + acceptance_rate: float | None = None + official_url: str | None = None + similarity: float | None = None + tier: str + match_reason: str + risk_note: str + + +class TierBox(BaseModel): + label: str + items: list[JournalItem] + + +class RecommendResponse(BaseModel): + ok: bool + tiers: dict[str, TierBox] | None = None + meta: dict[str, Any] | None = None + error: dict[str, Any] | None = None + + +# ---- GET /api/review/journals（只读列表）---- + +class JournalBrief(BaseModel): + """期刊公开字段（journals 表全列；不含 vstore_doc_id 内部映射）。""" + id: str + name: str + scope_desc: str = "" + impact_factor: float | None = None + review_time_weeks: float | None = None + acceptance_rate: float | None = None + submission_deadline: str | None = None + official_url: str | None = None + last_updated: str = "" + + +class JournalsListResponse(BaseModel): + ok: bool + items: list[JournalBrief] = [] + meta: dict[str, Any] | None = None + error: dict[str, Any] | None = None + + +# ?field= 排序白名单（列名无法参数化，白名单防注入；非法值 → VALIDATION 信封） +_SORT_FIELDS = frozenset({ + "id", + "name", + "impact_factor", + "review_time_weeks", + "acceptance_rate", + "last_updated", +}) +_DATA_SOURCE_NOTE = ( + "IF/审稿周期为公开近似值，acceptance_rate 为 null 表示暂无公开数据，需人工核验。" +) + + +def _sort_rows(rows: list[dict[str, Any]], field: str, order: str) -> list[dict[str, Any]]: + """按白名单字段排序；NULL 恒排最后；次序不稳定时按 db 默认序（稳定排序保留）。""" + present = [r for r in rows if r.get(field) is not None] + missing = [r for r in rows if r.get(field) is None] + present.sort(key=lambda r: r.get(field), reverse=(order == "desc")) + return present + missing + + +def _list_error(message: str) -> JournalsListResponse: + return JournalsListResponse( + ok=False, items=[], meta=None, + error={"code": "VALIDATION", "message": message, "degraded": False}, + ) + + +@router.get("/journals", response_model=JournalsListResponse) +async def journals_list( + field: str = "name", + order: str = "asc", +) -> JournalsListResponse: + """只读：journals 表全量列表（无写/删/改接口）。 + + - ?field= 排序字段白名单：id/name/impact_factor/review_time_weeks/ + acceptance_rate/last_updated，默认 name； + - ?order=asc|desc，默认 asc； + - 空表 → items=[] + meta.count=0（HTTP 200，不报错）。种子装载由 + POST /recommend 首次调用幂等完成，本端点严格只读。 + """ + col = (field or "").strip() + if col not in _SORT_FIELDS: + return _list_error( + f"不支持的排序字段 field={field!r}，可选：{', '.join(sorted(_SORT_FIELDS))}" + ) + ord_norm = (order or "asc").strip().lower() + if ord_norm not in ("asc", "desc"): + return _list_error(f"不支持的排序方向 order={order!r}，可选：asc, desc") + + rows = await list_journals() + rows = _sort_rows(rows, col, ord_norm) + + items = [ + JournalBrief( + id=str(r.get("id") or ""), + name=str(r.get("name") or ""), + scope_desc=str(r.get("scope_desc") or ""), + impact_factor=r.get("impact_factor"), + review_time_weeks=r.get("review_time_weeks"), + acceptance_rate=r.get("acceptance_rate"), + submission_deadline=r.get("submission_deadline"), + official_url=r.get("official_url"), + last_updated=str(r.get("last_updated") or ""), + ) + for r in rows + ] + latest = max((str(r.get("last_updated") or "") for r in rows), default="") + meta: dict[str, Any] = { + "count": len(items), + "last_updated": latest or None, + "data_source_note": _DATA_SOURCE_NOTE, + } + return JournalsListResponse(ok=True, items=items, meta=meta, error=None) + + +def _error(code: str, message: str, *, degraded: bool = False) -> RecommendResponse: + return RecommendResponse( + ok=False, tiers=None, meta=None, + error={"code": code, "message": message, "degraded": degraded}, + ) + + +def _item_sort_key(it: JournalItem) -> tuple[int, float, float]: + return ( + 1 if it.similarity is None else 0, + -(it.similarity or 0.0), + -(it.impact_factor or 0.0), + ) + + +def _apply_soft_preferences(item: JournalItem, prefs: Preferences) -> None: + """软偏好：不满足只加风险提示，不删条目。""" + weeks = item.review_time_weeks + if prefs.max_review_weeks is not None and weeks is not None and weeks > prefs.max_review_weeks: + prefix = f"⚠ 审稿约 {weeks:g} 周，超出你 {prefs.max_review_weeks:g} 周的周期偏好；" + item.risk_note = prefix + ("无" if item.risk_note == "无" else item.risk_note) + inf = item.impact_factor + if prefs.min_impact_factor is not None: + if inf is None: + item.risk_note = "⚠ 影响因子未收录，无法确认满足你的要求；" + item.risk_note + elif inf < prefs.min_impact_factor: + prefix = f"⚠ 影响因子 {inf:g} 低于你 {prefs.min_impact_factor:g} 的要求；" + item.risk_note = prefix + ("无" if item.risk_note == "无" else item.risk_note) + + +@router.post("/recommend", response_model=RecommendResponse) +async def recommend(req: RecommendRequest) -> RecommendResponse: + settings = get_settings() + + # ---- 入参校验（手动，错误信封与本服务其余端点一致）---- + abstract = (req.abstract or "").strip() + if len(abstract) < 50: + return _error("VALIDATION", "abstract 至少需要 50 个字符（请提供摘要或正文片段）") + if len(abstract) > 20000: + return _error("VALIDATION", "abstract 超过 20000 字符上限") + + prefs = req.preferences + + # ---- 种子 + 期刊元数据 ---- + try: + await ensure_journals_seeded() + journals = await list_journals() + except SeedError as exc: + logger.error("recommend: 种子装载失败：%s", exc) + return _error("SEED_UNAVAILABLE", str(exc), degraded=True) + if not journals: + return _error("SEED_UNAVAILABLE", "期刊库为空", degraded=True) + + # ---- ① vstore 向量匹配（摘要 vs scope_desc）---- + try: + candidates, vector_ok = await journal_kb.match(abstract, journals) + except journal_kb.MatchError as exc: + logger.warning("recommend: 向量匹配降级：%s", exc) + candidates, vector_ok = [], False + if not candidates: + # 无有效语义分 → 全量候选（similarity=None），LLM 仍可按 scope 文本重排 + candidates = [{"journal": j, "similarity": None} for j in journals] + vector_ok = False + + # ---- ② LLM 重排（id 白名单硬校验）---- + prefs_dict = prefs.model_dump(exclude_none=True) + llm_ok = True + dropped = 0 + repaired = 0 + try: + rankings, dropped, repaired = await journal_rank.rerank(abstract, candidates, prefs_dict) + except (llm.LLMError, journal_rank.RankError, ValueError) as exc: + # ValueError：complete_json 最终 json.loads 修复仍失败（同 presubmit 宽容口径） + logger.warning("recommend: LLM 重排降级为规则分档：%s", exc) + rankings = journal_rank.fallback_rankings(candidates) + llm_ok = False + + # ---- ③ 三档组装（名称/指标全部来自 DB，不采信 LLM 文本）---- + by_id = {str(c["journal"]["id"]): c for c in candidates} + tiers: dict[str, list[JournalItem]] = {t: [] for t in journal_rank.TIERS} + for r in rankings: + cand = by_id.get(r.journal_id) + if cand is None: + continue + j = cand["journal"] + sim = cand.get("similarity") + item = JournalItem( + id=str(j["id"]), + name=str(j.get("name") or j["id"]), + impact_factor=j.get("impact_factor"), + review_time_weeks=j.get("review_time_weeks"), + acceptance_rate=j.get("acceptance_rate"), + official_url=j.get("official_url"), + similarity=round(float(sim), 4) if sim is not None else None, + tier=r.tier, + match_reason=r.match_reason, + risk_note=r.risk_note, + ) + _apply_soft_preferences(item, prefs) + tiers.setdefault(r.tier, []).append(item) + + # 档内排序：相似度（None 靠后）→ 影响因子 + for box in tiers.values(): + box.sort(key=_item_sort_key) + + # tier_focus：软展示聚焦（默认 all；聚焦时其余档 items 置空，label 保留） + out_tiers: dict[str, TierBox] = {} + for t in journal_rank.TIERS: + items = tiers.get(t, []) + if prefs.tier_focus != "all" and t != prefs.tier_focus: + items = [] + out_tiers[t] = TierBox(label=journal_rank.TIER_LABELS[t], items=items) + + meta: dict[str, Any] = { + "model": llm._llm_config()["model"], + "vector_ok": vector_ok, + "llm_ok": llm_ok, + "seed_count": len(journals), + "candidate_count": len(candidates), + "dropped_hallucinated": dropped, + "repaired": repaired, + "preferences": prefs_dict, + "abstract_chars": len(abstract), + } + return RecommendResponse(ok=True, tiers=out_tiers, meta=meta, error=None) diff --git a/review_service/app/routers/presubmit.py b/review_service/app/routers/presubmit.py new file mode 100644 index 0000000..f15b69e --- /dev/null +++ b/review_service/app/routers/presubmit.py @@ -0,0 +1,87 @@ +"""routers/presubmit.py —— 投稿前自审稿（POST /api/review/presubmit）。 + +流程：上传 PDF → paper_parse（vstore 8094 解析）→ LLM 结构化审稿 +（DeepSeek json_object，System Prompt 为植物基因组学六维专属清单， +见 services/self_review.py）→ 归一化为 SelfReviewReport。 +""" +from __future__ import annotations + +import logging + +from fastapi import APIRouter, File, Form, UploadFile +from pydantic import BaseModel + +from ..config import get_settings +from ..services import llm, paper_parse, self_review + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/review", tags=["review-presubmit"]) + + +class PresubmitResponse(BaseModel): + ok: bool + report: self_review.SelfReviewReport | None = None + error: dict[str, object] | None = None + meta: dict[str, object] | None = None + + +def _error(code: str, message: str, *, degraded: bool = False) -> PresubmitResponse: + return PresubmitResponse( + ok=False, report=None, + error={"code": code, "message": message, "degraded": degraded}, + ) + + +@router.post("/presubmit", response_model=PresubmitResponse) +async def presubmit_review( + file: UploadFile = File(..., description="待审稿 PDF"), + target_journal: str = Form("", description="目标期刊（可选，影响格式类意见）"), +) -> PresubmitResponse: + settings = get_settings() + + filename = file.filename or "paper.pdf" + if not filename.lower().endswith(".pdf"): + return _error("VALIDATION", "仅支持 PDF 文件") + + content = await file.read() + if not content: + return _error("VALIDATION", "上传文件为空") + if len(content) > settings.max_pdf_bytes: + return _error("VALIDATION", "PDF 超过 50MB 上限") + + # ---- 解析（vstore 8094）---- + try: + text = await paper_parse.parse_pdf_bytes(content, filename=filename) + except paper_parse.ParseError as exc: + logger.warning("presubmit: 解析失败：%s", exc) + return _error("PARSE_FAILED", str(exc), degraded=True) + + truncated = len(text) > settings.max_text_chars + feed_text = text[: settings.max_text_chars] + + # ---- LLM 结构化自审（植物基因组学六维清单）---- + try: + payload = await llm.complete_json( + self_review.SYSTEM_PROMPT, + self_review.build_user_prompt(feed_text, (target_journal or "").strip()), + temperature=0.2, + max_tokens=8000, + timeout=settings.presubmit_read_timeout, + ) + except llm.LLMError as exc: + logger.error("presubmit: LLM 审稿失败：%s", exc) + return _error("UPSTREAM_DOWN", f"审稿模型调用失败：{exc}", degraded=True) + + report = self_review.coerce_report(payload) + return PresubmitResponse( + ok=True, + report=report, + meta={ + "filename": filename, + "target_journal": (target_journal or "").strip(), + "text_chars": len(feed_text), + "truncated": truncated, + "model": llm._llm_config()["model"], + }, + ) diff --git a/review_service/app/services/__init__.py b/review_service/app/services/__init__.py new file mode 100644 index 0000000..da95dd6 --- /dev/null +++ b/review_service/app/services/__init__.py @@ -0,0 +1 @@ +"""review 服务服务层包。""" diff --git a/review_service/app/services/journal_kb.py b/review_service/app/services/journal_kb.py new file mode 100644 index 0000000..b64c210 --- /dev/null +++ b/review_service/app/services/journal_kb.py @@ -0,0 +1,223 @@ +"""services/journal_kb.py —— 期刊 scope_desc → vstore 入库 + 向量匹配（只读消费 8094）。 + +链路（全部复用 vstore 既有 API，零改动）： +1. ensure_scope_docs：各期刊 scope_desc 以 .txt 入库（category=journal_scope， + name=期刊名），document_id 回写 journals.vstore_doc_id（幂等：已有 id 不重灌）。 +2. match：GET /api/vstore/search-chunks?q=<摘要截断>&doc_ids=&top_k=N + → 按 doc_id 映射回期刊，取该期刊最高 chunk 分数为 similarity。 +3. 自愈：白名单命中 0 条 → GET /documents/{id} 探活 → 失效的重灌（vstore 去重 + 语义下同内容重灌会换 doc_id，需回写）→ 重试一次 → 仍 0 视为语义低相似。 + +错误语义：连通性/上游异常抛 MatchError（路由降级为全量候选）； +返回 ([], False) 表示检索成功但无有效匹配（路由同样降级）。 +""" +from __future__ import annotations + +import logging +from typing import Any + +import httpx + +from ..config import get_settings +from ..db import set_vstore_doc_id + +logger = logging.getLogger(__name__) + +SCOPE_CATEGORY = "journal_scope" + + +class MatchError(Exception): + """vstore 向量匹配失败（不可达/响应异常）。""" + + +def _timeout() -> httpx.Timeout: + t = get_settings().vstore_match_timeout + return httpx.Timeout(connect=5.0, read=t, write=t, pool=5.0) + + +def _err_body(resp: httpx.Response) -> str: + try: + payload = resp.json() + except ValueError: + return (resp.text or "").strip()[:500] or "（空响应体）" + if isinstance(payload, dict): + err = payload.get("error") + if isinstance(err, dict) and err.get("message"): + return str(err["message"]) + if isinstance(err, str): + return err + return str(payload)[:500] + + +async def _ingest_scope(client: httpx.AsyncClient, base: str, journal: dict[str, Any]) -> str: + """单本期刊 scope_desc 入库，返回 document_id。""" + content = str(journal.get("scope_desc") or "").strip() + if not content: + raise MatchError(f"期刊 scope_desc 为空：{journal.get('id')}") + jid = str(journal.get("id") or "") + try: + resp = await client.post( + f"{base}/api/vstore/ingest", + files={"file": (f"{jid}.txt", content.encode("utf-8"), "text/plain")}, + data={ + "name": str(journal.get("name") or jid), + "category": SCOPE_CATEGORY, + "description": "review_service 期刊范围种子（scope_desc 向量匹配）", + }, + ) + except httpx.HTTPError as exc: + raise MatchError(f"vstore 不可达（{base}）：{exc}") from exc + if resp.status_code >= 400: + raise MatchError(f"vstore ingest 失败（HTTP {resp.status_code}）：{_err_body(resp)}") + try: + payload = resp.json() + except ValueError as exc: + raise MatchError(f"vstore ingest 响应非 JSON：{exc}") from exc + if not isinstance(payload, dict) or not payload.get("ok"): + raise MatchError( + str((payload.get("error") or {}).get("message") if isinstance(payload, dict) else "") or "vstore ingest 失败" + ) + doc_id = str((payload.get("item") or {}).get("document_id") or "") + if not doc_id: + raise MatchError("vstore ingest 响应缺少 document_id") + return doc_id + + +async def ensure_scope_docs(journals: list[dict[str, Any]]) -> list[dict[str, Any]]: + """补齐缺失的 scope 入库映射（就地回写 vstore_doc_id 并持久到 DB）。""" + settings = get_settings() + missing = [j for j in journals if not j.get("vstore_doc_id")] + if not missing: + return journals + async with httpx.AsyncClient(timeout=_timeout()) as client: + for j in missing: + try: + doc_id = await _ingest_scope(client, settings.vstore_base, j) + except MatchError as exc: + # 单本失败不阻断其余（该期刊将不进白名单，由路由降级兜底） + logger.warning("journal_kb: scope 入库失败（%s）：%s", j.get("id"), exc) + continue + j["vstore_doc_id"] = doc_id + await set_vstore_doc_id(str(j["id"]), doc_id) + return journals + + +async def _search( + client: httpx.AsyncClient, base: str, query: str, doc_ids: list[str] +) -> list[dict[str, Any]]: + """chunk 级白名单检索。连通/上游错误抛 MatchError。""" + if not doc_ids: + # doc_ids 为空时 vstore 侧等于不过滤 → 会命中用户私有库，必须拦死 + raise MatchError("白名单为空，拒绝全库检索") + try: + resp = await client.get( + f"{base}/api/vstore/search-chunks", + params={ + "q": query, + "top_k": get_settings().recommend_scope_top_k, + "doc_ids": ",".join(doc_ids), + }, + ) + except httpx.HTTPError as exc: + raise MatchError(f"vstore search-chunks 不可达：{exc}") from exc + if resp.status_code >= 400: + raise MatchError(f"vstore search-chunks 失败（HTTP {resp.status_code}）：{_err_body(resp)}") + try: + payload = resp.json() + except ValueError as exc: + raise MatchError(f"vstore search-chunks 响应非 JSON：{exc}") from exc + if not isinstance(payload, dict) or not payload.get("ok"): + raise MatchError( + str((payload.get("error") or {}).get("message") if isinstance(payload, dict) else "") or "vstore search-chunks 失败" + ) + items = payload.get("items") + return items if isinstance(items, list) else [] + + +async def _doc_alive(client: httpx.AsyncClient, base: str, doc_id: str) -> bool: + try: + resp = await client.get(f"{base}/api/vstore/documents/{doc_id}") + except httpx.HTTPError: + return False + if resp.status_code != 200: + return False + try: + payload = resp.json() + except ValueError: + return False + return bool(isinstance(payload, dict) and payload.get("ok")) + + +async def _refresh_dead_docs( + client: httpx.AsyncClient, base: str, journals: list[dict[str, Any]] +) -> None: + """白名单 0 命中时探活：失效文档重灌并回写新 id。""" + for j in journals: + doc_id = str(j.get("vstore_doc_id") or "") + if not doc_id or await _doc_alive(client, base, doc_id): + continue + try: + new_id = await _ingest_scope(client, base, j) + except MatchError as exc: + logger.warning("journal_kb: scope 重灌失败（%s）：%s", j.get("id"), exc) + continue + j["vstore_doc_id"] = new_id + await set_vstore_doc_id(str(j["id"]), new_id) + + +async def match( + abstract: str, journals: list[dict[str, Any]] +) -> tuple[list[dict[str, Any]], bool]: + """摘要 vs scope_desc 向量匹配。 + + 返回 (candidates, vector_ok)；candidates 元素： + {"journal": , "similarity": float} + 按 similarity 降序。MatchError → 路由降级；([], False) → 语义低相似。 + """ + settings = get_settings() + query = abstract.strip()[: settings.recommend_query_chars] + if not query: + return [], False + + await ensure_scope_docs(journals) + whitelisted = [j for j in journals if j.get("vstore_doc_id")] + if not whitelisted: + raise MatchError("无任何期刊 scope 入库成功（vstore_doc_id 为空）") + + doc_map = {str(j["vstore_doc_id"]): j for j in whitelisted} + timeout = _timeout() + async with httpx.AsyncClient(timeout=timeout) as client: + items = await _search(client, settings.vstore_base, query, list(doc_map)) + if not items: + await _refresh_dead_docs(client, settings.vstore_base, whitelisted) + doc_map = {str(j["vstore_doc_id"]): j for j in whitelisted if j.get("vstore_doc_id")} + if not doc_map: + return [], False + items = await _search(client, settings.vstore_base, query, list(doc_map)) + + # 聚合：每本期刊取最高 chunk 分 + best: dict[str, float] = {} + for item in items: + if not isinstance(item, dict): + continue + j = doc_map.get(str(item.get("doc_id") or "")) + if j is None: + continue + try: + score = float(item.get("score")) + except (TypeError, ValueError): + continue + jid = str(j["id"]) + if jid not in best or score > best[jid]: + best[jid] = score + if not best: + return [], False + + by_id = {str(j["id"]): j for j in journals} + candidates = [ + {"journal": by_id[jid], "similarity": score} + for jid, score in best.items() + if jid in by_id + ] + candidates.sort(key=lambda c: c["similarity"], reverse=True) + return candidates, bool(candidates) diff --git a/review_service/app/services/journal_rank.py b/review_service/app/services/journal_rank.py new file mode 100644 index 0000000..d478369 --- /dev/null +++ b/review_service/app/services/journal_rank.py @@ -0,0 +1,210 @@ +"""services/journal_rank.py —— LLM 重排 + 幻觉防护 + 三档归类。 + +职责： +1. rerank：把带语义分的候选期刊（来自 journal_kb）+ 用户偏好交给 LLM， + 要求按 id 白名单返回 {rankings:[{id, tier, match_reason, risk_note}]}。 +2. validate：硬校验——白名单外 id 丢弃（幻觉拦截）；tier 非法/LLM 缺漏的 + 期刊按规则补位（绝不因 LLM 输出丢失候选期刊）；期刊名称/指标一律不采信 + LLM，最终展示全部用 DB 字段（防止改名型幻觉）。 +3. fallback_rankings：LLM 不可用时的确定性规则分档（按 impact_factor）。 + +三档语义：challenge 挑战型（高影响/慢审稿）、fit 适配型（均衡）、 +safe 稳妥型（快速/门槛低）。 +""" +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Any + +from ..config import get_settings +from . import llm + +TIERS = ("challenge", "fit", "safe") +TIER_LABELS = { + "challenge": "挑战型（高影响因子/慢审稿）", + "fit": "适配型（中等影响/周期适中）", + "safe": "稳妥型（快速发表/门槛较低）", +} + +# 规则分档阈值（近似值，可后续配置化） +_CHALLENGE_MIN_IF = 9.0 +_SAFE_MAX_IF = 4.0 + +# tier 同义词归一（LLM 偶发中文标签） +_TIER_ALIASES = { + "挑战": "challenge", + "挑战型": "challenge", + "高风险": "challenge", + "适配": "fit", + "适配型": "fit", + "稳妥": "safe", + "稳妥型": "safe", + "低风险": "safe", +} + +SYSTEM_PROMPT = """你是学术投稿推荐专家（植物基因组学方向）。根据给定候选期刊、论文摘要与用户偏好，为**每一个**候选期刊输出推荐分档与理由。 + +硬规则（违反即视为无效输出）： +1. id 白名单：只能使用输入 candidates 中出现的 id；绝对禁止添加、替换或臆造列表之外的任何期刊。 +2. 每个输入 id 恰好输出一次，不得遗漏、不得重复；期刊名称等元数据由系统填充，你只输出 id。 +3. tier 只能取 challenge | fit | safe： + - challenge（挑战型）：高影响因子、审稿较慢或竞争激烈的期刊 + - fit（适配型）：影响力与发表周期均衡、与论文契合的期刊 + - safe（稳妥型）：审稿较快、门槛较低、命中概率较高的期刊 +4. match_reason：中文，1-2 句，结合摘要的研究主题与该刊 scope_desc 及指标（如有 similarity 一并参考），说明为何匹配/不匹配。 +5. risk_note：中文，1-2 句，指出投稿风险（超出用户偏好、IF 门槛偏高、范围匹配偏弱、审稿周期长等）；无明显风险写「无」。 +6. 用户 preferences 为软约束：不满足偏好的期刊仍必须输出，并在 risk_note 中明确说明。 +7. 不确定时宁可保守陈述，禁止编造统计数据（录用率、审稿天数等）。 + +输出：仅一个 JSON 对象，格式： +{"rankings":[{"id":"...","tier":"challenge","match_reason":"...","risk_note":"..."}]} +不要输出任何 JSON 之外的内容。""" + + +class RankError(Exception): + """LLM 输出形态异常（无法按 JSON 解析/结构非法）。""" + + +@dataclass +class Ranking: + journal_id: str + tier: str + match_reason: str + risk_note: str + + +def rule_tier(journal: dict[str, Any]) -> str: + """确定性分档规则（LLM 缺漏/不可用时兜底）。""" + inf = journal.get("impact_factor") + weeks = journal.get("review_time_weeks") + if isinstance(inf, (int, float)): + if inf >= _CHALLENGE_MIN_IF: + return "challenge" + if inf <= _SAFE_MAX_IF: + return "safe" + elif isinstance(weeks, (int, float)) and weeks <= 4: + return "safe" + return "fit" + + +def build_user_prompt( + abstract: str, + candidates: list[dict[str, Any]], + preferences: dict[str, Any], +) -> str: + """候选 + 偏好 + 摘要 → 用户消息（JSON）。""" + payload = { + "abstract": abstract.strip()[:4000], + "preferences": preferences or {}, + "candidates": [ + { + "id": str(c["journal"]["id"]), + "name": str(c["journal"].get("name") or ""), + "impact_factor": c["journal"].get("impact_factor"), + "review_time_weeks": c["journal"].get("review_time_weeks"), + "acceptance_rate": c["journal"].get("acceptance_rate"), + "similarity": c.get("similarity"), + "scope_desc": str(c["journal"].get("scope_desc") or "")[:400], + } + for c in candidates + ], + } + return json.dumps(payload, ensure_ascii=False) + + +def _normalize_tier(raw: Any) -> str | None: + tier = str(raw or "").strip().lower() + tier = _TIER_ALIASES.get(str(raw or "").strip(), tier) + return tier if tier in TIERS else None + + +def validate_rankings( + payload: Any, candidates: list[dict[str, Any]] +) -> tuple[list[Ranking], int, int]: + """校验 LLM 输出。 + + 返回 (rankings, dropped_hallucinated, repaired)： + - 白名单外 id / 重复 id → 丢弃并计数（repaired 中重复不计）； + - tier 非法、LLM 缺漏的 id → 按规则补位（repaired 计数）。 + """ + if not isinstance(payload, dict) or not isinstance(payload.get("rankings"), list): + raise RankError("LLM 输出缺少 rankings 数组") + + by_id = {str(c["journal"]["id"]): c["journal"] for c in candidates} + input_ids = list(by_id) + seen: set[str] = set() + rankings: list[Ranking] = [] + dropped = 0 + repaired = 0 + + for entry in payload["rankings"]: + if not isinstance(entry, dict): + dropped += 1 + continue + jid = str(entry.get("id") or "").strip() + if jid in seen: + continue + if jid not in by_id: + dropped += 1 # 幻觉期刊：绝不进输出 + continue + seen.add(jid) + tier = _normalize_tier(entry.get("tier")) + if tier is None: + tier = rule_tier(by_id[jid]) + repaired += 1 + reason = str(entry.get("match_reason") or "").strip() or "基于期刊范围与指标的匹配" + risk = str(entry.get("risk_note") or "").strip() or "无" + rankings.append(Ranking(journal_id=jid, tier=tier, match_reason=reason, risk_note=risk)) + + for jid in input_ids: + if jid in seen: + continue + rankings.append( + Ranking( + journal_id=jid, + tier=rule_tier(by_id[jid]), + match_reason="LLM 未给出该期刊条目，已按影响因子/周期规则补位", + risk_note="未提供（LLM 输出缺漏，规则兜底）", + ) + ) + repaired += 1 + return rankings, dropped, repaired + + +async def rerank( + abstract: str, + candidates: list[dict[str, Any]], + preferences: dict[str, Any], +) -> tuple[list[Ranking], int, int]: + """LLM 重排 + 校验。LLMError / RankError 上抛（路由降级为规则分档）。""" + settings = get_settings() + payload = await llm.complete_json( + SYSTEM_PROMPT, + build_user_prompt(abstract, candidates, preferences), + temperature=0.2, + max_tokens=3000, + timeout=settings.recommend_read_timeout, + ) + return validate_rankings(payload, candidates) + + +def fallback_rankings(candidates: list[dict[str, Any]]) -> list[Ranking]: + """LLM 不可用时的确定性分档（不丢任何候选期刊）。""" + rankings: list[Ranking] = [] + for c in candidates: + j = c["journal"] + inf = j.get("impact_factor") + weeks = j.get("review_time_weeks") + rankings.append( + Ranking( + journal_id=str(j["id"]), + tier=rule_tier(j), + match_reason=( + f"规则分档：影响因子 {inf if inf is not None else '未收录'}、" + f"审稿约 {weeks if weeks is not None else '未收录'} 周" + ), + risk_note="LLM 不可用，本条按影响因子/周期规则估算", + ) + ) + return rankings diff --git a/review_service/app/services/llm.py b/review_service/app/services/llm.py new file mode 100644 index 0000000..06c8e43 --- /dev/null +++ b/review_service/app/services/llm.py @@ -0,0 +1,203 @@ +"""services/llm.py —— LLM 调用（httpx 裸调 DeepSeek，模式同 paper_writer_service）。 + +- complete_text(system, user, *, temperature=0.3, max_tokens=8000, timeout=None) -> str +- complete_json(...)（response_format json_object + 一次 JSON 修复重试） +- 4 次尝试（1.5s / 3s / 6s 指数退避，429 / 5xx / 超时） +- env 键（优先 REVIEW_LLM_*，回退 DOCGEN_LLM_*，再回退内置默认）： + *_LLM_BASE_URL / *_LLM_API_KEY / *_LLM_MODEL / *_LLM_TIMEOUT +""" +from __future__ import annotations + +import asyncio +import json +import logging +import os +from typing import Any + +import httpx + +logger = logging.getLogger(__name__) + +DEFAULT_LLM_BASE_URL = "https://api.deepseek.com/v1" +DEFAULT_LLM_MODEL = "deepseek-chat" +DEFAULT_LLM_TIMEOUT = 300.0 +BACKOFF_SECONDS = (1.5, 3.0, 6.0) +MAX_ATTEMPTS = 4 + + +class LLMError(Exception): + """LLM 调用失败（重试尽/响应异常）。""" + + +def _env(keys: tuple[str, ...], default: str = "") -> str: + for key in keys: + val = os.environ.get(key) + if val: + return val + return default + + +def _llm_config() -> dict[str, Any]: + return { + "base_url": _env( + ("REVIEW_LLM_BASE_URL", "DOCGEN_LLM_BASE_URL"), DEFAULT_LLM_BASE_URL + ).rstrip("/"), + "api_key": _env(("REVIEW_LLM_API_KEY", "DOCGEN_LLM_API_KEY"), ""), + "model": _env(("REVIEW_LLM_MODEL", "DOCGEN_LLM_MODEL"), DEFAULT_LLM_MODEL), + "timeout": float( + _env(("REVIEW_LLM_TIMEOUT", "DOCGEN_LLM_TIMEOUT"), str(DEFAULT_LLM_TIMEOUT)) + ), + } + + +def _extract_text(payload: dict[str, Any]) -> str: + try: + choices = payload.get("choices") or [] + if choices: + content = choices[0].get("message", {}).get("content") + if isinstance(content, str): + return content + if isinstance(content, list): + return "".join( + str(c.get("text", "")) if isinstance(c, dict) else str(c) for c in content + ) + except Exception: + pass + return "" + + +async def _chat_once( + client: httpx.AsyncClient, + cfg: dict[str, Any], + system: str, + user: str, + *, + temperature: float, + max_tokens: int, + json_mode: bool, + timeout: float | None, +) -> str: + headers = {"Content-Type": "application/json"} + if cfg["api_key"]: + headers["Authorization"] = f"Bearer {cfg['api_key']}" + body: dict[str, Any] = { + "model": cfg["model"], + "messages": [ + {"role": "system", "content": system}, + {"role": "user", "content": user}, + ], + "temperature": temperature, + "max_tokens": max_tokens, + } + if json_mode: + body["response_format"] = {"type": "json_object"} + resp = await client.post( + f"{cfg['base_url']}/chat/completions", + headers=headers, + json=body, + timeout=timeout if timeout is not None else cfg["timeout"], + ) + resp.raise_for_status() + payload = resp.json() + if not isinstance(payload, dict): + raise LLMError("LLM 响应形态异常（非 JSON 对象）") + text = _extract_text(payload) + if not text: + raise LLMError("LLM 响应缺少 choices[0].message.content") + return text + + +async def _complete( + system: str, + user: str, + *, + temperature: float, + max_tokens: int, + json_mode: bool, + timeout: float | None, +) -> str: + cfg = _llm_config() + last_exc: BaseException | None = None + async with httpx.AsyncClient() as client: + for attempt in range(MAX_ATTEMPTS): + try: + return await _chat_once( + client, cfg, system, user, + temperature=temperature, max_tokens=max_tokens, json_mode=json_mode, + timeout=timeout, + ) + except LLMError: + raise + except (httpx.HTTPStatusError, httpx.HTTPError, httpx.TimeoutException) as exc: + last_exc = exc + status = getattr(getattr(exc, "response", None), "status_code", None) + retryable = status is None or status == 429 or status >= 500 + if not retryable or attempt == MAX_ATTEMPTS - 1: + raise LLMError(f"LLM 调用失败（HTTP {status or '网络'}）：{exc}") from exc + await asyncio.sleep(BACKOFF_SECONDS[min(attempt, len(BACKOFF_SECONDS) - 1)]) + raise LLMError(f"LLM 调用失败：{last_exc}") + + +async def complete_text( + system: str, + user: str, + *, + temperature: float = 0.3, + max_tokens: int = 8000, + timeout: float | None = None, +) -> str: + """文本补全。失败抛 LLMError。""" + return await _complete( + system, user, temperature=temperature, max_tokens=max_tokens, json_mode=False, + timeout=timeout, + ) + + +def _repair_json(text: str) -> str: + text = text.strip() + if text.startswith("```"): + text = text.strip("`") + if text.lower().startswith("json"): + text = text[4:] + text = text.strip() + if not text: + raise LLMError("LLM 返回空内容") + if text.startswith("{"): + return text + start = text.find("{") + end = text.rfind("}") + if start >= 0 and end > start: + return text[start:end + 1] + raise LLMError("LLM 输出中未找到 JSON 对象") + + +async def complete_json( + system: str, + user: str, + *, + temperature: float = 0.3, + max_tokens: int = 8000, + timeout: float | None = None, +) -> dict[str, Any]: + """JSON 补全（json_object + 一次修复重试）。""" + text = await _complete( + system, user, temperature=temperature, max_tokens=max_tokens, json_mode=True, + timeout=timeout, + ) + try: + return json.loads(text) + except ValueError: + pass + text2 = await _complete( + system + "\n（重要：你的回答必须是单个合法 JSON 对象，不要输出其他任何内容）", + user, + temperature=temperature, + max_tokens=max_tokens, + json_mode=True, + timeout=timeout, + ) + try: + return json.loads(text2) + except ValueError: + repaired = _repair_json(text2) + return json.loads(repaired) diff --git a/review_service/app/services/paper_parse.py b/review_service/app/services/paper_parse.py new file mode 100644 index 0000000..cb7c3f5 --- /dev/null +++ b/review_service/app/services/paper_parse.py @@ -0,0 +1,139 @@ +"""paper_parse.py —— PDF → 纯文本（经 vstore 8094 现有解析接口，零改动复用）。 + +链路（全部只读消费 vstore 既有 API，不修改 vstore 代码）： +1. POST {vstore}/api/vstore/ingest multipart file（vstore 内部 pymupdf 解析） +2. GET {vstore}/api/vstore/documents/{doc_id} → document.paragraphs 拼接纯文本 +3. DELETE {vstore}/api/vstore/documents/{doc_id} （临时解析件 best-effort 清理， + REVIEW_PARSE_DELETE_AFTER=0 可关闭——保留文档便于人工排查） + +为何不用 methodology 8092：其 /extract-upload 是 MinerU 分钟级异步 job +（MinerU 子进程 + LLM 抽取两阶段），不适配同步自审稿；vstore 段落解析秒级 +返回，满足 presubmit 场景。methodology 地址在 config 中预留供后续扩展。 +""" +from __future__ import annotations + +import logging +from typing import Any + +import httpx + +from ..config import get_settings + +logger = logging.getLogger(__name__) + + +class ParseError(Exception): + """PDF 解析失败（上游不可达/响应异常/空文本）。""" + + +async def parse_pdf(path: str) -> str: + """接收 PDF 文件路径，返回拼接后的纯文本。失败抛 ParseError。""" + try: + with open(path, "rb") as fh: + file_bytes = fh.read() + except OSError as exc: + raise ParseError(f"读取 PDF 失败：{exc}") from exc + return await parse_pdf_bytes(file_bytes, filename=str(path).rsplit("/", 1)[-1] or "paper.pdf") + + +async def parse_pdf_bytes(file_bytes: bytes, filename: str = "paper.pdf") -> str: + """接收 PDF 字节流，返回拼接后的纯文本。失败抛 ParseError。""" + if not file_bytes: + raise ParseError("PDF 内容为空") + + settings = get_settings() + base = settings.vstore_base + timeout = httpx.Timeout( + connect=5.0, read=settings.parse_timeout, write=settings.parse_timeout, pool=5.0 + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + # ---- 1) ingest（vstore 内部 pymupdf 解析 + 嵌入 + 加密落盘）---- + try: + resp = await client.post( + f"{base}/api/vstore/ingest", + files={"file": (filename, file_bytes, "application/pdf")}, + data={ + "name": filename, + "category": "review_presubmit", + "description": "review_service 临时解析件", + }, + ) + except httpx.HTTPError as exc: + raise ParseError(f"vstore 不可达（{base}）：{exc}") from exc + + if resp.status_code >= 400: + # 读上游错误体（信封 JSON 或裸文本），区分“不可达”与“上游报错” + raise ParseError(f"vstore ingest 失败（HTTP {resp.status_code}）：{_err_body(resp)}") + + try: + payload = resp.json() + except ValueError as exc: + raise ParseError(f"vstore ingest 响应非 JSON：{exc}") from exc + + if not isinstance(payload, dict) or not payload.get("ok"): + message = _err_message(payload) or "vstore ingest 失败" + raise ParseError(message) + doc_id = str((payload.get("item") or {}).get("document_id") or "") + if not doc_id: + raise ParseError("vstore ingest 响应缺少 document_id") + + # ---- 2) 取明文全文（paragraphs 拼接）+ 3) 清理临时件 ---- + try: + text = await _fetch_text(client, base, doc_id) + finally: + if settings.parse_delete_after: + await _best_effort_delete(client, base, doc_id) + + text = text.strip() + if not text: + raise ParseError("PDF 解析结果为空（可能是扫描件/图片型 PDF，vstore 无 OCR）") + return text + + +async def _fetch_text(client: httpx.AsyncClient, base: str, doc_id: str) -> str: + """取 vstore 文档明文并按段落拼接纯文本。""" + try: + resp = await client.get(f"{base}/api/vstore/documents/{doc_id}") + except httpx.HTTPError as exc: + raise ParseError(f"vstore 取文档失败（{doc_id}）：{exc}") from exc + if resp.status_code >= 400: + raise ParseError( + f"vstore 取文档失败（{doc_id}，HTTP {resp.status_code}）：{_err_body(resp)}" + ) + try: + payload = resp.json() + except ValueError as exc: + raise ParseError(f"vstore 取文档响应非 JSON（{doc_id}）：{exc}") from exc + if not isinstance(payload, dict) or not payload.get("ok"): + raise ParseError(_err_message(payload) or f"vstore 取文档失败（{doc_id}）") + + document: dict[str, Any] = payload.get("document") or {} + paragraphs = document.get("paragraphs") or [] + return "\n".join(str(p) for p in paragraphs if str(p).strip()) + + +async def _best_effort_delete(client: httpx.AsyncClient, base: str, doc_id: str) -> None: + try: + await client.delete(f"{base}/api/vstore/documents/{doc_id}") + except Exception: # noqa: BLE001 - 清理失败不阻断（文档级 secure_delete 已尽力） + logger.warning("paper_parse: vstore 临时文档清理失败（%s）", doc_id, exc_info=True) + + +def _err_message(payload: Any) -> str: + if isinstance(payload, dict): + err = payload.get("error") + if isinstance(err, dict): + return str(err.get("message") or "") + if isinstance(err, str): + return err + return "" + + +def _err_body(resp: httpx.Response) -> str: + """上游错误体：优先 JSON 信封 message，回退裸文本（截断 500 字符）。""" + try: + payload = resp.json() + except ValueError: + return (resp.text or "").strip()[:500] or "（空响应体）" + return _err_message(payload) or str(payload)[:500] diff --git a/review_service/app/services/self_review.py b/review_service/app/services/self_review.py new file mode 100644 index 0000000..0def3bd --- /dev/null +++ b/review_service/app/services/self_review.py @@ -0,0 +1,155 @@ +"""services/self_review.py —— 植物基因组学专属自审稿（System Prompt + 报告模型）。 + +六大领域维度检查清单硬编码在 SYSTEM_PROMPT（发给 DeepSeek）： +1. 基因组组装质量（N50/L50/BUSCO 完整格式/Hi-C 挂载率，缺项即 Major） +2. 基因注释可靠性（MAKER/AUGUSTUS 流程、RNA-seq 证据、基因数与已知范围一致性） +3. 比较基因组学与进化（Bootstrap/后验概率、OrthoFinder 等直系同源策略） +4. 数据可用性合规（SRA/BioProject + 组装/GFF3 Accession，缺失即 Major） +5. 统计与实验设计（GWAS/QTL 样本量·多重校正·群体结构；转录组生物学重复） +6. 伦理与合规·植物特异性（凭证标本 voucher、ABS 惠益分享声明） + +Major/Minor 每条必须给 Evidence（论文章节位置或具体指标数值），由 coerce +宽容归一化（缺失补空串，不丢条目）。 +""" +from __future__ import annotations + +from pydantic import BaseModel, Field + +RECOMMENDATIONS = ("accept", "minor_revision", "major_revision", "reject") + +SYSTEM_PROMPT = """你是一位植物基因组学领域资深审稿人，长期为 Plant Cell、Nature Plants、 +Genome Biology、Molecular Plant、Plant Journal 等期刊审稿。请对用户提供的论文正文做 +投稿前自审（pre-submission self review），严格按以下六个植物基因组学专属维度逐项检查， +输出结构化审稿意见。所有条目使用中文撰写（专业术语、指标名、工具名保留英文）。 + +【维度一：基因组组装质量】必查指标： +- contig N50 / scaffold N50 / L50（报了哪个要写明哪个） +- BUSCO 完整度：必须按完整格式报告——例如 + "Complete 95.3% [S:93.1%, D:2.2%], Missing 4.7%, n=1614 (embryophyta_odb10)" + （含 Complete/Single-copy/Duplicated/Missing 四项与数据集版本、n 值） +- Hi-C 挂载率（anchoring rate，chromosome-level 比例）及挂载到染色体的 scaffold 比例 +判定规则：若论文只报 N50 而未报 BUSCO 完整度或 Hi-C 挂载率（组装类论文），直接列为 +Major Weakness。同时检查组装是否为 pseudo-chromosome 级别、污染评估（如 k-mer/coverage +检查）与 gap 比例。 + +【维度二：基因注释可靠性】检查： +- 是否使用标准注释流程（MAKER、AUGUSTUS、BRAKER、GeMoMa、EVidenceModeler 等），参数与版本是否交代 +- 是否有转录组证据支持：RNA-seq（组织/时期/物种内或近缘种）比对支撑比例，或蛋白同源证据 +- 基因数量是否与该物种已知基因数范围一致（如水稻 ~37-40k、拟南芥 ~27k、玉米 ~39-45k、 + 小麦 ~107k、大豆 ~46k；近缘种偏差过大须说明）；有无转座元件基因误注释的排查 +- 注释质量评估：BRAKER/AED 分值、BUSCO 在蛋白集上的复核 + +【维度三：比较基因组学与进化】检查： +- 系统发育树：每个节点必须标注 Bootstrap 支持率（建议 ≥1000 次重复）或贝叶斯后验概率； + 未标注或仅"高/低支持"的描述 → Major 或 Minor（视严重程度） +- 直系同源基因鉴定策略是否说明清楚：OrthoFinder/OrthoMCL/RBX 等工具、版本、参数、 + 聚类阈值（inflation/coverage）、single-copy 标记基因的筛选方式 +- 分子进化分析是否规范：dN/dS 检验方法与多重校正、共线性（synteny）分析参数、 + 分化时间估算是否校准（化石/二次校准点） + +【维度四：数据可用性合规——极其重要】必查： +- 原始测序数据是否上传 NCBI SRA / BioProject（或 ENA/DDBJ），是否给出 Accession 号 + （PRJNA/PRJEB/ SRP 开头） +- 基因组组装是否有 GenBank/ENA accession（GCA_/GCF_），GFF3 注释文件、蛋白/CDS 文件 + 是否有可公开获取的位置（仓库 accession 或补充数据链接） +- 转录组数据是否同样有 SRA accession +判定规则：组装/注释类论文若缺失上述任何一环，直接列为 Major Weakness。 + +【维度五：统计与实验设计】检查： +- GWAS/QTL 定位：样本量/群体大小是否充分；多重检验校正是否明确（Bonferroni/FDR/ + permutation 阈值/PCA 显著性水平）；群体结构控制（PCA、亲缘关系矩阵 K、分层分析） + 是否交代；表型测量描述是否充分 +- 转录组分析：生物学重复数量（≥3 为底线）、差异分析工具（DESeq2/edgeR 等）与 + 校正方法、批次效应处理、|log2FC| 与 padj 阈值 +- 其他统计：t 检验/ANOVA 是否满足前提、重复类型（生物学 vs 技术重复）写明与否 + +【维度六：伦理与合规（植物特异性）】检查： +- 野生植物材料：是否提供凭证标本（voucher specimen）信息——标本馆编号、存放机构、 + 采集地点（GPS）与采集许可；无凭证标本的野生资源取样 → Major 或 Minor（视用途） +- 遗传资源获取与惠益分享（ABS）合规性：是否声明符合《名古屋议定书》/本国遗传资源 + 获取条例（如中国《生物安全法》/《种子法》相关条款）、是否说明材料来源与授权 +- 农业转基因/基因编辑材料的合规声明（若涉及） + +输出要求（必须输出单个 JSON 对象，不要输出其他任何内容）： +{ + "summary": "总体评价（3-5 句，须点明论文类型与所适用的维度）", + "major_weaknesses": [{"title": "...", "detail": "...", "evidence": "...", "suggestion": "..."}], + "minor_weaknesses": [{"title": "...", "detail": "...", "evidence": "...", "suggestion": "..."}], + "presentation_format": [{"title": "...", "detail": "...", "evidence": "...", "suggestion": "..."}], + "questions_for_authors": ["..."], + "recommendation": "major_revision" +} + +硬性规则： +1. major_weaknesses 3-8 条、minor_weaknesses 3-8 条；上述维度中命中的硬性判定规则 + （组装缺 BUSCO/挂载率、缺数据 Accession）必须出现。 +2. **每条 major_weaknesses 的 evidence 字段必填**：必须引用论文的具体章节位置 + （如 "Methods 2.1"、"Table 2"、"Results 第 3 段"、"Data Availability 一句话未见"） + 或具体指标数值（如 "仅报 scaffold N50=42.6 Mb，未提供 BUSCO 与 Hi-C 挂载率"）。 + evidence 严禁编造：引用不到具体位置时，写明"正文未见相关报告"。 +3. minor_weaknesses/presentation_format 的 evidence 尽量给出，无法定位时可留空字符串。 +4. 论文不涉及的维度（如纯 RNA-seq 论文不谈组装）在 summary 中说明并跳过，不硬凑条目。 +5. recommendation 只能是 accept / minor_revision / major_revision / reject 之一， + 缺失关键合规项（维度四）时通常不高于 major_revision。 +6. presentation_format 覆盖：图表规范（图注含单位/误差棒说明）、参考文献格式、 + 字数与结构、英文语法（若投英文刊）。 +""" + + +class WeaknessItem(BaseModel): + title: str + detail: str = "" + evidence: str = "" + suggestion: str = "" + + +class SelfReviewReport(BaseModel): + summary: str = "" + major_weaknesses: list[WeaknessItem] = Field(default_factory=list) + minor_weaknesses: list[WeaknessItem] = Field(default_factory=list) + presentation_format: list[WeaknessItem] = Field(default_factory=list) + questions_for_authors: list[str] = Field(default_factory=list) + recommendation: str = "major_revision" + + +def build_user_prompt(text: str, target_journal: str) -> str: + header = f"目标期刊：{target_journal}\n\n" if target_journal else "" + return f"{header}论文正文：\n\n{text}" + + +def _coerce_items(raw: object) -> list[WeaknessItem]: + items: list[WeaknessItem] = [] + if not isinstance(raw, list): + return items + for entry in raw: + if isinstance(entry, str) and entry.strip(): + items.append(WeaknessItem(title=entry.strip())) + elif isinstance(entry, dict): + title = str(entry.get("title") or "").strip() + detail = str(entry.get("detail") or "").strip() + if not title and not detail: + continue + items.append(WeaknessItem( + title=title or "（未命名）", + detail=detail, + evidence=str(entry.get("evidence") or "").strip(), + suggestion=str(entry.get("suggestion") or "").strip(), + )) + return items + + +def coerce_report(payload: dict) -> SelfReviewReport: + questions = payload.get("questions_for_authors") or [] + if not isinstance(questions, list): + questions = [] + recommendation = str(payload.get("recommendation") or "").strip().lower() + if recommendation not in RECOMMENDATIONS: + recommendation = "major_revision" + return SelfReviewReport( + summary=str(payload.get("summary") or "").strip(), + major_weaknesses=_coerce_items(payload.get("major_weaknesses")), + minor_weaknesses=_coerce_items(payload.get("minor_weaknesses")), + presentation_format=_coerce_items(payload.get("presentation_format")), + questions_for_authors=[str(q).strip() for q in questions if str(q).strip()], + recommendation=recommendation, + ) diff --git a/review_service/data/journals/journals.json b/review_service/data/journals/journals.json new file mode 100644 index 0000000..a5c0ee1 --- /dev/null +++ b/review_service/data/journals/journals.json @@ -0,0 +1,62 @@ +{ + "version": 1, + "updated": "2026-10-09", + "notes": "植物基因组学领域种子期刊。acceptance_rate 无公开权威数据一律为 null（不编造）；impact_factor / review_time_weeks 为公开 JCR/期刊官网常识近似值，以 last_updated 标注刷新时间，接入真实数据源前需人工核验。scope_desc 为英文范围描述，供 vstore 向量匹配使用。", + "journals": [ + { + "id": "plant-cell", + "name": "The Plant Cell", + "scope_desc": "Broad-interest plant biology with emphasis on molecular, cellular and physiological mechanisms, including genomics, systems biology, gene regulation, signaling, and plant-microbe interactions. Publishes genome-scale and functional genomics studies with strong mechanistic insight.", + "impact_factor": 11.6, + "review_time_weeks": 6.0, + "acceptance_rate": null, + "submission_deadline": null, + "official_url": "https://academic.oup.com/plcell", + "last_updated": "2026-10-09" + }, + { + "id": "genome-biology", + "name": "Genome Biology", + "scope_desc": "High-impact open-access journal for genome-scale research across all organisms, including genome assembly and annotation, comparative and evolutionary genomics, computational methods, single-cell and multi-omics studies, and reproducible bioinformatics.", + "impact_factor": 10.1, + "review_time_weeks": 4.0, + "acceptance_rate": null, + "submission_deadline": null, + "official_url": "https://genomebiology.biomedcentral.com", + "last_updated": "2026-10-09" + }, + { + "id": "plant-journal", + "name": "The Plant Journal", + "scope_desc": "Molecular and physiological plant sciences, including genome biology and genomics, gene function, cell and developmental biology, stress and environmental responses, and plant biotechnology; favors experimental studies with clear mechanistic advance.", + "impact_factor": 7.2, + "review_time_weeks": 5.0, + "acceptance_rate": null, + "submission_deadline": null, + "official_url": "https://onlinelibrary.wiley.com/journal/1365313X", + "last_updated": "2026-10-09" + }, + { + "id": "theoretical-applied-genetics", + "name": "Theoretical and Applied Genetics", + "scope_desc": "Plant breeding and genetics research combining theoretical and applied approaches, including quantitative genetics, QTL mapping, genome-wide association studies, genomic selection, population genetics, and molecular marker applications for crop improvement.", + "impact_factor": 5.3, + "review_time_weeks": 5.0, + "acceptance_rate": null, + "submission_deadline": null, + "official_url": "https://link.springer.com/journal/338", + "last_updated": "2026-10-09" + }, + { + "id": "plant-genome", + "name": "The Plant Genome", + "scope_desc": "Plant and crop genome research with practical breeding applications, including genome assembly and annotation, structural variation, GWAS and QTL analysis, marker-assisted and genomic selection, and translational genomics for crop improvement.", + "impact_factor": 3.3, + "review_time_weeks": 4.0, + "acceptance_rate": null, + "submission_deadline": null, + "official_url": "https://acsess.onlinelibrary.wiley.com/journal/1940-3372", + "last_updated": "2026-10-09" + } + ] +} diff --git a/review_service/pytest.ini b/review_service/pytest.ini new file mode 100644 index 0000000..1f69640 --- /dev/null +++ b/review_service/pytest.ini @@ -0,0 +1,6 @@ +[pytest] +python_files = test_*.py +pythonpath = . +markers = + review_unit: 审稿服务单元测试 +asyncio_mode = auto diff --git a/review_service/restart.sh b/review_service/restart.sh new file mode 100755 index 0000000..80b522d --- /dev/null +++ b/review_service/restart.sh @@ -0,0 +1,33 @@ +#!/usr/bin/env bash +# restart.sh —— REVIEW 服务重启脚本（8100；不触碰生产服务 8090/8091/8092/8093/8094/8097/8098）。 +# 用法: restart.sh [python] [port] +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +# 默认 venv：本机统一 venv 固定路径（原先 $REPO_ROOT/../ 推导在本工作区不存在） +PYTHON="${1:-/home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python}" +PORT="${2:-8100}" +HOST="${REVIEW_BIND_HOST:-127.0.0.1}" + +[[ -x "$PYTHON" ]] || { echo "Python not found: $PYTHON" >&2; exit 2; } + +APP_MODULE="app.main:app" +if command -v lsof >/dev/null 2>&1; then + PID="$(lsof -ti tcp:"$PORT" -sTCP:LISTEN 2>/dev/null || true)" + if [[ -n "$PID" ]]; then + echo "[restart] 停止 $PORT 旧进程 pid=$PID" + kill $PID 2>/dev/null || true + sleep 2 + fi +fi + +cd "$SCRIPT_DIR" +# 加载运行时配置（.env 不入 git）：services/llm.py 直接读 REVIEW_LLM_*/DOCGEN_LLM_* +# 裸键，paper_parse.py 读 REVIEW_*——需 set -a 显式导出（同 paper_writer_service 先例）。 +set -a +[ -f .env ] && source .env +set +a +echo "[restart] 启动审稿服务 :$PORT（$PYTHON）" +nohup "$PYTHON" -m uvicorn "$APP_MODULE" --host "$HOST" --port "$PORT" \ + > "$SCRIPT_DIR/review.log" 2>&1 & +echo "[restart] 完成；日志 review_service/review.log" diff --git a/review_service/tests/test_presubmit.py b/review_service/tests/test_presubmit.py new file mode 100644 index 0000000..5d4a1de --- /dev/null +++ b/review_service/tests/test_presubmit.py @@ -0,0 +1,163 @@ +"""tests/test_presubmit.py —— presubmit 离线单元测试（mock 解析与 LLM，不依赖上游）。 + +覆盖：非 PDF 校验 / 报告归一化 / LLM 失败降级 / 解析失败降级 / 非法 recommendation 回退。 +运行: cd review_service && /bin/python -m pytest -q +""" +from __future__ import annotations + +import pytest +from fastapi.testclient import TestClient + +from app.main import app +from app.services import llm, paper_parse + +FAKE_PDF = b"%PDF-1.4 fake-bytes-for-test" + +LLM_PAYLOAD = { + "summary": "研究问题清晰，但统计与证据链存在重大缺口。", + "major_weaknesses": [ + { + "title": "统计方法不当", + "detail": "未做多重检验校正", + "evidence": "Methods 2.1 仅报 p<0.05，未见 FDR/Bonferroni", + "suggestion": "补充 FDR 校正", + }, + "纯字符串条目", # 宽容归一化：字符串 → WeaknessItem(title=...) + ], + # minor 故意不带 evidence 字段 → 归一化应补空串 + "minor_weaknesses": [{"detail": "缺少缩写表", "title": "缩写"}], + "presentation_format": [{"title": "图注不完整", "detail": "Fig.2 无单位", "suggestion": "补单位"}], + "questions_for_authors": ["群体结构如何校正？", ""], + "recommendation": "MAJOR_REVISION", # 应归一化为小写 +} + + +@pytest.fixture() +def client() -> TestClient: + return TestClient(app) + + +@pytest.fixture(autouse=True) +def _mock_upstreams(monkeypatch): + async def fake_parse(file_bytes: bytes, filename: str = "paper.pdf") -> str: + assert file_bytes == FAKE_PDF + return "论文正文……" * 10 + + async def fake_llm_json(system: str, user: str, **kwargs): + assert "论文正文" in user + return dict(LLM_PAYLOAD) + + async def boom_llm(*args, **kwargs): + raise llm.LLMError("余额不足") + + async def boom_parse(*args, **kwargs): + raise paper_parse.ParseError("PDF 解析结果为空") + + monkeypatch.setattr(paper_parse, "parse_pdf_bytes", fake_parse) + monkeypatch.setattr(llm, "complete_json", fake_llm_json) + yield + # monkeypatch 自动还原 + + +def _post(client: TestClient, filename: str, content: bytes, journal: str = ""): + data = {"target_journal": journal} if journal else {} + return client.post( + "/api/review/presubmit", + files={"file": (filename, content, "application/pdf")}, + data=data, + ) + + +def test_non_pdf_rejected(client: TestClient): + r = client.post( + "/api/review/presubmit", + files={"file": ("a.txt", b"x", "text/plain")}, + ) + body = r.json() + assert body["ok"] is False + assert body["error"]["code"] == "VALIDATION" + + +def test_empty_file_rejected(client: TestClient): + body = _post(client, "p.pdf", b"").json() + assert body["ok"] is False + assert body["error"]["code"] == "VALIDATION" + + +def test_normal_report_normalized(client: TestClient): + body = _post(client, "p.pdf", FAKE_PDF, journal="Plant Cell").json() + assert body["ok"] is True + report = body["report"] + # recommendation 大写 → 小写 + assert report["recommendation"] == "major_revision" + # 字符串条目宽容归一化 + assert len(report["major_weaknesses"]) == 2 + assert report["major_weaknesses"][1]["title"] == "纯字符串条目" + # Evidence 字段：有则透传，字符串条目/缺失则为空串 + assert report["major_weaknesses"][0]["evidence"] == "Methods 2.1 仅报 p<0.05，未见 FDR/Bonferroni" + assert report["major_weaknesses"][1]["evidence"] == "" + assert report["minor_weaknesses"][0]["evidence"] == "" + # 空问题剔除 + assert report["questions_for_authors"] == ["群体结构如何校正？"] + assert report["minor_weaknesses"] and report["presentation_format"] and report["summary"] + # meta 透传 + assert body["meta"]["target_journal"] == "Plant Cell" + assert body["meta"]["text_chars"] > 0 + assert body["meta"]["truncated"] is False + + +def test_llm_failure_degrades(client: TestClient, monkeypatch): + async def boom(*args, **kwargs): + raise llm.LLMError("余额不足") + + monkeypatch.setattr(llm, "complete_json", boom) + body = _post(client, "p.pdf", FAKE_PDF).json() + assert body["ok"] is False + assert body["error"]["code"] == "UPSTREAM_DOWN" + assert body["error"]["degraded"] is True + + +def test_parse_failure_degrades(client: TestClient, monkeypatch): + async def boom(*args, **kwargs): + raise paper_parse.ParseError("PDF 解析结果为空") + + monkeypatch.setattr(paper_parse, "parse_pdf_bytes", boom) + body = _post(client, "p.pdf", FAKE_PDF).json() + assert body["ok"] is False + assert body["error"]["code"] == "PARSE_FAILED" + assert body["error"]["degraded"] is True + + +def test_invalid_recommendation_falls_back(client: TestClient, monkeypatch): + async def weird(*args, **kwargs): + return {"summary": "x", "recommendation": "strong_accept", "major_weaknesses": "not-a-list"} + + monkeypatch.setattr(llm, "complete_json", weird) + body = _post(client, "p.pdf", FAKE_PDF).json() + assert body["ok"] is True + assert body["report"]["recommendation"] == "major_revision" + assert body["report"]["major_weaknesses"] == [] + + +def test_domain_system_prompt_wired(client: TestClient, monkeypatch): + """植物基因组学六维清单 + Evidence 规则必须注入 System Prompt。""" + captured: dict[str, str] = {} + + async def spy(system: str, user: str, **kwargs): + captured["system"] = system + return dict(LLM_PAYLOAD) + + monkeypatch.setattr(llm, "complete_json", spy) + body = _post(client, "p.pdf", FAKE_PDF, journal="Plant Cell").json() + assert body["ok"] is True + system = captured["system"] + for marker in ( + "BUSCO", "N50", "L50", "Hi-C", "挂载率", # 维度1 组装质量 + "MAKER", "AUGUSTUS", "RNA-seq", # 维度2 注释可靠性 + "Bootstrap", "OrthoFinder", # 维度3 比较基因组 + "SRA", "BioProject", "GFF3", "accession", # 维度4 数据合规 + "GWAS", "多重检验校正", "生物学重复", # 维度5 统计设计 + "凭证标本", "ABS", "惠益分享", # 维度6 伦理合规 + "evidence", "章节", # Evidence 硬性要求 + ): + assert marker in system, f"System Prompt 缺少领域标记: {marker}" diff --git a/review_service/tests/test_recommend.py b/review_service/tests/test_recommend.py new file mode 100644 index 0000000..1f6e50f --- /dev/null +++ b/review_service/tests/test_recommend.py @@ -0,0 +1,271 @@ +"""tests/test_recommend.py —— /api/review/recommend 离线单元测试。 + +mock vstore 向量匹配（journal_kb.match）与 LLM（llm.complete_json），不依赖上游。 +核心回归：LLM 幻觉期刊名绝不进输出（白名单 id 硬校验 + 名称一律取 DB）。 + +运行: cd review_service && /bin/python -m pytest -q +""" +from __future__ import annotations + +import asyncio + +import pytest +from fastapi.testclient import TestClient + +from app import db as db_mod +from app.main import app +from app.services import journal_kb, journal_rank, llm + +ABSTRACT = ( + "We report a chromosome-level genome assembly of wild rice Oryza rufipogon " + "using PacBio HiFi and Hi-C sequencing, annotate 38,412 protein-coding genes " + "with RNA-seq evidence, and perform comparative genomics and GWAS analyses " + "to reveal the evolutionary basis of agronomic traits in rice." +) + +SEED_NAMES = { + "The Plant Cell", + "Genome Biology", + "The Plant Journal", + "Theoretical and Applied Genetics", + "The Plant Genome", +} + +# 覆盖全部 5 本种子期刊的合法重排结果 +ALL_RANKINGS = [ + {"id": "plant-cell", "tier": "challenge", "match_reason": "功能基因组机制研究契合", "risk_note": "竞争激烈"}, + {"id": "genome-biology", "tier": "challenge", "match_reason": "组装与比较基因组学范围契合", "risk_note": "IF 门槛高"}, + {"id": "plant-journal", "tier": "fit", "match_reason": "植物分子生物学范围契合", "risk_note": "无"}, + {"id": "theoretical-applied-genetics", "tier": "fit", "match_reason": "GWAS/驯化分析契合", "risk_note": "偏重育种"}, + {"id": "plant-genome", "tier": "safe", "match_reason": "作物基因组组装契合", "risk_note": "IF 较低"}, +] + + +def make_fake_match(scores: dict[str, float] | None = None, fail: bool = False): + async def fake(abstract: str, journals: list[dict]): + if fail: + raise journal_kb.MatchError("vstore 不可达（mock）") + out = [ + {"journal": j, "similarity": (scores or {}).get(str(j["id"]), 0.5)} + for j in journals + ] + return out, True + + return fake + + +def make_fake_llm(rankings: list[dict] | None = None, fail: bool = False, payload=None): + async def fake(system: str, user: str, **kwargs): + if fail: + raise llm.LLMError("余额不足（mock）") + if payload is not None: + return payload + return {"rankings": rankings if rankings is not None else []} + + return fake + + +@pytest.fixture() +def client(tmp_path, monkeypatch): + db_mod.set_path(tmp_path / "review_test.db") # 进 lifespan 前注入 tmp 路径 + monkeypatch.setattr(journal_kb, "match", make_fake_match()) + monkeypatch.setattr(llm, "complete_json", make_fake_llm(ALL_RANKINGS)) + with TestClient(app) as c: # CM 触发 lifespan：init_db(注入路径) + 收尾 close + yield c + + +def _all_items(body: dict) -> list[dict]: + items: list[dict] = [] + for tier in ("challenge", "fit", "safe"): + items.extend(body["tiers"][tier]["items"]) + return items + + +def test_recommend_happy_path(client: TestClient): + r = client.post("/api/review/recommend", json={"abstract": ABSTRACT}) + assert r.status_code == 200 + body = r.json() + assert body["ok"] is True + assert set(body["tiers"]) == {"challenge", "fit", "safe"} + assert "挑战型" in body["tiers"]["challenge"]["label"] + assert "适配型" in body["tiers"]["fit"]["label"] + assert "稳妥型" in body["tiers"]["safe"]["label"] + + items = _all_items(body) + assert len(items) == 5 + assert {it["name"] for it in items} == SEED_NAMES + for it in items: + assert it["match_reason"], f"缺少匹配理由：{it['name']}" + assert it["risk_note"], f"缺少风险提示：{it['name']}" + assert it["similarity"] == 0.5 + # 三档内容与 LLM 输出一致 + assert [it["name"] for it in body["tiers"]["challenge"]["items"]][0] == "The Plant Cell" + assert body["tiers"]["safe"]["items"][0]["name"] == "The Plant Genome" + + meta = body["meta"] + assert meta["llm_ok"] is True and meta["vector_ok"] is True + assert meta["dropped_hallucinated"] == 0 + assert meta["seed_count"] == 5 + + +def test_seed_has_no_fabricated_acceptance_rate(client: TestClient): + """种子诚实性：无公开数据的 acceptance_rate 必须为 null。""" + body = client.post("/api/review/recommend", json={"abstract": ABSTRACT}).json() + for it in _all_items(body): + assert it["acceptance_rate"] is None, f"编造了录用率：{it['name']}" + + +def test_hallucinated_journal_dropped(client: TestClient, monkeypatch): + """LLM 幻觉期刊（白名单外 id）绝不进输出；缺漏期刊规则补位。""" + hallucinated = { + "id": "nature-genetics", + "tier": "challenge", + "match_reason": "幻觉期刊，输入列表中并不存在", + "risk_note": "x", + } + monkeypatch.setattr( + llm, + "complete_json", + make_fake_llm([hallucinated, ALL_RANKINGS[1]]), # 1 幻觉 + 1 真实，其余缺漏 + ) + body = client.post("/api/review/recommend", json={"abstract": ABSTRACT}).json() + assert body["ok"] is True + + items = _all_items(body) + raw = str(body) + assert "nature-genetics" not in raw and "Nature Genetics" not in raw + assert "幻觉期刊" not in raw # 幻觉条目的理由同样被丢弃 + assert {it["name"] for it in items} <= SEED_NAMES + assert len(items) == 5 # 4 本缺漏由规则补位，不丢候选 + assert body["meta"]["dropped_hallucinated"] == 1 + assert body["meta"]["repaired"] == 4 + + +def test_llm_name_tamper_ignored(client: TestClient, monkeypatch): + """LLM 篡改期刊名不采信：输出名称一律取 DB。""" + tampered = [ + {"id": "plant-cell", "tier": "fit", "name": "Fake Journal X", + "match_reason": "被篡改的名称", "risk_note": "无"} + ] + monkeypatch.setattr(llm, "complete_json", make_fake_llm(tampered)) + body = client.post("/api/review/recommend", json={"abstract": ABSTRACT}).json() + assert body["ok"] is True + raw = str(body) + assert "Fake Journal X" not in raw + plant_cell = [it for it in _all_items(body) if it["id"] == "plant-cell"] + assert plant_cell and plant_cell[0]["name"] == "The Plant Cell" + + +def test_llm_failure_fallback(client: TestClient, monkeypatch): + """LLM 失败 → 规则分档，推荐仍可用（永不 5xx）。""" + monkeypatch.setattr(llm, "complete_json", make_fake_llm(fail=True)) + body = client.post("/api/review/recommend", json={"abstract": ABSTRACT}).json() + assert body["ok"] is True + assert body["meta"]["llm_ok"] is False + items = _all_items(body) + assert len(items) == 5 + assert {it["name"] for it in items} == SEED_NAMES + # 规则分档：IF≥9 → challenge（The Plant Cell 11.6 / Genome Biology 10.1） + challenge = {it["name"] for it in body["tiers"]["challenge"]["items"]} + assert challenge == {"The Plant Cell", "Genome Biology"} + for it in items: + assert it["match_reason"] and it["risk_note"] + + +def test_vector_failure_degrades(client: TestClient, monkeypatch): + """向量匹配失败 → 全量候选 + similarity 为 null，不报错。""" + monkeypatch.setattr(journal_kb, "match", make_fake_match(fail=True)) + body = client.post("/api/review/recommend", json={"abstract": ABSTRACT}).json() + assert body["ok"] is True + assert body["meta"]["vector_ok"] is False + assert body["meta"]["candidate_count"] == 5 + items = _all_items(body) + assert len(items) == 5 + assert all(it["similarity"] is None for it in items) + + +def test_invalid_tier_repaired(client: TestClient, monkeypatch): + """非法 tier 按规则补位，不丢期刊。""" + bad = [ + {"id": "plant-genome", "tier": "超越顶刊", "match_reason": "tier 非法", "risk_note": "无"}, + {"id": "plant-cell", "tier": "challenge", "match_reason": "合法", "risk_note": "无"}, + ] + monkeypatch.setattr(llm, "complete_json", make_fake_llm(bad)) + body = client.post("/api/review/recommend", json={"abstract": ABSTRACT}).json() + assert body["ok"] is True + assert len(_all_items(body)) == 5 + # plant-genome 规则分档 → IF 3.3 ≤ 4 → safe + safe_ids = [it["id"] for it in body["tiers"]["safe"]["items"]] + assert "plant-genome" in safe_ids + assert body["meta"]["repaired"] >= 1 + + +def test_short_abstract_validation(client: TestClient): + body = client.post("/api/review/recommend", json={"abstract": "too short"}).json() + assert body["ok"] is False + assert body["error"]["code"] == "VALIDATION" + assert body["tiers"] is None + + +def test_tier_focus_filters_other_tiers(client: TestClient): + body = client.post( + "/api/review/recommend", + json={"abstract": ABSTRACT, "preferences": {"tier_focus": "safe"}}, + ).json() + assert body["ok"] is True + assert body["tiers"]["challenge"]["items"] == [] + assert body["tiers"]["fit"]["items"] == [] + assert body["tiers"]["safe"]["items"] # 聚焦档保留 + assert body["meta"]["preferences"]["tier_focus"] == "safe" + + +def test_soft_preference_risk_note(client: TestClient): + """软偏好：超周期偏好只加风险提示，不删条目。""" + body = client.post( + "/api/review/recommend", + json={"abstract": ABSTRACT, "preferences": {"max_review_weeks": 5}}, + ).json() + assert body["ok"] is True + items = _all_items(body) + assert len(items) == 5 # 不删 + # The Plant Cell 审稿约 6 周 > 5 周 → 带 ⚠ 提示 + pc = next(it for it in items if it["id"] == "plant-cell") + assert "超出你 5 周" in pc["risk_note"] + + +def test_seed_idempotent(tmp_path): + """种子装载幂等：连跑两次仍恰好 5 行，且不编造 acceptance_rate。""" + + async def work(): + await db_mod.init_db(tmp_path / "seed.db") + try: + await db_mod.ensure_journals_seeded() + await db_mod.ensure_journals_seeded() + return await db_mod.list_journals() + finally: + await db_mod.close_db() + + rows = asyncio.run(work()) + assert len(rows) == 5 + assert {r["id"] for r in rows} == { + "plant-cell", "genome-biology", "plant-journal", + "theoretical-applied-genetics", "plant-genome", + } + assert all(r["acceptance_rate"] is None for r in rows) + assert all(r["scope_desc"] for r in rows) + assert all(r["official_url"] for r in rows) + + +def test_rule_fallback_covers_all_tiers(): + """纯规则分档覆盖三档（不依赖 LLM 与网络）。""" + candidates = [ + {"journal": {"id": "plant-cell", "impact_factor": 11.6, "review_time_weeks": 6.0}, "similarity": None}, + {"journal": {"id": "plant-journal", "impact_factor": 7.2, "review_time_weeks": 5.0}, "similarity": None}, + {"journal": {"id": "plant-genome", "impact_factor": 3.3, "review_time_weeks": 4.0}, "similarity": None}, + ] + rankings = journal_rank.fallback_rankings(candidates) + assert {r.journal_id for r in rankings} == {"plant-cell", "plant-journal", "plant-genome"} + by_id = {r.journal_id: r.tier for r in rankings} + assert by_id["plant-cell"] == "challenge" + assert by_id["plant-journal"] == "fit" + assert by_id["plant-genome"] == "safe" diff --git a/sdk/app/config.py b/sdk/app/config.py index caa9685..8a07ef1 100644 --- a/sdk/app/config.py +++ b/sdk/app/config.py @@ -81,6 +81,11 @@ class Settings(BaseSettings): paper_writer_url: str = "http://127.0.0.1:8097" paper_writer_read_timeout: float = 360.0 # write_paper_section 同步长调用读超时（秒） + # ---- 科研审稿服务（review_service 8100；REVIEW_ENABLED 为假时路由不注册）---- + review_enabled: bool = False + review_service_url: str = "http://127.0.0.1:8100" + review_read_timeout: float = 360.0 # presubmit 上传+LLM 长调用读超时（秒） + # ---- 生产形态：BFF 静态托管 ui/dist（ARCHITECTURE.md §3，M6 挂载）---- frontend_dist_dir: str = "ui/dist" # 生产形态：静态托管 ui/dist diff --git a/sdk/app/main.py b/sdk/app/main.py index d15fac9..1a5fdcd 100644 --- a/sdk/app/main.py +++ b/sdk/app/main.py @@ -26,6 +26,7 @@ from .routers import capabilities, docgen, extraction, figures, health, kg, lega from .routers import bioflow_proxy from .routers import paper from .routers import private_docs +from .routers import review_proxy from .upstream import close_client logging.basicConfig(level=logging.INFO) @@ -62,6 +63,8 @@ def create_app() -> FastAPI: app.include_router(bioflow_proxy.router) # 生信反代（BIOFLOW_ENABLED 开关，FR-BIO-028） if get_settings().paper_writer_enabled: app.include_router(paper.router) # 论文撰写（PAPER_WRITER_ENABLED 开关） + if get_settings().review_enabled: + app.include_router(review_proxy.router) # 科研审稿反代（REVIEW_ENABLED 开关） @app.exception_handler(RequestValidationError) async def _validation_handler(_: Request, exc: RequestValidationError) -> JSONResponse: diff --git a/sdk/app/routers/review_proxy.py b/sdk/app/routers/review_proxy.py new file mode 100644 index 0000000..7adbfc7 --- /dev/null +++ b/sdk/app/routers/review_proxy.py @@ -0,0 +1,85 @@ +"""routers/review_proxy.py —— BFF 科研审稿反代（REVIEW_ENABLED 条件注册）。 + +- GET /api/review/health → 探活 review 服务 /health（离线返回 offline 标记， + 前端据此隐藏审稿入口；模式同 routers/paper.py）。 +- /api/review/{path:path} → 透明反代 review_service（8100）；presubmit 为 + 上传+LLM 长调用，读超时取 settings.review_read_timeout。 +- REVIEW_ENABLED 为假时路由不注册（main.py 按开关 include，三形态部署）。 + +约束：本文件是 BFF 层对审稿模块的唯一新增面（新文件，不改既有路由逻辑）。 +review_service 失败统一 503 UPSTREAM_DOWN 信封（经 app.upstream.envelope）。 +""" +from __future__ import annotations + +import logging +from typing import Any + +import httpx +from fastapi import APIRouter, Request +from fastapi.responses import JSONResponse, Response + +from ..config import get_settings +from ..upstream import envelope + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/review", tags=["review-proxy"]) + +PROXY_CONNECT_TIMEOUT = 5.0 + + +def _service_base() -> str: + return get_settings().review_service_url.rstrip("/") + + +@router.get("/health") +async def review_health() -> dict[str, Any]: + """探活反代 8100 /health；离线返回 offline 标记。""" + s = get_settings() + try: + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.get(f"{s.review_service_url.rstrip('/')}/health") + resp.raise_for_status() + return resp.json() + except Exception: # noqa: BLE001 + return {"ok": False, "status": "offline"} + + +async def _proxy(request: Request, path: str) -> Any: + """httpx 透传反代（非 SSE；presubmit 长读超时）；失败→503 信封。""" + s = get_settings() + base = _service_base() + url = f"{base}/api/review/{path.lstrip('/')}" + if request.url.query: + url = f"{url}?{request.url.query}" + body = await request.body() + headers = {k: v for k, v in request.headers.items() + if k.lower() not in ("host", "content-length", "connection")} + timeout = httpx.Timeout( + connect=PROXY_CONNECT_TIMEOUT, + read=s.review_read_timeout, + write=s.review_read_timeout, + pool=5.0, + ) + async with httpx.AsyncClient(timeout=timeout) as client: + try: + upstream = await client.build_request( + request.method, url, content=body, headers=headers + ) + resp = await client.send(upstream) + except httpx.HTTPError as exc: + logger.warning("review_proxy: 审稿服务不可达 %s: %s", url, exc) + payload, status = envelope( + "UPSTREAM_DOWN", message="审稿服务当前不可用，请稍后重试", + degraded=True, status=503, + ) + return JSONResponse(status_code=status, content=payload) + content = await resp.aread() + + media_type = resp.headers.get("content-type", "") or "application/json" + return Response(status_code=resp.status_code, content=content, media_type=media_type) + + +@router.api_route("/{path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE"]) +async def proxy_all(path: str, request: Request) -> Any: + return await _proxy(request, path)