"""Small shared helpers + constants for the notes package: display-name derivation, board-filter narrowing, the owner-scoped fetch, and filename/slug sanitizers used by both the attachment routes and the importer.""" from __future__ import annotations import os import re from quart import g from sqlalchemy import select from ..models.note import Note from ..responses import parse_uuid ALLOWED_IMAGE_MIMES = {"image/png": ".png", "image/jpeg": ".jpg", "image/gif": ".gif", "image/webp": ".webp"} VALID_FILTERS = {"active", "archived", "trash"} DISPLAY_TITLE_CAP = 200 def derive_display_title(title: str | None, body: str | None) -> str: """The note's display NAME: the explicit title if set, else the first non-empty line of the body (trimmed, length-capped). Persisted as notes.display_title so a body-only note is still nameable/searchable/linkable — the user never has to type a title. Deterministic (literal first line, no AI).""" if title and title.strip(): return title.strip()[:DISPLAY_TITLE_CAP] for line in (body or "").splitlines(): stripped = line.strip() if stripped: return stripped[:DISPLAY_TITLE_CAP] return "" def is_empty_note(title: str | None, body: str | None) -> bool: return not (title or "").strip() and not (body or "").strip() 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): """Narrow a notes query to one board view. 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 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. """ if filter_name == "archived": return stmt.where(Note.deleted_at.is_(None), Note.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), Note.archived.is_(False)) async def _get_owned(db, note_id: str) -> Note | None: """Fetch a note the current user OWNS (mutations are owner-only in M1/M2). 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. """ nid = parse_uuid(note_id) if nid is None: return None return await db.scalar( select(Note).where(Note.id == nid, Note.owner_id == g.user_id, Note.purged_at.is_(None)) ) def _escape_like(s: str) -> str: """Escape LIKE wildcards so user input matches literally (escape char = \\).""" return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") 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 _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"