222 lines
8.6 KiB
Python
222 lines
8.6 KiB
Python
"""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 os
|
||
import shutil
|
||
import subprocess
|
||
import tempfile
|
||
from functools import lru_cache
|
||
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
|
||
|
||
|
||
_FAMILY_SUFFIXES = {
|
||
"Writer": frozenset({".doc", ".docx", ".odt"}),
|
||
"Calc": frozenset({".xls", ".xlsx", ".ods"}),
|
||
"Impress": frozenset({".ppt", ".pptx", ".odp"}),
|
||
}
|
||
_KNOWN_SUFFIXES = frozenset().union(*_FAMILY_SUFFIXES.values())
|
||
_DEBIAN_PACKAGES = {
|
||
"Writer": ("libreoffice-writer", "libreoffice-writer-nogui"),
|
||
"Calc": ("libreoffice-calc", "libreoffice-calc-nogui"),
|
||
"Impress": ("libreoffice-impress", "libreoffice-impress-nogui"),
|
||
}
|
||
_DEFAULT_TIMEOUT = 120
|
||
|
||
|
||
def _dpkg_package_installed(package: str) -> bool:
|
||
try:
|
||
proc = subprocess.run(
|
||
["dpkg-query", "-W", "-f=${Status}", package],
|
||
capture_output=True,
|
||
timeout=5,
|
||
check=False,
|
||
text=True,
|
||
)
|
||
except (OSError, subprocess.TimeoutExpired):
|
||
return False
|
||
return proc.returncode == 0 and "install ok installed" in proc.stdout
|
||
|
||
|
||
def _debian_office_families(soffice: str) -> Optional[frozenset[str]]:
|
||
"""标准 Debian/Ubuntu 包安装时返回已安装组件；其他安装形态返回 None。"""
|
||
if os.name != "posix" or shutil.which("dpkg-query") is None:
|
||
return None
|
||
try:
|
||
resolved = Path(soffice).resolve()
|
||
except OSError:
|
||
return None
|
||
if not resolved.is_relative_to("/usr"):
|
||
return None
|
||
return frozenset(
|
||
family
|
||
for family, packages in _DEBIAN_PACKAGES.items()
|
||
if any(_dpkg_package_installed(package) for package in packages)
|
||
)
|
||
|
||
|
||
@lru_cache(maxsize=1)
|
||
def office_supported_suffixes() -> frozenset[str]:
|
||
"""返回 host 当前 LibreOffice 组件实际支持的输入后缀。"""
|
||
try:
|
||
soffice = find_soffice()
|
||
except SofficeNotFoundError:
|
||
return frozenset()
|
||
families = _debian_office_families(soffice)
|
||
if families is None:
|
||
# Windows/macOS/自带 tar 包通常是完整套件；无法可靠拆包探测时按完整安装处理。
|
||
return _KNOWN_SUFFIXES
|
||
return frozenset().union(*(_FAMILY_SUFFIXES[family] for family in families))
|
||
|
||
|
||
def soffice_available() -> bool:
|
||
"""Host 至少有一个可用 LibreOffice 组件时才向 agent 注册工具。"""
|
||
return bool(office_supported_suffixes())
|
||
|
||
|
||
class OfficeToPdfTool(Tool):
|
||
name = "office_to_pdf"
|
||
description = "Convert an existing Office document to PDF using backend host LibreOffice."
|
||
parameters = {
|
||
"type": "object",
|
||
"properties": {
|
||
"source": {
|
||
"type": "string",
|
||
"description": (
|
||
"Existing Office file path: task-relative, user-root-relative as shown in chat, "
|
||
"or an absolute path 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 __init__(
|
||
self,
|
||
base_dir: Optional[Path] = None,
|
||
user_root: Optional[Path] = None,
|
||
) -> None:
|
||
super().__init__(base_dir=base_dir, user_root=user_root)
|
||
self.supported_suffixes = office_supported_suffixes()
|
||
supported = ", ".join(suffix.lstrip(".").upper() for suffix in sorted(self.supported_suffixes))
|
||
self.description = (
|
||
"Convert an existing Office document in the current user's workspace to PDF using "
|
||
f"LibreOffice on the backend host. Formats available on this host: {supported or 'none'}. "
|
||
"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."
|
||
)
|
||
|
||
def _resolve_office_path(self, raw: str, *, existing: bool) -> Path:
|
||
"""兼容 task 相对路径与界面展示的 `<working_dir>/...` 用户根相对路径。"""
|
||
p = Path(raw)
|
||
if p.is_absolute() or raw == "/workspace" or raw.startswith("/workspace/"):
|
||
return self._resolve_user_file(raw)
|
||
|
||
base_candidate = self._resolve_user_file(raw)
|
||
if existing and base_candidate.exists():
|
||
return base_candidate
|
||
if self.user_root is None:
|
||
return base_candidate
|
||
|
||
root = self.user_root.resolve()
|
||
root_candidate = (root / p).resolve()
|
||
try:
|
||
root_candidate.relative_to(root)
|
||
task_rel = self.base_dir.resolve().relative_to(root)
|
||
except ValueError:
|
||
return base_candidate
|
||
|
||
has_task_prefix = p.parts[:len(task_rel.parts)] == task_rel.parts
|
||
if (existing and root_candidate.exists()) or has_task_prefix:
|
||
return root_candidate
|
||
return base_candidate
|
||
|
||
def execute(self, source: str, output: Optional[str] = None) -> str:
|
||
try:
|
||
src = self._resolve_office_path(source, existing=True)
|
||
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}"
|
||
suffix = src.suffix.lower()
|
||
if suffix not in _KNOWN_SUFFIXES:
|
||
known = ", ".join(sorted(_KNOWN_SUFFIXES))
|
||
return f"[Error] unsupported Office format {src.suffix!r}; known formats: {known}"
|
||
if suffix not in self.supported_suffixes:
|
||
family = next(name for name, suffixes in _FAMILY_SUFFIXES.items() if suffix in suffixes)
|
||
return (
|
||
f"[Error] backend host LibreOffice lacks the {family} component required for "
|
||
f"{suffix}; ask the administrator to install that component"
|
||
)
|
||
|
||
raw_out = output.strip() if isinstance(output, str) and output.strip() else ""
|
||
try:
|
||
out = self._resolve_office_path(raw_out, existing=False) 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)"
|