| 1 | from __future__ import annotations |
| 2 | |
| 3 | from typing import Any |
| 4 | |
| 5 | from helpers.ws import WsHandler |
| 6 | from helpers.ws_manager import WsResult |
| 7 | from plugins._editor.helpers import markdown_sessions |
| 8 | from plugins._office.helpers import document_store |
| 9 | |
| 10 | |
| 11 | class WsEditor(WsHandler): |
| 12 | async def on_disconnect(self, sid: str) -> None: |
| 13 | markdown_sessions.get_manager().close_sid(sid) |
| 14 | |
| 15 | async def process(self, event: str, data: dict[str, Any], sid: str) -> dict[str, Any] | WsResult | None: |
| 16 | if not event.startswith("editor_"): |
| 17 | return None |
| 18 | try: |
| 19 | if event == "editor_open": |
| 20 | return self._open(data, sid) |
| 21 | if event == "editor_input": |
| 22 | return markdown_sessions.get_manager().input( |
| 23 | str(data.get("session_id") or ""), |
| 24 | text=data.get("text") if "text" in data else None, |
| 25 | patch=data.get("patch") if isinstance(data.get("patch"), dict) else None, |
| 26 | ) |
| 27 | if event == "editor_save": |
| 28 | return markdown_sessions.get_manager().save( |
| 29 | str(data.get("session_id") or ""), |
| 30 | text=data.get("text") if "text" in data else None, |
| 31 | ) |
| 32 | if event == "editor_activate": |
| 33 | return markdown_sessions.get_manager().activate(str(data.get("session_id") or "")) |
| 34 | if event == "editor_close": |
| 35 | return markdown_sessions.get_manager().close(str(data.get("session_id") or "")) |
| 36 | except FileNotFoundError as exc: |
| 37 | return WsResult.error(code="EDITOR_SESSION_NOT_FOUND", message=str(exc), correlation_id=data.get("correlationId")) |
| 38 | except Exception as exc: |
| 39 | return WsResult.error(code="EDITOR_ERROR", message=str(exc), correlation_id=data.get("correlationId")) |
| 40 | |
| 41 | return WsResult.error( |
| 42 | code="UNKNOWN_EDITOR_EVENT", |
| 43 | message=f"Unknown editor event: {event}", |
| 44 | correlation_id=data.get("correlationId"), |
| 45 | ) |
| 46 | |
| 47 | def _open(self, data: dict[str, Any], sid: str) -> dict[str, Any] | WsResult: |
| 48 | context_id = str(data.get("ctxid") or data.get("context_id") or "") |
| 49 | file_id = str(data.get("file_id") or "").strip() |
| 50 | path = str(data.get("path") or "").strip() |
| 51 | if file_id: |
| 52 | doc = document_store.get_document(file_id) |
| 53 | elif path: |
| 54 | doc = document_store.register_document(path, context_id=context_id) |
| 55 | else: |
| 56 | fmt = str(data.get("format") or "md").lower().strip().lstrip(".") |
| 57 | if fmt not in document_store.EDITOR_TEXT_EXTENSIONS: |
| 58 | return WsResult.error( |
| 59 | code="UNSUPPORTED_EDITOR_DOCUMENT", |
| 60 | message=f"Editor can only create Markdown (.md) and text (.txt) documents, not .{fmt}.", |
| 61 | correlation_id=data.get("correlationId"), |
| 62 | ) |
| 63 | doc = document_store.create_document( |
| 64 | kind="document", |
| 65 | title=str(data.get("title") or "Untitled"), |
| 66 | fmt=fmt, |
| 67 | content=str(data.get("content") or ""), |
| 68 | context_id=context_id, |
| 69 | ) |
| 70 | return markdown_sessions.get_manager().open( |
| 71 | doc, |
| 72 | sid=sid, |
| 73 | context_id=context_id, |
| 74 | refresh=data.get("refresh") is True, |
| 75 | ) |