""" LibreOffice Excel→PDF 转换壳。 本文件职责:探测 soffice、在「调用方已占用 LibreOffice 单槽」的前提下同步转 PDF。 禁止:在企微回调 / H5 线程调用;禁止本进程同时起两台 soffice(由 SlotPool size=1 保证)。 """ from __future__ import annotations import logging import subprocess from dataclasses import dataclass from pathlib import Path from typing import Optional from agent.config import Settings, get_settings logger = logging.getLogger(__name__) @dataclass class LibreOfficeResult: """转换结果;ok=False 时 error 说明原因,不抛到 Worker 主循环。""" ok: bool pdf_path: str = "" error: str = "" returncode: int = 0 @dataclass class LibreOfficeConverter: """ soffice 无界面转换器。 调用约定:仅从 WorkerRuntime.libreoffice 槽内进入 convert()。 """ settings: Settings @classmethod def from_settings(cls, settings: Optional[Settings] = None) -> "LibreOfficeConverter": return cls(settings=settings or get_settings()) def executable(self) -> Path: return Path(self.settings.libreoffice_path) def available(self) -> bool: """本机是否存在配置的 soffice 可执行文件。""" return self.executable().is_file() def convert_xlsx_to_pdf(self, xlsx_path: str | Path) -> LibreOfficeResult: """ 将 Excel 转为同目录 PDF。 副作用:启动 soffice 子进程(超时见 settings);成功写 PDF 文件。 线程:必须已在 LibreOffice 单槽内。 """ source = Path(xlsx_path).resolve() if not source.is_file(): return LibreOfficeResult(ok=False, error="source_not_found") exe = self.executable() if not exe.is_file(): return LibreOfficeResult(ok=False, error="libreoffice_not_installed") out_dir = source.parent target = source.with_suffix(".pdf") try: completed = subprocess.run( [ str(exe), "--headless", "--convert-to", # calc 导出才认 Excel 的 A4/居中/页宽缩放;裸 pdf 会裁顶、偏左 "pdf:calc_pdf_Export", "--outdir", str(out_dir), str(source), ], capture_output=True, timeout=float(self.settings.libreoffice_timeout_seconds), check=False, ) except (OSError, subprocess.TimeoutExpired) as exc: logger.warning("LibreOffice 执行失败: %s", exc) return LibreOfficeResult(ok=False, error=f"exec_failed:{exc}") if completed.returncode != 0 or not target.is_file(): stderr = (completed.stderr or b"")[:400] logger.warning( "LibreOffice 未生成 PDF rc=%s stderr=%s", completed.returncode, stderr, ) return LibreOfficeResult( ok=False, error="pdf_not_generated", returncode=int(completed.returncode), ) head = target.read_bytes()[:5] if head != b"%PDF-": target.unlink(missing_ok=True) return LibreOfficeResult(ok=False, error="pdf_header_invalid") logger.info("LibreOffice 转 PDF 成功 path=%s", target) return LibreOfficeResult(ok=True, pdf_path=str(target), returncode=0)