vibe-sketcher/Documentation

Source code for vibesketcher.document.io

"""Read and write ``.vsk`` files.

A document is UTF-8 JSON, indented by two spaces, with a trailing newline,
so a ``.vsk`` diffs cleanly under version control. Writing is atomic: the
file is written beside its target and renamed over it, so an interrupted
save never leaves half a document.
"""

from __future__ import annotations

import json
import os
import stat
import tempfile
import threading
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from pydantic import ValidationError

from vibesketcher.document.model import Document, Edit, Provenance

__all__ = [
    "ABSENT",
    "DocumentError",
    "SaveRefused",
    "check_output",
    "digest_of",
    "dumps",
    "load",
    "loads",
    "new_document",
    "save",
    "validation_errors",
    "with_edit",
    "write_output",
]

#: The file extension of a vibe-sketcher document.
SUFFIX = ".vsk"




class DocumentError(ValueError):
    """A document that is not valid JSON or not a valid vibe-sketcher document.

    ``errors`` lists every problem as ``"location: message"``, the form the
    CLI prints and the REST API returns.
    """

    def __init__(self, errors: list[str]) -> None:
        super().__init__("invalid vibe-sketcher document:\n  " + "\n  ".join(errors))
        self.errors = errors





class SaveRefused(ValueError):
    """Writing would discard QVF result data or replace a changed target."""





def validation_errors(error: ValidationError) -> list[str]:
    """Readable ``location: message`` lines for a pydantic validation error."""
    lines = []
    for item in error.errors(include_url=False):
        location = ".".join(str(part) for part in item["loc"]) or "document"
        lines.append(f"{location}: {item['msg']}")
    return lines



def from_data(data: Any) -> Document:
    """Validate already-parsed JSON data as a document, within :mod:`vibesketcher.limits`."""
    from vibesketcher.limits import check_depth

    try:
        check_depth(data, "the document")
    except ValueError as error:
        raise DocumentError([f"document: {error}"]) from None
    try:
        return Document.model_validate(data)
    except ValidationError as error:
        raise DocumentError(validation_errors(error)) from error




def loads(text: str) -> Document:
    """Parse and validate a document from JSON text, within :mod:`vibesketcher.limits`."""
    from vibesketcher.limits import parse_json

    try:
        data = parse_json(text, "the document")
    except ValueError as error:
        raise DocumentError([f"document: {error}"]) from error
    return from_data(data)





def load(path: str | os.PathLike[str]) -> Document:
    """Read and validate a document: a ``.qvf`` archive or a ``.vsk`` file."""
    return _read_snapshot(path).document



@dataclass(frozen=True)
class _DocumentSnapshot:
    """One bounded read: parsed contents and their exact source, including absence.

    The digest is derived from the bytes parsed below, never from a subsequent
    filesystem lookup. A Store may bind this receipt without rereading the file.
    """

    document: Document
    path: Path
    destination: Path
    digest: str
    result: dict[str, Any] | None = None


def _read_snapshot(
    path: str | os.PathLike[str], *, allow_missing: bool = False
) -> _DocumentSnapshot:
    import hashlib

    from vibesketcher.limits import MAX_INPUT_BYTES, MAX_QVF_BYTES, LimitExceeded, read_file

    source = Path(path).absolute()
    destination = source.resolve()
    qvf = source.suffix.lower() == ".qvf"
    try:
        content = read_file(
            destination, source.name, limit=MAX_QVF_BYTES if qvf else MAX_INPUT_BYTES
        )
    except FileNotFoundError:
        if not allow_missing:
            raise
        # Do not look again: a file created after this observation must not
        # become the baseline of a new, empty document's first save.
        _check_destination(source, destination)
        return _DocumentSnapshot(new_document(), source, destination, ABSENT)
    except LimitExceeded as error:
        raise DocumentError([f"{source.name}: {error}"]) from None
    result = None
    if qvf:
        from vibesketcher.document.qvf import QvfError, from_qvf

        try:
            derived = from_qvf(content)
        except QvfError as error:
            raise DocumentError([f"{source.name}: {error}"]) from None
        document, result = derived.document, derived.result
    else:
        try:
            document = loads(content.decode("utf-8"))
        except UnicodeDecodeError as error:
            raise DocumentError([f"{source.name}: not UTF-8 text ({error})"]) from None
    _check_destination(source, destination)
    return _DocumentSnapshot(
        document, source, destination, hashlib.sha256(content).hexdigest(), result
    )




