From 25a2a5f20c28a143375e537d64abd505e61fa743 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Fri, 1 May 2026 03:42:03 +0000 Subject: [PATCH] Separate OnlyOffice async handling, add office_previews WS flag, framed worker protocol - Move OnlyOffice conversion out of worker subprocesses into async event loop using httpx.AsyncClient with shared client and clean shutdown hook - Add OOConversionManager with in-flight deduplication (asyncio.Future) and configurable concurrency limit (OO_MAX_CONCURRENT=2) - Add 10s total timeout for office previews via asyncio.wait_for - Return 503 when OnlyOffice is unavailable, 504 on timeout - Remove office handling from worker dispatch(); workers now only do images, video, and PDFs - Replace PNG tempfile bridge with framed binary input protocol on worker stdin: (json_size)(data_size)(json)(raw_data) - Add process_image_buffer() for in-memory AVIF conversion via pyvips - Add office_previews to WS server message, cached every 30s from OO availability check - Frontend gates only office document previews on office_previews flag; images, video, PDFs remain unconditional - Change default OnlyOffice port from 8080 to 8988 --- cista/api.py | 3 + cista/app.py | 3 +- cista/onlyoffice.py | 126 ++++++++++++++++- cista/preview.py | 172 +++++++++++++++++++---- cista/preview_worker.py | 39 ++++- frontend/src/components/MediaPreview.vue | 125 +++++++++------- frontend/src/repositories/Document.ts | 17 ++- frontend/src/stores/main.ts | 2 +- 8 files changed, 396 insertions(+), 91 deletions(-) diff --git a/cista/api.py b/cista/api.py index fb9eeb8..dc87c82 100644 --- a/cista/api.py +++ b/cista/api.py @@ -55,6 +55,8 @@ async def watch(req, ws): "privileged": req.ctx.user.privileged, } + from cista import onlyoffice + await ws.send( msgspec.json.encode( { @@ -63,6 +65,7 @@ async def watch(req, ws): "version": __version__, "public": config.config.public, "paskia": sso.paskia_enabled(), + "office_previews": await onlyoffice.is_available_cached(), }, "user": user_info, } diff --git a/cista/app.py b/cista/app.py index d184b71..ae991e6 100644 --- a/cista/app.py +++ b/cista/app.py @@ -16,7 +16,7 @@ from setproctitle import setproctitle from stream_zip import ZIP_AUTO, stream_zip from zstandard import ZstdCompressor -from cista import auth, config, fileserver, preview, session, sharefs, sso, watching +from cista import auth, config, fileserver, onlyoffice, preview, session, sharefs, sso, watching from cista.api import bp from cista.preview import shutdown_preview_workers, start_preview_workers from cista.sanic_logging import ( @@ -130,6 +130,7 @@ async def main_start(app): @app.before_server_stop async def main_stop(app): watching.stop(app) + await onlyoffice.close_oo_client() await shutdown_preview_workers() app.ctx.threadexec.shutdown() app.ctx.zipexec.shutdown(cancel_futures=True) diff --git a/cista/onlyoffice.py b/cista/onlyoffice.py index eb28eee..832d032 100644 --- a/cista/onlyoffice.py +++ b/cista/onlyoffice.py @@ -10,6 +10,7 @@ Environment requirements: reachable from the container (usually the docker bridge IP). """ +import asyncio import json import os import socket @@ -23,6 +24,7 @@ from pathlib import Path from time import perf_counter from urllib.parse import quote +import httpx import jwt from sanic.log import logger @@ -31,8 +33,11 @@ from sanic.log import logger # --------------------------------------------------------------------------- +_httpx_client: httpx.AsyncClient | None = None + + def _get_onlyoffice_url() -> str: - return os.environ.get("ONLYOFFICE_URL", "http://localhost:8080") + return os.environ.get("ONLYOFFICE_URL", "http://localhost:8988") def _get_jwt_secret() -> str | None: @@ -62,6 +67,27 @@ def _get_callback_host() -> str: return "127.0.0.1" +# --------------------------------------------------------------------------- +# Async HTTP client +# --------------------------------------------------------------------------- + + +def get_httpx_client() -> httpx.AsyncClient: + """Return the shared async HTTP client for OnlyOffice requests.""" + 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 # --------------------------------------------------------------------------- @@ -69,14 +95,42 @@ def _get_callback_host() -> str: def is_available() -> bool: """Return True if the configured OnlyOffice Document Server is reachable.""" - url = _get_onlyoffice_url() + url = _get_onlyoffice_url().rstrip("/") + "/ConvertService.ashx" try: with urllib.request.urlopen(url, timeout=3) as resp: # noqa: S310 - return resp.status == 200 + return resp.status in (200, 405) # 405 Method Not Allowed is fine, means endpoint exists except Exception: return False +async def is_available_async(timeout: float = 2.0) -> bool: + """Return True if the configured OnlyOffice Document Server is reachable.""" + url = _get_onlyoffice_url().rstrip("/") + "/ConvertService.ashx" + client = get_httpx_client() + try: + response = await client.get(url, timeout=timeout) + return response.status_code in (200, 405) + except Exception: + return False + + +_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.""" + 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() + _oo_available_cache = (result, now) + return result + + # --------------------------------------------------------------------------- # Temporary HTTP server so OnlyOffice can download the file # --------------------------------------------------------------------------- @@ -121,7 +175,7 @@ def _build_jwt_token(payload: dict) -> str | None: return jwt.encode(payload, secret, algorithm="HS256") -def convert_to_png(file_path: Path, timeout: float = 30.0) -> bytes: +def convert_to_png(file_path: Path, timeout: float = 5.0) -> bytes: """Convert *file_path* to PNG using OnlyOffice Document Server. Returns the PNG bytes. Raises RuntimeError on failure. @@ -182,3 +236,67 @@ def convert_to_png(file_path: Path, timeout: float = 30.0) -> bytes: return png_resp.read() finally: httpd.shutdown() + + +async def convert_to_png_async(file_path: Path, timeout: float = 5.0) -> bytes: + """Convert *file_path* to PNG using OnlyOffice Document Server (async). + + Returns the PNG bytes. Raises RuntimeError on failure. + """ + 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"cista_{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=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=timeout) + png_response.raise_for_status() + return png_response.content + finally: + await asyncio.to_thread(httpd.shutdown) diff --git a/cista/preview.py b/cista/preview.py index 46763ba..e8a7245 100644 --- a/cista/preview.py +++ b/cista/preview.py @@ -10,7 +10,7 @@ import urllib.parse from collections import OrderedDict from dataclasses import dataclass from multiprocessing import cpu_count -from pathlib import PurePosixPath +from pathlib import Path, PurePosixPath from time import perf_counter from urllib.parse import unquote from wsgiref.handlers import format_date_time @@ -112,24 +112,25 @@ class _PreviewWorker: def __init__(self, proc: asyncio.subprocess.Process): self.proc = proc - async def request(self, filepath, quality: int, maxsize: int, maxzoom: float): + 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") - line = ( - msgspec.json.encode( - PreviewRequest( - path=str(filepath), - quality=quality, - maxsize=maxsize, - maxzoom=maxzoom, - ) + meta = msgspec.json.encode( + PreviewRequest( + path=str(filepath), + quality=quality, + maxsize=maxsize, + maxzoom=maxzoom, ) - + b"\n" ) - self.proc.stdin.write(line) + payload = data or b"" + packet = struct.pack(" bytes: + """Return PNG bytes for *filepath*, deduplicating concurrent requests.""" + stat = 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 + asyncio.create_task(self._do_convert(filepath, key, future)) + + return await future + + async def _do_convert( + self, filepath: Path, key: str, future: asyncio.Future[bytes] + ) -> None: + try: + async with self._semaphore: + oo = _get_onlyoffice() + if oo is False: + raise OnlyOfficeUnavailableError("OnlyOffice is not installed") + png_bytes = await oo.convert_to_png_async(filepath, timeout=5.0) + except Exception as e: + future.set_exception(e) + async with self._lock: + self._in_flight.pop(key, None) + else: + 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 + + +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.""" + oo = _get_onlyoffice() + if oo is False: + raise OnlyOfficeUnavailableError("OnlyOffice is not installed") + if not await oo.is_available_async(): + raise OnlyOfficeUnavailableError("OnlyOffice Document Server is not reachable") + + 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_process( + 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_process( - filepath, quality: int, maxsize: int, maxzoom: float + 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 PreviewError(f"preview worker pool unavailable for {filepath.name}") - return await _preview_pool.run(filepath, quality, maxsize, maxzoom) + return await _preview_pool.run(filepath, quality, maxsize, maxzoom, data) DOC_PREVIEW_SUFFIXES = {".pdf", ".xps", ".epub", ".mobi"} @@ -492,9 +583,19 @@ async def preview(req, path): # Generate preview try: - img, preview_resp = await _run_preview_process( - filepath, quality, maxsize, maxzoom - ) + if filepath.suffix.lower() in OFFICE_PREVIEW_SUFFIXES: + img, preview_resp = await asyncio.wait_for( + _generate_office_preview(filepath, quality, maxsize, maxzoom), + timeout=PREVIEW_TIMEOUT, + ) + else: + img, preview_resp = await _run_preview_process( + filepath, quality, maxsize, maxzoom + ) + except asyncio.TimeoutError: + return empty(504) + except OnlyOfficeUnavailableError: + return empty(503) except PreviewTimeoutError: return empty(504) except PreviewError as e: @@ -539,18 +640,16 @@ async def preview(req, path): return raw(img, headers=headers) -def dispatch(path, quality, maxsize, maxzoom): +def dispatch(path, quality, maxsize, maxzoom, data=None): backend = "unknown" try: + if data is not None: + 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) - if suffix in OFFICE_PREVIEW_SUFFIXES: - backend = "onlyoffice" - return process_office( - path, quality=quality, maxsize=maxsize, maxzoom=maxzoom - ) mime_type, _ = mimetypes.guess_type(path.name) if mime_type and mime_type.startswith("video/"): backend = "video" @@ -592,6 +691,29 @@ def process_image_pyvips(path, *, maxsize, quality): ) +def process_image_buffer(data: bytes, *, quality, maxsize, maxzoom): + t_start = perf_counter() + img = pyvips.Image.new_from_buffer(data, "") + img = img.autorot() + 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, + strip=True, + ) + t_end = perf_counter() + + return ret, PreviewResponse( + ok=True, + mime="image/avif", + backend="pyvips", + timings=[round((t_end - t_start) * 1000, 1)], + ) + + def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0): t_load_start = perf_counter() pdf = fitz.open(path) diff --git a/cista/preview_worker.py b/cista/preview_worker.py index 2d3dd57..087209c 100644 --- a/cista/preview_worker.py +++ b/cista/preview_worker.py @@ -2,9 +2,12 @@ Two modes are supported: 1) Legacy one-shot mode: argv has path/quality/maxsize/maxzoom. -2) Long-lived mode: read JSONL commands from stdin and write framed responses. +2) Long-lived mode: read framed requests from stdin and write framed responses. -Framed response format: +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). """ @@ -40,6 +43,30 @@ _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: meta_bytes = _enc.encode(resp) packet = struct.pack(" None: from cista.preview import dispatch while True: - line = sys.stdin.buffer.readline() - if not line: + 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): - req = _dec_req.decode(line) result, resp = dispatch( - Path(req.path), req.quality, req.maxsize, req.maxzoom + Path(req.path), req.quality, req.maxsize, req.maxzoom, data ) if not resp.ok: captured = stderr_capture.getvalue().strip() diff --git a/frontend/src/components/MediaPreview.vue b/frontend/src/components/MediaPreview.vue index 74b7e77..da3bd22 100644 --- a/frontend/src/components/MediaPreview.vue +++ b/frontend/src/components/MediaPreview.vue @@ -20,6 +20,7 @@