"""Office 文档转 PDF(host-side LibreOffice)。 Docker 沙盒内不安装 LibreOffice;本工具运行在 backend host,把用户工作区内已有的 DOCX/PPTX/XLSX/ODF 文件交给 host `soffice --headless` 转成 PDF。新写的 Markdown 报告不走这里,统一由沙盒平台渲染器 `rendering/render.py --format pdf` 直出。 """ from __future__ import annotations import shutil import subprocess import tempfile from pathlib import Path from typing import Optional from uuid import uuid4 from tools.base import FileOutOfBounds, Tool from web.pptx_render import SofficeNotFoundError, find_soffice _ALLOWED_SUFFIXES = { ".doc", ".docx", ".ppt", ".pptx", ".xls", ".xlsx", ".odt", ".odp", ".ods", } _DEFAULT_TIMEOUT = 120 def soffice_available() -> bool: """Host 装有 LibreOffice 时才向 agent 注册工具。""" try: find_soffice() except SofficeNotFoundError: return False return True class OfficeToPdfTool(Tool): name = "office_to_pdf" description = ( "Convert an existing Office document in the current user's workspace to PDF using " "LibreOffice on the backend host. Supports DOC/DOCX, PPT/PPTX, XLS/XLSX and ODF. " "Use this only when an Office file already exists; for a new Markdown report, call " "the platform rendering/render.py command with --format pdf directly." ) parameters = { "type": "object", "properties": { "source": { "type": "string", "description": "Existing Office file path, relative to task_dir or under /workspace.", }, "output": { "type": "string", "description": ( "Optional output .pdf path in the user's workspace. " "Defaults to the source path with a .pdf suffix." ), }, }, "required": ["source"], } def execute(self, source: str, output: Optional[str] = None) -> str: try: src = self._resolve_user_file(source) except FileOutOfBounds: return f"[Error] source path out of user workspace: {source}" if not src.is_file(): return f"[Error] source file not found: {source}" if src.suffix.lower() not in _ALLOWED_SUFFIXES: allowed = ", ".join(sorted(_ALLOWED_SUFFIXES)) return f"[Error] unsupported Office format {src.suffix!r}; allowed: {allowed}" raw_out = output.strip() if isinstance(output, str) and output.strip() else "" try: out = self._resolve_user_file(raw_out) if raw_out else src.with_suffix(".pdf") except FileOutOfBounds: return f"[Error] output path out of user workspace: {output}" if out.suffix.lower() != ".pdf": return f"[Error] output must end with .pdf: {output}" try: soffice = find_soffice() except SofficeNotFoundError as e: return f"[Error] LibreOffice unavailable on backend host: {e}" out.parent.mkdir(parents=True, exist_ok=True) try: with tempfile.TemporaryDirectory(prefix="office-pdf-") as tmp: tmp_dir = Path(tmp) profile_uri = (tmp_dir / "profile").resolve().as_uri() cmd = [ soffice, "--headless", "--norestore", "--nolockcheck", "--nodefault", f"-env:UserInstallation={profile_uri}", "--convert-to", "pdf", "--outdir", str(tmp_dir), str(src), ] proc = subprocess.run( cmd, capture_output=True, timeout=_DEFAULT_TIMEOUT, check=False, ) converted = tmp_dir / f"{src.stem}.pdf" if proc.returncode != 0 or not converted.is_file() or converted.stat().st_size == 0: detail = (proc.stderr or proc.stdout or b"").decode("utf-8", "replace")[-500:] return f"[Error] LibreOffice conversion failed (rc={proc.returncode}): {detail}" staged = out.with_name(f".{out.name}.{uuid4().hex}.tmp") try: shutil.copyfile(converted, staged) staged.replace(out) finally: staged.unlink(missing_ok=True) except subprocess.TimeoutExpired: return f"[Error] LibreOffice conversion timed out after {_DEFAULT_TIMEOUT}s: {source}" except OSError as e: return f"[Error] failed to publish PDF: {type(e).__name__}: {e}" return f"[OK] PDF created: {self._display(out)} ({out.stat().st_size} bytes)"