def dumps(document: Document) -> str:
    """The document as ``.vsk`` text."""
    return json.dumps(document.to_json_data(), indent=2, ensure_ascii=False) + "\n"



_UMASK_LOCK = threading.Lock()


def _new_file_mode() -> int:
    """The mode a newly created file gets under the current umask."""
    with _UMASK_LOCK:  # reading the umask means setting it; keep threads out
        umask = os.umask(0o022)
        os.umask(umask)
    return 0o666 & ~umask


#: What :func:`digest_of` says of a file that does not exist.
ABSENT = "absent"

_CHANGED = (
    "The file changed outside vibe-sketcher since it was opened or last saved, so saving "
    "would discard that change. Reopen the file to see it, or keep this version with "
    "Save As or a download under another name."
)




def digest_of(path: str | os.PathLike[str]) -> str:
    """The SHA-256 of a file's bytes, or :data:`ABSENT` when there is no file."""
    import hashlib

    try:
        digest = hashlib.sha256()
        with Path(path).open("rb") as source:
            for chunk in iter(lambda: source.read(1024 * 1024), b""):
                digest.update(chunk)
        return digest.hexdigest()
    except FileNotFoundError:
        return ABSENT





def save(document: Document, path: str | os.PathLike[str], *, expect: str | None = None) -> Path:
    """Write ``document`` to ``path`` atomically; returns the path written.

    A ``.qvf`` path gets a QVF archive (raising
    :class:`~vibesketcher.document.qvf.QvfUnavailable` without qvf-writer), any
    other a ``.vsk``. A file that exists keeps its permissions; a new one gets
    the umask's. QVF results and archives with data outside the native sketch
    raise :class:`SaveRefused`; save the sketch to another path instead.
    With ``expect`` (a :func:`digest_of`), a file that no longer holds those
    bytes is not replaced either.
    """
    return _save_with_digest(document, path, expect=expect)[0]



def _save_with_digest(
    document: Document,
    path: str | os.PathLike[str],
    *,
    expect: str | None = None,
    destination: Path | None = None,
) -> tuple[Path, str]:
    """Save once and bind the receipt to the serialized bytes, not a later file read."""
    import hashlib

    target = Path(path)
    if destination is not None:
        _check_destination(target, destination)
    if target.suffix.lower() == ".qvf":
        from vibesketcher.document.qvf import to_qvf

        content = to_qvf(document)
    else:
        content = dumps(document).encode("utf-8")
    written = _write_output(target, content, expect=expect, destination=destination)
    digest = hashlib.sha256(content).hexdigest()
    if digest_of(written) != digest:
        raise SaveRefused(_CHANGED)
    return written, digest


def _qvf_snapshot(target: Path) -> bytes | None:
    from vibesketcher.limits import MAX_QVF_BYTES, LimitExceeded, read_file

    try:
        return read_file(target, target.name, limit=MAX_QVF_BYTES)
    except FileNotFoundError:
        return None
    except LimitExceeded as error:
        raise SaveRefused(f"Cannot inspect the QVF destination: {error}. Use Save As.") from None




def check_output(path: str | os.PathLike[str]) -> None:
    """Refuse a protected QVF destination before an operation with side effects.

    :func:`write_output` checks again at the actual write; this preflight is
    useful before submitting jobs whose run records must be saved in place.
    """
    target = Path(path)
    if ".qvf" in (target.suffix.lower(), target.resolve().suffix.lower()):
        from vibesketcher.document.qvf import overwrite_refusal

        existing = _qvf_snapshot(target)
        if existing is not None and (reason := overwrite_refusal(existing)):
            raise SaveRefused(reason)





