| 1 | from __future__ import annotations |
| 2 | |
| 3 | from pathlib import Path |
| 4 | from typing import Any |
| 5 | |
| 6 | from helpers.ws import WsHandler |
| 7 | from helpers.ws_manager import WsResult |
| 8 | from plugins._desktop.helpers import desktop_session |
| 9 | from plugins._office.helpers import document_store |
| 10 | |
| 11 | |
| 12 | class WsOffice(WsHandler): |
| 13 | async def on_disconnect(self, sid: str) -> None: |
| 14 | return None |
| 15 | |
| 16 | async def process(self, event: str, data: dict[str, Any], sid: str) -> dict[str, Any] | WsResult | None: |
| 17 | if not event.startswith("office_"): |
| 18 | return None |
| 19 | try: |
| 20 | if event == "office_open": |
| 21 | return self._open(data, sid) |
| 22 | if event in {"office_input", "office_save", "office_close"}: |
| 23 | return { |
| 24 | "ok": False, |
| 25 | "error": "Office WebSocket editing is not available for text documents; use the Editor surface.", |
| 26 | } |
| 27 | except FileNotFoundError as exc: |
| 28 | return WsResult.error(code="OFFICE_SESSION_NOT_FOUND", message=str(exc), correlation_id=data.get("correlationId")) |
| 29 | except Exception as exc: |
| 30 | return WsResult.error(code="OFFICE_ERROR", message=str(exc), correlation_id=data.get("correlationId")) |
| 31 | |
| 32 | return WsResult.error( |
| 33 | code="UNKNOWN_OFFICE_EVENT", |
| 34 | message=f"Unknown office event: {event}", |
| 35 | correlation_id=data.get("correlationId"), |
| 36 | ) |
| 37 | |
| 38 | def _open(self, data: dict[str, Any], sid: str) -> dict[str, Any] | WsResult: |
| 39 | context_id = str(data.get("ctxid") or data.get("context_id") or "") |
| 40 | file_id = str(data.get("file_id") or "").strip() |
| 41 | path = str(data.get("path") or "").strip() |
| 42 | if file_id: |
| 43 | doc = document_store.get_document(file_id) |
| 44 | elif path: |
| 45 | doc = document_store.register_document(path, context_id=context_id) |
| 46 | else: |
| 47 | fmt = str(data.get("format") or "odt").lower().strip().lstrip(".") |
| 48 | if fmt not in desktop_session.OFFICIAL_EXTENSIONS: |
| 49 | return WsResult.error( |
| 50 | code="UNSUPPORTED_OFFICE_DOCUMENT", |
| 51 | message=f"Office can only create LibreOffice formats, not .{fmt}.", |
| 52 | correlation_id=data.get("correlationId"), |
| 53 | ) |
| 54 | doc = document_store.create_document( |
| 55 | kind=str(data.get("kind") or "document"), |
| 56 | title=str(data.get("title") or "Untitled"), |
| 57 | fmt=fmt, |
| 58 | content=str(data.get("content") or ""), |
| 59 | context_id=context_id, |
| 60 | ) |
| 61 | ext = str(doc.get("extension") or "").lower() |
| 62 | if ext in document_store.EDITOR_TEXT_EXTENSIONS: |
| 63 | return WsResult.error( |
| 64 | code="UNSUPPORTED_OFFICE_DOCUMENT", |
| 65 | message="Text documents use the Editor surface.", |
| 66 | correlation_id=data.get("correlationId"), |
| 67 | ) |
| 68 | if ext in desktop_session.OFFICIAL_EXTENSIONS: |
| 69 | return { |
| 70 | "ok": True, |
| 71 | "requires_desktop": True, |
| 72 | "file_id": doc["file_id"], |
| 73 | "title": doc["basename"], |
| 74 | "extension": doc["extension"], |
| 75 | "path": doc["path"], |
| 76 | "document": _public_doc(doc), |
| 77 | "version": document_store.item_version(doc), |
| 78 | } |
| 79 | return WsResult.error( |
| 80 | code="UNSUPPORTED_OFFICE_DOCUMENT", |
| 81 | message=f".{ext} documents are not supported by LibreOffice.", |
| 82 | correlation_id=data.get("correlationId"), |
| 83 | ) |
| 84 | |
| 85 | |
| 86 | def _public_doc(doc: dict[str, Any]) -> dict[str, Any]: |
| 87 | return { |
| 88 | "file_id": doc["file_id"], |
| 89 | "path": document_store.display_path(doc["path"]), |
| 90 | "basename": doc["basename"], |
| 91 | "title": doc["basename"], |
| 92 | "extension": doc["extension"], |
| 93 | "size": doc["size"], |
| 94 | "version": document_store.item_version(doc), |
| 95 | "last_modified": doc["last_modified"], |
| 96 | "exists": Path(doc["path"]).exists(), |
| 97 | } |