327 lines
10 KiB
Python
327 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
import uuid
|
|
|
|
import aiosqlite
|
|
|
|
from backend.config import settings
|
|
from backend.database import get_user_id
|
|
|
|
TAG_RE = re.compile(r"#[A-Za-z0-9_-]+")
|
|
|
|
|
|
def extract_tags(text: str) -> list[str]:
|
|
seen: list[str] = []
|
|
for match in TAG_RE.finditer(text):
|
|
tag = match.group(0)[1:].lower()
|
|
if tag not in seen:
|
|
seen.append(tag)
|
|
return seen
|
|
|
|
|
|
def extract_title(text: str) -> str:
|
|
for line in text.splitlines():
|
|
stripped = line.strip()
|
|
if not stripped:
|
|
continue
|
|
if stripped.startswith("# "):
|
|
title = stripped[2:].strip()
|
|
else:
|
|
title = stripped
|
|
if title:
|
|
return title[:200]
|
|
return "Untitled"
|
|
|
|
|
|
def note_path(username: str, note_id: str) -> str:
|
|
return f"{username}/{note_id}.md"
|
|
|
|
|
|
async def ensure_user_dir(username: str) -> None:
|
|
import os
|
|
|
|
os.makedirs(os.path.join(settings.data_dir, username), exist_ok=True)
|
|
|
|
|
|
async def write_note_file(username: str, note_id: str, body: str) -> str:
|
|
from pathlib import Path
|
|
|
|
await ensure_user_dir(username)
|
|
rel = note_path(username, note_id)
|
|
full = Path(settings.data_dir) / rel
|
|
full.write_text(body, encoding="utf-8")
|
|
return rel
|
|
|
|
|
|
async def read_note_file(rel_path: str) -> str:
|
|
from pathlib import Path
|
|
|
|
full = Path(settings.data_dir) / rel_path
|
|
if not full.is_file():
|
|
raise FileNotFoundError(rel_path)
|
|
return full.read_text(encoding="utf-8")
|
|
|
|
|
|
async def delete_note_file(rel_path: str) -> None:
|
|
from pathlib import Path
|
|
|
|
full = Path(settings.data_dir) / rel_path
|
|
if full.is_file():
|
|
full.unlink()
|
|
|
|
|
|
def new_note_id() -> str:
|
|
return uuid.uuid4().hex
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Note CRUD (metadata + tags + FTS)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def create_note(
|
|
username: str,
|
|
body: str,
|
|
ip: str | None = None,
|
|
ua: str | None = None,
|
|
) -> dict:
|
|
title = extract_title(body)
|
|
tags = extract_tags(body)
|
|
note_id = new_note_id()
|
|
rel = await write_note_file(username, note_id, body)
|
|
|
|
async with aiosqlite.connect(settings.database_path) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
user_id = await get_user_id(db, username)
|
|
if user_id is None:
|
|
raise ValueError(f"unknown user: {username}")
|
|
|
|
await db.execute(
|
|
"INSERT INTO notes (id, user_id, title, path) VALUES (?, ?, ?, ?)",
|
|
(note_id, user_id, title, rel),
|
|
)
|
|
await _set_tags(db, note_id, tags)
|
|
await _index_note(db, note_id, user_id, title, body)
|
|
await _record_history(db, note_id, user_id, "create", ip, ua)
|
|
await db.commit()
|
|
|
|
row = await db.execute(
|
|
"SELECT n.id, n.title, n.path, n.created_at, n.updated_at,"
|
|
" u.name AS username"
|
|
" FROM notes n JOIN users u ON u.id = n.user_id WHERE n.id = ?",
|
|
(note_id,),
|
|
)
|
|
note = dict(await row.fetchone())
|
|
note["tags"] = tags
|
|
note["body"] = body
|
|
return note
|
|
|
|
|
|
async def update_note(
|
|
note_id: str,
|
|
body: str,
|
|
ip: str | None = None,
|
|
ua: str | None = None,
|
|
) -> dict | None:
|
|
title = extract_title(body)
|
|
tags = extract_tags(body)
|
|
|
|
async with aiosqlite.connect(settings.database_path) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
row = await db.execute(
|
|
"SELECT n.id, n.user_id, n.path, u.name AS username"
|
|
" FROM notes n JOIN users u ON u.id = n.user_id WHERE n.id = ?",
|
|
(note_id,),
|
|
)
|
|
row = await row.fetchone()
|
|
if row is None:
|
|
return None
|
|
|
|
await write_note_file(row["username"], note_id, body)
|
|
await db.execute(
|
|
"UPDATE notes SET title = ?, updated_at = datetime('now') WHERE id = ?",
|
|
(title, note_id),
|
|
)
|
|
await _set_tags(db, note_id, tags)
|
|
await _index_note(db, note_id, row["user_id"], title, body)
|
|
await _record_history(db, note_id, row["user_id"], "update", ip, ua)
|
|
await db.commit()
|
|
|
|
return await get_note(note_id)
|
|
|
|
|
|
async def get_note(note_id: str) -> dict | None:
|
|
async with aiosqlite.connect(settings.database_path) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
row = await db.execute(
|
|
"SELECT n.id, n.title, n.path, n.created_at, n.updated_at,"
|
|
" u.name AS username"
|
|
" FROM notes n JOIN users u ON u.id = n.user_id"
|
|
" WHERE n.id = ? AND n.deleted_at IS NULL",
|
|
(note_id,),
|
|
)
|
|
row = await row.fetchone()
|
|
if row is None:
|
|
return None
|
|
note = dict(row)
|
|
note["tags"] = await get_note_tags(note_id)
|
|
note["body"] = await read_note_file(note["path"])
|
|
return note
|
|
|
|
|
|
async def delete_note(
|
|
note_id: str,
|
|
ip: str | None = None,
|
|
ua: str | None = None,
|
|
) -> bool:
|
|
async with aiosqlite.connect(settings.database_path) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
row = await db.execute(
|
|
"SELECT user_id, path FROM notes WHERE id = ? AND deleted_at IS NULL",
|
|
(note_id,),
|
|
)
|
|
row = await row.fetchone()
|
|
if row is None:
|
|
return False
|
|
|
|
await delete_note_file(row["path"])
|
|
await db.execute(
|
|
"UPDATE notes SET deleted_at = datetime('now') WHERE id = ?",
|
|
(note_id,),
|
|
)
|
|
await _record_history(db, note_id, row["user_id"], "delete", ip, ua)
|
|
await db.commit()
|
|
return True
|
|
|
|
|
|
async def list_notes(username: str, limit: int = 100) -> list[dict]:
|
|
async with aiosqlite.connect(settings.database_path) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
rows = await db.execute(
|
|
"SELECT n.id, n.title, n.updated_at, n.created_at, u.name AS username"
|
|
" FROM notes n JOIN users u ON u.id = n.user_id"
|
|
" WHERE n.deleted_at IS NULL AND u.name = ?"
|
|
" ORDER BY n.updated_at DESC LIMIT ?",
|
|
(username, limit),
|
|
)
|
|
notes = [dict(r) for r in await rows.fetchall()]
|
|
for note in notes:
|
|
note["tags"] = await get_note_tags(note["id"])
|
|
return notes
|
|
|
|
|
|
async def list_recent(limit: int = 30) -> list[dict]:
|
|
async with aiosqlite.connect(settings.database_path) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
rows = await db.execute(
|
|
"SELECT n.id, n.title, n.updated_at, n.created_at, u.name AS username"
|
|
" FROM notes n JOIN users u ON u.id = n.user_id"
|
|
" WHERE n.deleted_at IS NULL"
|
|
" ORDER BY n.updated_at DESC LIMIT ?",
|
|
(limit,),
|
|
)
|
|
notes = [dict(r) for r in await rows.fetchall()]
|
|
for note in notes:
|
|
note["tags"] = await get_note_tags(note["id"])
|
|
return notes
|
|
|
|
|
|
async def get_note_tags(note_id: str) -> list[str]:
|
|
async with aiosqlite.connect(settings.database_path) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
rows = await db.execute(
|
|
"SELECT t.name FROM tags t JOIN note_tags nt ON nt.tag_id = t.id"
|
|
" WHERE nt.note_id = ? ORDER BY t.name",
|
|
(note_id,),
|
|
)
|
|
return [r["name"] for r in await rows.fetchall()]
|
|
|
|
|
|
async def tag_counts() -> list[dict]:
|
|
async with aiosqlite.connect(settings.database_path) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
rows = await db.execute(
|
|
"SELECT t.name, COUNT(nt.note_id) AS count"
|
|
" FROM tags t JOIN note_tags nt ON nt.tag_id = t.id"
|
|
" JOIN notes n ON n.id = nt.note_id AND n.deleted_at IS NULL"
|
|
" GROUP BY t.id ORDER BY count DESC, t.name LIMIT 50"
|
|
)
|
|
return [dict(r) for r in await rows.fetchall()]
|
|
|
|
|
|
async def search_notes(query: str, limit: int = 50) -> list[dict]:
|
|
async with aiosqlite.connect(settings.database_path) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
rows = await db.execute(
|
|
"SELECT f.note_id AS id, f.title, f.body, n.updated_at,"
|
|
" n.created_at, u.name AS username, bm25(notes_fts) AS rank"
|
|
" FROM notes_fts f"
|
|
" JOIN notes n ON n.id = f.note_id AND n.deleted_at IS NULL"
|
|
" JOIN users u ON u.id = n.user_id"
|
|
" WHERE notes_fts MATCH ? ORDER BY rank LIMIT ?",
|
|
(query, limit),
|
|
)
|
|
notes = [dict(r) for r in await rows.fetchall()]
|
|
for note in notes:
|
|
note["tags"] = await get_note_tags(note["id"])
|
|
note.pop("body", None)
|
|
return notes
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def _set_tags(db: aiosqlite.Connection, note_id: str, tags: list[str]) -> None:
|
|
await db.execute("DELETE FROM note_tags WHERE note_id = ?", (note_id,))
|
|
for tag in tags:
|
|
await db.execute("INSERT OR IGNORE INTO tags (name) VALUES (?)", (tag,))
|
|
row = await db.execute("SELECT id FROM tags WHERE name = ?", (tag,))
|
|
row = await row.fetchone()
|
|
await db.execute(
|
|
"INSERT OR IGNORE INTO note_tags (note_id, tag_id) VALUES (?, ?)",
|
|
(note_id, row["id"]),
|
|
)
|
|
|
|
|
|
async def _index_note(
|
|
db: aiosqlite.Connection,
|
|
note_id: str,
|
|
user_id: int,
|
|
title: str,
|
|
body: str,
|
|
) -> None:
|
|
await db.execute("DELETE FROM notes_fts WHERE note_id = ?", (note_id,))
|
|
await db.execute(
|
|
"INSERT INTO notes_fts (note_id, user_id, title, body) VALUES (?, ?, ?, ?)",
|
|
(note_id, user_id, title, body),
|
|
)
|
|
|
|
|
|
async def _record_history(
|
|
db: aiosqlite.Connection,
|
|
note_id: str,
|
|
user_id: int,
|
|
action: str,
|
|
ip: str | None,
|
|
ua: str | None,
|
|
) -> None:
|
|
await db.execute(
|
|
"INSERT INTO history (note_id, user_id, action, ip, ua) VALUES (?, ?, ?, ?, ?)",
|
|
(note_id, user_id, action, ip, ua),
|
|
)
|
|
|
|
|
|
async def get_history(note_id: str) -> list[dict]:
|
|
async with aiosqlite.connect(settings.database_path) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
rows = await db.execute(
|
|
"SELECT h.id, h.action, h.at, h.ip, h.ua, u.name AS username"
|
|
" FROM history h JOIN users u ON u.id = h.user_id"
|
|
" WHERE h.note_id = ? ORDER BY h.at DESC",
|
|
(note_id,),
|
|
)
|
|
return [dict(r) for r in await rows.fetchall()]
|