def write_output(
    path: str | os.PathLike[str], content: bytes, *, expect: str | None = None
) -> Path:
    """Atomically write exported bytes, protecting existing QVF result files.

    This also guards explicit format overrides (such as SVG sent to a QVF
    filename). Symlinks retain their link and write through to their target.
    With ``expect``, the file must still hold the bytes that :func:`digest_of`
    measured (or still be absent), before writing and again at the rename;
    otherwise :class:`SaveRefused`, and the file is left as it is.

    The new bytes reach the disk before they replace the old ones, and the
    directory entry after, so a crash or a full disk leaves either the old file
    or the new one, never a mixture. A crash at the wrong moment can leave a
    hidden ``.NAME.*.tmp`` beside it, which is safe to delete.
    """
    return _write_output(path, content, expect=expect)



def _write_output(
    path: str | os.PathLike[str],
    content: bytes,
    *,
    expect: str | None = None,
    destination: Path | None = None,
) -> Path:
    """Use a caller's persistent destination when supplied, without resolving a new one."""
    original = Path(path)
    target = original.resolve() if destination is None else destination
    _check_destination(original, target)
    if expect is not None and digest_of(target) != expect:
        raise SaveRefused(_CHANGED)
    guarded = ".qvf" in (original.suffix.lower(), target.suffix.lower())
    previous = _qvf_snapshot(target) if guarded else None
    if guarded and previous is not None:
        from vibesketcher.document.qvf import overwrite_refusal

        if reason := overwrite_refusal(previous):
            raise SaveRefused(reason)
    mode = stat.S_IMODE(target.stat().st_mode) if target.exists() else _new_file_mode()
    descriptor, temporary = tempfile.mkstemp(
        prefix=f".{target.name}.", suffix=".tmp", dir=target.parent
    )
    try:
        with os.fdopen(descriptor, "wb") as stream:
            stream.write(content)
            stream.flush()
            os.fsync(stream.fileno())
        os.chmod(temporary, mode)
        if guarded and _qvf_snapshot(target) != previous:
            raise SaveRefused("The QVF destination changed during saving. Use Save As.")
        if expect is not None and digest_of(target) != expect:
            raise SaveRefused(_CHANGED)
        _check_destination(original, target)
        os.replace(temporary, target)
    except BaseException:
        Path(temporary).unlink(missing_ok=True)
        raise
    _sync_directory(target.parent)
    _check_destination(original, target)
    return original


def _check_destination(original: Path, target: Path) -> None:
    """A selected alias or symlinked parent must still identify the same destination."""
    try:
        unchanged = original.resolve() == target
    except (OSError, RuntimeError):
        unchanged = False
    if not unchanged:
        raise SaveRefused("The save destination changed during saving. Reopen it or use Save As.")


def _sync_directory(directory: Path) -> None:
    """Make a rename in ``directory`` durable; a no-op where directories cannot be synced."""
    flags = getattr(os, "O_DIRECTORY", None)
    if flags is None:
        return
    try:
        descriptor = os.open(directory, os.O_RDONLY | flags)
    except OSError:
        return
    try:
        os.fsync(descriptor)
    except OSError:
        pass  # some file systems refuse; the file itself is already synced
    finally:
        os.close(descriptor)




def new_document(app: str | None = None) -> Document:
    """An empty document stamped with this application and the current time."""
    if app is None:
        from vibesketcher import __version__

        app = f"vibe-sketcher {__version__}"
    return Document(provenance=Provenance(app=app))





def with_edit(document: Document, *, actor: str, client: str, summary: str) -> Document:
    """``document`` with an edit recorded for its next revision.

    A run of alike edits (same actor, client and summary) folds into one
    entry, so a drawing session does not leave hundreds. The revision itself
    is the caller's to advance: the server's store does it on commit.
    """
    from datetime import UTC, datetime

    now = datetime.now(UTC).replace(microsecond=0)
    edits = list(document.provenance.edits)
    revision = document.revision + 1
    if edits and (edits[-1].actor, edits[-1].client, edits[-1].summary) == (actor, client, summary):
        edits[-1] = edits[-1].model_copy(update={"revision": revision, "at": now})
    else:
        fields = {"actor": actor, "client": client, "summary": summary}
        edits.append(Edit.model_validate({"revision": revision, "at": now, **fields}))
    provenance = document.provenance.model_copy(update={"edits": edits})
    return document.model_copy(update={"provenance": provenance})