"""File-backed text documents with dirty tracking and a buffer registry."""
from __future__ import annotations
import logging
from pathlib import Path
from .descriptors import Signal
log = logging.getLogger(__name__)
# Language detection from file extension
_EXTENSION_LANGUAGES = {
".py": "python",
".pyw": "python",
".js": "javascript",
".mjs": "javascript",
".ts": "typescript",
".tsx": "typescript",
".c": "c",
".h": "c",
".cpp": "cpp",
".hpp": "cpp",
".cc": "cpp",
".rs": "rust",
".go": "go",
".java": "java",
".rb": "ruby",
".lua": "lua",
".sh": "shell",
".bash": "shell",
".zsh": "shell",
".json": "json",
".toml": "toml",
".yaml": "yaml",
".yml": "yaml",
".xml": "xml",
".html": "html",
".htm": "html",
".css": "css",
".md": "markdown",
".txt": "text",
".glsl": "glsl",
".vert": "glsl",
".frag": "glsl",
}
[docs]
class Document:
"""File-backed text document with dirty tracking."""
def __init__(self, path: Path | str | None = None, encoding: str = "utf-8"):
self.path: Path | None = Path(path) if path else None
self.encoding = encoding
self.language: str = ""
self._content: str = ""
self._saved_content: str = ""
# Signals
self.content_changed = Signal()
self.dirty_changed = Signal()
self.saved = Signal()
# Auto-detect language from path
if self.path:
self.language = _EXTENSION_LANGUAGES.get(self.path.suffix.lower(), "")
@property
def content(self) -> str:
return self._content
@content.setter
def content(self, value: str):
if value == self._content:
return
was_dirty = self.dirty
self._content = value
self.content_changed.emit()
if was_dirty != self.dirty:
self.dirty_changed.emit()
@property
def dirty(self) -> bool:
return self._content != self._saved_content
@property
def title(self) -> str:
name = self.path.name if self.path else "untitled"
return f"{name}*" if self.dirty else name
[docs]
def load(self) -> bool:
"""Load content from file. Returns True on success."""
if not self.path or not self.path.exists():
return False
try:
self._content = self.path.read_text(encoding=self.encoding)
self._saved_content = self._content
self.language = _EXTENSION_LANGUAGES.get(self.path.suffix.lower(), "")
self.content_changed.emit()
self.dirty_changed.emit()
return True
except (OSError, UnicodeDecodeError) as e:
log.error("Failed to load %s: %s", self.path, e)
return False
[docs]
def save(self, path: Path | str | None = None) -> bool:
"""Save content to file. Returns True on success."""
if path:
self.path = Path(path)
if not self.path:
return False
try:
self.path.write_text(self._content, encoding=self.encoding)
self._saved_content = self._content
self.dirty_changed.emit()
self.saved.emit()
return True
except OSError as e:
log.error("Failed to save %s: %s", self.path, e)
return False
[docs]
class BufferRegistry:
"""Manages open documents with deduplication."""
def __init__(self):
self._buffers: list[Document] = []
self._active: Document | None = None
self._path_map: dict[Path, Document] = {}
# Signals
self.buffer_opened = Signal()
self.buffer_closed = Signal()
self.active_changed = Signal()
[docs]
def open(self, path: Path | str) -> Document:
"""Open a file, deduplicating by resolved path."""
resolved = Path(path).resolve()
if resolved in self._path_map:
doc = self._path_map[resolved]
self.active = doc
return doc
doc = Document(resolved)
doc.load()
self._buffers.append(doc)
self._path_map[resolved] = doc
self.buffer_opened.emit(doc)
self.active = doc
return doc
[docs]
def create(self, content: str = "") -> Document:
"""Create a new untitled buffer."""
doc = Document()
doc.content = content
doc._saved_content = content # not dirty initially
self._buffers.append(doc)
self.buffer_opened.emit(doc)
self.active = doc
return doc
[docs]
def close(self, doc: Document) -> bool:
"""Close a document. Returns False if dirty (caller should prompt)."""
if doc.dirty:
return False
if doc in self._buffers:
self._buffers.remove(doc)
if doc.path:
resolved = doc.path.resolve()
self._path_map.pop(resolved, None)
self.buffer_closed.emit(doc)
if self._active is doc:
self._active = self._buffers[-1] if self._buffers else None
self.active_changed.emit(self._active)
return True
@property
def active(self) -> Document | None:
return self._active
@active.setter
def active(self, doc: Document | None):
if doc is self._active:
return
self._active = doc
self.active_changed.emit(doc)
@property
def buffers(self) -> list[Document]:
return list(self._buffers)
@property
def has_dirty(self) -> bool:
return any(d.dirty for d in self._buffers)