Jeffrey Smith
6acfeab456
All checks were successful
CI/CD / detect-changes (pull_request) Successful in 22s
CI/CD / test-frontend (pull_request) Successful in 5s
CI/CD / test-go-pg (pull_request) Successful in 2m48s
CI/CD / test-sqlite (pull_request) Successful in 3m5s
CI/CD / build-and-deploy (pull_request) Successful in 2m29s
Canvas force-directed graph view for notes surface. GET /graph API endpoint returns all non-archived notes as nodes (with tags) and resolved wikilinks as edges. Force simulation with repulsion, attraction, and center gravity. Click focus dims unconnected nodes, shift+click chains neighborhoods, double-click opens note. Hover tooltips show title, tags, and connection count. Nodes colored by folder (10-color palette), tag filter dims non-matching. Orphan nodes shown with dashed stroke and hide toggle. Zoom/pan via scroll wheel and mouse drag. Retina-aware canvas rendering with CSS variable resolution via getComputedStyle. Notes package 0.7.0 → 0.8.0. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
669 lines
22 KiB
Plaintext
669 lines
22 KiB
Plaintext
# Notes — Starlark Backend (v0.4.0)
|
|
#
|
|
# Markdown notes surface using ext_data.
|
|
#
|
|
# Entry points:
|
|
# on_request(req) → surface API routes
|
|
#
|
|
# Modules: db, json, settings
|
|
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# Helpers
|
|
# ═══════════════════════════════════════════════
|
|
|
|
def _resp(status, data):
|
|
return {"status": status, "body": json.encode(data), "headers": {"Content-Type": "application/json"}}
|
|
|
|
def _str(v):
|
|
if v == None:
|
|
return ""
|
|
return str(v)
|
|
|
|
def _int(v):
|
|
if v == None:
|
|
return 0
|
|
s = str(v)
|
|
if not s:
|
|
return 0
|
|
return int(s)
|
|
|
|
def _snippet(body, max_len = 120):
|
|
"""First max_len chars of body, stripped of leading #/whitespace."""
|
|
if not body:
|
|
return ""
|
|
line = body.split("\n")[0]
|
|
# strip leading markdown heading markers
|
|
line = line.lstrip("#").strip()
|
|
if len(line) > max_len:
|
|
return line[:max_len]
|
|
return line
|
|
|
|
def _tags_for_notes(note_ids):
|
|
"""Batch-fetch tags for a list of note IDs. Returns {note_id: [tag, ...]}."""
|
|
if not note_ids:
|
|
return {}
|
|
all_tags = db.query("tags", limit=5000)
|
|
result = {}
|
|
for t in (all_tags or []):
|
|
nid = _str(t.get("note_id", ""))
|
|
tag = _str(t.get("tag", ""))
|
|
if nid and tag:
|
|
if nid not in result:
|
|
result[nid] = []
|
|
result[nid].append(tag)
|
|
return result
|
|
|
|
|
|
def _extract_wikilinks(text):
|
|
"""Parse [[...]] patterns from text. Returns deduplicated list of link_text strings."""
|
|
links = []
|
|
seen = {}
|
|
parts = text.split("[[")
|
|
# parts[0] is before any wikilink; parts[1:] each start after "[["
|
|
for i in range(1, len(parts)):
|
|
chunk = parts[i]
|
|
end = chunk.find("]]")
|
|
if end > 0:
|
|
link_text = chunk[:end].strip()
|
|
lt_lower = link_text.lower()
|
|
if link_text and lt_lower not in seen:
|
|
seen[lt_lower] = True
|
|
links.append(link_text)
|
|
return links
|
|
|
|
|
|
def _sync_links(source_id, body):
|
|
"""Extract wikilinks from body, resolve to note IDs, replace link rows."""
|
|
raw_links = _extract_wikilinks(_str(body))
|
|
if not raw_links and not db.query("links", filters={"source_id": source_id}, limit=1):
|
|
return # nothing to do
|
|
|
|
# resolve titles to note IDs (case-insensitive)
|
|
all_notes = db.query("notes", filters={"archived": 0}, limit=5000)
|
|
title_map = {}
|
|
for n in (all_notes or []):
|
|
t = _str(n.get("title", "")).lower().strip()
|
|
if t:
|
|
title_map[t] = n.get("id", "")
|
|
|
|
# delete existing links for this source
|
|
old_links = db.query("links", filters={"source_id": source_id}, limit=500)
|
|
for link in (old_links or []):
|
|
db.delete("links", link["id"])
|
|
|
|
# insert new links
|
|
for lt in raw_links:
|
|
target_id = title_map.get(lt.lower().strip(), "")
|
|
# skip self-links
|
|
if target_id == source_id:
|
|
target_id = ""
|
|
db.insert("links", {
|
|
"source_id": source_id,
|
|
"target_id": target_id,
|
|
"link_text": lt,
|
|
})
|
|
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# Surface API routes (/s/notes/api/*)
|
|
# ═══════════════════════════════════════════════
|
|
|
|
def on_request(req):
|
|
path = req["path"]
|
|
method = req["method"]
|
|
|
|
# GET /notes — list
|
|
if method == "GET" and path == "/notes":
|
|
return _list_notes(req)
|
|
|
|
# POST /notes — create
|
|
if method == "POST" and path == "/notes":
|
|
return _create_note(req)
|
|
|
|
# GET /stats
|
|
if method == "GET" and path == "/stats":
|
|
return _get_stats()
|
|
|
|
# GET /search
|
|
if method == "GET" and path == "/search":
|
|
return _search_notes(req)
|
|
|
|
# POST /notes/move (must be before /notes/:id catch-all)
|
|
if method == "POST" and path == "/notes/move":
|
|
return _move_note(req)
|
|
|
|
# GET /graph — full graph data for visualization
|
|
if method == "GET" and path == "/graph":
|
|
return _get_graph()
|
|
|
|
# ── Link routes ──────────────────────────
|
|
# GET /links/:note_id — outgoing links
|
|
if method == "GET" and path.startswith("/links/"):
|
|
return _get_links(path[len("/links/"):])
|
|
|
|
# GET /backlinks/:note_id — incoming links
|
|
if method == "GET" and path.startswith("/backlinks/"):
|
|
return _get_backlinks(path[len("/backlinks/"):])
|
|
|
|
# ── Tag routes ────────────────────────────
|
|
# GET /tags — all unique tags
|
|
if method == "GET" and path == "/tags":
|
|
return _list_tags()
|
|
|
|
# GET /tags/:note_id
|
|
if method == "GET" and path.startswith("/tags/"):
|
|
return _get_note_tags(path[len("/tags/"):])
|
|
|
|
# PUT /tags/:note_id
|
|
if method == "PUT" and path.startswith("/tags/"):
|
|
return _set_note_tags(path[len("/tags/"):], req)
|
|
|
|
# ── Note routes ───────────────────────────
|
|
# GET /notes/:id
|
|
if method == "GET" and path.startswith("/notes/"):
|
|
return _get_note(path[len("/notes/"):])
|
|
|
|
# PUT /notes/:id
|
|
if method == "PUT" and path.startswith("/notes/"):
|
|
return _update_note(path[len("/notes/"):], req)
|
|
|
|
# DELETE /notes/:id
|
|
if method == "DELETE" and path.startswith("/notes/"):
|
|
return _delete_note(path[len("/notes/"):], req)
|
|
|
|
# ── Folder routes ─────────────────────────
|
|
# GET /folders
|
|
if method == "GET" and path == "/folders":
|
|
return _list_folders()
|
|
|
|
# POST /folders
|
|
if method == "POST" and path == "/folders":
|
|
return _create_folder(req)
|
|
|
|
# PUT /folders/:id
|
|
if method == "PUT" and path.startswith("/folders/"):
|
|
return _update_folder(path[len("/folders/"):], req)
|
|
|
|
# DELETE /folders/:id
|
|
if method == "DELETE" and path.startswith("/folders/"):
|
|
return _delete_folder(path[len("/folders/"):])
|
|
|
|
return _resp(404, {"error": "not found"})
|
|
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# Notes CRUD
|
|
# ═══════════════════════════════════════════════
|
|
|
|
def _list_notes(req):
|
|
q = req.get("query", {})
|
|
filters = {}
|
|
|
|
folder = _str(q.get("folder_id", ""))
|
|
if folder:
|
|
filters["folder_id"] = folder
|
|
|
|
pinned = _str(q.get("pinned", ""))
|
|
if pinned:
|
|
filters["pinned"] = int(pinned)
|
|
|
|
archived = _str(q.get("archived", ""))
|
|
if archived:
|
|
filters["archived"] = int(archived)
|
|
else:
|
|
# default: exclude archived
|
|
filters["archived"] = 0
|
|
|
|
creator = _str(q.get("creator_id", ""))
|
|
if creator:
|
|
filters["creator_id"] = creator
|
|
|
|
order = _str(q.get("order", "")) or "updated_at"
|
|
limit_str = _str(q.get("limit", ""))
|
|
limit = int(limit_str) if limit_str else 200
|
|
|
|
rows = db.query("notes", filters=filters, order=order, limit=limit)
|
|
|
|
# batch-fetch tags for all notes
|
|
note_ids = []
|
|
for r in (rows or []):
|
|
note_ids.append(r.get("id", ""))
|
|
tags_map = _tags_for_notes(note_ids)
|
|
|
|
# lightweight projection: strip body for list view
|
|
items = []
|
|
for r in (rows or []):
|
|
nid = r.get("id", "")
|
|
items.append({
|
|
"id": nid,
|
|
"title": r.get("title", ""),
|
|
"snippet": _snippet(r.get("body", "")),
|
|
"folder_id": r.get("folder_id", ""),
|
|
"creator_id": r.get("creator_id", ""),
|
|
"updated_at": r.get("updated_at", ""),
|
|
"pinned": _int(r.get("pinned", 0)),
|
|
"archived": _int(r.get("archived", 0)),
|
|
"created_at": r.get("created_at", ""),
|
|
"tags": tags_map.get(nid, []),
|
|
})
|
|
return _resp(200, {"data": items})
|
|
|
|
|
|
def _create_note(req):
|
|
body = json.decode(req.get("body", "{}"))
|
|
user_id = req.get("user_id", "")
|
|
|
|
title = _str(body.get("title", ""))
|
|
if not title:
|
|
title = "Untitled"
|
|
|
|
body_text = _str(body.get("body", ""))
|
|
row = db.insert("notes", {
|
|
"title": title,
|
|
"body": body_text,
|
|
"folder_id": _str(body.get("folder_id", "")),
|
|
"creator_id": user_id,
|
|
"updated_at": "",
|
|
"pinned": _int(body.get("pinned", 0)),
|
|
"archived": 0,
|
|
})
|
|
_sync_links(row["id"], body_text)
|
|
return _resp(201, row)
|
|
|
|
|
|
def _get_note(note_id):
|
|
rows = db.query("notes", filters={"id": note_id}, limit=1)
|
|
if not rows:
|
|
return _resp(404, {"error": "note not found"})
|
|
note = rows[0]
|
|
# include tags
|
|
tag_rows = db.query("tags", filters={"note_id": note_id}, limit=100)
|
|
tags = []
|
|
for t in (tag_rows or []):
|
|
tag = _str(t.get("tag", ""))
|
|
if tag:
|
|
tags.append(tag)
|
|
note["tags"] = tags
|
|
return _resp(200, note)
|
|
|
|
|
|
def _update_note(note_id, req):
|
|
body = json.decode(req.get("body", "{}"))
|
|
|
|
existing = db.query("notes", filters={"id": note_id}, limit=1)
|
|
if not existing:
|
|
return _resp(404, {"error": "note not found"})
|
|
|
|
updates = {}
|
|
for key in ["title", "body", "folder_id"]:
|
|
if key in body:
|
|
updates[key] = _str(body[key])
|
|
|
|
if "pinned" in body:
|
|
updates["pinned"] = _int(body["pinned"])
|
|
if "archived" in body:
|
|
updates["archived"] = _int(body["archived"])
|
|
|
|
# touch updated_at
|
|
updates["updated_at"] = "now"
|
|
|
|
ok = db.update("notes", note_id, updates)
|
|
if not ok:
|
|
return _resp(500, {"error": "update failed"})
|
|
|
|
# sync wikilinks when body changes
|
|
if "body" in body:
|
|
_sync_links(note_id, _str(body["body"]))
|
|
|
|
rows = db.query("notes", filters={"id": note_id}, limit=1)
|
|
return _resp(200, rows[0] if rows else {})
|
|
|
|
|
|
def _delete_note(note_id, req):
|
|
q = req.get("query", {})
|
|
hard = _str(q.get("hard", ""))
|
|
|
|
if hard == "1":
|
|
# cascade-delete tags for this note
|
|
note_tags = db.query("tags", filters={"note_id": note_id}, limit=100)
|
|
for t in (note_tags or []):
|
|
db.delete("tags", t["id"])
|
|
# cascade-delete outgoing links FROM this note
|
|
out_links = db.query("links", filters={"source_id": note_id}, limit=500)
|
|
for link in (out_links or []):
|
|
db.delete("links", link["id"])
|
|
# cascade-delete incoming backlinks TO this note
|
|
in_links = db.query("links", filters={"target_id": note_id}, limit=500)
|
|
for link in (in_links or []):
|
|
db.delete("links", link["id"])
|
|
ok = db.delete("notes", note_id)
|
|
if not ok:
|
|
return _resp(404, {"error": "note not found"})
|
|
return _resp(200, {"deleted": True})
|
|
|
|
# soft delete: archive
|
|
ok = db.update("notes", note_id, {"archived": 1, "updated_at": "now"})
|
|
if not ok:
|
|
return _resp(404, {"error": "note not found"})
|
|
return _resp(200, {"archived": True})
|
|
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# Move note between folders
|
|
# ═══════════════════════════════════════════════
|
|
|
|
def _move_note(req):
|
|
body = json.decode(req.get("body", "{}"))
|
|
note_id = _str(body.get("note_id", ""))
|
|
folder_id = _str(body.get("folder_id", ""))
|
|
if not note_id:
|
|
return _resp(400, {"error": "note_id required"})
|
|
existing = db.query("notes", filters={"id": note_id}, limit=1)
|
|
if not existing:
|
|
return _resp(404, {"error": "note not found"})
|
|
ok = db.update("notes", note_id, {"folder_id": folder_id, "updated_at": "now"})
|
|
if not ok:
|
|
return _resp(500, {"error": "move failed"})
|
|
rows = db.query("notes", filters={"id": note_id}, limit=1)
|
|
return _resp(200, rows[0] if rows else {})
|
|
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# Tags CRUD
|
|
# ═══════════════════════════════════════════════
|
|
|
|
def _list_tags():
|
|
"""Return sorted list of all unique tag strings."""
|
|
all_tags = db.query("tags", limit=5000)
|
|
seen = {}
|
|
for t in (all_tags or []):
|
|
tag = _str(t.get("tag", "")).strip()
|
|
if tag:
|
|
seen[tag] = True
|
|
tags = sorted(seen.keys())
|
|
return _resp(200, {"data": tags})
|
|
|
|
|
|
def _get_note_tags(note_id):
|
|
"""Return tags for a specific note."""
|
|
rows = db.query("tags", filters={"note_id": note_id}, limit=100)
|
|
tags = []
|
|
for t in (rows or []):
|
|
tag = _str(t.get("tag", ""))
|
|
if tag:
|
|
tags.append(tag)
|
|
return _resp(200, {"data": tags})
|
|
|
|
|
|
def _set_note_tags(note_id, req):
|
|
"""Replace all tags for a note (delete + reinsert)."""
|
|
body = json.decode(req.get("body", "{}"))
|
|
new_tags_raw = body.get("tags", [])
|
|
|
|
# validate note exists
|
|
existing = db.query("notes", filters={"id": note_id}, limit=1)
|
|
if not existing:
|
|
return _resp(404, {"error": "note not found"})
|
|
|
|
# normalize: lowercase, strip, deduplicate, skip empty
|
|
seen = {}
|
|
clean_tags = []
|
|
for raw in new_tags_raw:
|
|
tag = _str(raw).strip().lower()
|
|
if tag and tag not in seen:
|
|
seen[tag] = True
|
|
clean_tags.append(tag)
|
|
|
|
# delete all existing tags for this note
|
|
old_tags = db.query("tags", filters={"note_id": note_id}, limit=100)
|
|
for t in (old_tags or []):
|
|
db.delete("tags", t["id"])
|
|
|
|
# insert new tags
|
|
for tag in clean_tags:
|
|
db.insert("tags", {"note_id": note_id, "tag": tag})
|
|
|
|
return _resp(200, {"data": clean_tags})
|
|
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# Links (backlinks + wikilinks)
|
|
# ═══════════════════════════════════════════════
|
|
|
|
def _get_links(note_id):
|
|
"""Return outgoing links for a note."""
|
|
rows = db.query("links", filters={"source_id": note_id}, limit=500)
|
|
items = []
|
|
for r in (rows or []):
|
|
items.append({
|
|
"source_id": r.get("source_id", ""),
|
|
"target_id": r.get("target_id", ""),
|
|
"link_text": r.get("link_text", ""),
|
|
})
|
|
return _resp(200, {"data": items})
|
|
|
|
|
|
def _get_backlinks(note_id):
|
|
"""Return notes that link TO this note (backlinks), enriched with source titles."""
|
|
rows = db.query("links", filters={"target_id": note_id}, limit=500)
|
|
if not rows:
|
|
return _resp(200, {"data": []})
|
|
|
|
# batch-fetch source note titles
|
|
all_notes = db.query("notes", filters={"archived": 0}, limit=5000)
|
|
note_map = {}
|
|
for n in (all_notes or []):
|
|
note_map[n.get("id", "")] = n.get("title", "Untitled")
|
|
|
|
items = []
|
|
for r in rows:
|
|
sid = r.get("source_id", "")
|
|
items.append({
|
|
"source_id": sid,
|
|
"source_title": note_map.get(sid, "Unknown"),
|
|
"link_text": r.get("link_text", ""),
|
|
})
|
|
return _resp(200, {"data": items})
|
|
|
|
|
|
def _get_graph():
|
|
"""Return all non-archived notes as nodes and resolved links as edges."""
|
|
all_notes = db.query("notes", filters={"archived": 0}, limit=5000)
|
|
notes = all_notes or []
|
|
|
|
# batch-fetch tags
|
|
note_ids = [n.get("id", "") for n in notes]
|
|
tags_map = _tags_for_notes(note_ids)
|
|
|
|
# build nodes + id set for edge filtering
|
|
nodes = []
|
|
note_id_set = {}
|
|
for n in notes:
|
|
nid = n.get("id", "")
|
|
note_id_set[nid] = True
|
|
nodes.append({
|
|
"id": nid,
|
|
"title": n.get("title", "Untitled"),
|
|
"folder_id": _str(n.get("folder_id", "")),
|
|
"tags": tags_map.get(nid, []),
|
|
})
|
|
|
|
# fetch links, keep only resolved (both endpoints exist)
|
|
all_links = db.query("links", limit=10000)
|
|
edges = []
|
|
for link in (all_links or []):
|
|
src = _str(link.get("source_id", ""))
|
|
tgt = _str(link.get("target_id", ""))
|
|
if src and tgt and src in note_id_set and tgt in note_id_set:
|
|
edges.append({
|
|
"source": src,
|
|
"target": tgt,
|
|
"text": _str(link.get("link_text", "")),
|
|
})
|
|
|
|
return _resp(200, {"nodes": nodes, "edges": edges})
|
|
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# Folder CRUD
|
|
# ═══════════════════════════════════════════════
|
|
|
|
def _list_folders():
|
|
rows = db.query("folders", order="sort_order", limit=500)
|
|
items = []
|
|
for r in (rows or []):
|
|
items.append({
|
|
"id": r.get("id", ""),
|
|
"name": r.get("name", ""),
|
|
"parent_id": r.get("parent_id", ""),
|
|
"sort_order": _int(r.get("sort_order", 0)),
|
|
"creator_id": r.get("creator_id", ""),
|
|
"created_at": r.get("created_at", ""),
|
|
})
|
|
return _resp(200, {"data": items})
|
|
|
|
|
|
def _create_folder(req):
|
|
body = json.decode(req.get("body", "{}"))
|
|
user_id = req.get("user_id", "")
|
|
name = _str(body.get("name", ""))
|
|
if not name:
|
|
return _resp(400, {"error": "name required"})
|
|
row = db.insert("folders", {
|
|
"name": name,
|
|
"parent_id": _str(body.get("parent_id", "")),
|
|
"creator_id": user_id,
|
|
"sort_order": _int(body.get("sort_order", 0)),
|
|
})
|
|
return _resp(201, row)
|
|
|
|
|
|
def _update_folder(folder_id, req):
|
|
body = json.decode(req.get("body", "{}"))
|
|
existing = db.query("folders", filters={"id": folder_id}, limit=1)
|
|
if not existing:
|
|
return _resp(404, {"error": "folder not found"})
|
|
updates = {}
|
|
for key in ["name", "parent_id"]:
|
|
if key in body:
|
|
updates[key] = _str(body[key])
|
|
if "sort_order" in body:
|
|
updates["sort_order"] = _int(body["sort_order"])
|
|
if not updates:
|
|
return _resp(200, existing[0])
|
|
ok = db.update("folders", folder_id, updates)
|
|
if not ok:
|
|
return _resp(500, {"error": "update failed"})
|
|
rows = db.query("folders", filters={"id": folder_id}, limit=1)
|
|
return _resp(200, rows[0] if rows else {})
|
|
|
|
|
|
def _delete_folder(folder_id):
|
|
existing = db.query("folders", filters={"id": folder_id}, limit=1)
|
|
if not existing:
|
|
return _resp(404, {"error": "folder not found"})
|
|
parent_id = _str(existing[0].get("parent_id", ""))
|
|
|
|
# reparent child folders to deleted folder's parent
|
|
children = db.query("folders", filters={"parent_id": folder_id}, limit=500)
|
|
for child in (children or []):
|
|
db.update("folders", child["id"], {"parent_id": parent_id})
|
|
|
|
# orphan notes in this folder (move to unfiled)
|
|
notes_in_folder = db.query("notes", filters={"folder_id": folder_id}, limit=1000)
|
|
for n in (notes_in_folder or []):
|
|
db.update("notes", n["id"], {"folder_id": ""})
|
|
|
|
ok = db.delete("folders", folder_id)
|
|
if not ok:
|
|
return _resp(500, {"error": "delete failed"})
|
|
return _resp(200, {"deleted": True, "orphaned_notes": len(notes_in_folder or [])})
|
|
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# Search
|
|
# ═══════════════════════════════════════════════
|
|
|
|
def _search_notes(req):
|
|
q = req.get("query", {})
|
|
term = _str(q.get("q", ""))
|
|
if not term:
|
|
return _resp(400, {"error": "q parameter required"})
|
|
|
|
# fetch all non-archived notes and filter in Starlark
|
|
# (ext_data db.query doesn't support LIKE — filter client-side)
|
|
rows = db.query("notes", filters={"archived": 0}, order="updated_at", limit=1000)
|
|
term_lower = term.lower()
|
|
|
|
# batch-fetch tags for tag-based matching
|
|
tags_map = _tags_for_notes([r.get("id", "") for r in (rows or [])])
|
|
|
|
matches = []
|
|
for r in (rows or []):
|
|
title = _str(r.get("title", "")).lower()
|
|
body = _str(r.get("body", "")).lower()
|
|
nid = r.get("id", "")
|
|
|
|
# check title and body
|
|
found = term_lower in title or term_lower in body
|
|
|
|
# check tags
|
|
if not found:
|
|
note_tags = tags_map.get(nid, [])
|
|
for tg in note_tags:
|
|
if term_lower in tg.lower():
|
|
found = True
|
|
|
|
if found:
|
|
matches.append({
|
|
"id": nid,
|
|
"title": r.get("title", ""),
|
|
"snippet": _snippet(r.get("body", "")),
|
|
"folder_id": r.get("folder_id", ""),
|
|
"updated_at": r.get("updated_at", ""),
|
|
"pinned": _int(r.get("pinned", 0)),
|
|
"created_at": r.get("created_at", ""),
|
|
"tags": tags_map.get(nid, []),
|
|
})
|
|
return _resp(200, {"data": matches})
|
|
|
|
|
|
# ═══════════════════════════════════════════════
|
|
# Stats
|
|
# ═══════════════════════════════════════════════
|
|
|
|
def _get_stats():
|
|
all_notes = db.query("notes", limit=10000)
|
|
notes = all_notes or []
|
|
total = 0
|
|
pinned = 0
|
|
archived = 0
|
|
unfiled = 0
|
|
for n in notes:
|
|
if _int(n.get("archived", 0)) == 1:
|
|
archived = archived + 1
|
|
else:
|
|
total = total + 1
|
|
if _int(n.get("pinned", 0)) == 1:
|
|
pinned = pinned + 1
|
|
if not _str(n.get("folder_id", "")):
|
|
unfiled = unfiled + 1
|
|
all_folders = db.query("folders", limit=10000)
|
|
folder_count = len(all_folders or [])
|
|
|
|
# count unique tags
|
|
all_tags = db.query("tags", limit=10000)
|
|
tag_seen = {}
|
|
for t in (all_tags or []):
|
|
tag = _str(t.get("tag", ""))
|
|
if tag:
|
|
tag_seen[tag] = True
|
|
tag_count = len(tag_seen)
|
|
|
|
# count links
|
|
all_links = db.query("links", limit=10000)
|
|
link_count = len(all_links or [])
|
|
|
|
return _resp(200, {"total": total, "pinned": pinned, "archived": archived, "unfiled": unfiled, "folders": folder_count, "tags": tag_count, "links": link_count})
|