inkwell/src/inkwell/notes/helpers.py

"""Small shared helpers + constants for the notes package: display-name derivation,
board-filter narrowing, the owner-scoped fetch, filename/slug sanitizers, and the one
place an attachment's bytes are written — shared by the upload route, the importer
and sync."""
from __future__ import annotations

from .checklist import strip_marker

import hashlib
import logging
import os
import re
import uuid

from quart import g
from sqlalchemy import func, select

from ..acl import EDIT, visible_to_user
from ..config import Config
from ..models.note import Note
from ..models.note_attachment import NoteAttachment
from ..responses import parse_uuid

logger = logging.getLogger(__name__)

ALLOWED_IMAGE_MIMES = {"image/png": ".png", "image/jpeg": ".jpg", "image/gif": ".gif", "image/webp": ".webp"}

VALID_FILTERS = {"active", "archived", "trash", "all"}

DISPLAY_TITLE_CAP = 200


def derive_display_title(body: str | None) -> str:
    """The note's display NAME: the first line of the body that says anything.

    Stored as notes.display_title for search results and export filenames. A task
    line is named without its marker ("milk", not "- [ ] milk"), and an empty item
    is skipped.

    Lines split on "\n" alone, like every other reader of the grammar (`splitlines`
    would also break on "\r" and U+2028). Mirrors `derive::display_title` in the
    core; both run the `display_titles` cases in core/testdata/grammar.json.
    """
    for line in (body or "").split("\n"):
        stripped = strip_marker(line.strip()).strip()
        if stripped:
            return stripped[:DISPLAY_TITLE_CAP]
    return ""


def is_empty_note(body: str | None, items: list | None = None) -> bool:
    """Nothing worth keeping: no body text and no checklist items.

    `items` is separate because create accepts them separately (the importer holds a
    list); they are folded into the body after this check.
    """
    return not (body or "").strip() and not items


def parse_list_items(raw: object) -> list[str]:
    """Trimmed, non-empty checklist item texts from a create payload's `items`."""
    if not isinstance(raw, list):
        return []
    return [s.strip() for s in raw if isinstance(s, str) and s.strip()]


def apply_filter(stmt, filter_name: str, archived):
    """Narrow a notes query to one board view.

    `archived` is the viewer's archive flag (`note_state.archived_for`): a note
    someone shared with you is archived on your board when you archived it.

    Every branch excludes purge tombstones — content-less rows kept only so the sync
    feed can tell offline clients a note is gone (see `retention.purge_note`). The
    active/archived/all branches get that for free from `deleted_at IS NULL`, since
    a tombstone keeps the timestamp; Trash is the one view that has to say so.

    `all` is the board and the Archive together: what a search or a tag looks
    through, so finished work stays findable after it has been put away.
    """
    if filter_name == "all":
        return stmt.where(Note.deleted_at.is_(None))
    if filter_name == "archived":
        return stmt.where(Note.deleted_at.is_(None), archived.is_(True))
    if filter_name == "trash":
        return stmt.where(Note.deleted_at.is_not(None), Note.purged_at.is_(None))
    return stmt.where(Note.deleted_at.is_(None), archived.is_(False))


async def top_position(db, owner_id) -> int:
    """The highest manual position among `owner_id`'s live notes, 0 for none. A new
    note goes one above it, to the top of the board."""
    return int(
        await db.scalar(
            select(func.coalesce(func.max(Note.position), 0)).where(
                Note.owner_id == owner_id, Note.deleted_at.is_(None)
            )
        )
    )


def note_visible(permission: str | None = None):
    """Predicate: the current user may see the note, as its owner or through a share
    (at `permission`, when one is given). Every note read scoped to a viewer asks
    through here."""
    return visible_to_user("note", Note.owner_id, Note.id, g.user_id, permission=permission)


async def _fetch_note(db, note_id: str, gate) -> Note | None:
    """The one shape of a per-note fetch: a well-formed id, a note that isn't purged,
    and `gate` saying who may have it. None for anything else, so the route 404s."""
    nid = parse_uuid(note_id)
    if nid is None:
        return None
    return await db.scalar(select(Note).where(Note.id == nid, gate, Note.purged_at.is_(None)))


async def _get_owned(db, note_id: str) -> Note | None:
    """Fetch a note the current user OWNS, for every change only an owner may make:
    trash, delete, labels, reminders, attachments, previews, history and sharing.
    Pin and archive are everyone's own.

    Someone the note is shared with gets None, and so a 404, even when they can read
    it: a 403 would confirm the note exists to a caller who only has its id.

    A purged note reads as absent: the REST API must treat it as gone, so opening,
    editing or restoring one 404s. The sync push path looks rows up directly rather
    than through here, which is what still lets a client re-create an id it owns.
    """
    return await _fetch_note(db, note_id, Note.owner_id == g.user_id)


async def _get_visible(db, note_id: str) -> Note | None:
    """Fetch a note the current user can see: theirs, or shared with them at any
    level. What a recipient may do to it from here is their own pin, archive and
    place; `_get_editable` and `_get_owned` gate the rest."""
    return await _fetch_note(db, note_id, note_visible())


async def _get_editable(db, note_id: str) -> Note | None:
    """Fetch a note the current user may change the TEXT of: its body and its
    checklist. The owner, or someone it is shared with at `edit`; a view share gets
    None, and so a 404, like a stranger."""
    return await _fetch_note(db, note_id, note_visible(EDIT))


def _slugify(text: str) -> str:
    """A filesystem-safe slug from a note's display name (for the .md filename)."""
    s = re.sub(r"[^\w\s-]", "", (text or "").strip().lower())
    s = re.sub(r"[\s_-]+", "-", s).strip("-")
    return s[:60] or "note"


def _safe_filename(name: str | None) -> str:
    """The upload's original name reduced to a safe basename (display + download)."""
    base = os.path.basename((name or "").strip().replace("\\", "/"))
    return base[:255] or "file"


def _attachment_ext(filename: str, mime: str) -> str:
    """Storage extension: the original file's extension, else a known image ext."""
    ext = os.path.splitext(filename)[1].lower()
    if ext and len(ext) <= 12:
        return ext
    return ALLOWED_IMAGE_MIMES.get(mime, "")


def normalize_mime(raw: str | None) -> str:
    """A declared content type reduced to its bare, lowercase media type."""
    return (raw or "application/octet-stream").split(";")[0].strip().lower() or "application/octet-stream"


def store_attachment(
    db, note: Note, raw: bytes, filename: str | None, mime: str | None, att_id: uuid.UUID | None = None
) -> NoteAttachment:
    """Write one attachment's bytes under the media root and add its row (not committed).

    The single writer for the upload route, the importer and sync, so all three
    sanitize the name, pick the extension and record the hash the same way. A sync
    client passes `att_id` so a file attached offline keeps the id it was given there.
    """
    filename = _safe_filename(filename)
    mime = normalize_mime(mime)
    att_id = att_id or uuid.uuid4()
    rel = os.path.join(str(note.id), f"{att_id}{_attachment_ext(filename, mime)}")
    dest = Config.media_root() / rel
    dest.parent.mkdir(parents=True, exist_ok=True)
    dest.write_bytes(raw)
    row = NoteAttachment(
        id=att_id,
        note_id=note.id,
        path=rel,
        filename=filename,
        mime=mime,
        size=len(raw),
        sha256=hashlib.sha256(raw).hexdigest(),
    )
    db.add(row)
    return row


def _header_filename(name: str) -> str:
    """Sanitize a filename for a Content-Disposition header (drop quotes/newlines)."""
    return re.sub(r'[\r\n"]', "", name or "")[:255] or "file"