| 1 | from __future__ import annotations |
| 2 | |
| 3 | import time |
| 4 | import uuid |
| 5 | from dataclasses import dataclass, field |
| 6 | from pathlib import Path |
| 7 | from typing import Any |
| 8 | |
| 9 | from plugins._office.helpers import document_store |
| 10 | |
| 11 | |
| 12 | @dataclass |
| 13 | class MarkdownSession: |
| 14 | session_id: str |
| 15 | file_id: str |
| 16 | sid: str |
| 17 | context_id: str |
| 18 | extension: str |
| 19 | path: str |
| 20 | title: str |
| 21 | text: str = "" |
| 22 | dirty: bool = False |
| 23 | active: bool = False |
| 24 | base_sha256: str = "" |
| 25 | base_version: str = "" |
| 26 | external_modified: bool = False |
| 27 | external_version: str = "" |
| 28 | opened_at: float = field(default_factory=time.time) |
| 29 | updated_at: float = field(default_factory=time.time) |
| 30 | last_active_at: float = field(default_factory=time.time) |
| 31 | |
| 32 | |
| 33 | class MarkdownSessionManager: |
| 34 | """Owns native Editor sessions for Markdown and plain text documents.""" |
| 35 | |
| 36 | def __init__(self) -> None: |
| 37 | self._sessions: dict[str, MarkdownSession] = {} |
| 38 | self._active_by_context: dict[str, str] = {} |
| 39 | |
| 40 | def open(self, doc: dict[str, Any], sid: str = "", context_id: str = "", refresh: bool = False) -> dict[str, Any]: |
| 41 | ext = str(doc["extension"]).lower() |
| 42 | if ext not in document_store.EDITOR_TEXT_EXTENSIONS: |
| 43 | raise ValueError(f"Editor is only available for Markdown and text files. Open .{ext} files in the Desktop.") |
| 44 | |
| 45 | normalized_context = str(context_id or "") |
| 46 | if refresh: |
| 47 | try: |
| 48 | doc = document_store.register_document(doc["path"], context_id=normalized_context) |
| 49 | except Exception: |
| 50 | pass |
| 51 | |
| 52 | for session in self._sessions.values(): |
| 53 | if session.file_id != doc["file_id"] or session.context_id != normalized_context: |
| 54 | continue |
| 55 | if sid: |
| 56 | session.sid = sid |
| 57 | doc_sha = str(doc.get("sha256") or "") |
| 58 | should_reload = not session.dirty and (refresh or (doc_sha and doc_sha != session.base_sha256)) |
| 59 | if should_reload: |
| 60 | session.text = document_store.read_text_for_editor(doc) |
| 61 | session.dirty = False |
| 62 | _set_session_base(session, doc) |
| 63 | elif session.dirty and doc_sha and doc_sha != session.base_sha256: |
| 64 | _mark_session_external(session, doc) |
| 65 | session.path = doc["path"] |
| 66 | session.title = doc["basename"] |
| 67 | session.extension = ext |
| 68 | session.updated_at = time.time() |
| 69 | self.activate(session.session_id) |
| 70 | return self._payload(session, doc) |
| 71 | |
| 72 | session = MarkdownSession( |
| 73 | session_id=uuid.uuid4().hex, |
| 74 | file_id=doc["file_id"], |
| 75 | sid=sid, |
| 76 | context_id=normalized_context, |
| 77 | extension=ext, |
| 78 | path=doc["path"], |
| 79 | title=doc["basename"], |
| 80 | text=document_store.read_text_for_editor(doc), |
| 81 | ) |
| 82 | _set_session_base(session, doc) |
| 83 | self._sessions[session.session_id] = session |
| 84 | self.activate(session.session_id) |
| 85 | return self._payload(session, doc) |
| 86 | |
| 87 | def input(self, session_id: str, text: str | None = None, patch: dict[str, Any] | None = None) -> dict[str, Any]: |
| 88 | session = self._require(session_id) |
| 89 | if text is not None: |
| 90 | session.text = str(text) |
| 91 | elif patch: |
| 92 | session.text = _apply_text_patch(session.text, patch) |
| 93 | session.dirty = True |
| 94 | session.updated_at = time.time() |
| 95 | self.activate(session.session_id) |
| 96 | return {"ok": True, "session_id": session.session_id} |
| 97 | |
| 98 | def save(self, session_id: str, text: str | None = None) -> dict[str, Any]: |
| 99 | session = self._require(session_id) |
| 100 | if text is not None: |
| 101 | session.text = str(text) |
| 102 | |
| 103 | conflict = self._save_conflict(session) |
| 104 | if conflict is not None: |
| 105 | return conflict |
| 106 | |
| 107 | updated = document_store.write_text_document(session.file_id, session.text) |
| 108 | session.updated_at = time.time() |
| 109 | session.dirty = False |
| 110 | session.path = updated["path"] |
| 111 | session.title = updated["basename"] |
| 112 | session.extension = str(updated.get("extension") or session.extension).lower() |
| 113 | _set_session_base(session, updated) |
| 114 | self._refresh_file_sessions( |
| 115 | updated, |
| 116 | text=session.text, |
| 117 | dirty=False, |
| 118 | source_session_id=session.session_id, |
| 119 | ) |
| 120 | return { |
| 121 | "ok": True, |
| 122 | "document": _public_doc(updated), |
| 123 | "version": document_store.item_version(updated), |
| 124 | } |
| 125 | |
| 126 | def save_as(self, session_id: str, path: str, text: str | None = None) -> dict[str, Any]: |
| 127 | session = self._require(session_id) |
| 128 | if text is not None: |
| 129 | session.text = str(text) |
| 130 | |
| 131 | updated = document_store.save_text_document_as( |
| 132 | session.file_id, |
| 133 | path, |
| 134 | session.text, |
| 135 | context_id=session.context_id, |
| 136 | ) |
| 137 | old_file_id = session.file_id |
| 138 | session.file_id = updated["file_id"] |
| 139 | session.updated_at = time.time() |
| 140 | session.dirty = False |
| 141 | session.path = updated["path"] |
| 142 | session.title = updated["basename"] |
| 143 | session.extension = str(updated.get("extension") or session.extension).lower() |
| 144 | _set_session_base(session, updated) |
| 145 | return { |
| 146 | "ok": True, |
| 147 | "previous_file_id": old_file_id, |
| 148 | "document": _public_doc(updated), |
| 149 | "version": document_store.item_version(updated), |
| 150 | } |
| 151 | |
| 152 | def activate(self, session_id: str) -> dict[str, Any]: |
| 153 | session = self._require(session_id) |
| 154 | now = time.time() |
| 155 | previous_id = self._active_by_context.get(session.context_id) |
| 156 | if previous_id and previous_id in self._sessions: |
| 157 | self._sessions[previous_id].active = False |
| 158 | session.active = True |
| 159 | session.last_active_at = now |
| 160 | session.updated_at = now |
| 161 | self._active_by_context[session.context_id] = session.session_id |
| 162 | return {"ok": True, "session_id": session.session_id} |
| 163 | |
| 164 | def renamed(self, file_id: str, doc: dict[str, Any], text: str | None = None) -> dict[str, Any]: |
| 165 | updated = self._refresh_file_sessions(doc, text=text, dirty=False, refresh_dirty=True) |
| 166 | return {"ok": True, "updated": updated, "file_id": file_id} |
| 167 | |
| 168 | def refresh_document(self, file_id: str) -> dict[str, Any]: |
| 169 | normalized = str(file_id or "").strip() |
| 170 | if not normalized: |
| 171 | return {"ok": True, "refreshed": 0, "sessions": []} |
| 172 | try: |
| 173 | doc = document_store.get_document(normalized) |
| 174 | except Exception: |
| 175 | return {"ok": False, "refreshed": 0, "sessions": []} |
| 176 | if str(doc.get("extension") or "").lower() not in document_store.EDITOR_TEXT_EXTENSIONS: |
| 177 | return {"ok": True, "refreshed": 0, "sessions": []} |
| 178 | |
| 179 | refreshed = self._refresh_file_sessions( |
| 180 | doc, |
| 181 | text=document_store.read_text_for_editor(doc), |
| 182 | dirty=False, |
| 183 | ) |
| 184 | return {"ok": True, "refreshed": len(refreshed), "sessions": refreshed} |
| 185 | |
| 186 | def sync_external_file_mutations(self, paths: list[str] | tuple[str, ...] | str, context_id: str = "") -> dict[str, Any]: |
| 187 | raw_paths = [paths] if isinstance(paths, str) else list(paths or []) |
| 188 | normalized_paths = [str(path or "").strip() for path in raw_paths if str(path or "").strip()] |
| 189 | if not normalized_paths: |
| 190 | return {"ok": True, "matched": 0, "sessions": []} |
| 191 | |
| 192 | matched_file_ids: set[str] = set() |
| 193 | matched_sessions: list[str] = [] |
| 194 | for session in list(self._sessions.values()): |
| 195 | if not any(_paths_match(path, session.path, session.context_id) for path in normalized_paths): |
| 196 | continue |
| 197 | matched_file_ids.add(session.file_id) |
| 198 | matched_sessions.append(session.session_id) |
| 199 | |
| 200 | for file_id in matched_file_ids: |
| 201 | sessions = [session for session in self._sessions.values() if session.file_id == file_id] |
| 202 | if not sessions: |
| 203 | continue |
| 204 | session = sessions[0] |
| 205 | try: |
| 206 | doc = document_store.register_document(session.path, context_id=session.context_id) |
| 207 | except Exception: |
| 208 | try: |
| 209 | doc = document_store.get_document(file_id) |
| 210 | except Exception: |
| 211 | continue |
| 212 | for target in sessions: |
| 213 | _mark_session_external(target, doc) |
| 214 | continue |
| 215 | self.refresh_document(file_id) |
| 216 | |
| 217 | return { |
| 218 | "ok": True, |
| 219 | "matched": len(matched_file_ids), |
| 220 | "sessions": matched_sessions, |
| 221 | } |
| 222 | |
| 223 | def list_open(self, context_id: str = "", limit: int = 20) -> list[dict[str, Any]]: |
| 224 | context_id = str(context_id or "") |
| 225 | sessions = [session for session in self._sessions.values() if session.context_id == context_id] |
| 226 | sessions.sort(key=lambda item: (not item.active, -item.last_active_at, -item.updated_at)) |
| 227 | |
| 228 | grouped: dict[str, MarkdownSession] = {} |
| 229 | counts: dict[str, int] = {} |
| 230 | for session in sessions: |
| 231 | key = session.file_id or session.path |
| 232 | counts[key] = counts.get(key, 0) + 1 |
| 233 | current = grouped.get(key) |
| 234 | if current is None or session.active or session.last_active_at > current.last_active_at: |
| 235 | grouped[key] = session |
| 236 | |
| 237 | result = [] |
| 238 | for session in grouped.values(): |
| 239 | try: |
| 240 | doc = document_store.get_document(session.file_id) |
| 241 | version = document_store.item_version(doc) |
| 242 | size = doc.get("size", 0) |
| 243 | last_modified = doc.get("last_modified", "") |
| 244 | path = document_store.display_path(doc.get("path", session.path)) |
| 245 | title = doc.get("basename") or session.title |
| 246 | except Exception: |
| 247 | version = "" |
| 248 | size = 0 |
| 249 | last_modified = "" |
| 250 | path = document_store.display_path(session.path) |
| 251 | title = session.title |
| 252 | result.append({ |
| 253 | "session_id": session.session_id, |
| 254 | "file_id": session.file_id, |
| 255 | "title": title, |
| 256 | "extension": session.extension, |
| 257 | "path": path, |
| 258 | "version": version, |
| 259 | "size": size, |
| 260 | "last_modified": last_modified, |
| 261 | "dirty": session.dirty, |
| 262 | "active": session.active, |
| 263 | "external_modified": session.external_modified, |
| 264 | "external_version": session.external_version, |
| 265 | "open_sessions": counts.get(session.file_id or session.path, 1), |
| 266 | "last_active_at": session.last_active_at, |
| 267 | }) |
| 268 | |
| 269 | result.sort(key=lambda item: (not item["active"], -float(item["last_active_at"] or 0))) |
| 270 | safe_limit = max(1, int(limit or 20)) |
| 271 | return result[:safe_limit] |
| 272 | |
| 273 | def close(self, session_id: str) -> dict[str, Any]: |
| 274 | session = self._sessions.pop(str(session_id or ""), None) |
| 275 | if not session: |
| 276 | return {"ok": True, "closed": 0} |
| 277 | if self._active_by_context.get(session.context_id) == session.session_id: |
| 278 | replacement = sorted( |
| 279 | [item for item in self._sessions.values() if item.context_id == session.context_id], |
| 280 | key=lambda item: item.last_active_at, |
| 281 | reverse=True, |
| 282 | ) |
| 283 | if replacement: |
| 284 | self.activate(replacement[0].session_id) |
| 285 | else: |
| 286 | self._active_by_context.pop(session.context_id, None) |
| 287 | return {"ok": True, "closed": 1, "session_id": session_id} |
| 288 | |
| 289 | def close_sid(self, sid: str) -> int: |
| 290 | doomed = [session_id for session_id, session in self._sessions.items() if session.sid == sid] |
| 291 | for session_id in doomed: |
| 292 | self.close(session_id) |
| 293 | return len(doomed) |
| 294 | |
| 295 | def _save_conflict(self, session: MarkdownSession) -> dict[str, Any] | None: |
| 296 | try: |
| 297 | doc = document_store.get_document(session.file_id) |
| 298 | except Exception as exc: |
| 299 | return { |
| 300 | "ok": False, |
| 301 | "code": "editor_document_missing", |
| 302 | "error": f"Editor save failed because the document metadata is missing: {exc}", |
| 303 | } |
| 304 | |
| 305 | path = Path(doc["path"]) |
| 306 | desired = str(session.text or "").encode("utf-8") |
| 307 | desired_sha = document_store.sha256_bytes(desired) |
| 308 | current_exists = path.exists() |
| 309 | current = path.read_bytes() if current_exists else b"" |
| 310 | current_sha = document_store.sha256_bytes(current) if current_exists else "" |
| 311 | expected_sha = session.base_sha256 or str(doc.get("sha256") or "") |
| 312 | |
| 313 | if expected_sha and current_sha != expected_sha: |
| 314 | if current_exists and desired_sha == current_sha: |
| 315 | updated = document_store.register_document(path, context_id=session.context_id) |
| 316 | session.dirty = False |
| 317 | session.path = updated["path"] |
| 318 | session.title = updated["basename"] |
| 319 | _set_session_base(session, updated) |
| 320 | self._refresh_file_sessions( |
| 321 | updated, |
| 322 | text=session.text, |
| 323 | dirty=False, |
| 324 | source_session_id=session.session_id, |
| 325 | ) |
| 326 | return { |
| 327 | "ok": True, |
| 328 | "document": _public_doc(updated), |
| 329 | "version": document_store.item_version(updated), |
| 330 | } |
| 331 | |
| 332 | latest_doc = _refresh_registered_doc(doc, context_id=session.context_id) |
| 333 | _mark_session_external(session, latest_doc) |
| 334 | return { |
| 335 | "ok": False, |
| 336 | "code": "external_change_conflict", |
| 337 | "error": ( |
| 338 | "This file changed on disk since the Editor loaded it. " |
| 339 | "Reload it before saving to avoid overwriting the newer file." |
| 340 | ), |
| 341 | "document": _public_doc(latest_doc), |
| 342 | "version": document_store.item_version(latest_doc), |
| 343 | } |
| 344 | |
| 345 | return None |
| 346 | |
| 347 | def _refresh_file_sessions( |
| 348 | self, |
| 349 | doc: dict[str, Any], |
| 350 | text: str | None = None, |
| 351 | dirty: bool | None = None, |
| 352 | *, |
| 353 | source_session_id: str = "", |
| 354 | refresh_dirty: bool = False, |
| 355 | ) -> list[str]: |
| 356 | file_id = str(doc.get("file_id") or "").strip() |
| 357 | refreshed: list[str] = [] |
| 358 | for session in self._sessions.values(): |
| 359 | if session.file_id != file_id: |
| 360 | continue |
| 361 | can_replace_text = ( |
| 362 | text is not None |
| 363 | and ( |
| 364 | refresh_dirty |
| 365 | or not session.dirty |
| 366 | or (source_session_id and session.session_id == source_session_id) |
| 367 | ) |
| 368 | ) |
| 369 | if can_replace_text: |
| 370 | session.text = str(text) |
| 371 | session.path = doc["path"] |
| 372 | session.title = doc["basename"] |
| 373 | session.extension = str(doc.get("extension") or session.extension).lower() |
| 374 | if dirty is not None and (can_replace_text or refresh_dirty or not session.dirty): |
| 375 | session.dirty = dirty |
| 376 | if can_replace_text or not session.dirty: |
| 377 | _set_session_base(session, doc) |
| 378 | elif text is not None: |
| 379 | _mark_session_external(session, doc) |
| 380 | session.updated_at = time.time() |
| 381 | refreshed.append(session.session_id) |
| 382 | return refreshed |
| 383 | |
| 384 | def _payload(self, session: MarkdownSession, doc: dict[str, Any]) -> dict[str, Any]: |
| 385 | return { |
| 386 | "ok": True, |
| 387 | "session_id": session.session_id, |
| 388 | "file_id": session.file_id, |
| 389 | "title": session.title, |
| 390 | "extension": session.extension, |
| 391 | "path": session.path, |
| 392 | "text": session.text, |
| 393 | "dirty": session.dirty, |
| 394 | "active": session.active, |
| 395 | "external_modified": session.external_modified, |
| 396 | "external_version": session.external_version, |
| 397 | "context_id": session.context_id, |
| 398 | "document": _public_doc(doc), |
| 399 | "version": document_store.item_version(doc), |
| 400 | } |
| 401 | |
| 402 | def _require(self, session_id: str) -> MarkdownSession: |
| 403 | normalized = str(session_id or "").strip() |
| 404 | session = self._sessions.get(normalized) |
| 405 | if not session: |
| 406 | raise FileNotFoundError(f"Editor session not found: {normalized}") |
| 407 | return session |
| 408 | |
| 409 | |
| 410 | def get_manager() -> MarkdownSessionManager: |
| 411 | global _manager |
| 412 | try: |
| 413 | return _manager |
| 414 | except NameError: |
| 415 | _manager = MarkdownSessionManager() |
| 416 | return _manager |
| 417 | |
| 418 | |
| 419 | def _public_doc(doc: dict[str, Any]) -> dict[str, Any]: |
| 420 | return { |
| 421 | "file_id": doc["file_id"], |
| 422 | "path": document_store.display_path(doc["path"]), |
| 423 | "basename": doc["basename"], |
| 424 | "title": doc["basename"], |
| 425 | "extension": doc["extension"], |
| 426 | "size": doc["size"], |
| 427 | "version": document_store.item_version(doc), |
| 428 | "last_modified": doc["last_modified"], |
| 429 | "exists": Path(doc["path"]).exists(), |
| 430 | } |
| 431 | |
| 432 | |
| 433 | def _set_session_base(session: MarkdownSession, doc: dict[str, Any]) -> None: |
| 434 | session.base_sha256 = str(doc.get("sha256") or "") |
| 435 | session.base_version = document_store.item_version(doc) |
| 436 | session.external_modified = False |
| 437 | session.external_version = "" |
| 438 | |
| 439 | |
| 440 | def _mark_session_external(session: MarkdownSession, doc: dict[str, Any]) -> None: |
| 441 | session.external_modified = True |
| 442 | try: |
| 443 | session.external_version = document_store.item_version(doc) |
| 444 | except Exception: |
| 445 | session.external_version = "" |
| 446 | |
| 447 | |
| 448 | def _refresh_registered_doc(doc: dict[str, Any], context_id: str = "") -> dict[str, Any]: |
| 449 | try: |
| 450 | path = Path(doc["path"]) |
| 451 | if path.exists(): |
| 452 | return document_store.register_document(path, context_id=context_id) |
| 453 | except Exception: |
| 454 | pass |
| 455 | return doc |
| 456 | |
| 457 | |
| 458 | def _paths_match(left: str, right: str, context_id: str = "") -> bool: |
| 459 | left_path = _normalize_path_for_compare(left, context_id=context_id) |
| 460 | right_path = _normalize_path_for_compare(right, context_id=context_id) |
| 461 | return bool(left_path and right_path and left_path == right_path) |
| 462 | |
| 463 | |
| 464 | def _normalize_path_for_compare(path: str, context_id: str = "") -> str: |
| 465 | value = str(path or "").strip() |
| 466 | if not value: |
| 467 | return "" |
| 468 | try: |
| 469 | return str(document_store.normalize_path(value, context_id=context_id)) |
| 470 | except Exception: |
| 471 | pass |
| 472 | try: |
| 473 | return str(document_store._path_from_a0(value).resolve(strict=False)) |
| 474 | except Exception: |
| 475 | return str(Path(value).expanduser().resolve(strict=False)) |
| 476 | |
| 477 | |
| 478 | def _apply_text_patch(text: str, patch: dict[str, Any]) -> str: |
| 479 | if "content" in patch: |
| 480 | return str(patch.get("content") or "") |
| 481 | start = int(patch.get("start") or 0) |
| 482 | end = int(patch.get("end") if patch.get("end") is not None else start) |
| 483 | replacement = str(patch.get("text") or "") |
| 484 | start = max(0, min(len(text), start)) |
| 485 | end = max(start, min(len(text), end)) |
| 486 | return text[:start] + replacement + text[end:] |