zcbot/web/pptx_render.py

150 lines
5.6 KiB
Python

"""pptx_render.py: 把 .pptx 高保真转成 PDF,供前端在线预览复用现成的 PDF iframe。
设计见 DESIGN §8.3。要点:
- 跑在 backend host(`web/app.py` 进程),**不进执行沙盒**;面向 user_root 下任意 pptx。
- 调 `soffice --headless --convert-to pdf`;**每次独立 `-env:UserInstallation`** 临时 profile,
绕开 LibreOffice 单 profile 锁(否则并发转换互斥),用完即删。
- 转换结果缓存到源 pptx 同目录的隐藏 `.preview/<stem>.<hash>.pdf`;hash 由 mtime+size 派生,
源文件一改 hash 变 → 自然失效。`.preview/` 是 dotdir,`_enumerate_files` 已跳过。
- soffice 缺失 → 抛 `SofficeNotFoundError`(端点回 501);超时/失败 → 抛 `PptxConvertError`。
本模块是**同步**的(subprocess 阻塞);端点侧用 `run_in_executor` 不堵事件循环、用
per-path `asyncio.Lock` 防同一文件并发重复转换。
"""
from __future__ import annotations
import hashlib
import shutil
import subprocess
import tempfile
from pathlib import Path
# Windows 常见安装路径 + Linux 部署路径;都没有再查 PATH。
_SOFFICE_CANDIDATES = [
r"C:\Program Files\LibreOffice\program\soffice.exe",
r"C:\Program Files (x86)\LibreOffice\program\soffice.exe",
"/usr/bin/soffice",
"/usr/bin/libreoffice",
"/opt/libreoffice/program/soffice",
]
_PREVIEW_DIRNAME = ".preview"
_DEFAULT_TIMEOUT = 60 # soffice 冷启 + 转换;大 deck 靠它兜底,超时 kill
class SofficeNotFoundError(RuntimeError):
"""本机/镜像未装 LibreOffice。端点据此回 501 + 提示下载。"""
class PptxConvertError(RuntimeError):
"""soffice 转换失败 / 超时 / 没产出 PDF。端点据此回 500 + 前端回退下载。"""
def find_soffice() -> str:
"""定位 soffice 可执行文件;复用 render_bg.py 的「候选路径 + PATH 兜底」思路。"""
for c in _SOFFICE_CANDIDATES:
if Path(c).exists():
return c
for name in ("soffice", "soffice.exe", "libreoffice"):
p = shutil.which(name)
if p:
return p
raise SofficeNotFoundError(
"服务器未装 LibreOffice(soffice),无法在线预览 pptx。请安装 "
"libreoffice-impress 或下载原文件查看。"
)
def _cache_pdf_path(pptx_path: Path) -> Path:
"""缓存落点:`<pptx 同目录>/.preview/<stem>.<hash>.pdf`。
hash 由源文件 (mtime_ns, size) 派生 —— 源一改 hash 变,旧缓存自然失效(换名),
无需显式比对新鲜度。
"""
st = pptx_path.stat()
sig = f"{st.st_mtime_ns}-{st.st_size}".encode("utf-8")
digest = hashlib.sha1(sig).hexdigest()[:12]
return pptx_path.parent / _PREVIEW_DIRNAME / f"{pptx_path.stem}.{digest}.pdf"
def _prune_stale(cache_dir: Path, stem: str, keep: Path) -> None:
"""删同一 stem 的旧 hash 缓存(源文件已变),只留 keep。兜底磁盘,best-effort。"""
try:
for old in cache_dir.glob(f"{stem}.*.pdf"):
if old != keep:
old.unlink(missing_ok=True)
except OSError:
pass
def _run_soffice(soffice: str, pptx_path: Path, outdir: Path, timeout: int) -> Path:
"""跑一次 soffice 转换,产出 PDF 落到 outdir,返回该 PDF 路径。
`-env:UserInstallation` 指向**独立临时 profile**,避免与宿主 LibreOffice 或其他并发
转换抢单 profile 锁。`--convert-to` 模式默认不执行文档宏(安全边界,见 DESIGN §8.3)。
"""
with tempfile.TemporaryDirectory(prefix="lo-profile-") as profile:
profile_uri = Path(profile).resolve().as_uri()
cmd = [
soffice,
"--headless",
"--norestore",
"--nolockcheck",
"--nodefault",
f"-env:UserInstallation={profile_uri}",
"--convert-to",
"pdf",
"--outdir",
str(outdir),
str(pptx_path),
]
try:
proc = subprocess.run(
cmd,
capture_output=True,
timeout=timeout,
check=False,
)
except subprocess.TimeoutExpired as e:
raise PptxConvertError(
f"pptx 转换超时(>{timeout}s):{pptx_path.name}"
) from e
out_pdf = outdir / f"{pptx_path.stem}.pdf"
if proc.returncode != 0 or not out_pdf.exists():
tail = (proc.stderr or proc.stdout or b"").decode("utf-8", "replace")[-500:]
raise PptxConvertError(
f"soffice 转换失败(rc={proc.returncode}):{pptx_path.name}\n{tail}"
)
return out_pdf
def pptx_to_pdf(pptx_path: Path | str, *, timeout: int = _DEFAULT_TIMEOUT) -> Path:
"""把 pptx 转成 PDF 并返回 PDF 路径(同步、带缓存)。
命中且新鲜的缓存直接返回;否则调 soffice 转换,产物原子落到 `.preview/`。
soffice 缺失 → `SofficeNotFoundError`;转换失败/超时 → `PptxConvertError`。
"""
pptx_path = Path(pptx_path)
if not pptx_path.is_file():
raise PptxConvertError(f"pptx 不存在:{pptx_path}")
cache_pdf = _cache_pdf_path(pptx_path)
if cache_pdf.exists():
return cache_pdf
soffice = find_soffice() # 先解析 soffice,缺失早抛(不白建临时目录)
cache_dir = cache_pdf.parent
cache_dir.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(prefix="pptx-pdf-") as tmp:
out_pdf = _run_soffice(soffice, pptx_path, Path(tmp), timeout)
# 原子落位:同分区 replace;跨分区(临时目录在别处)退回 copy。
try:
out_pdf.replace(cache_pdf)
except OSError:
shutil.copyfile(out_pdf, cache_pdf)
_prune_stale(cache_dir, pptx_path.stem, keep=cache_pdf)
return cache_pdf