From 1c73320f1db90b42552add07258612f23c496fc9 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Tue, 11 Aug 2026 22:50:47 +0000 Subject: [PATCH] Standalone mediapreview project --- .gitignore | 6 + README.md | 58 +++ mediapreview/__init__.py | 34 ++ mediapreview/__main__.py | 10 + mediapreview/backends/__init__.py | 57 +++ mediapreview/backends/image.py | 188 +++++++++ mediapreview/backends/pdf.py | 48 +++ mediapreview/backends/video.py | 154 +++++++ mediapreview/cache.py | 44 ++ mediapreview/docker/Dockerfile | 70 ++++ mediapreview/docker/converter-wrapper.sh | 19 + mediapreview/docker/entrypoint.sh | 8 + mediapreview/formats.py | 94 +++++ mediapreview/office.py | 422 ++++++++++++++++++++ mediapreview/pool.py | 488 +++++++++++++++++++++++ mediapreview/protocol.py | 21 + mediapreview/util/__init__.py | 1 + mediapreview/util/logformat.py | 33 ++ mediapreview/worker.py | 175 ++++++++ pyproject.toml | 98 +++++ tests/test_pool.py | 175 ++++++++ 21 files changed, 2203 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 mediapreview/__init__.py create mode 100644 mediapreview/__main__.py create mode 100644 mediapreview/backends/__init__.py create mode 100644 mediapreview/backends/image.py create mode 100644 mediapreview/backends/pdf.py create mode 100644 mediapreview/backends/video.py create mode 100644 mediapreview/cache.py create mode 100644 mediapreview/docker/Dockerfile create mode 100644 mediapreview/docker/converter-wrapper.sh create mode 100644 mediapreview/docker/entrypoint.sh create mode 100644 mediapreview/formats.py create mode 100644 mediapreview/office.py create mode 100644 mediapreview/pool.py create mode 100644 mediapreview/protocol.py create mode 100644 mediapreview/util/__init__.py create mode 100644 mediapreview/util/logformat.py create mode 100644 mediapreview/worker.py create mode 100644 pyproject.toml create mode 100644 tests/test_pool.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..463e1fb --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +.* +*.lock +!.gitignore +__pycache__/ +*.egg-info/ +/dist diff --git a/README.md b/README.md new file mode 100644 index 0000000..d47789a --- /dev/null +++ b/README.md @@ -0,0 +1,58 @@ +# mediapreview + +Low-level media preview converters plus an optional persistent worker pool +framework. All converters produce AVIF output. + +## Layout + +| Module | Purpose | +|--------|---------| +| `mediapreview.backends.image` | Image → AVIF via pyvips (ffmpeg for HEIC/HEIF/AVIF) | +| `mediapreview.backends.video` | Video frame → AVIF via PyAV (HDR preserved) | +| `mediapreview.backends.pdf` | PDF/XPS/EPUB page → AVIF via PyMuPDF + pyvips | +| `mediapreview.backends` | `dispatch()` — pick a backend by path/mimetype | +| `mediapreview.formats` | Suffix sets, previewability and priority classification | +| `mediapreview.office` | OnlyOffice Document Server client + Docker bootstrap | +| `mediapreview.docker/` | Patched OnlyOffice image build context (ships in the wheel) | +| `mediapreview.pool` | Async persistent subprocess worker pool (optional) | +| `mediapreview.worker` | Worker subprocess entry point (framed stdin/stdout protocol) | +| `mediapreview.protocol` | msgspec wire structs (`PreviewRequest` / `PreviewResponse`) | +| `mediapreview.cache` | Thread-safe LRU cache for preview responses | + +## Extras + +```bash +pip install mediapreview # image converter only (pyvips) +pip install mediapreview[pdf] # + PDF/XPS/EPUB (pymupdf) +pip install mediapreview[video] # + video (av, numpy) +pip install mediapreview[office] # + OnlyOffice client (httpx, pyjwt) +pip install mediapreview[worker] # + worker pool (blake3, tracerite) +pip install mediapreview[standard] # everything +``` + +## Usage + +Low-level, in-process: + +```python +from mediapreview import dispatch + +avif_bytes, resp = dispatch(path, quality=60, maxsize=512, maxzoom=2.0) +``` + +Worker pool (isolates heavy imports and native crashes from the async loop): + +```python +from mediapreview.pool import start_preview_workers, shutdown_preview_workers +``` + +## OnlyOffice Docker bootstrap + +A patched OnlyOffice image (configurable converter worker count) ships as +package data and can be built/started with: + +```python +from mediapreview.office import setup_docker + +setup_docker() # builds + runs "onlyoffice-mediapreview" on port 8988 +``` diff --git a/mediapreview/__init__.py b/mediapreview/__init__.py new file mode 100644 index 0000000..62dabd0 --- /dev/null +++ b/mediapreview/__init__.py @@ -0,0 +1,34 @@ +"""Media preview framework — low-level converters and a worker pool. + +Optional functionality is gated by extras: + + pip install mediapreview[standard] # all preview backends + worker pool + pip install mediapreview[worker] # persistent subprocess worker pool + pip install mediapreview[pdf] # PDF previews + pip install mediapreview[video] # video previews + pip install mediapreview[office] # OnlyOffice document conversion +""" + +from mediapreview.backends import ( + dispatch, + process_image, + process_image_buffer, + process_pdf, + process_video, +) +from mediapreview.cache import CachedPreview, PreviewCache +from mediapreview.formats import is_previewable_path +from mediapreview.protocol import PreviewRequest, PreviewResponse + +__all__ = [ + "CachedPreview", + "PreviewCache", + "PreviewRequest", + "PreviewResponse", + "dispatch", + "is_previewable_path", + "process_image", + "process_image_buffer", + "process_pdf", + "process_video", +] diff --git a/mediapreview/__main__.py b/mediapreview/__main__.py new file mode 100644 index 0000000..faa12f9 --- /dev/null +++ b/mediapreview/__main__.py @@ -0,0 +1,10 @@ +"""CLI entry point: delegates to the preview worker subprocess. + +A richer CLI for one-shot preview generation and pool management can be +added later. +""" + +from mediapreview.worker import main + +if __name__ == "__main__": + main() diff --git a/mediapreview/backends/__init__.py b/mediapreview/backends/__init__.py new file mode 100644 index 0000000..e2b86a6 --- /dev/null +++ b/mediapreview/backends/__init__.py @@ -0,0 +1,57 @@ +"""Low-level preview conversion backends. + +Each backend module exposes a `process_*` function taking a path (or buffer) +plus quality/size parameters and returning `(avif_bytes, PreviewResponse)`. +`dispatch` picks the right backend for a path. +""" + +import logging +import mimetypes + +from mediapreview.backends.image import ( + process_image, + process_image_buffer, + process_image_pyvips, +) +from mediapreview.backends.pdf import process_pdf +from mediapreview.backends.video import process_video +from mediapreview.formats import DOC_PREVIEW_SUFFIXES +from mediapreview.protocol import PreviewResponse + +__all__ = [ + "dispatch", + "process_image", + "process_image_buffer", + "process_image_pyvips", + "process_pdf", + "process_video", +] + +logger = logging.getLogger(__name__) + + +def dispatch(path, quality, maxsize, maxzoom, data=None): + backend = "unknown" + try: + if data: + backend = "pyvips" + return process_image_buffer( + data, quality=quality, maxsize=maxsize, maxzoom=maxzoom + ) + suffix = path.suffix.lower() + if suffix in DOC_PREVIEW_SUFFIXES: + backend = "pdf" + return process_pdf(path, quality=quality, maxsize=maxsize, maxzoom=maxzoom) + mime_type, _ = mimetypes.guess_type(path.name) + if mime_type and mime_type.startswith("video/"): + backend = "video" + return process_video(path, quality=quality, maxsize=maxsize) + if mime_type and mime_type.startswith("image/"): + backend = "pyvips" + return process_image(path, quality=quality, maxsize=maxsize) + except ValueError as e: + return None, PreviewResponse(ok=False, backend=backend, error=str(e)) + except Exception as e: + logger.exception("Preview dispatch failed for %s", path) + return None, PreviewResponse(ok=False, backend=backend, error=str(e)) + return None, PreviewResponse(ok=False, backend=backend, error="preview unsupported") diff --git a/mediapreview/backends/image.py b/mediapreview/backends/image.py new file mode 100644 index 0000000..e693823 --- /dev/null +++ b/mediapreview/backends/image.py @@ -0,0 +1,188 @@ +"""Image preview conversion via pyvips (and ffmpeg for HEIC/HEIF/AVIF).""" + +import shlex +import subprocess +import tempfile +from pathlib import Path +from time import perf_counter + +import pyvips + +from mediapreview.protocol import PreviewResponse + +AVIF_FAST_EFFORT = 0 + + +def process_image(path, *, maxsize, quality): + return process_image_pyvips(path, maxsize=maxsize, quality=quality) + + +def _get_image_dimensions(path: Path) -> tuple[int, int] | None: + """Probe image dimensions. + + pyvips can read the header of most formats (including HEIC) without + fully decoding the image. + """ + try: + img = pyvips.Image.new_from_file(str(path)) + img = img.autorot() + except pyvips.error.Error: + return None + else: + return img.width, img.height + + +def _image_via_ffmpeg(path: Path, maxsize: int, quality: int) -> bytes: + """Convert any image to AVIF using ffmpeg CLI. + + ffmpeg handles HEIC tile assembly, HDR metadata and ICC profile embedding + automatically. Note: -vf cannot be used here — HEIC tile assembly feeds + the stream from a complex filtergraph, which conflicts with simple -vf + filtering; scaling must use the -s output option instead. + """ + dims = _get_image_dimensions(path) + crf = int(63 * (1 - quality / 100) ** 2) + with tempfile.NamedTemporaryFile(suffix=".avif", delete=False) as tmp_f: + tmp_path = tmp_f.name + cmd = [ + "ffmpeg", + # Keep error messages, drop the banner/config/stream-mapping spam. + "-hide_banner", + "-loglevel", + "error", + "-nostats", + # No interactive keyboard prompts ("Press [q] to stop ..."). + "-nostdin", + "-y", + "-i", + str(path), + "-frames:v", + "1", + "-c:v", + "av1", + "-crf", + str(crf), + "-cpu-used", + "8", + tmp_path, + ] + if dims is not None: + w, h = dims + if max(w, h) > maxsize: + scale = min(maxsize / w, maxsize / h) + new_w = int(w * scale) + new_h = int(h * scale) + # insert -s right after the input file + input_index = cmd.index(str(path)) + 1 + cmd.insert(input_index, "-s") + cmd.insert(input_index + 1, f"{new_w}x{new_h}") + try: + try: + # stdin=DEVNULL is critical: ffmpeg must not inherit the worker's + # stdin, which carries the framed request protocol. An inherited + # stdin lets ffmpeg eat protocol bytes and, if the worker is + # killed mid-conversion, keeps the orphaned ffmpeg holding the + # pipe open so the parent's proc.wait() hangs forever. + subprocess.run( # noqa: S603 + cmd, + capture_output=True, + check=True, + shell=False, + stdin=subprocess.DEVNULL, + ) + except subprocess.CalledProcessError as e: + shell_cmd = shlex.join(cmd) + stderr = (e.stderr or b"").decode(errors="replace").strip() + if stderr: + raise RuntimeError( + f"ffmpeg failed (exit {e.returncode}):\n{shell_cmd}\n{stderr}" + ) from e + raise RuntimeError( + f"ffmpeg failed (exit {e.returncode}):\n{shell_cmd}" + ) from e + with Path(tmp_path).open("rb") as f: + return f.read() + finally: + Path(tmp_path).unlink(missing_ok=True) + + +def process_image_pyvips(path, *, maxsize, quality): + t_start = perf_counter() + suffix = path.suffix.lower() + + # HEIC/HEIF/AVIF: ffmpeg handles tile assembly and HDR correctly; + # skip pyvips entirely (pyvips drops CICP colour metadata, turning + # HDR sources into washed-out SDR previews). + if suffix in (".heic", ".heif", ".avif"): + dims = _get_image_dimensions(path) + width, height = dims or (None, None) + ret = _image_via_ffmpeg(path, maxsize, quality) + t_end = perf_counter() + return ret, PreviewResponse( + ok=True, + mime="image/avif", + backend="ffmpeg", + timings=[round((t_end - t_start) * 1000, 1)], + width=width, + height=height, + ) + + # Other image formats: pyvips only. ffmpeg is not a useful fallback + # here — when pyvips cannot decode a file, ffmpeg's image decoders + # cannot either, and their failure output is far noisier. + try: + img = pyvips.Image.new_from_file(str(path), access="sequential") + if img.get_typeof("orientation") and img.get("orientation") != 1: + # autorot's rot90 reads pixels out of order, which sequential + # access cannot do — reopen with random access when rotating. + img = pyvips.Image.new_from_file(str(path)).autorot() + orig_w, orig_h = img.width, img.height + scale = min(maxsize / img.width, maxsize / img.height, 1.0) + if scale < 1.0: + img = img.resize(scale) + ret = img.write_to_buffer( + ".avif", + Q=quality, + effort=AVIF_FAST_EFFORT, + keep="none", + ) + except pyvips.error.Error as e: + raise ValueError(f"cannot decode image: {e}") from e + backend = "pyvips" + t_end = perf_counter() + + return ret, PreviewResponse( + ok=True, + mime="image/avif", + backend=backend, + timings=[round((t_end - t_start) * 1000, 1)], + width=orig_w, + height=orig_h, + ) + + +def process_image_buffer(data: bytes, *, quality, maxsize, maxzoom): + _ = maxzoom + t_start = perf_counter() + img = pyvips.Image.new_from_buffer(data, "") + img = img.autorot() + orig_w, orig_h = img.width, img.height + scale = min(maxsize / img.width, maxsize / img.height, 1.0) + if scale < 1.0: + img = img.resize(scale) + ret = img.write_to_buffer( + ".avif", + Q=quality, + effort=AVIF_FAST_EFFORT, + keep="none", + ) + t_end = perf_counter() + + return ret, PreviewResponse( + ok=True, + mime="image/avif", + backend="pyvips", + timings=[round((t_end - t_start) * 1000, 1)], + width=orig_w, + height=orig_h, + ) diff --git a/mediapreview/backends/pdf.py b/mediapreview/backends/pdf.py new file mode 100644 index 0000000..85300e3 --- /dev/null +++ b/mediapreview/backends/pdf.py @@ -0,0 +1,48 @@ +"""PDF/XPS/EPUB preview conversion via PyMuPDF + pyvips.""" + +from time import perf_counter + +import pyvips + +from mediapreview.backends.image import AVIF_FAST_EFFORT +from mediapreview.protocol import PreviewResponse + +try: + import pymupdf +except ImportError: # pragma: no cover - optional pdf extra + pymupdf = None + + +def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0): + if pymupdf is None: + raise ImportError( + "PDF previews require the 'pdf' extra: pip install mediapreview[pdf]" + ) + t_load_start = perf_counter() + with pymupdf.open(path) as pdf: + page = pdf.load_page(page_number) + w, h = page.rect[2:4] + zoom = min(maxsize / w, maxsize / h, maxzoom) + mat = pymupdf.Matrix(zoom, zoom) + pix = page.get_pixmap(matrix=mat) + t_load_end = perf_counter() + + t_save_start = perf_counter() + img = pyvips.Image.new_from_memory( + pix.samples_mv, pix.width, pix.height, pix.n, "uchar" + ) + ret = img.write_to_buffer(".avif", Q=quality, effort=AVIF_FAST_EFFORT, keep="none") + backend = "pdf+pyvips" + t_save_end = perf_counter() + + return ret, PreviewResponse( + ok=True, + mime="image/avif", + backend=backend, + timings=[ + round((t_load_end - t_load_start) * 1000, 1), + round((t_save_end - t_save_start) * 1000, 1), + ], + width=round(w), + height=round(h), + ) diff --git a/mediapreview/backends/video.py b/mediapreview/backends/video.py new file mode 100644 index 0000000..60521b7 --- /dev/null +++ b/mediapreview/backends/video.py @@ -0,0 +1,154 @@ +"""Video preview conversion via PyAV (AV1/AVIF output, HDR preserved).""" + +import gc +import io +import os + +# Silence the SVT-AV1 encoder's stderr spam, set log level ERROR +os.environ.setdefault("SVT_LOG", "1") + +from time import perf_counter + +from mediapreview.protocol import PreviewResponse + +try: + import av + import numpy as np +except ImportError: # pragma: no cover - optional video extra + av = None + np = None + + +def _rotate_frame_yuv(frame, k): + """Rotate a planar YUV420 frame by k*90° counter-clockwise, keeping its format. + + Rotating each plane independently preserves the pixel format (including + 10-bit HDR formats like yuv420p10le, which PyAV exposes as uint16 planes) + and the source colorspace. + """ + if av is None or np is None: + raise ImportError( + "Video previews require the 'video' extra: pip install mediapreview[video]" + ) + fmt = frame.format + w, h = frame.width, frame.height + if ( + not fmt.is_planar + or fmt.chroma_width(w) * 2 != w + or fmt.chroma_height(h) * 2 != h + ): + raise ValueError(f"unsupported format for YUV rotation: {fmt.name}") + planes = frame.to_ndarray() + y, u, v = ( + planes[:h], + planes[h : h + h // 4].reshape(h // 2, w // 2), + planes[h + h // 4 :].reshape(h // 2, w // 2), + ) + planes = np.hstack( + [p.flat for p in (np.rot90(y, k), np.rot90(u, k), np.rot90(v, k))] + ) + new_width = w if k % 2 == 0 else h + return av.VideoFrame.from_ndarray(planes.reshape(-1, new_width), format=fmt.name) + + +def process_video(path, *, maxsize, quality): + if av is None or np is None: + raise ImportError( + "Video previews require the 'video' extra: pip install mediapreview[video]" + ) + frame = None + imgdata = io.BytesIO() + istream = ostream = icc = occ = frame = None + t_load_start = perf_counter() + # Initialize to avoid "possibly unbound" in static analysis when exceptions occur + t_load_end = t_load_start + t_save_start = t_load_start + t_save_end = t_load_start + with ( + av.open( + str(path), + options={ + "analyzeduration": "1000000", # 1 second (in microseconds) + "fflags": "fastseek", + }, + ) as icontainer, + av.open(imgdata, "w", format="avif") as ocontainer, + ): + istream = icontainer.streams.video[0] + istream.codec_context.skip_frame = "NONKEY" + icontainer.seek((icontainer.duration or 0) // 8) + for frame in icontainer.decode(istream): + if frame.dts is not None: + break + else: + raise RuntimeError("No frames found in video") + + # Resize frame to thumbnail size + # Capture display dimensions before resize (accounting for rotation) + disp_w = frame.width + disp_h = frame.height + if frame.rotation in (90, 270): + disp_w, disp_h = disp_h, disp_w + if frame.width > maxsize or frame.height > maxsize: + scale_factor = min(maxsize / frame.width, maxsize / frame.height) + new_width = int(frame.width * scale_factor) + new_height = int(frame.height * scale_factor) + frame = frame.reformat(width=new_width, height=new_height) + + # Apply display-matrix rotation if present + if frame.rotation: + # frame.rotation indicates clockwise rotation needed to display correctly + k = (frame.rotation // 90) % 4 # Convert to counter-clockwise rotations + frame = _rotate_frame_yuv(frame, k) + + # libsvtav1 rejects full-range JPEG-style YUV pixel formats such as + # yuvj420p, so normalize them before opening the encoder. + if frame.format.name.startswith("yuvj"): + frame = frame.reformat(format="yuv420p") + t_load_end = perf_counter() + + t_save_start = perf_counter() + crf = str(int(63 * (1 - quality / 100) ** 2)) # Closely matching PIL quality-% + ostream = ocontainer.add_stream( + "av1", + options={ + "crf": crf, + "usage": "realtime", + "cpu-used": "8", + "threads": "1", + }, + ) + if not isinstance(ostream, av.VideoStream): + raise TypeError("failed to initialize AV1 video stream") + ostream.width = frame.width + ostream.height = frame.height + ostream.pix_fmt = frame.format.name + icc = istream.codec_context + occ = ostream.codec_context + + # Copy HDR metadata from input video stream + occ.color_primaries = icc.color_primaries + occ.color_trc = icc.color_trc + occ.colorspace = icc.colorspace + occ.color_range = icc.color_range + + ocontainer.mux(ostream.encode(frame)) + ocontainer.mux(ostream.encode(None)) # Flush the stream + t_save_end = perf_counter() + + # Capture result before cleanup + ret = imgdata.getvalue() + resp = PreviewResponse( + ok=True, + mime="image/avif", + backend="video", + timings=[ + round((t_load_end - t_load_start) * 1000, 1), + round((t_save_end - t_save_start) * 1000, 1), + ], + width=disp_w, + height=disp_h, + ) + del imgdata, istream, ostream, icc, occ, frame + gc.collect() + return ret, resp diff --git a/mediapreview/cache.py b/mediapreview/cache.py new file mode 100644 index 0000000..76e5826 --- /dev/null +++ b/mediapreview/cache.py @@ -0,0 +1,44 @@ +"""In-memory LRU cache for preview responses.""" + +from collections import OrderedDict +from dataclasses import dataclass +from threading import Lock + + +@dataclass(slots=True) +class CachedPreview: + """Cached preview with headers and body.""" + + headers: dict[str, str] + body: bytes + + +class PreviewCache: + """Thread-safe LRU cache for preview responses.""" + + def __init__(self, capacity: int = 500): + self.capacity = capacity + self._cache: OrderedDict[str, CachedPreview] = OrderedDict() + self._lock = Lock() + + def get(self, key: str) -> CachedPreview | None: + """Get cached preview, moving it to end (most recently used).""" + with self._lock: + if key in self._cache: + self._cache.move_to_end(key) + return self._cache[key] + return None + + def set(self, key: str, value: CachedPreview) -> None: + """Cache preview, evicting oldest if at capacity.""" + with self._lock: + if key in self._cache: + self._cache.move_to_end(key) + else: + if len(self._cache) >= self.capacity: + self._cache.popitem(last=False) + self._cache[key] = value + + def __len__(self) -> int: + with self._lock: + return len(self._cache) diff --git a/mediapreview/docker/Dockerfile b/mediapreview/docker/Dockerfile new file mode 100644 index 0000000..ce3b327 --- /dev/null +++ b/mediapreview/docker/Dockerfile @@ -0,0 +1,70 @@ +# Patched OnlyOffice Document Server with configurable converter worker count. +# +# The Community Edition hardcodes the document converter to 1 worker, +# which creates a severe bottleneck under concurrent load. +# This image patches the open-source license.js to spawn a configurable +# number of converter workers (default 8). +# +# Build: +# docker build -t onlyoffice-mediapreview mediapreview/mediapreview/docker +# +# Run: +# docker run -d -p 8988:80 \ +# -e WORKERS=16 \ +# -e JWT_SECRET=your-strong-secret \ +# --name onlyoffice-mediapreview onlyoffice-mediapreview +# +# JWT: +# Set JWT_SECRET to the same value you pass to Cista as ONLYOFFICE_JWT_SECRET. +# OnlyOffice will enable token validation automatically. +# +# The ONLYOFFICE_VERSION build arg lets you target a specific release. + +ARG ONLYOFFICE_VERSION=9.3.1 + +FROM onlyoffice/documentserver:${ONLYOFFICE_VERSION} + +# Prevent interactive apt prompts +ENV DEBIAN_FRONTEND=noninteractive + +# Install Node.js, npm, and git so we can run the FileConverter from source. +RUN apt-get update -qq && \ + apt-get install -y -qq --no-install-recommends \ + nodejs \ + npm \ + git \ + ca-certificates && \ + rm -rf /var/lib/apt/lists/* + +# Clone the open-source server components (shallow, ~15 MB). +# The master branch is used because the Linux/web tags are not published +# in the server repo; the license.js file has been stable for years. +RUN git clone --depth 1 https://github.com/ONLYOFFICE/server.git /opt/oo-server + +# Patch license.js so the converter worker count is read from an env var +# instead of being hardcoded to 1. +RUN sed -i \ + 's/count: 1,/count: parseInt(process.env.WORKERS, 10) || 8,/' \ + /opt/oo-server/Common/sources/license.js + +# Install npm dependencies for the modules the FileConverter touches. +# DocService deps are also needed because converter.js pulls in baseConnector. +RUN cd /opt/oo-server/Common && npm ci --no-audit --no-fund +RUN cd /opt/oo-server/FileConverter && npm ci --no-audit --no-fund +RUN cd /opt/oo-server/DocService && npm ci --no-audit --no-fund + +# Back up the compiled pkg binary and replace it with our wrapper. +RUN mv /var/www/onlyoffice/documentserver/server/FileConverter/converter \ + /var/www/onlyoffice/documentserver/server/FileConverter/converter.orig + +COPY converter-wrapper.sh /var/www/onlyoffice/documentserver/server/FileConverter/converter +RUN chmod +x /var/www/onlyoffice/documentserver/server/FileConverter/converter + +# Default worker count (override at runtime with -e WORKERS=16). +ENV WORKERS=8 + +# Use our custom entrypoint to persist the env var to a file that the +# non-root converter process (user=ds) can read. +COPY entrypoint.sh /app/ds/run-document-server-patched.sh +RUN chmod +x /app/ds/run-document-server-patched.sh +ENTRYPOINT ["/app/ds/run-document-server-patched.sh"] diff --git a/mediapreview/docker/converter-wrapper.sh b/mediapreview/docker/converter-wrapper.sh new file mode 100644 index 0000000..bd79138 --- /dev/null +++ b/mediapreview/docker/converter-wrapper.sh @@ -0,0 +1,19 @@ +#!/bin/bash +# Wrapper that runs the OnlyOffice FileConverter from patched Node.js source. +# Replaces the compiled pkg binary shipped with the Community Edition. + +# The env var is not passed through supervisor to the 'ds' user, so we read +# it from a file written by the custom entrypoint. +if [ -z "${WORKERS}" ] && [ -r /tmp/oo-converter-workers.txt ]; then + export WORKERS=$(cat /tmp/oo-converter-workers.txt) +fi + +cd /opt/oo-server/FileConverter || exit 1 + +export NODE_ENV=production-linux +export NODE_CONFIG_DIR=/etc/onlyoffice/documentserver +export NODE_DISABLE_COLORS=1 +export APPLICATION_NAME=onlyoffice +export LD_LIBRARY_PATH=/var/www/onlyoffice/documentserver/server/FileConverter/bin + +exec node sources/convertermaster.js "$@" diff --git a/mediapreview/docker/entrypoint.sh b/mediapreview/docker/entrypoint.sh new file mode 100644 index 0000000..8eea7d1 --- /dev/null +++ b/mediapreview/docker/entrypoint.sh @@ -0,0 +1,8 @@ +#!/bin/bash +# Custom entrypoint that persists WORKERS to a file readable by +# the non-root user that supervisor uses to run the converter. + +echo "${WORKERS:-8}" > /tmp/oo-converter-workers.txt +chmod 644 /tmp/oo-converter-workers.txt + +exec /app/ds/run-document-server.sh "$@" diff --git a/mediapreview/formats.py b/mediapreview/formats.py new file mode 100644 index 0000000..b188158 --- /dev/null +++ b/mediapreview/formats.py @@ -0,0 +1,94 @@ +"""File-type classification for preview generation. + +Suffix sets and helpers that decide which backend handles a given path, +how previewable it is, and how jobs should be prioritised. +""" + +import mimetypes +from pathlib import Path + +DOC_PREVIEW_SUFFIXES = {".pdf", ".xps", ".epub", ".mobi"} + +OFFICE_PREVIEW_SUFFIXES = { + ".doc", + ".dot", + ".docx", + ".docm", + ".dotx", + ".dotm", + ".rtf", + ".odt", + ".ott", + ".txt", + ".md", + ".mhtml", + ".mht", + ".html", + ".htm", + ".xml", + ".wps", + ".wri", + # Spreadsheets + ".xls", + ".xlsx", + ".xlsm", + ".xlsb", + ".xltx", + ".xltm", + ".ods", + ".ots", + ".csv", + # Presentations + ".ppt", + ".pptx", + ".pptm", + ".pps", + ".ppsx", + ".pot", + ".potx", + ".odp", + ".otp", +} + + +def is_previewable_path(path) -> bool: + suffix = path.suffix.lower() + if suffix in DOC_PREVIEW_SUFFIXES or suffix in OFFICE_PREVIEW_SUFFIXES: + return True + mime_type, _ = mimetypes.guess_type(path.name) + if not mime_type: + return False + return mime_type.startswith(("image/", "video/")) + + +def preview_job_priority(path) -> int: + """Return priority for preview job (lower=higher priority). + + Priority order: images (0) < video (1) < PDF (2) < office (3) < unknown (4) + """ + suffix = path.suffix.lower() + if suffix in DOC_PREVIEW_SUFFIXES: + return 2 + if suffix in OFFICE_PREVIEW_SUFFIXES: + return 3 + mime_type, _ = mimetypes.guess_type(path.name) + if mime_type and mime_type.startswith("image/"): + return 0 + if mime_type and mime_type.startswith("video/"): + return 1 + return 4 + + +def expected_backend(path: Path) -> str: + """Best-effort backend label used for timeout/access logging.""" + suffix = path.suffix.lower() + if suffix in OFFICE_PREVIEW_SUFFIXES: + return "onlyoffice" + if suffix in DOC_PREVIEW_SUFFIXES: + return "pdf" + mime_type, _ = mimetypes.guess_type(path.name) + if mime_type and mime_type.startswith("video/"): + return "video" + if mime_type and mime_type.startswith("image/"): + return "pyvips" + return "preview" diff --git a/mediapreview/office.py b/mediapreview/office.py new file mode 100644 index 0000000..7d73c69 --- /dev/null +++ b/mediapreview/office.py @@ -0,0 +1,422 @@ +"""OnlyOffice Document Server integration for office document preview. + +Provides server-side conversion of office documents to PNG via the +OnlyOffice Document Server /ConvertService.ashx API. The resulting PNG +is passed through pyvips for AVIF compression. + +Environment requirements: + - OnlyOffice Document Server must be running and reachable. + - If Document Server runs in Docker, the callback host IP must be + reachable from the container (usually the docker bridge IP). +""" + +import asyncio +import json +import logging +import os +import secrets +import socket +import socketserver +import subprocess +import threading +import urllib.error +import urllib.request +from functools import lru_cache, partial +from http.server import SimpleHTTPRequestHandler +from multiprocessing import cpu_count +from pathlib import Path +from time import perf_counter +from urllib.parse import quote + +try: + import httpx + import jwt +except ImportError: # pragma: no cover - optional office extra + httpx = None + jwt = None + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Configuration helpers +# --------------------------------------------------------------------------- + + +_httpx_client: httpx.AsyncClient | None = None + + +def _get_onlyoffice_url() -> str: + return os.environ.get( + "ONLYOFFICE_URL", + os.environ.get("ONLYOFFICE_CISTA_URL", "http://localhost:8988"), + ) + + +def _get_jwt_secret() -> str: + return os.environ.get("ONLYOFFICE_JWT_SECRET", "") + + +@lru_cache(maxsize=1) +def _get_callback_host() -> str: + """Return the host IP that OnlyOffice (usually in Docker) can use to reach us.""" + if host := os.environ.get("ONLYOFFICE_CALLBACK_HOST"): + return host + # Try to auto-detect docker bridge IP + try: + result = subprocess.run( + ["/sbin/ip", "-4", "addr", "show", "docker0"], + capture_output=True, + text=True, + timeout=2, + check=False, + ) + for line in result.stdout.splitlines(): + if "inet " in line: + parts = line.strip().split() + addr_part = parts[1] # e.g. 172.17.0.1/16 + return addr_part.split("/")[0] + except Exception: + logger.debug("Failed to auto-detect docker bridge IP") + return "127.0.0.1" + + +def onlyoffice_error_short_text(detail: str) -> str: + """Short human-readable label for an OnlyOffice failure, for log annotation.""" + if detail.startswith("OnlyOffice conversion error:"): + code = detail.rsplit(":", 1)[-1].strip() + return { + "-8": "onlyoffice jwt error", + "-4": "onlyoffice input error", + "-2": "onlyoffice timeout error", + "-1": "onlyoffice unknown error", + }.get(code, f"onlyoffice {code} error") + if "OnlyOffice response did not contain FileUrl" in detail: + return "onlyoffice no-fileurl error" + return "onlyoffice error" + + +# --------------------------------------------------------------------------- +# Async HTTP client +# --------------------------------------------------------------------------- + + +def get_httpx_client() -> httpx.AsyncClient: + """Return the shared async HTTP client for OnlyOffice requests.""" + if httpx is None: + raise ImportError( + "OnlyOffice integration requires the 'office' extra: pip install mediapreview[office]" + ) + global _httpx_client + if _httpx_client is None: + _httpx_client = httpx.AsyncClient() + return _httpx_client + + +async def close_oo_client() -> None: + """Close the shared async HTTP client.""" + global _httpx_client + if _httpx_client is not None: + await _httpx_client.aclose() + _httpx_client = None + + +# --------------------------------------------------------------------------- +# Availability check +# --------------------------------------------------------------------------- + + +def _probe_status() -> tuple[bool, bool, str | None]: + """Return (ok, responded, detail) for a lightweight reachability probe.""" + url = _get_onlyoffice_url().rstrip("/") + "/ConvertService.ashx" + try: + with urllib.request.urlopen(url, timeout=2) as resp: # noqa: S310 + status = resp.status + except urllib.error.HTTPError as e: + status = e.code + except Exception: + return False, False, None + + if status in (200, 405): + return True, True, None + if status >= 500: + return False, True, f"HTTP {status}" + return False, True, f"HTTP {status}" + + +def log_reachable_info() -> None: + """Log info on success, warning on responded probe errors, silent on no-response.""" + ok, responded, detail = _probe_status() + if ok: + logger.info("Using OnlyOffice document server at %s", _get_onlyoffice_url()) + elif responded: + suffix = f": {detail}" if detail else "" + logger.warning("OnlyOffice probe failed%s", suffix) + + +def setup_docker(name: str = "onlyoffice-mediapreview", port: int = 8988) -> int: + """Build and run the patched OnlyOffice Docker image. + + Uses ONLYOFFICE_JWT_SECRET if set, otherwise generates a random secret. + The Docker build context ships inside the package at `mediapreview/docker`. + """ + secret = _get_jwt_secret() or secrets.token_hex(16) + docker_dir = Path(__file__).parent / "docker" + if not docker_dir.is_dir(): + raise FileNotFoundError( + f"Docker files not found at {docker_dir}. Is the package installed correctly?" + ) + + logger.info("Building OnlyOffice image") + build_cmd = ["docker", "build", "-t", name, str(docker_dir)] + logger.info("%s", " ".join(build_cmd)) + result = subprocess.run(build_cmd, check=False, shell=False) # noqa: S603 + if result.returncode != 0: + raise RuntimeError("Failed to build OnlyOffice image") + + logger.info("Starting OnlyOffice container") + run_cmd = [ + "docker", + "run", + "-d", + "-p", + f"{port}:80", + "-e", + f"JWT_SECRET={secret}", + "-e", + "WORKERS=8", + "--name", + name, + "--restart", + "unless-stopped", + name, + ] + logger.info("%s", " ".join(run_cmd)) + result = subprocess.run(run_cmd, check=False, shell=False) # noqa: S603 + if result.returncode != 0: + raise RuntimeError("Failed to start OnlyOffice container") + logger.info("OnlyOffice is running on http://localhost:%d", port) + return 0 + + +async def is_available_async(request_timeout: float = 2.0) -> bool: + """Return True if the configured OnlyOffice Document Server is reachable.""" + if httpx is None: + raise ImportError( + "OnlyOffice integration requires the 'office' extra: pip install mediapreview[office]" + ) + url = _get_onlyoffice_url().rstrip("/") + "/ConvertService.ashx" + client = get_httpx_client() + try: + response = await client.get(url, timeout=request_timeout) + except Exception: + return False + else: + return response.status_code in (200, 405) + + +_oo_available_cache: tuple[bool, float] | None = None +OO_AVAILABILITY_CACHE_TTL = 30.0 + + +async def is_available_cached() -> bool: + """Return cached OnlyOffice availability, refreshed every 30 seconds. + + State transitions are logged, so an unreachable server is reported once + instead of on every preview attempt. + """ + global _oo_available_cache + now = perf_counter() + if _oo_available_cache is not None: + result, timestamp = _oo_available_cache + if now - timestamp < OO_AVAILABILITY_CACHE_TTL: + return result + result = await is_available_async() + if _oo_available_cache is None or _oo_available_cache[0] != result: + if result: + logger.info( + "OnlyOffice document server available at %s", _get_onlyoffice_url() + ) + else: + logger.warning( + "OnlyOffice document server not reachable at %s", _get_onlyoffice_url() + ) + _oo_available_cache = (result, now) + return result + + +# --------------------------------------------------------------------------- +# Temporary HTTP server so OnlyOffice can download the file +# --------------------------------------------------------------------------- + + +class _QuietHandler(SimpleHTTPRequestHandler): + def log_message(self, fmt, *args) -> None: + pass + + +def _get_free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("0.0.0.0", 0)) # noqa: S104 + return s.getsockname()[1] + + +def _serve_file_temporarily(file_path: Path): + """Start a temporary HTTP server for *file_path* and return (url, server).""" + directory = str(file_path.parent) + filename = file_path.name + port = _get_free_port() + + handler = partial(_QuietHandler, directory=directory) + httpd = socketserver.TCPServer(("0.0.0.0", port), handler) # noqa: S104 + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + + host = _get_callback_host() + url = f"http://{host}:{port}/{quote(filename)}" + return url, httpd + + +# --------------------------------------------------------------------------- +# OnlyOffice conversion client +# --------------------------------------------------------------------------- + + +def _build_jwt_token(payload: dict) -> str | None: + secret = _get_jwt_secret() + if not secret: + return None + return jwt.encode(payload, secret, algorithm="HS256") + + +async def convert_to_png_async(file_path: Path, request_timeout: float = 5.0) -> bytes: + """Convert *file_path* to PNG using OnlyOffice Document Server (async). + + Returns the PNG bytes. Raises RuntimeError on failure. + """ + if httpx is None or jwt is None: + raise ImportError( + "OnlyOffice integration requires the 'office' extra: pip install mediapreview[office]" + ) + oo_url = _get_onlyoffice_url().rstrip("/") + convert_url = f"{oo_url}/ConvertService.ashx" + client = get_httpx_client() + + # Start temporary HTTP server so OnlyOffice can fetch the file + doc_url, httpd = await asyncio.to_thread(_serve_file_temporarily, file_path) + try: + suffix = file_path.suffix.lstrip(".").lower() + payload = { + "async": False, + "filetype": suffix, + "key": f"mediapreview_{(await asyncio.to_thread(file_path.stat)).st_mtime_ns}", + "outputtype": "png", + "title": file_path.name, + "url": doc_url, + } + + headers = {"Content-Type": "application/json"} + token = _build_jwt_token(payload) + if token: + # Conversion API expects JWT in request body when token checks are enabled. + payload["token"] = token + headers["Authorization"] = token + + t_start = perf_counter() + response = await client.post( + convert_url, + content=json.dumps(payload).encode(), + headers=headers, + timeout=request_timeout, + ) + response.raise_for_status() + body = response.content + t_end = perf_counter() + + # Parse XML response + text = body.decode("utf-8", errors="replace") + if "" in text: + code = "unknown" + if "" in text and "" in text: + code = text.split("")[1].split("")[0] + raise RuntimeError(f"OnlyOffice conversion error: {code}") + + if "" not in text: + raise RuntimeError("OnlyOffice response did not contain FileUrl") + + file_url = text.split("")[1].split("")[0] + file_url = file_url.replace("&", "&") + + logger.debug("OnlyOffice converted in %.2fs: %s", t_end - t_start, file_url) + + # Download converted PNG + png_response = await client.get(file_url, timeout=request_timeout) + png_response.raise_for_status() + return png_response.content + finally: + await asyncio.to_thread(httpd.shutdown) + + +# --------------------------------------------------------------------------- +# Conversion manager (pulled from the preview orchestrator) +# --------------------------------------------------------------------------- + +# Max concurrent OnlyOffice conversion requests. OO has its own queue; +# we must not flood it. This is intentionally small. +OO_MAX_CONCURRENT = max(2, min(8, cpu_count())) + + +class OOConversionManager: + """Manages async OnlyOffice conversions with deduplication and concurrency limits.""" + + def __init__(self, max_concurrent: int = OO_MAX_CONCURRENT): + self._semaphore = asyncio.Semaphore(max_concurrent) + self._in_flight: dict[str, asyncio.Future[bytes]] = {} + self._tasks: set[asyncio.Task[None]] = set() + self._lock = asyncio.Lock() + + async def convert(self, filepath: Path) -> bytes: + """Return PNG bytes for *filepath*, deduplicating concurrent requests.""" + if not await is_available_cached(): + raise RuntimeError("OnlyOffice server not reachable") + stat = await asyncio.to_thread(filepath.stat) + key = f"{filepath}:{stat.st_mtime_ns}" + + async with self._lock: + if key in self._in_flight: + future = self._in_flight[key] + else: + future = asyncio.get_running_loop().create_future() + self._in_flight[key] = future + task = asyncio.create_task(self._do_convert(filepath, key, future)) + self._tasks.add(task) + task.add_done_callback(self._tasks.discard) + + return await future + + async def _do_convert( + self, filepath: Path, key: str, future: asyncio.Future[bytes] + ) -> None: + try: + async with self._semaphore: + png_bytes = await convert_to_png_async(filepath, request_timeout=5.0) + except Exception as e: + if not future.done(): + future.set_exception(e) + async with self._lock: + self._in_flight.pop(key, None) + else: + if not future.done(): + future.set_result(png_bytes) + async with self._lock: + self._in_flight.pop(key, None) + + +_oo_manager: OOConversionManager | None = None + + +def get_oo_manager() -> OOConversionManager: + """Return the singleton OOConversionManager.""" + global _oo_manager + if _oo_manager is None: + _oo_manager = OOConversionManager(max_concurrent=OO_MAX_CONCURRENT) + return _oo_manager diff --git a/mediapreview/pool.py b/mediapreview/pool.py new file mode 100644 index 0000000..cdf2621 --- /dev/null +++ b/mediapreview/pool.py @@ -0,0 +1,488 @@ +"""Async preview worker pool framework.""" + +import asyncio +import contextlib +import logging +import os +import signal +import struct +import sys +from multiprocessing import cpu_count +from pathlib import Path +from time import perf_counter + +import msgspec +from blake3 import blake3 + +from mediapreview.formats import ( + expected_backend as _expected_preview_backend, +) +from mediapreview.formats import ( + is_previewable_path, +) +from mediapreview.formats import ( + preview_job_priority as _preview_job_priority, +) +from mediapreview.office import get_oo_manager +from mediapreview.protocol import PreviewRequest, PreviewResponse + +__all__ = [ + "PREVIEW_TIMEOUT", + "PreviewError", + "PreviewPoolClosedError", + "PreviewTimeoutError", + "generate_office_preview", + "is_previewable_path", + "run_preview", + "shutdown_preview_workers", + "start_preview_workers", +] + +logger = logging.getLogger(__name__) + + +class PreviewTimeoutError(Exception): + """Raised when the preview subprocess exceeds PREVIEW_TIMEOUT.""" + + def __init__(self, message: str, *, backend: str | None = None): + super().__init__(message) + self.backend = backend + + +class PreviewError(Exception): + """Raised when the preview subprocess exits with a non-zero status.""" + + def __init__( + self, + message: str, + *, + stderr: str | None = None, + backend: str | None = None, + ): + super().__init__(message) + self.stderr = stderr + self.backend = backend + + +class PreviewPoolClosedError(PreviewError): + """The preview worker pool has been shut down.""" + + +class WorkerChecksumError(Exception): + """Raised when worker response checksum does not match the packet.""" + + +class WorkerProtocolError(Exception): + """Raised when worker response packet is malformed.""" + + +PREVIEW_TIMEOUT = 10.0 # seconds until preview subprocess is killed +PREVIEW_WORKERS = max(2, min(8, cpu_count())) +WORKER_KILL_GRACE = 5.0 # max seconds to wait for a killed worker to be reaped +WORKER_RESPAWN_DELAY = 1.0 # initial delay before retrying a failed worker spawn +WORKER_RESPAWN_DELAY_MAX = 30.0 +WORKER_CHECKSUM_BYTES = 32 +WORKER_MAX_JSON_BYTES = 1_000_000 + +_active_procs: set[asyncio.subprocess.Process] = set() +_preview_pool = None +_preview_pool_lock = asyncio.Lock() +_pool_stopped = False + + +class _PreviewWorker: + def __init__(self, proc: asyncio.subprocess.Process): + self.proc = proc + + async def request( + self, + filepath, + quality: int, + maxsize: int, + maxzoom: float, + data: bytes | None = None, + ): + if self.proc.returncode is not None: + raise WorkerProtocolError("worker already exited") + if self.proc.stdin is None or self.proc.stdout is None: + raise WorkerProtocolError("worker streams not available") + + meta = msgspec.json.encode( + PreviewRequest( + path=str(filepath), + quality=quality, + maxsize=maxsize, + maxzoom=maxzoom, + ) + ) + payload = data or b"" + packet = struct.pack(" WORKER_MAX_JSON_BYTES: + raise WorkerProtocolError(f"worker JSON too large: {json_size}") + meta_raw = await self.proc.stdout.readexactly(json_size) + payload = await self.proc.stdout.readexactly(data_size) + packet = header + meta_raw + payload + if blake3(packet).digest() != checksum: + raise WorkerChecksumError("worker checksum mismatch") + + resp = msgspec.json.decode(meta_raw, type=PreviewResponse) + if not resp.ok: + raise PreviewError( + resp.error or "preview worker error", + stderr=resp.stderr, + backend=resp.backend, + ) + return payload or None, resp + + async def kill(self) -> None: + try: + if self.proc.returncode is None: + # Safe to hard-kill: the worker is stateless per request. + # Kill the whole process group (worker is the group leader, + # spawned with start_new_session) so that an in-flight ffmpeg + # grandchild cannot be orphaned by the worker's SIGKILL. + # proc.wait() must not be awaited unaided: if a pipe + # transport is flow-control paused (e.g. an undrained stderr + # pipe), asyncio may never resolve wait() even after SIGKILL, + # which would permanently wedge the calling dispatcher. + with contextlib.suppress(ProcessLookupError): + os.killpg(self.proc.pid, signal.SIGKILL) + try: + await asyncio.wait_for(self.proc.wait(), timeout=WORKER_KILL_GRACE) + except TimeoutError: + logger.exception( + "Preview worker pid=%s not reaped within %ds of kill", + self.proc.pid, + int(WORKER_KILL_GRACE), + ) + finally: + _active_procs.discard(self.proc) + + +class _PreviewWorkerPool: + def __init__(self, size: int): + self.size = size + self._idle: asyncio.Queue[_PreviewWorker] = asyncio.Queue() + self._pending: asyncio.PriorityQueue[tuple[int, int, asyncio.Future, tuple]] = ( + asyncio.PriorityQueue() + ) + self._workers: set[_PreviewWorker] = set() + self._dispatchers: list[asyncio.Task] = [] + self._in_flight: set[asyncio.Future] = set() + self._seq = 0 + self._closed = False + + async def _spawn_worker(self) -> _PreviewWorker: + # stderr is inherited, not piped: a piped stderr that nobody drains + # eventually fills its OS buffer, blocking the worker mid-request, + # and its flow-control-paused transport makes proc.wait() hang even + # after kill() — together this used to permanently wedge the pool. + # Inheriting sends worker diagnostics straight to the server log. + proc = await asyncio.create_subprocess_exec( + sys.executable, + "-m", + "mediapreview.worker", + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=None, + # Own process group so kill() can SIGKILL the worker together with + # any grandchild (e.g. ffmpeg) it may have spawned. + start_new_session=True, + ) + _active_procs.add(proc) + try: + ready = await asyncio.wait_for(proc.stdout.readexactly(1), timeout=30.0) + except TimeoutError as err: + with contextlib.suppress(ProcessLookupError): + os.killpg(proc.pid, signal.SIGKILL) + with contextlib.suppress(Exception): + await proc.wait() + raise WorkerProtocolError( + "preview worker failed to become ready" + " (worker stderr goes to the server log)" + ) from err + except asyncio.IncompleteReadError as err: + raise WorkerProtocolError( + "preview worker exited before signalling readiness" + " (worker stderr goes to the server log)" + ) from err + if ready != b"\x01": + raise WorkerProtocolError(f"preview worker ready signal invalid: {ready!r}") + return _PreviewWorker(proc) + + async def _add_worker(self) -> None: + worker = await self._spawn_worker() + self._workers.add(worker) + await self._idle.put(worker) + + async def _replace_worker(self, worker: _PreviewWorker) -> None: + self._workers.discard(worker) + try: + await worker.kill() + except Exception: + logger.exception("Failed to kill preview worker pid=%s", worker.proc.pid) + # Keep retrying until a replacement is up: a pool that silently + # shrinks degrades all preview traffic to timeouts. + delay = WORKER_RESPAWN_DELAY + while not self._closed: + try: + await self._add_worker() + except Exception: + logger.exception( + "Failed to replace preview worker (pool %d/%d); retrying in %ds", + len(self._workers), + self.size, + int(delay), + ) + await asyncio.sleep(delay) + delay = min(delay * 2, WORKER_RESPAWN_DELAY_MAX) + else: + return + + async def _dispatch_loop(self) -> None: + # Nothing may escape the loop body: a dispatcher that dies silently + # permanently shrinks pool capacity and degrades all preview + # traffic to timeouts. + while True: + try: + await self._dispatch_one() + except asyncio.CancelledError: + return + except Exception: + logger.exception("Preview dispatcher error; continuing") + + async def _dispatch_one(self) -> None: + _priority, _seq, future, args = await self._pending.get() + + if future.cancelled(): + return + + try: + worker = await asyncio.wait_for(self._idle.get(), timeout=PREVIEW_TIMEOUT) + except TimeoutError: + logger.warning( + "Preview worker unavailable (%ds) for %s", + int(PREVIEW_TIMEOUT), + args[0].name, + ) + if not future.done(): + future.set_exception( + PreviewTimeoutError( + args[0].name, + backend=_expected_preview_backend(args[0]), + ) + ) + return + + filepath = args[0] + replace = False + try: + out, resp = await asyncio.wait_for( + worker.request(*args), + timeout=PREVIEW_TIMEOUT, + ) + if not future.done(): + future.set_result((out, resp)) + except TimeoutError: + replace = True + logger.warning( + "Preview worker pid=%s timed out (%ds) on %s; replacing it", + worker.proc.pid, + int(PREVIEW_TIMEOUT), + filepath.name, + ) + if not future.done(): + future.set_exception( + PreviewTimeoutError( + filepath.name, + backend=_expected_preview_backend(filepath), + ) + ) + except WorkerChecksumError: + replace = True + logger.exception( + "Preview checksum mismatch for %s (worker pid=%s); replacing it", + filepath.name, + worker.proc.pid, + ) + if not future.done(): + future.set_exception( + PreviewError(f"worker checksum mismatch for {filepath.name}") + ) + except PreviewError as e: + if not future.done(): + future.set_exception(e) + except ( + WorkerProtocolError, + asyncio.IncompleteReadError, + BrokenPipeError, + ConnectionResetError, + OSError, + ValueError, + msgspec.DecodeError, + ) as e: + replace = True + logger.warning( + "Preview worker pid=%s protocol failure for %s: %s", + worker.proc.pid, + filepath.name, + e, + ) + if not future.done(): + future.set_exception( + PreviewError(f"worker protocol failure for {filepath.name}: {e}") + ) + except Exception: + replace = True + logger.exception("Unexpected preview worker error for %s", filepath.name) + if not future.done(): + future.set_exception( + PreviewError(f"unexpected worker error for {filepath.name}") + ) + finally: + if replace: + await self._replace_worker(worker) + elif worker.proc.returncode is None: + await self._idle.put(worker) + else: + await self._replace_worker(worker) + + async def start(self) -> None: + workers = await asyncio.gather( + *(self._spawn_worker() for _ in range(self.size)) + ) + for worker in workers: + self._workers.add(worker) + await self._idle.put(worker) + for _ in range(self.size): + self._dispatchers.append(asyncio.create_task(self._dispatch_loop())) + + async def run( + self, + filepath, + quality: int, + maxsize: int, + maxzoom: float, + data: bytes | None = None, + ): + if self._closed: + raise PreviewPoolClosedError("preview worker pool closed") + loop = asyncio.get_running_loop() + future = loop.create_future() + self._in_flight.add(future) + self._seq += 1 + await self._pending.put( + ( + _preview_job_priority(filepath), + self._seq, + future, + (filepath, quality, maxsize, maxzoom, data), + ) + ) + try: + return await future + finally: + self._in_flight.discard(future) + + async def close(self) -> None: + self._closed = True + for task in self._dispatchers: + task.cancel() + if self._dispatchers: + await asyncio.gather(*self._dispatchers, return_exceptions=True) + self._dispatchers.clear() + workers = list(self._workers) + self._workers.clear() + # Fail every future still waiting on a result — pending and in-flight + # alike — so request handlers finish immediately instead of waiting + # out their timeouts during server shutdown. + for future in list(self._in_flight): + if not future.done(): + future.set_exception( + PreviewPoolClosedError("preview worker pool closed") + ) + while not self._pending.empty(): + try: + _priority, _seq, future, _args = self._pending.get_nowait() + except asyncio.QueueEmpty: + break + if not future.done(): + future.set_exception( + PreviewPoolClosedError("preview worker pool closed") + ) + while not self._idle.empty(): + try: + self._idle.get_nowait() + except asyncio.QueueEmpty: + break + await asyncio.gather( + *(worker.kill() for worker in workers), return_exceptions=True + ) + + +async def start_preview_workers() -> None: + """Warm up persistent preview workers.""" + global _preview_pool + if _preview_pool is not None or _pool_stopped: + return + async with _preview_pool_lock: + if _preview_pool is not None or _pool_stopped: + return + pool = _PreviewWorkerPool(PREVIEW_WORKERS) + await pool.start() + _preview_pool = pool + logger.info("Started %d persistent preview workers", PREVIEW_WORKERS) + + +async def shutdown_preview_workers() -> None: + """Kill persistent preview workers.""" + global _preview_pool, _pool_stopped + _pool_stopped = True + async with _preview_pool_lock: + pool = _preview_pool + _preview_pool = None + if pool is not None: + await pool.close() + if not _active_procs: + return + for proc in list(_active_procs): + with contextlib.suppress(ProcessLookupError): + os.killpg(proc.pid, signal.SIGKILL) + await asyncio.gather( + *(proc.wait() for proc in list(_active_procs)), return_exceptions=True + ) + _active_procs.clear() + + +async def generate_office_preview( + filepath: Path, quality: int, maxsize: int, maxzoom: float +) -> tuple[bytes | None, PreviewResponse | None]: + """Generate a preview for an office file using OnlyOffice + worker AVIF conversion.""" + manager = get_oo_manager() + t_oo_start = perf_counter() + png_bytes = await manager.convert(filepath) + t_oo_end = perf_counter() + + img, resp = await run_preview(filepath, quality, maxsize, maxzoom, data=png_bytes) + + if resp is not None: + resp.backend = "onlyoffice+" + (resp.backend or "pyvips") + if resp.timings: + resp.timings = [round((t_oo_end - t_oo_start) * 1000, 1), *resp.timings] + return img, resp + + +async def run_preview( + filepath, quality: int, maxsize: int, maxzoom: float, data: bytes | None = None +) -> tuple[bytes | None, PreviewResponse | None]: + """Run preview request in a persistent worker process.""" + await start_preview_workers() + if _preview_pool is None: + raise PreviewPoolClosedError("preview worker pool closed") + return await _preview_pool.run(filepath, quality, maxsize, maxzoom, data) diff --git a/mediapreview/protocol.py b/mediapreview/protocol.py new file mode 100644 index 0000000..f4ae49f --- /dev/null +++ b/mediapreview/protocol.py @@ -0,0 +1,21 @@ +"""Wire protocol structs for the preview worker pool.""" + +import msgspec + + +class PreviewRequest(msgspec.Struct, omit_defaults=True): + path: str + quality: int + maxsize: int + maxzoom: float + + +class PreviewResponse(msgspec.Struct, omit_defaults=True): + ok: bool + mime: str | None = None + backend: str | None = None + timings: list[float] | None = None + error: str | None = None + stderr: str | None = None + width: int | None = None + height: int | None = None diff --git a/mediapreview/util/__init__.py b/mediapreview/util/__init__.py new file mode 100644 index 0000000..96675f0 --- /dev/null +++ b/mediapreview/util/__init__.py @@ -0,0 +1 @@ +"""Utility helpers for the mediapreview package.""" diff --git a/mediapreview/util/logformat.py b/mediapreview/util/logformat.py new file mode 100644 index 0000000..140d3cb --- /dev/null +++ b/mediapreview/util/logformat.py @@ -0,0 +1,33 @@ +"""Shared log formatting helpers.""" + +import logging +import unicodedata + +LEVEL_EMOJI = { + logging.DEBUG: "🔍", + logging.INFO: "ℹ️", # noqa: RUF001 + logging.WARNING: "⚠️", + logging.ERROR: "🛑", + logging.CRITICAL: "🛑", +} + + +def display_width(text: str) -> int: + return sum( + 1 + (unicodedata.east_asian_width(c) in "FW") + for c in text + if unicodedata.category(c) != "Mn" + ) + + +def format_level_prefix(levelno: int) -> str: + emoji = LEVEL_EMOJI.get(levelno, "▪️") + prefix = f"{emoji} " + return prefix + (" " * max(0, 3 - display_width(prefix))) + + +class EmojiFormatter(logging.Formatter): + """Compact formatter: emoji + message, no timestamp/level text/logger name.""" + + def format(self, record: logging.LogRecord) -> str: + return format_level_prefix(record.levelno) + record.getMessage() diff --git a/mediapreview/worker.py b/mediapreview/worker.py new file mode 100644 index 0000000..1facfe6 --- /dev/null +++ b/mediapreview/worker.py @@ -0,0 +1,175 @@ +"""Preview generation worker subprocess entry point. + +The actual conversion logic lives in `mediapreview.backends`; this module +only implements the worker process shell around it. + +Two modes are supported: +1) Legacy one-shot mode: argv has path/quality/maxsize/maxzoom. +2) Long-lived mode: read framed requests from stdin and write framed responses. + +Framed request format (stdin): + (uint32 json size)(uint32 data size)(json)(binary data) + +Framed response format (stdout): + (blake3(packet))(uint32 json size)(uint32 payload size)(json)(binary payload) +where packet = (uint32 json size)(uint32 payload size)(json)(binary payload). +""" + +import contextlib +import io +import logging +import os +import signal +import struct +import sys +from pathlib import Path + +import msgspec +import tracerite +from blake3 import blake3 + +from mediapreview.backends import dispatch +from mediapreview.protocol import PreviewRequest, PreviewResponse +from mediapreview.util.logformat import format_level_prefix + +logger = logging.getLogger(__name__) + + +class _WorkerLogFormatter(logging.Formatter): + """Emoji level prefix like the main process, tagged with the worker pid.""" + + def format(self, record: logging.LogRecord) -> str: + prefix = format_level_prefix(record.levelno) + return f"{prefix}worker[{os.getpid()}]: {record.getMessage()}" + + +_enc = msgspec.json.Encoder() +_dec_req = msgspec.json.Decoder(PreviewRequest) + + +def _read_exactly(f, n: int) -> bytes: + buf = b"" + while len(buf) < n: + chunk = f.read(n - len(buf)) + if not chunk: + raise EOFError + buf += chunk + return buf + + +def _read_request() -> tuple[PreviewRequest, bytes] | None: + try: + header = _read_exactly(sys.stdin.buffer, 8) + except EOFError: + return None + json_size, data_size = struct.unpack(" None: + out = _protocol_out if _protocol_out is not None else sys.stdout.buffer + meta_bytes = _enc.encode(resp) + packet = struct.pack(" None: + if len(sys.argv) != 5: + sys.stderr.write(f"Usage: {sys.argv[0]} \n") + sys.exit(1) + + path = Path(sys.argv[1]) + quality = int(sys.argv[2]) + maxsize = int(sys.argv[3]) + maxzoom = float(sys.argv[4]) + result, _ = dispatch(path, quality, maxsize, maxzoom) + if result: + sys.stdout.buffer.write(result) + sys.stdout.buffer.flush() + + +def _run_loop() -> None: + while True: + result = _read_request() + if result is None: + return + req, data = result + stderr_capture = io.StringIO() + handler = logging.StreamHandler(stderr_capture) + root_logger = logging.getLogger() + root_logger.addHandler(handler) + try: + with contextlib.redirect_stderr(stderr_capture): + result, resp = dispatch( + Path(req.path), req.quality, req.maxsize, req.maxzoom, data + ) + if not resp.ok: + captured = stderr_capture.getvalue().strip() + if captured: + resp = PreviewResponse( + ok=False, + backend=resp.backend, + error=resp.error, + stderr=captured, + ) + _write_response(resp, result or b"") + except Exception as e: + logger.exception("Preview worker error for %s", req.path) + captured = stderr_capture.getvalue().strip() + _write_response( + PreviewResponse(ok=False, error=str(e), stderr=captured or None), b"" + ) + finally: + root_logger.removeHandler(handler) + handler.close() + + +def main() -> None: + # Format tracebacks like the main process. + tracerite.load() + # Configure all log output to stderr before any imports that may emit + # logs. stderr is inherited by the parent, so this lands in the server + # log, formatted like the main process and tagged with the worker pid. + handler = logging.StreamHandler(sys.stderr) + handler.setFormatter(_WorkerLogFormatter()) + logging.basicConfig(level=logging.INFO, handlers=[handler]) + # pyvips is chatty at INFO ("threadpool completed ..." per operation). + logging.getLogger("pyvips").setLevel(logging.WARNING) + # NOTE: standalone package no longer depends on cista config loading. + # Consumers can load their own configuration before starting workers. + if len(sys.argv) > 1: + _run_once() + return + # Ctrl-C SIGINTs the whole process group; the parent pool terminates us + # (and our stdin EOF exits us) — don't dump KeyboardInterrupt tracebacks. + signal.signal(signal.SIGINT, signal.SIG_IGN) + # The command channel is a binary protocol on fd 1. Anything printed to + # stdout by Python code (e.g. a library emitting a warning via print()) + # would corrupt the protocol, so redirect Python-level stdout to stderr + # (the server log) and keep the raw buffer solely for protocol traffic. + global _protocol_out + _protocol_out = sys.stdout.buffer + sys.stdout = sys.stderr + # Eagerly import heavy modules before signalling readiness so the parent + # does not hand us a request while we are still initialising. + _protocol_out.write(b"\x01") + _protocol_out.flush() + _run_loop() + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..1ad57e5 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,98 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "mediapreview" +version = "0.1.0" +description = "Low-level media preview converters and a worker pool framework" +readme = "README.md" +requires-python = ">=3.11" +dependencies = [ + "msgspec>=0.19.0", + "pyvips[binary]>=3.1.1", +] + +[project.optional-dependencies] +worker = [ + "blake3>=1.0.5", + "tracerite>=2.3.1", +] +pdf = [ + "pymupdf>=1.26.3", +] +video = [ + "av>=15.0.0", + "numpy>=2.3.2", +] +office = [ + "httpx>=0.28.0", + "pyjwt>=2.10.1", +] +standard = [ + "mediapreview[worker]", + "mediapreview[pdf]", + "mediapreview[video]", + "mediapreview[office]", +] +dev = [ + "pytest>=8.4.1", + "pytest-asyncio>=0.25.0", + "pytest-cov>=6.0.0", + "ruff>=0.8.0", + "mypy>=1.13.0", +] + +[tool.hatch.build.targets.wheel] +packages = ["mediapreview"] + +[tool.ruff] +target-version = "py311" + +[tool.ruff.lint] +select = ["ALL"] +ignore = [ + "COM812", + "ISC001", + "ANN001", + "ANN002", + "ANN003", + "ANN201", + "ANN202", + "ANN204", + "ANN205", + "BLE001", + "C901", + "CPY", + "D100", + "D101", + "D102", + "D103", + "D104", + "D105", + "D107", + "D200", + "D203", + "D212", + "D213", + "D400", + "D401", + "D413", + "D415", + "E501", + "EM101", + "EM102", + "INP001", + "PLR0911", + "PLR0912", + "PLR0913", + "PLR0915", + "PLR2004", + "PLW0603", + "TRY003", +] +isort.known-first-party = ["mediapreview"] +per-file-ignores."tests/*" = ["S", "ANN", "D", "INP", "PLR2004", "ARG001", "SLF001"] + +[tool.pytest.ini_options] +asyncio_mode = "auto" diff --git a/tests/test_pool.py b/tests/test_pool.py new file mode 100644 index 0000000..575d153 --- /dev/null +++ b/tests/test_pool.py @@ -0,0 +1,175 @@ +"""Tests for the preview worker pool resilience. + +Regression context: a piped worker stderr that nobody drains used to block +the worker mid-request once the OS pipe buffer filled, and asyncio's +proc.wait() then never resolved even after kill() — wedging one dispatcher +per stuck worker until all preview traffic timed out permanently. +""" + +import asyncio +import sys +import textwrap +import time +from pathlib import Path +from unittest.mock import AsyncMock, Mock + +import pytest + +from mediapreview import pool + +FAKE_WORKER = textwrap.dedent( + """ + import json + import struct + import sys + + import blake3 + + + def read_exact(n): + buf = b"" + while len(buf) < n: + chunk = sys.stdin.buffer.read(n - len(buf)) + if not chunk: + raise EOFError + buf += chunk + return buf + + + sys.stdout.buffer.write(b"\\x01") + sys.stdout.buffer.flush() + while True: + header = sys.stdin.buffer.read(8) + if not header or len(header) < 8: + break + meta_len, payload_len = struct.unpack("