140 lines
5.6 KiB
Python
140 lines
5.6 KiB
Python
"""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]
|