M9 S1d: split the notes.py monolith into a cohesive package
The 1574-line notes.py becomes a `notes/` package. The heavy shared logic moves
into focused modules; the route handlers + blueprint registration stay together
in __init__ so registration is trivially correct (most routes have no CI
auth-test that would otherwise catch a route silently dropping out):
- notes/_bp.py — the Blueprint (isolated so route modules could import it
without a cycle; also the seam for a later route split).
- notes/serialize.py — note (+labels/items/attachments/previews) serialization.
- notes/links.py — [[wiki-link]] + #tag parsing and reconciliation.
- notes/recurrence.py — recurring-reminder next-occurrence math.
- notes/helpers.py — display-title/empty/filter/owner-fetch + filename/slug utils.
- notes/import_export.py — export markdown + Keep/native import specs + zip budget.
- notes/__init__.py — the `/api/notes` routes + re-exports the external surface
(app.py imports `bp`; sync.py + tests import helpers).
Pure reorganization — no behavior change (routes/helpers moved verbatim). Callers
(app.py, sync.py, test_notes.py) are unchanged: `from thoughtsync.notes import X`
resolves via the package __init__ (rule 22 — the package replaces the module).
No import cycle (nothing in the package's dep chain imports notes; only app.py +
sync.py consume it). New test_all_note_routes_registered asserts all 29 route
endpoints are attached, so CI catches any module that fails to register. Runtime
DB behavior operator-verified on deploy (no Postgres CI lane).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FRgehjoz7Yv8LkUfADxACm
This commit is contained in:
Claude Opus 4.8
@@ -1,365 +1,104 @@
|
||||
"""Notes API (the `/api/notes` blueprint).
|
||||
|
||||
The bulk of the shared logic lives in cohesive sibling modules — serialization
|
||||
(`serialize`), wiki-links/tags (`links`), recurring reminders (`recurrence`),
|
||||
small text/query helpers (`helpers`), and export/import (`import_export`). The
|
||||
route handlers themselves stay here so blueprint registration is in one place, and
|
||||
`bp` is defined in `_bp` so every module can import it without a cycle.
|
||||
|
||||
External callers import from `thoughtsync.notes` (see `__all__`); those names are
|
||||
re-exported here so the package is a drop-in replacement for the old module."""
|
||||
from __future__ import annotations
|
||||
|
||||
import calendar
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import posixpath
|
||||
import re
|
||||
import uuid
|
||||
import zipfile
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from quart import Blueprint, Response, g, jsonify, request, send_file
|
||||
from sqlalchemy import case, delete, func, literal_column, select
|
||||
from quart import Response, g, jsonify, request, send_file
|
||||
from sqlalchemy import case, func, literal_column, select
|
||||
|
||||
from .acl import visible_to_user
|
||||
from .auth import login_required
|
||||
from .colors import NOTE_COLORS, normalize_color
|
||||
from .common import coerce_bool, iso, parse_dt
|
||||
from .config import Config
|
||||
from .db import session_scope
|
||||
from .labeling import reconcile_manual_labels, resolve_owned_label_ids
|
||||
from .responses import json_error, not_found, parse_uuid
|
||||
from .settings import get_setting
|
||||
from .unfurl import UnfurlError, unfurl
|
||||
from .models.label import Label, NoteLabel
|
||||
from .models.note import Note
|
||||
from .models.note_attachment import NoteAttachment
|
||||
from .models.note_item import NoteItem
|
||||
from .models.note_link import NoteLink
|
||||
from .models.note_link_preview import NoteLinkPreview
|
||||
from .models.note_revision import NoteRevision
|
||||
from ..acl import visible_to_user
|
||||
from ..auth import login_required
|
||||
from ..colors import NOTE_COLORS, normalize_color
|
||||
from ..common import coerce_bool, iso, parse_dt
|
||||
from ..config import Config
|
||||
from ..db import session_scope
|
||||
from ..labeling import reconcile_manual_labels, resolve_owned_label_ids
|
||||
from ..models.label import Label, NoteLabel
|
||||
from ..models.note import Note
|
||||
from ..models.note_attachment import NoteAttachment
|
||||
from ..models.note_item import NoteItem
|
||||
from ..models.note_link import NoteLink
|
||||
from ..models.note_link_preview import NoteLinkPreview
|
||||
from ..models.note_revision import NoteRevision
|
||||
from ..responses import json_error, not_found, parse_uuid
|
||||
from ..settings import get_setting
|
||||
from ..unfurl import UnfurlError, unfurl
|
||||
from ._bp import bp
|
||||
from .helpers import (
|
||||
ALLOWED_IMAGE_MIMES,
|
||||
VALID_FILTERS,
|
||||
_attachment_ext,
|
||||
_escape_like,
|
||||
_get_owned,
|
||||
_header_filename,
|
||||
_safe_filename,
|
||||
_slugify,
|
||||
apply_filter,
|
||||
derive_display_title,
|
||||
is_empty_note,
|
||||
parse_list_items,
|
||||
)
|
||||
from .import_export import (
|
||||
IMPORT_MAX_ENTRIES,
|
||||
_ImportBudget,
|
||||
_ImportTooLarge,
|
||||
_create_imported_note,
|
||||
_keep_spec,
|
||||
_native_spec,
|
||||
_note_markdown,
|
||||
_read_import_specs,
|
||||
_usec_to_dt,
|
||||
)
|
||||
from .links import (
|
||||
_reconcile_tags,
|
||||
_rename_inbound_links,
|
||||
_rewrite_links,
|
||||
parse_link_titles,
|
||||
parse_tags,
|
||||
rewrite_link_title,
|
||||
)
|
||||
from .recurrence import REMINDER_RECURRENCES, next_occurrence, normalize_recurrence
|
||||
from .serialize import _items_for_notes, _labels_for_notes, _serialize_note, _serialize_notes
|
||||
|
||||
ALLOWED_IMAGE_MIMES = {"image/png": ".png", "image/jpeg": ".jpg", "image/gif": ".gif", "image/webp": ".webp"}
|
||||
|
||||
_LINK_RE = re.compile(r"\[\[([^\[\]]+)\]\]")
|
||||
|
||||
# A #tag: `#` at the start of the body or after whitespace, then a word char and
|
||||
# word chars/hyphens. A URL fragment (foo#bar) or mid-word `#` is not preceded by
|
||||
# whitespace, so it won't match.
|
||||
_TAG_RE = re.compile(r"(?:^|(?<=\s))#(\w[\w-]*)")
|
||||
|
||||
|
||||
def parse_tags(body: str | None) -> list[str]:
|
||||
"""Distinct #hashtags from a note body, in order, deduped case-insensitively.
|
||||
A tag must contain a letter, so #2024 or #_ are ignored (avoids numeric noise)."""
|
||||
if not body:
|
||||
return []
|
||||
out: list[str] = []
|
||||
seen: set[str] = set()
|
||||
for match in _TAG_RE.finditer(body):
|
||||
tag = match.group(1)
|
||||
if not any(c.isalpha() for c in tag):
|
||||
continue
|
||||
norm = tag.lower()
|
||||
if norm not in seen:
|
||||
seen.add(norm)
|
||||
out.append(tag)
|
||||
return out
|
||||
|
||||
|
||||
def parse_link_titles(body: str | None) -> list[str]:
|
||||
"""Extract distinct normalized [[wiki-link]] titles from a note body."""
|
||||
if not body:
|
||||
return []
|
||||
out: list[str] = []
|
||||
for match in _LINK_RE.finditer(body):
|
||||
norm = match.group(1).strip().lower()
|
||||
if norm and norm not in out:
|
||||
out.append(norm)
|
||||
return out
|
||||
|
||||
bp = Blueprint("notes", __name__, url_prefix="/api/notes")
|
||||
|
||||
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."""
|
||||
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))
|
||||
return stmt.where(Note.deleted_at.is_(None), Note.archived.is_(False))
|
||||
|
||||
|
||||
async def _labels_for_notes(db, note_ids: list) -> dict:
|
||||
"""Map note_id -> [{id, name}] in one query (no lazy relationship loading)."""
|
||||
result: dict = {}
|
||||
if not note_ids:
|
||||
return result
|
||||
rows = await db.execute(
|
||||
select(NoteLabel.note_id, Label.id, Label.name, Label.color, NoteLabel.via_tag)
|
||||
.join(Label, Label.id == NoteLabel.label_id)
|
||||
.where(NoteLabel.note_id.in_(note_ids))
|
||||
.order_by(Label.name)
|
||||
)
|
||||
for note_id, label_id, name, color, via_tag in rows.all():
|
||||
result.setdefault(note_id, []).append(
|
||||
{"id": str(label_id), "name": name, "color": color, "via_tag": via_tag}
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def _serialize_item(item: NoteItem) -> dict:
|
||||
return {"id": str(item.id), "text": item.text, "checked": item.checked, "position": item.position}
|
||||
|
||||
|
||||
async def _items_for_notes(db, note_ids: list) -> dict:
|
||||
"""Map note_id -> [checklist items] in one query, ordered by position."""
|
||||
result: dict = {}
|
||||
if not note_ids:
|
||||
return result
|
||||
items = (
|
||||
await db.scalars(
|
||||
select(NoteItem).where(NoteItem.note_id.in_(note_ids)).order_by(NoteItem.position, NoteItem.created_at)
|
||||
)
|
||||
).all()
|
||||
for item in items:
|
||||
result.setdefault(item.note_id, []).append(_serialize_item(item))
|
||||
return result
|
||||
|
||||
|
||||
def _attachment_url(note_id, att_id) -> str:
|
||||
return f"/api/notes/{note_id}/attachments/{att_id}"
|
||||
|
||||
|
||||
async def _attachments_for_notes(db, note_ids: list) -> dict:
|
||||
result: dict = {}
|
||||
if not note_ids:
|
||||
return result
|
||||
rows = (
|
||||
await db.scalars(
|
||||
select(NoteAttachment).where(NoteAttachment.note_id.in_(note_ids)).order_by(NoteAttachment.created_at)
|
||||
)
|
||||
).all()
|
||||
for att in rows:
|
||||
result.setdefault(att.note_id, []).append(
|
||||
{
|
||||
"id": str(att.id),
|
||||
"url": _attachment_url(att.note_id, att.id),
|
||||
"filename": att.filename,
|
||||
"mime": att.mime,
|
||||
"size": att.size,
|
||||
"sha256": att.sha256,
|
||||
}
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def _serialize_preview(p: NoteLinkPreview) -> dict:
|
||||
return {
|
||||
"id": str(p.id),
|
||||
"url": p.url,
|
||||
"title": p.title,
|
||||
"description": p.description,
|
||||
"image_url": p.image_url,
|
||||
"site_name": p.site_name,
|
||||
}
|
||||
|
||||
|
||||
async def _previews_for_notes(db, note_ids: list) -> dict:
|
||||
"""Map note_id -> [link previews] in one query."""
|
||||
result: dict = {}
|
||||
if not note_ids:
|
||||
return result
|
||||
rows = (
|
||||
await db.scalars(
|
||||
select(NoteLinkPreview)
|
||||
.where(NoteLinkPreview.note_id.in_(note_ids))
|
||||
.order_by(NoteLinkPreview.created_at)
|
||||
)
|
||||
).all()
|
||||
for p in rows:
|
||||
result.setdefault(p.note_id, []).append(_serialize_preview(p))
|
||||
return result
|
||||
|
||||
|
||||
async def _serialize_note(db, note: Note) -> dict:
|
||||
data = note.serialize()
|
||||
labels = await _labels_for_notes(db, [note.id])
|
||||
data["labels"] = labels.get(note.id, [])
|
||||
items = await _items_for_notes(db, [note.id])
|
||||
data["items"] = items.get(note.id, [])
|
||||
attachments = await _attachments_for_notes(db, [note.id])
|
||||
data["attachments"] = attachments.get(note.id, [])
|
||||
previews = await _previews_for_notes(db, [note.id])
|
||||
data["previews"] = previews.get(note.id, [])
|
||||
return data
|
||||
|
||||
|
||||
async def _serialize_notes(db, notes: list) -> list:
|
||||
ids = [n.id for n in notes]
|
||||
labels_map = await _labels_for_notes(db, ids)
|
||||
items_map = await _items_for_notes(db, ids)
|
||||
attach_map = await _attachments_for_notes(db, ids)
|
||||
preview_map = await _previews_for_notes(db, ids)
|
||||
out = []
|
||||
for n in notes:
|
||||
data = n.serialize()
|
||||
data["labels"] = labels_map.get(n.id, [])
|
||||
data["items"] = items_map.get(n.id, [])
|
||||
data["attachments"] = attach_map.get(n.id, [])
|
||||
data["previews"] = preview_map.get(n.id, [])
|
||||
out.append(data)
|
||||
return out
|
||||
|
||||
|
||||
async def _get_owned(db, note_id: str) -> Note | None:
|
||||
"""Fetch a note the current user OWNS (mutations are owner-only in M1/M2)."""
|
||||
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))
|
||||
|
||||
|
||||
async def _rewrite_links(db, note: Note) -> None:
|
||||
"""Replace a note's outgoing wiki-links from its current body."""
|
||||
await db.execute(delete(NoteLink).where(NoteLink.source_id == note.id))
|
||||
for norm in parse_link_titles(note.body):
|
||||
db.add(NoteLink(source_id=note.id, target_norm=norm))
|
||||
|
||||
|
||||
async def _find_or_create_label(db, owner_id, name: str):
|
||||
"""Owner's label id for `name` (case-insensitive match), creating it if absent."""
|
||||
existing = await db.scalar(
|
||||
select(Label.id).where(Label.owner_id == owner_id, func.lower(Label.name) == name.lower())
|
||||
)
|
||||
if existing is not None:
|
||||
return existing
|
||||
label = Label(owner_id=owner_id, name=name)
|
||||
db.add(label)
|
||||
await db.flush()
|
||||
return label.id
|
||||
|
||||
|
||||
async def _reconcile_tags(db, note: Note) -> None:
|
||||
"""Sync tag-sourced labels (via_tag=True) with the #hashtags in the note body:
|
||||
attach labels for current tags, detach tag-labels whose #tag was removed. Manual
|
||||
picker labels (via_tag=False) are never touched."""
|
||||
tag_label_ids: set = set()
|
||||
for name in parse_tags(note.body):
|
||||
tag_label_ids.add(await _find_or_create_label(db, note.owner_id, name))
|
||||
rows = (await db.scalars(select(NoteLabel).where(NoteLabel.note_id == note.id))).all()
|
||||
attached_ids = {r.label_id for r in rows}
|
||||
# Detach tag-labels no longer backed by a #tag in the body.
|
||||
for r in rows:
|
||||
if r.via_tag and r.label_id not in tag_label_ids:
|
||||
await db.delete(r)
|
||||
attached_ids.discard(r.label_id)
|
||||
# Attach new tags — skip labels already attached (in any form) to respect the PK
|
||||
# and leave a manually-added label of the same name as-is.
|
||||
for lid in tag_label_ids:
|
||||
if lid not in attached_ids:
|
||||
db.add(NoteLabel(note_id=note.id, label_id=lid, via_tag=True))
|
||||
attached_ids.add(lid)
|
||||
|
||||
|
||||
def rewrite_link_title(body: str | None, old_norm: str, new_title: str) -> str:
|
||||
"""Repoint every [[token]] whose normalized form == old_norm to [[new_title]]."""
|
||||
if not body:
|
||||
return body or ""
|
||||
|
||||
def _sub(match: re.Match) -> str:
|
||||
return f"[[{new_title}]]" if match.group(1).strip().lower() == old_norm else match.group(0)
|
||||
|
||||
return _LINK_RE.sub(_sub, body)
|
||||
|
||||
|
||||
async def _rename_inbound_links(db, renamed: Note, old_title: str, new_title: str) -> None:
|
||||
"""Rewrite [[old title]] references (and their link rows) in every note that
|
||||
links to the renamed note, so its backlinks survive the title change."""
|
||||
old_norm = old_title.strip().lower()
|
||||
sources = (
|
||||
await db.scalars(
|
||||
select(Note)
|
||||
.join(NoteLink, NoteLink.source_id == Note.id)
|
||||
.where(
|
||||
NoteLink.target_norm == old_norm,
|
||||
Note.owner_id == renamed.owner_id,
|
||||
Note.deleted_at.is_(None),
|
||||
)
|
||||
)
|
||||
).all()
|
||||
seen: set = set()
|
||||
for source in sources:
|
||||
if source.id in seen:
|
||||
continue
|
||||
seen.add(source.id)
|
||||
source.body = rewrite_link_title(source.body, old_norm, new_title)
|
||||
await _rewrite_links(db, source)
|
||||
|
||||
|
||||
REMINDER_RECURRENCES = {"daily", "weekly", "monthly", "yearly"}
|
||||
|
||||
|
||||
def normalize_recurrence(value: object) -> str | None:
|
||||
return value if value in REMINDER_RECURRENCES else None
|
||||
|
||||
|
||||
def _add_months(dt: datetime, months: int) -> datetime:
|
||||
"""Shift a datetime by whole months, clamping the day to the target month's length
|
||||
(so Jan 31 + 1 month → Feb 28/29). Keeps the time-of-day."""
|
||||
m = dt.month - 1 + months
|
||||
year = dt.year + m // 12
|
||||
month = m % 12 + 1
|
||||
day = min(dt.day, calendar.monthrange(year, month)[1])
|
||||
return dt.replace(year=year, month=month, day=day)
|
||||
|
||||
|
||||
def _advance_once(dt: datetime, recurrence: str) -> datetime | None:
|
||||
if recurrence == "daily":
|
||||
return dt + timedelta(days=1)
|
||||
if recurrence == "weekly":
|
||||
return dt + timedelta(weeks=1)
|
||||
if recurrence == "monthly":
|
||||
return _add_months(dt, 1)
|
||||
if recurrence == "yearly":
|
||||
return _add_months(dt, 12)
|
||||
return None
|
||||
|
||||
|
||||
def next_occurrence(remind_at: datetime, recurrence: str, after: datetime) -> datetime | None:
|
||||
"""The next reminder fire time strictly after `after`, rolling a recurring reminder
|
||||
forward past any missed occurrences. None if `recurrence` isn't a known interval."""
|
||||
nxt = _advance_once(remind_at, recurrence)
|
||||
if nxt is None:
|
||||
return None
|
||||
while nxt <= after:
|
||||
step = _advance_once(nxt, recurrence)
|
||||
if step is None or step == nxt:
|
||||
break
|
||||
nxt = step
|
||||
return nxt
|
||||
__all__ = [
|
||||
"bp",
|
||||
"derive_display_title",
|
||||
"is_empty_note",
|
||||
"parse_list_items",
|
||||
"parse_tags",
|
||||
"parse_link_titles",
|
||||
"rewrite_link_title",
|
||||
"normalize_color",
|
||||
"normalize_recurrence",
|
||||
"next_occurrence",
|
||||
"_reconcile_tags",
|
||||
"_rename_inbound_links",
|
||||
"_rewrite_links",
|
||||
"_serialize_notes",
|
||||
"_escape_like",
|
||||
"_safe_filename",
|
||||
"_attachment_ext",
|
||||
"_header_filename",
|
||||
"_slugify",
|
||||
"_keep_spec",
|
||||
"_native_spec",
|
||||
"_usec_to_dt",
|
||||
]
|
||||
|
||||
|
||||
@bp.get("")
|
||||
@@ -508,41 +247,6 @@ async def snooze_reminder(note_id: str):
|
||||
return jsonify(await _serialize_note(db, note))
|
||||
|
||||
|
||||
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 _note_markdown(note: Note, labels: list, items: list) -> str:
|
||||
"""One note as a human-readable Markdown file with a small frontmatter block.
|
||||
The authoritative machine format is notes.json; this is for reading/portability."""
|
||||
fm = ["---"]
|
||||
if note.title:
|
||||
fm.append(f"title: {note.title}")
|
||||
fm.append(f"display_name: {note.display_title}")
|
||||
if labels:
|
||||
fm.append("labels: [" + ", ".join(lb["name"] for lb in labels) + "]")
|
||||
fm.append(f"color: {note.color}")
|
||||
if note.pinned:
|
||||
fm.append("pinned: true")
|
||||
if note.archived:
|
||||
fm.append("archived: true")
|
||||
if note.remind_at:
|
||||
fm.append(f"remind_at: {note.remind_at.isoformat()}")
|
||||
fm.append(f"created: {note.created_at.isoformat() if note.created_at else ''}")
|
||||
fm.append(f"updated: {note.updated_at.isoformat() if note.updated_at else ''}")
|
||||
fm.append("---")
|
||||
fm.append("")
|
||||
if note.kind == "list":
|
||||
for it in items:
|
||||
fm.append(f"- [{'x' if it['checked'] else ' '}] {it['text']}")
|
||||
else:
|
||||
fm.append(note.body)
|
||||
return "\n".join(fm) + "\n"
|
||||
|
||||
|
||||
@bp.get("/export")
|
||||
@login_required
|
||||
async def export_notes():
|
||||
@@ -619,284 +323,6 @@ async def export_notes():
|
||||
)
|
||||
|
||||
|
||||
# --- Import: ThoughtSync's own export (round-trip) OR a Google Keep Takeout zip ---
|
||||
|
||||
# Google Keep (Takeout) color enum → our palette. Keep has a few hues we don't
|
||||
# (BROWN/DARKBLUE/CERULEAN); map each to the nearest. Unknowns fall back to default.
|
||||
_KEEP_COLOR_MAP = {
|
||||
"DEFAULT": "default",
|
||||
"RED": "red",
|
||||
"ORANGE": "orange",
|
||||
"YELLOW": "yellow",
|
||||
"GREEN": "green",
|
||||
"TEAL": "teal",
|
||||
"CERULEAN": "teal",
|
||||
"BLUE": "blue",
|
||||
"DARKBLUE": "blue",
|
||||
"PURPLE": "purple",
|
||||
"PINK": "pink",
|
||||
"BROWN": "orange",
|
||||
"GRAY": "gray",
|
||||
}
|
||||
|
||||
# Reverse of ALLOWED_IMAGE_MIMES, for inferring an attachment's mime from its
|
||||
# filename when the source didn't record one (Keep usually does; be defensive).
|
||||
_EXT_MIME = {ext: mime for mime, ext in ALLOWED_IMAGE_MIMES.items()}
|
||||
_EXT_MIME[".jpeg"] = "image/jpeg"
|
||||
|
||||
|
||||
def _usec_to_dt(usec: object) -> datetime | None:
|
||||
"""Google Keep timestamps are integer MICROseconds since the Unix epoch (UTC)."""
|
||||
try:
|
||||
return datetime.fromtimestamp(int(usec) / 1_000_000, tz=timezone.utc)
|
||||
except (TypeError, ValueError, OverflowError, OSError):
|
||||
return None
|
||||
|
||||
|
||||
def _native_spec(n: dict) -> dict:
|
||||
"""Normalize one note from a ThoughtSync export's notes.json into the common
|
||||
import spec consumed by _create_imported_note."""
|
||||
return {
|
||||
"title": n.get("title"),
|
||||
"body": n.get("body") or "",
|
||||
"kind": n.get("kind"),
|
||||
"color": n.get("color"),
|
||||
"pinned": bool(n.get("pinned")),
|
||||
"archived": bool(n.get("archived")),
|
||||
"trashed": False, # export only includes live notes
|
||||
"remind_at": parse_dt(n.get("remind_at")),
|
||||
"recurrence": normalize_recurrence(n.get("recurrence")),
|
||||
"created_at": parse_dt(n.get("created_at")),
|
||||
"updated_at": parse_dt(n.get("updated_at")),
|
||||
"labels": [s for s in (n.get("labels") or []) if isinstance(s, str)],
|
||||
"items": [
|
||||
{"text": it.get("text"), "checked": bool(it.get("checked"))}
|
||||
for it in (n.get("items") or [])
|
||||
if isinstance(it, dict)
|
||||
],
|
||||
# export writes attachments[].file as the zip-internal path already.
|
||||
"attachments": [
|
||||
{"file": a.get("file"), "mime": a.get("mime")}
|
||||
for a in (n.get("attachments") or [])
|
||||
if isinstance(a, dict) and a.get("file")
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _keep_spec(kn: dict, keep_dir: str) -> dict:
|
||||
"""Normalize one Google Keep note (Takeout <note>.json) into the common import
|
||||
spec. `keep_dir` is the note JSON's folder, used to resolve attachment paths."""
|
||||
list_content = kn.get("listContent") if isinstance(kn.get("listContent"), list) else []
|
||||
is_list = bool(list_content)
|
||||
body = kn.get("textContent") or "" if not is_list else ""
|
||||
# Keep stores link annotations (e.g. shared URLs) separately from the text —
|
||||
# fold any URLs into the body so the content survives the move.
|
||||
urls = [
|
||||
ann.get("url")
|
||||
for ann in (kn.get("annotations") or [])
|
||||
if isinstance(ann, dict) and ann.get("url")
|
||||
]
|
||||
extra = "\n".join(u for u in urls if u and u not in body)
|
||||
if extra:
|
||||
body = f"{body}\n\n{extra}" if body.strip() else extra
|
||||
|
||||
attachments = []
|
||||
for a in kn.get("attachments") or []:
|
||||
if not isinstance(a, dict):
|
||||
continue
|
||||
fp = a.get("filePath")
|
||||
if not fp:
|
||||
continue
|
||||
zpath = posixpath.join(keep_dir, fp) if keep_dir else fp
|
||||
mime = a.get("mimetype") or _EXT_MIME.get(posixpath.splitext(fp)[1].lower())
|
||||
attachments.append({"file": zpath, "mime": mime})
|
||||
|
||||
return {
|
||||
"title": kn.get("title"),
|
||||
"body": body,
|
||||
"kind": "list" if is_list else "text",
|
||||
"color": _KEEP_COLOR_MAP.get(str(kn.get("color") or "DEFAULT").upper(), "default"),
|
||||
"pinned": bool(kn.get("isPinned")),
|
||||
"archived": bool(kn.get("isArchived")),
|
||||
"trashed": bool(kn.get("isTrashed")),
|
||||
"remind_at": None, # Keep reminders aren't in Takeout note JSON
|
||||
"created_at": _usec_to_dt(kn.get("createdTimestampUsec")),
|
||||
"updated_at": _usec_to_dt(kn.get("userEditedTimestampUsec")),
|
||||
"labels": [
|
||||
lb.get("name")
|
||||
for lb in (kn.get("labels") or [])
|
||||
if isinstance(lb, dict) and lb.get("name")
|
||||
],
|
||||
"items": [
|
||||
{"text": li.get("text"), "checked": bool(li.get("isChecked"))}
|
||||
for li in list_content
|
||||
if isinstance(li, dict)
|
||||
],
|
||||
"attachments": attachments,
|
||||
}
|
||||
|
||||
|
||||
IMPORT_MAX_ENTRIES = 10_000
|
||||
IMPORT_MAX_ENTRY_BYTES = 64 * 1024 * 1024 # 64 MB decompressed per file
|
||||
IMPORT_MAX_TOTAL_BYTES = 512 * 1024 * 1024 # 512 MB decompressed across the whole import
|
||||
|
||||
|
||||
class _ImportTooLarge(Exception):
|
||||
"""An import zip decompressed past the byte budget (a zip bomb, or just too big)."""
|
||||
|
||||
|
||||
class _ImportBudget:
|
||||
"""Caps DECOMPRESSED bytes pulled from an import zip — per entry and cumulatively.
|
||||
|
||||
zipfile inflates into memory on read, so an archive that's tiny on disk can expand
|
||||
to gigabytes. We stream each entry and read at most the remaining budget + 1 byte,
|
||||
so an oversized (or size-lying) entry is caught mid-read instead of after it has
|
||||
already been fully inflated."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.remaining = IMPORT_MAX_TOTAL_BYTES
|
||||
|
||||
def read(self, zf: zipfile.ZipFile, name: str) -> bytes:
|
||||
cap = min(IMPORT_MAX_ENTRY_BYTES, self.remaining)
|
||||
with zf.open(name) as fh:
|
||||
data = fh.read(cap + 1)
|
||||
if len(data) > cap:
|
||||
raise _ImportTooLarge()
|
||||
self.remaining -= len(data)
|
||||
return data
|
||||
|
||||
|
||||
def _read_import_specs(zf: zipfile.ZipFile, budget: _ImportBudget) -> tuple[list[dict], str]:
|
||||
"""Detect the archive format and return (specs, source). A ThoughtSync export
|
||||
is recognized by its notes.json (app == thoughtsync); otherwise each Keep-shaped
|
||||
<note>.json is imported. Returns ([], "") when nothing importable is found."""
|
||||
names = zf.namelist()
|
||||
for name in names:
|
||||
if posixpath.basename(name) == "notes.json":
|
||||
try:
|
||||
doc = json.loads(budget.read(zf, name))
|
||||
except (ValueError, KeyError):
|
||||
continue
|
||||
if isinstance(doc, dict) and doc.get("app") == "thoughtsync":
|
||||
specs = [_native_spec(n) for n in (doc.get("notes") or []) if isinstance(n, dict)]
|
||||
return specs, "thoughtsync"
|
||||
|
||||
keep_specs: list[dict] = []
|
||||
keep_keys = ("textContent", "listContent", "isPinned", "isArchived", "isTrashed", "userEditedTimestampUsec")
|
||||
for name in names:
|
||||
if not name.lower().endswith(".json") or posixpath.basename(name) == "notes.json":
|
||||
continue
|
||||
try:
|
||||
kn = json.loads(budget.read(zf, name))
|
||||
except (ValueError, KeyError):
|
||||
continue
|
||||
if isinstance(kn, dict) and any(k in kn for k in keep_keys):
|
||||
keep_specs.append(_keep_spec(kn, posixpath.dirname(name)))
|
||||
return (keep_specs, "keep") if keep_specs else ([], "")
|
||||
|
||||
|
||||
def _import_attachment(db, note: Note, zf: zipfile.ZipFile, att: dict, budget: _ImportBudget) -> bool:
|
||||
"""Copy one attachment (any type — incl. Keep audio memos) out of the zip into
|
||||
media storage and record it, preserving its filename + hash. Returns True if written."""
|
||||
zpath = att.get("file")
|
||||
if not zpath:
|
||||
return False
|
||||
try:
|
||||
raw = budget.read(zf, zpath)
|
||||
except KeyError:
|
||||
return False
|
||||
filename = _safe_filename(posixpath.basename(zpath))
|
||||
mime = (att.get("mime") or "application/octet-stream").split(";")[0].strip().lower()
|
||||
ext = _attachment_ext(filename, mime)
|
||||
att_id = uuid.uuid4()
|
||||
rel = os.path.join(str(note.id), f"{att_id}{ext}")
|
||||
dest = Config.media_root() / rel
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
dest.write_bytes(raw)
|
||||
db.add(
|
||||
NoteAttachment(
|
||||
id=att_id,
|
||||
note_id=note.id,
|
||||
path=rel,
|
||||
filename=filename,
|
||||
mime=mime,
|
||||
size=len(raw),
|
||||
sha256=hashlib.sha256(raw).hexdigest(),
|
||||
)
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
async def _create_imported_note(
|
||||
db, owner_id, spec: dict, zf: zipfile.ZipFile, position: int, budget: _ImportBudget
|
||||
) -> bool:
|
||||
"""Insert one imported note plus its items/labels/attachments, reusing the same
|
||||
display-title derivation + tag/link reconciliation as create_note. Returns False
|
||||
(nothing written) when the spec is empty."""
|
||||
title = (spec.get("title") or "").strip() or None
|
||||
body = spec.get("body") or ""
|
||||
kind = spec.get("kind") if spec.get("kind") in ("text", "list") else "text"
|
||||
items = spec.get("items") or []
|
||||
if kind == "list":
|
||||
if not (title or any((it.get("text") or "").strip() for it in items)):
|
||||
return False
|
||||
elif is_empty_note(title, body):
|
||||
return False
|
||||
|
||||
note = Note(
|
||||
owner_id=owner_id,
|
||||
title=title,
|
||||
display_title=derive_display_title(title, body),
|
||||
body=body,
|
||||
kind=kind,
|
||||
color=normalize_color(spec.get("color")),
|
||||
pinned=bool(spec.get("pinned")),
|
||||
archived=bool(spec.get("archived")),
|
||||
position=position,
|
||||
)
|
||||
if spec.get("remind_at"):
|
||||
note.remind_at = spec["remind_at"]
|
||||
if spec.get("recurrence"):
|
||||
note.recurrence = spec["recurrence"]
|
||||
if spec.get("trashed"):
|
||||
note.deleted_at = datetime.now(timezone.utc)
|
||||
# Preserve source timestamps: set before flush so they land in the INSERT
|
||||
# (updated_at's onupdate only fires on later UPDATEs, which we don't trigger).
|
||||
if spec.get("created_at"):
|
||||
note.created_at = spec["created_at"]
|
||||
if spec.get("updated_at"):
|
||||
note.updated_at = spec["updated_at"]
|
||||
db.add(note)
|
||||
await db.flush() # assign note.id before items/labels/attachments/links
|
||||
|
||||
if kind == "list":
|
||||
for pos, it in enumerate(items):
|
||||
text = (it.get("text") or "").strip()
|
||||
if text:
|
||||
db.add(NoteItem(note_id=note.id, text=text, checked=bool(it.get("checked")), position=pos))
|
||||
|
||||
# Explicit (picker-style) labels are manual — via_tag=False. Inline #tags in the
|
||||
# body are handled by _reconcile_tags below, same as a normal create.
|
||||
for name in spec.get("labels") or []:
|
||||
name = (name or "").strip()
|
||||
if not name:
|
||||
continue
|
||||
lid = await _find_or_create_label(db, owner_id, name)
|
||||
exists = await db.scalar(
|
||||
select(NoteLabel).where(NoteLabel.note_id == note.id, NoteLabel.label_id == lid)
|
||||
)
|
||||
if exists is None:
|
||||
db.add(NoteLabel(note_id=note.id, label_id=lid, via_tag=False))
|
||||
|
||||
for att in spec.get("attachments") or []:
|
||||
if isinstance(att, dict):
|
||||
_import_attachment(db, note, zf, att, budget)
|
||||
|
||||
await _rewrite_links(db, note)
|
||||
await _reconcile_tags(db, note)
|
||||
return True
|
||||
|
||||
|
||||
@bp.post("/import")
|
||||
@login_required
|
||||
async def import_notes():
|
||||
@@ -965,11 +391,6 @@ async def list_titles():
|
||||
)
|
||||
|
||||
|
||||
def _escape_like(s: str) -> str:
|
||||
"""Escape LIKE wildcards so user input matches literally (escape char = \\)."""
|
||||
return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
||||
|
||||
|
||||
@bp.get("/link-search")
|
||||
@login_required
|
||||
async def link_search():
|
||||
@@ -1345,25 +766,6 @@ async def reorder_items(note_id: str):
|
||||
return jsonify(await _serialize_note(db, 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"
|
||||
|
||||
|
||||
@bp.post("/<note_id>/attachments")
|
||||
@login_required
|
||||
async def upload_attachment(note_id: str):
|
||||
@@ -0,0 +1,8 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from quart import Blueprint
|
||||
|
||||
# The notes blueprint lives in its own tiny module so every route module in the
|
||||
# package can `from ._bp import bp` without importing the package __init__ (which
|
||||
# imports the route modules) — i.e. no import cycle.
|
||||
bp = Blueprint("notes", __name__, url_prefix="/api/notes")
|
||||
@@ -0,0 +1,92 @@
|
||||
"""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."""
|
||||
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))
|
||||
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)."""
|
||||
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))
|
||||
|
||||
|
||||
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"
|
||||
@@ -0,0 +1,338 @@
|
||||
"""Export/import helpers (the non-route logic). Export renders each note as Markdown;
|
||||
import normalizes a ThoughtSync export OR a Google Keep Takeout zip into a common spec
|
||||
and materializes notes — with a decompression-size budget so an import zip bomb can't
|
||||
exhaust memory/disk."""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import posixpath
|
||||
import uuid
|
||||
import zipfile
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from sqlalchemy import select
|
||||
|
||||
from ..colors import normalize_color
|
||||
from ..common import parse_dt
|
||||
from ..config import Config
|
||||
from ..models.label import NoteLabel
|
||||
from ..models.note import Note
|
||||
from ..models.note_attachment import NoteAttachment
|
||||
from ..models.note_item import NoteItem
|
||||
from .helpers import (
|
||||
ALLOWED_IMAGE_MIMES,
|
||||
_attachment_ext,
|
||||
_safe_filename,
|
||||
derive_display_title,
|
||||
is_empty_note,
|
||||
)
|
||||
from .links import _find_or_create_label, _reconcile_tags, _rewrite_links
|
||||
from .recurrence import normalize_recurrence
|
||||
|
||||
|
||||
def _note_markdown(note: Note, labels: list, items: list) -> str:
|
||||
"""One note as a human-readable Markdown file with a small frontmatter block.
|
||||
The authoritative machine format is notes.json; this is for reading/portability."""
|
||||
fm = ["---"]
|
||||
if note.title:
|
||||
fm.append(f"title: {note.title}")
|
||||
fm.append(f"display_name: {note.display_title}")
|
||||
if labels:
|
||||
fm.append("labels: [" + ", ".join(lb["name"] for lb in labels) + "]")
|
||||
fm.append(f"color: {note.color}")
|
||||
if note.pinned:
|
||||
fm.append("pinned: true")
|
||||
if note.archived:
|
||||
fm.append("archived: true")
|
||||
if note.remind_at:
|
||||
fm.append(f"remind_at: {note.remind_at.isoformat()}")
|
||||
fm.append(f"created: {note.created_at.isoformat() if note.created_at else ''}")
|
||||
fm.append(f"updated: {note.updated_at.isoformat() if note.updated_at else ''}")
|
||||
fm.append("---")
|
||||
fm.append("")
|
||||
if note.kind == "list":
|
||||
for it in items:
|
||||
fm.append(f"- [{'x' if it['checked'] else ' '}] {it['text']}")
|
||||
else:
|
||||
fm.append(note.body)
|
||||
return "\n".join(fm) + "\n"
|
||||
|
||||
|
||||
# --- Import: ThoughtSync's own export (round-trip) OR a Google Keep Takeout zip ---
|
||||
|
||||
# Google Keep (Takeout) color enum → our palette. Keep has a few hues we don't
|
||||
# (BROWN/DARKBLUE/CERULEAN); map each to the nearest. Unknowns fall back to default.
|
||||
_KEEP_COLOR_MAP = {
|
||||
"DEFAULT": "default",
|
||||
"RED": "red",
|
||||
"ORANGE": "orange",
|
||||
"YELLOW": "yellow",
|
||||
"GREEN": "green",
|
||||
"TEAL": "teal",
|
||||
"CERULEAN": "teal",
|
||||
"BLUE": "blue",
|
||||
"DARKBLUE": "blue",
|
||||
"PURPLE": "purple",
|
||||
"PINK": "pink",
|
||||
"BROWN": "orange",
|
||||
"GRAY": "gray",
|
||||
}
|
||||
|
||||
# Reverse of ALLOWED_IMAGE_MIMES, for inferring an attachment's mime from its
|
||||
# filename when the source didn't record one (Keep usually does; be defensive).
|
||||
_EXT_MIME = {ext: mime for mime, ext in ALLOWED_IMAGE_MIMES.items()}
|
||||
_EXT_MIME[".jpeg"] = "image/jpeg"
|
||||
|
||||
|
||||
def _usec_to_dt(usec: object) -> datetime | None:
|
||||
"""Google Keep timestamps are integer MICROseconds since the Unix epoch (UTC)."""
|
||||
try:
|
||||
return datetime.fromtimestamp(int(usec) / 1_000_000, tz=timezone.utc)
|
||||
except (TypeError, ValueError, OverflowError, OSError):
|
||||
return None
|
||||
|
||||
|
||||
def _native_spec(n: dict) -> dict:
|
||||
"""Normalize one note from a ThoughtSync export's notes.json into the common
|
||||
import spec consumed by _create_imported_note."""
|
||||
return {
|
||||
"title": n.get("title"),
|
||||
"body": n.get("body") or "",
|
||||
"kind": n.get("kind"),
|
||||
"color": n.get("color"),
|
||||
"pinned": bool(n.get("pinned")),
|
||||
"archived": bool(n.get("archived")),
|
||||
"trashed": False, # export only includes live notes
|
||||
"remind_at": parse_dt(n.get("remind_at")),
|
||||
"recurrence": normalize_recurrence(n.get("recurrence")),
|
||||
"created_at": parse_dt(n.get("created_at")),
|
||||
"updated_at": parse_dt(n.get("updated_at")),
|
||||
"labels": [s for s in (n.get("labels") or []) if isinstance(s, str)],
|
||||
"items": [
|
||||
{"text": it.get("text"), "checked": bool(it.get("checked"))}
|
||||
for it in (n.get("items") or [])
|
||||
if isinstance(it, dict)
|
||||
],
|
||||
# export writes attachments[].file as the zip-internal path already.
|
||||
"attachments": [
|
||||
{"file": a.get("file"), "mime": a.get("mime")}
|
||||
for a in (n.get("attachments") or [])
|
||||
if isinstance(a, dict) and a.get("file")
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _keep_spec(kn: dict, keep_dir: str) -> dict:
|
||||
"""Normalize one Google Keep note (Takeout <note>.json) into the common import
|
||||
spec. `keep_dir` is the note JSON's folder, used to resolve attachment paths."""
|
||||
list_content = kn.get("listContent") if isinstance(kn.get("listContent"), list) else []
|
||||
is_list = bool(list_content)
|
||||
body = kn.get("textContent") or "" if not is_list else ""
|
||||
# Keep stores link annotations (e.g. shared URLs) separately from the text —
|
||||
# fold any URLs into the body so the content survives the move.
|
||||
urls = [
|
||||
ann.get("url")
|
||||
for ann in (kn.get("annotations") or [])
|
||||
if isinstance(ann, dict) and ann.get("url")
|
||||
]
|
||||
extra = "\n".join(u for u in urls if u and u not in body)
|
||||
if extra:
|
||||
body = f"{body}\n\n{extra}" if body.strip() else extra
|
||||
|
||||
attachments = []
|
||||
for a in kn.get("attachments") or []:
|
||||
if not isinstance(a, dict):
|
||||
continue
|
||||
fp = a.get("filePath")
|
||||
if not fp:
|
||||
continue
|
||||
zpath = posixpath.join(keep_dir, fp) if keep_dir else fp
|
||||
mime = a.get("mimetype") or _EXT_MIME.get(posixpath.splitext(fp)[1].lower())
|
||||
attachments.append({"file": zpath, "mime": mime})
|
||||
|
||||
return {
|
||||
"title": kn.get("title"),
|
||||
"body": body,
|
||||
"kind": "list" if is_list else "text",
|
||||
"color": _KEEP_COLOR_MAP.get(str(kn.get("color") or "DEFAULT").upper(), "default"),
|
||||
"pinned": bool(kn.get("isPinned")),
|
||||
"archived": bool(kn.get("isArchived")),
|
||||
"trashed": bool(kn.get("isTrashed")),
|
||||
"remind_at": None, # Keep reminders aren't in Takeout note JSON
|
||||
"created_at": _usec_to_dt(kn.get("createdTimestampUsec")),
|
||||
"updated_at": _usec_to_dt(kn.get("userEditedTimestampUsec")),
|
||||
"labels": [
|
||||
lb.get("name")
|
||||
for lb in (kn.get("labels") or [])
|
||||
if isinstance(lb, dict) and lb.get("name")
|
||||
],
|
||||
"items": [
|
||||
{"text": li.get("text"), "checked": bool(li.get("isChecked"))}
|
||||
for li in list_content
|
||||
if isinstance(li, dict)
|
||||
],
|
||||
"attachments": attachments,
|
||||
}
|
||||
|
||||
|
||||
IMPORT_MAX_ENTRIES = 10_000
|
||||
IMPORT_MAX_ENTRY_BYTES = 64 * 1024 * 1024 # 64 MB decompressed per file
|
||||
IMPORT_MAX_TOTAL_BYTES = 512 * 1024 * 1024 # 512 MB decompressed across the whole import
|
||||
|
||||
|
||||
class _ImportTooLarge(Exception):
|
||||
"""An import zip decompressed past the byte budget (a zip bomb, or just too big)."""
|
||||
|
||||
|
||||
class _ImportBudget:
|
||||
"""Caps DECOMPRESSED bytes pulled from an import zip — per entry and cumulatively.
|
||||
|
||||
zipfile inflates into memory on read, so an archive that's tiny on disk can expand
|
||||
to gigabytes. We stream each entry and read at most the remaining budget + 1 byte,
|
||||
so an oversized (or size-lying) entry is caught mid-read instead of after it has
|
||||
already been fully inflated."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.remaining = IMPORT_MAX_TOTAL_BYTES
|
||||
|
||||
def read(self, zf: zipfile.ZipFile, name: str) -> bytes:
|
||||
cap = min(IMPORT_MAX_ENTRY_BYTES, self.remaining)
|
||||
with zf.open(name) as fh:
|
||||
data = fh.read(cap + 1)
|
||||
if len(data) > cap:
|
||||
raise _ImportTooLarge()
|
||||
self.remaining -= len(data)
|
||||
return data
|
||||
|
||||
|
||||
def _read_import_specs(zf: zipfile.ZipFile, budget: _ImportBudget) -> tuple[list[dict], str]:
|
||||
"""Detect the archive format and return (specs, source). A ThoughtSync export
|
||||
is recognized by its notes.json (app == thoughtsync); otherwise each Keep-shaped
|
||||
<note>.json is imported. Returns ([], "") when nothing importable is found."""
|
||||
names = zf.namelist()
|
||||
for name in names:
|
||||
if posixpath.basename(name) == "notes.json":
|
||||
try:
|
||||
doc = json.loads(budget.read(zf, name))
|
||||
except (ValueError, KeyError):
|
||||
continue
|
||||
if isinstance(doc, dict) and doc.get("app") == "thoughtsync":
|
||||
specs = [_native_spec(n) for n in (doc.get("notes") or []) if isinstance(n, dict)]
|
||||
return specs, "thoughtsync"
|
||||
|
||||
keep_specs: list[dict] = []
|
||||
keep_keys = ("textContent", "listContent", "isPinned", "isArchived", "isTrashed", "userEditedTimestampUsec")
|
||||
for name in names:
|
||||
if not name.lower().endswith(".json") or posixpath.basename(name) == "notes.json":
|
||||
continue
|
||||
try:
|
||||
kn = json.loads(budget.read(zf, name))
|
||||
except (ValueError, KeyError):
|
||||
continue
|
||||
if isinstance(kn, dict) and any(k in kn for k in keep_keys):
|
||||
keep_specs.append(_keep_spec(kn, posixpath.dirname(name)))
|
||||
return (keep_specs, "keep") if keep_specs else ([], "")
|
||||
|
||||
|
||||
def _import_attachment(db, note: Note, zf: zipfile.ZipFile, att: dict, budget: _ImportBudget) -> bool:
|
||||
"""Copy one attachment (any type — incl. Keep audio memos) out of the zip into
|
||||
media storage and record it, preserving its filename + hash. Returns True if written."""
|
||||
zpath = att.get("file")
|
||||
if not zpath:
|
||||
return False
|
||||
try:
|
||||
raw = budget.read(zf, zpath)
|
||||
except KeyError:
|
||||
return False
|
||||
filename = _safe_filename(posixpath.basename(zpath))
|
||||
mime = (att.get("mime") or "application/octet-stream").split(";")[0].strip().lower()
|
||||
ext = _attachment_ext(filename, mime)
|
||||
att_id = uuid.uuid4()
|
||||
rel = os.path.join(str(note.id), f"{att_id}{ext}")
|
||||
dest = Config.media_root() / rel
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
dest.write_bytes(raw)
|
||||
db.add(
|
||||
NoteAttachment(
|
||||
id=att_id,
|
||||
note_id=note.id,
|
||||
path=rel,
|
||||
filename=filename,
|
||||
mime=mime,
|
||||
size=len(raw),
|
||||
sha256=hashlib.sha256(raw).hexdigest(),
|
||||
)
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
async def _create_imported_note(
|
||||
db, owner_id, spec: dict, zf: zipfile.ZipFile, position: int, budget: _ImportBudget
|
||||
) -> bool:
|
||||
"""Insert one imported note plus its items/labels/attachments, reusing the same
|
||||
display-title derivation + tag/link reconciliation as create_note. Returns False
|
||||
(nothing written) when the spec is empty."""
|
||||
title = (spec.get("title") or "").strip() or None
|
||||
body = spec.get("body") or ""
|
||||
kind = spec.get("kind") if spec.get("kind") in ("text", "list") else "text"
|
||||
items = spec.get("items") or []
|
||||
if kind == "list":
|
||||
if not (title or any((it.get("text") or "").strip() for it in items)):
|
||||
return False
|
||||
elif is_empty_note(title, body):
|
||||
return False
|
||||
|
||||
note = Note(
|
||||
owner_id=owner_id,
|
||||
title=title,
|
||||
display_title=derive_display_title(title, body),
|
||||
body=body,
|
||||
kind=kind,
|
||||
color=normalize_color(spec.get("color")),
|
||||
pinned=bool(spec.get("pinned")),
|
||||
archived=bool(spec.get("archived")),
|
||||
position=position,
|
||||
)
|
||||
if spec.get("remind_at"):
|
||||
note.remind_at = spec["remind_at"]
|
||||
if spec.get("recurrence"):
|
||||
note.recurrence = spec["recurrence"]
|
||||
if spec.get("trashed"):
|
||||
note.deleted_at = datetime.now(timezone.utc)
|
||||
# Preserve source timestamps: set before flush so they land in the INSERT
|
||||
# (updated_at's onupdate only fires on later UPDATEs, which we don't trigger).
|
||||
if spec.get("created_at"):
|
||||
note.created_at = spec["created_at"]
|
||||
if spec.get("updated_at"):
|
||||
note.updated_at = spec["updated_at"]
|
||||
db.add(note)
|
||||
await db.flush() # assign note.id before items/labels/attachments/links
|
||||
|
||||
if kind == "list":
|
||||
for pos, it in enumerate(items):
|
||||
text = (it.get("text") or "").strip()
|
||||
if text:
|
||||
db.add(NoteItem(note_id=note.id, text=text, checked=bool(it.get("checked")), position=pos))
|
||||
|
||||
# Explicit (picker-style) labels are manual — via_tag=False. Inline #tags in the
|
||||
# body are handled by _reconcile_tags below, same as a normal create.
|
||||
for name in spec.get("labels") or []:
|
||||
name = (name or "").strip()
|
||||
if not name:
|
||||
continue
|
||||
lid = await _find_or_create_label(db, owner_id, name)
|
||||
exists = await db.scalar(
|
||||
select(NoteLabel).where(NoteLabel.note_id == note.id, NoteLabel.label_id == lid)
|
||||
)
|
||||
if exists is None:
|
||||
db.add(NoteLabel(note_id=note.id, label_id=lid, via_tag=False))
|
||||
|
||||
for att in spec.get("attachments") or []:
|
||||
if isinstance(att, dict):
|
||||
_import_attachment(db, note, zf, att, budget)
|
||||
|
||||
await _rewrite_links(db, note)
|
||||
await _reconcile_tags(db, note)
|
||||
return True
|
||||
@@ -0,0 +1,126 @@
|
||||
"""[[wiki-links]] and #tags — parsing note bodies and keeping the derived
|
||||
note_links / tag-sourced note_labels rows in sync with the text. Manual (picker)
|
||||
labels are NOT touched here (see the labeling module)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
from sqlalchemy import delete, func, select
|
||||
|
||||
from ..models.label import Label, NoteLabel
|
||||
from ..models.note import Note
|
||||
from ..models.note_link import NoteLink
|
||||
|
||||
_LINK_RE = re.compile(r"\[\[([^\[\]]+)\]\]")
|
||||
|
||||
# A #tag: `#` at the start of the body or after whitespace, then a word char and
|
||||
# word chars/hyphens. A URL fragment (foo#bar) or mid-word `#` is not preceded by
|
||||
# whitespace, so it won't match.
|
||||
_TAG_RE = re.compile(r"(?:^|(?<=\s))#(\w[\w-]*)")
|
||||
|
||||
|
||||
def parse_tags(body: str | None) -> list[str]:
|
||||
"""Distinct #hashtags from a note body, in order, deduped case-insensitively.
|
||||
A tag must contain a letter, so #2024 or #_ are ignored (avoids numeric noise)."""
|
||||
if not body:
|
||||
return []
|
||||
out: list[str] = []
|
||||
seen: set[str] = set()
|
||||
for match in _TAG_RE.finditer(body):
|
||||
tag = match.group(1)
|
||||
if not any(c.isalpha() for c in tag):
|
||||
continue
|
||||
norm = tag.lower()
|
||||
if norm not in seen:
|
||||
seen.add(norm)
|
||||
out.append(tag)
|
||||
return out
|
||||
|
||||
|
||||
def parse_link_titles(body: str | None) -> list[str]:
|
||||
"""Extract distinct normalized [[wiki-link]] titles from a note body."""
|
||||
if not body:
|
||||
return []
|
||||
out: list[str] = []
|
||||
for match in _LINK_RE.finditer(body):
|
||||
norm = match.group(1).strip().lower()
|
||||
if norm and norm not in out:
|
||||
out.append(norm)
|
||||
return out
|
||||
|
||||
|
||||
async def _rewrite_links(db, note: Note) -> None:
|
||||
"""Replace a note's outgoing wiki-links from its current body."""
|
||||
await db.execute(delete(NoteLink).where(NoteLink.source_id == note.id))
|
||||
for norm in parse_link_titles(note.body):
|
||||
db.add(NoteLink(source_id=note.id, target_norm=norm))
|
||||
|
||||
|
||||
async def _find_or_create_label(db, owner_id, name: str):
|
||||
"""Owner's label id for `name` (case-insensitive match), creating it if absent."""
|
||||
existing = await db.scalar(
|
||||
select(Label.id).where(Label.owner_id == owner_id, func.lower(Label.name) == name.lower())
|
||||
)
|
||||
if existing is not None:
|
||||
return existing
|
||||
label = Label(owner_id=owner_id, name=name)
|
||||
db.add(label)
|
||||
await db.flush()
|
||||
return label.id
|
||||
|
||||
|
||||
async def _reconcile_tags(db, note: Note) -> None:
|
||||
"""Sync tag-sourced labels (via_tag=True) with the #hashtags in the note body:
|
||||
attach labels for current tags, detach tag-labels whose #tag was removed. Manual
|
||||
picker labels (via_tag=False) are never touched."""
|
||||
tag_label_ids: set = set()
|
||||
for name in parse_tags(note.body):
|
||||
tag_label_ids.add(await _find_or_create_label(db, note.owner_id, name))
|
||||
rows = (await db.scalars(select(NoteLabel).where(NoteLabel.note_id == note.id))).all()
|
||||
attached_ids = {r.label_id for r in rows}
|
||||
# Detach tag-labels no longer backed by a #tag in the body.
|
||||
for r in rows:
|
||||
if r.via_tag and r.label_id not in tag_label_ids:
|
||||
await db.delete(r)
|
||||
attached_ids.discard(r.label_id)
|
||||
# Attach new tags — skip labels already attached (in any form) to respect the PK
|
||||
# and leave a manually-added label of the same name as-is.
|
||||
for lid in tag_label_ids:
|
||||
if lid not in attached_ids:
|
||||
db.add(NoteLabel(note_id=note.id, label_id=lid, via_tag=True))
|
||||
attached_ids.add(lid)
|
||||
|
||||
|
||||
def rewrite_link_title(body: str | None, old_norm: str, new_title: str) -> str:
|
||||
"""Repoint every [[token]] whose normalized form == old_norm to [[new_title]]."""
|
||||
if not body:
|
||||
return body or ""
|
||||
|
||||
def _sub(match: re.Match) -> str:
|
||||
return f"[[{new_title}]]" if match.group(1).strip().lower() == old_norm else match.group(0)
|
||||
|
||||
return _LINK_RE.sub(_sub, body)
|
||||
|
||||
|
||||
async def _rename_inbound_links(db, renamed: Note, old_title: str, new_title: str) -> None:
|
||||
"""Rewrite [[old title]] references (and their link rows) in every note that
|
||||
links to the renamed note, so its backlinks survive the title change."""
|
||||
old_norm = old_title.strip().lower()
|
||||
sources = (
|
||||
await db.scalars(
|
||||
select(Note)
|
||||
.join(NoteLink, NoteLink.source_id == Note.id)
|
||||
.where(
|
||||
NoteLink.target_norm == old_norm,
|
||||
Note.owner_id == renamed.owner_id,
|
||||
Note.deleted_at.is_(None),
|
||||
)
|
||||
)
|
||||
).all()
|
||||
seen: set = set()
|
||||
for source in sources:
|
||||
if source.id in seen:
|
||||
continue
|
||||
seen.add(source.id)
|
||||
source.body = rewrite_link_title(source.body, old_norm, new_title)
|
||||
await _rewrite_links(db, source)
|
||||
@@ -0,0 +1,48 @@
|
||||
"""Recurring-reminder math — validate a recurrence interval and roll a reminder
|
||||
forward to its next occurrence (used when a recurring reminder is completed)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import calendar
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
REMINDER_RECURRENCES = {"daily", "weekly", "monthly", "yearly"}
|
||||
|
||||
|
||||
def normalize_recurrence(value: object) -> str | None:
|
||||
return value if value in REMINDER_RECURRENCES else None
|
||||
|
||||
|
||||
def _add_months(dt: datetime, months: int) -> datetime:
|
||||
"""Shift a datetime by whole months, clamping the day to the target month's length
|
||||
(so Jan 31 + 1 month → Feb 28/29). Keeps the time-of-day."""
|
||||
m = dt.month - 1 + months
|
||||
year = dt.year + m // 12
|
||||
month = m % 12 + 1
|
||||
day = min(dt.day, calendar.monthrange(year, month)[1])
|
||||
return dt.replace(year=year, month=month, day=day)
|
||||
|
||||
|
||||
def _advance_once(dt: datetime, recurrence: str) -> datetime | None:
|
||||
if recurrence == "daily":
|
||||
return dt + timedelta(days=1)
|
||||
if recurrence == "weekly":
|
||||
return dt + timedelta(weeks=1)
|
||||
if recurrence == "monthly":
|
||||
return _add_months(dt, 1)
|
||||
if recurrence == "yearly":
|
||||
return _add_months(dt, 12)
|
||||
return None
|
||||
|
||||
|
||||
def next_occurrence(remind_at: datetime, recurrence: str, after: datetime) -> datetime | None:
|
||||
"""The next reminder fire time strictly after `after`, rolling a recurring reminder
|
||||
forward past any missed occurrences. None if `recurrence` isn't a known interval."""
|
||||
nxt = _advance_once(remind_at, recurrence)
|
||||
if nxt is None:
|
||||
return None
|
||||
while nxt <= after:
|
||||
step = _advance_once(nxt, recurrence)
|
||||
if step is None or step == nxt:
|
||||
break
|
||||
nxt = step
|
||||
return nxt
|
||||
@@ -0,0 +1,134 @@
|
||||
"""Note serialization — turn a Note (+ its labels/items/attachments/previews) into
|
||||
the JSON dict the API returns. The bulk loaders (`*_for_notes`) fetch each child
|
||||
collection for a batch of notes in one query, so list endpoints avoid N+1s."""
|
||||
from __future__ import annotations
|
||||
|
||||
from sqlalchemy import select
|
||||
|
||||
from ..models.label import Label, NoteLabel
|
||||
from ..models.note import Note
|
||||
from ..models.note_attachment import NoteAttachment
|
||||
from ..models.note_item import NoteItem
|
||||
from ..models.note_link_preview import NoteLinkPreview
|
||||
|
||||
|
||||
async def _labels_for_notes(db, note_ids: list) -> dict:
|
||||
"""Map note_id -> [{id, name}] in one query (no lazy relationship loading)."""
|
||||
result: dict = {}
|
||||
if not note_ids:
|
||||
return result
|
||||
rows = await db.execute(
|
||||
select(NoteLabel.note_id, Label.id, Label.name, Label.color, NoteLabel.via_tag)
|
||||
.join(Label, Label.id == NoteLabel.label_id)
|
||||
.where(NoteLabel.note_id.in_(note_ids))
|
||||
.order_by(Label.name)
|
||||
)
|
||||
for note_id, label_id, name, color, via_tag in rows.all():
|
||||
result.setdefault(note_id, []).append(
|
||||
{"id": str(label_id), "name": name, "color": color, "via_tag": via_tag}
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def _serialize_item(item: NoteItem) -> dict:
|
||||
return {"id": str(item.id), "text": item.text, "checked": item.checked, "position": item.position}
|
||||
|
||||
|
||||
async def _items_for_notes(db, note_ids: list) -> dict:
|
||||
"""Map note_id -> [checklist items] in one query, ordered by position."""
|
||||
result: dict = {}
|
||||
if not note_ids:
|
||||
return result
|
||||
items = (
|
||||
await db.scalars(
|
||||
select(NoteItem).where(NoteItem.note_id.in_(note_ids)).order_by(NoteItem.position, NoteItem.created_at)
|
||||
)
|
||||
).all()
|
||||
for item in items:
|
||||
result.setdefault(item.note_id, []).append(_serialize_item(item))
|
||||
return result
|
||||
|
||||
|
||||
def _attachment_url(note_id, att_id) -> str:
|
||||
return f"/api/notes/{note_id}/attachments/{att_id}"
|
||||
|
||||
|
||||
async def _attachments_for_notes(db, note_ids: list) -> dict:
|
||||
result: dict = {}
|
||||
if not note_ids:
|
||||
return result
|
||||
rows = (
|
||||
await db.scalars(
|
||||
select(NoteAttachment).where(NoteAttachment.note_id.in_(note_ids)).order_by(NoteAttachment.created_at)
|
||||
)
|
||||
).all()
|
||||
for att in rows:
|
||||
result.setdefault(att.note_id, []).append(
|
||||
{
|
||||
"id": str(att.id),
|
||||
"url": _attachment_url(att.note_id, att.id),
|
||||
"filename": att.filename,
|
||||
"mime": att.mime,
|
||||
"size": att.size,
|
||||
"sha256": att.sha256,
|
||||
}
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def _serialize_preview(p: NoteLinkPreview) -> dict:
|
||||
return {
|
||||
"id": str(p.id),
|
||||
"url": p.url,
|
||||
"title": p.title,
|
||||
"description": p.description,
|
||||
"image_url": p.image_url,
|
||||
"site_name": p.site_name,
|
||||
}
|
||||
|
||||
|
||||
async def _previews_for_notes(db, note_ids: list) -> dict:
|
||||
"""Map note_id -> [link previews] in one query."""
|
||||
result: dict = {}
|
||||
if not note_ids:
|
||||
return result
|
||||
rows = (
|
||||
await db.scalars(
|
||||
select(NoteLinkPreview)
|
||||
.where(NoteLinkPreview.note_id.in_(note_ids))
|
||||
.order_by(NoteLinkPreview.created_at)
|
||||
)
|
||||
).all()
|
||||
for p in rows:
|
||||
result.setdefault(p.note_id, []).append(_serialize_preview(p))
|
||||
return result
|
||||
|
||||
|
||||
async def _serialize_note(db, note: Note) -> dict:
|
||||
data = note.serialize()
|
||||
labels = await _labels_for_notes(db, [note.id])
|
||||
data["labels"] = labels.get(note.id, [])
|
||||
items = await _items_for_notes(db, [note.id])
|
||||
data["items"] = items.get(note.id, [])
|
||||
attachments = await _attachments_for_notes(db, [note.id])
|
||||
data["attachments"] = attachments.get(note.id, [])
|
||||
previews = await _previews_for_notes(db, [note.id])
|
||||
data["previews"] = previews.get(note.id, [])
|
||||
return data
|
||||
|
||||
|
||||
async def _serialize_notes(db, notes: list) -> list:
|
||||
ids = [n.id for n in notes]
|
||||
labels_map = await _labels_for_notes(db, ids)
|
||||
items_map = await _items_for_notes(db, ids)
|
||||
attach_map = await _attachments_for_notes(db, ids)
|
||||
preview_map = await _previews_for_notes(db, ids)
|
||||
out = []
|
||||
for n in notes:
|
||||
data = n.serialize()
|
||||
data["labels"] = labels_map.get(n.id, [])
|
||||
data["items"] = items_map.get(n.id, [])
|
||||
data["attachments"] = attach_map.get(n.id, [])
|
||||
data["previews"] = preview_map.get(n.id, [])
|
||||
out.append(data)
|
||||
return out
|
||||
@@ -31,6 +31,27 @@ def app():
|
||||
return create_app()
|
||||
|
||||
|
||||
def test_all_note_routes_registered(app):
|
||||
# Guards the notes package split: every route handler must still be attached to the
|
||||
# blueprint. A route whose module isn't imported by notes/__init__ would silently
|
||||
# 404 at runtime, and most routes have no auth-guard test to otherwise catch it.
|
||||
registered = {r.endpoint for r in app.url_map.iter_rules()}
|
||||
expected = {
|
||||
f"notes.{name}"
|
||||
for name in (
|
||||
"list_notes", "search_notes", "list_reminders", "complete_reminder",
|
||||
"snooze_reminder", "export_notes", "import_notes", "list_titles",
|
||||
"link_search", "note_backlinks", "reorder_notes", "create_note",
|
||||
"get_note", "update_note", "list_revisions", "restore_revision",
|
||||
"set_note_labels", "add_item", "update_item", "delete_item",
|
||||
"reorder_items", "upload_attachment", "get_attachment",
|
||||
"delete_attachment", "unfurl_link", "delete_preview", "trash_note",
|
||||
"restore_note", "delete_note",
|
||||
)
|
||||
}
|
||||
assert expected <= registered, f"unregistered note routes: {expected - registered}"
|
||||
|
||||
|
||||
def test_is_empty_note():
|
||||
assert is_empty_note(None, None)
|
||||
assert is_empty_note("", " ")
|
||||
|
||||
Reference in New Issue
Block a user