File size: 5,301 Bytes
4fd1234 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | import json
from datetime import date
from db import get_cursor
import curator
def save_page(user_id: str, raw_text: str, annotations: dict) -> tuple[str, str]:
"""Curates the page (title/summary/type/resources), inserts into DB. Returns (page_id, title)."""
import nlp as _nlp
curated = curator.curate_page(raw_text)
category = _nlp.detect_category(raw_text[:300])
title = curated["title"]
metadata = {
"category": category,
"summary": curated["summary"],
"page_type": curated["page_type"],
"links": curated["links"],
"books": curated["books"],
}
with get_cursor() as cur:
cur.execute(
"""INSERT INTO pages (user_id, title, date, raw_text, annotations, metadata)
VALUES (%s, %s, %s, %s, %s, %s) RETURNING id::text""",
(user_id, title, date.today(), raw_text,
json.dumps(annotations), json.dumps(metadata)),
)
page_id = cur.fetchone()["id"]
return page_id, title
def list_pages(user_id: str) -> list[dict]:
"""Sidebar list — returns [{id, title, date, category, page_type, preview}] newest-first, max 50.
Fetches the first 300 chars of raw_text to detect category and build a hover
preview without pulling the full lesson text over the wire.
"""
import nlp as _nlp
with get_cursor() as cur:
cur.execute(
"SELECT id::text, title, date::text, "
"LEFT(raw_text, 300) AS snippet, "
"COALESCE(metadata->>'category', '') AS stored_category, "
"COALESCE(metadata->>'page_type', 'lesson') AS page_type "
"FROM pages WHERE user_id = %s ORDER BY created_at DESC LIMIT 50",
(user_id,),
)
rows = [dict(r) for r in cur.fetchall()]
for row in rows:
snippet = row.pop("snippet", "") or ""
stored_cat = row.pop("stored_category", "") or ""
row["category"] = stored_cat if stored_cat else _nlp.detect_category(snippet)
preview = snippet[:100]
row["preview"] = preview + ("…" if len(snippet) > 100 else "")
return rows
def list_resources(user_id: str) -> list[dict]:
"""Return resource-type pages: [{id, title, links, books}], newest-first."""
with get_cursor() as cur:
cur.execute(
"SELECT id::text, title, "
"COALESCE(metadata->'links', '[]'::jsonb) AS links, "
"COALESCE(metadata->'books', '[]'::jsonb) AS books "
"FROM pages WHERE user_id = %s AND metadata->>'page_type' = 'resource' "
"ORDER BY created_at DESC",
(user_id,),
)
return [dict(r) for r in cur.fetchall()]
def update_resource_links(page_id: str, user_id: str, links: list[dict]) -> bool:
"""Overwrite metadata->'links' on a resource page. Returns True if a row was updated."""
with get_cursor() as cur:
cur.execute(
"UPDATE pages SET metadata = jsonb_set(metadata, '{links}', %s::jsonb) "
"WHERE id = %s AND user_id = %s AND metadata->>'page_type' = 'resource' "
"RETURNING id",
(json.dumps(links), page_id, user_id),
)
return cur.fetchone() is not None
def update_title(page_id: str, user_id: str, title: str) -> str:
"""Persist a user-edited title (overrides the auto-generated one). Returns the saved title."""
title = (title or "").strip()[:80] or "Untitled Lesson"
with get_cursor() as cur:
cur.execute(
"UPDATE pages SET title = %s WHERE id = %s AND user_id = %s",
(title, page_id, user_id),
)
return title
def get_page(page_id: str, user_id: str) -> dict | None:
with get_cursor() as cur:
cur.execute(
"SELECT id::text, title, raw_text, annotations "
"FROM pages WHERE id = %s AND user_id = %s",
(page_id, user_id),
)
row = cur.fetchone()
return dict(row) if row else None
def update_annotations(page_id: str, annotations: dict) -> None:
"""Persist LLM-enriched annotations back to the page row."""
with get_cursor() as cur:
cur.execute(
"UPDATE pages SET annotations = %s WHERE id = %s",
(json.dumps(annotations), page_id),
)
def update_page(page_id: str, user_id: str, raw_text: str, annotations: dict) -> str:
"""Update page content in-place, keep existing title. Returns the title."""
with get_cursor() as cur:
cur.execute(
"SELECT title FROM pages WHERE id = %s AND user_id = %s",
(page_id, user_id),
)
row = cur.fetchone()
title = row["title"] if row else raw_text.split("\n")[0][:60]
with get_cursor() as cur:
cur.execute(
"UPDATE pages SET raw_text = %s, annotations = %s WHERE id = %s AND user_id = %s",
(raw_text, json.dumps(annotations), page_id, user_id),
)
return title
def delete_page(page_id: str, user_id: str) -> bool:
"""Delete page row (exercises cascade). Returns True if a row was deleted."""
with get_cursor() as cur:
cur.execute(
"DELETE FROM pages WHERE id = %s AND user_id = %s RETURNING id",
(page_id, user_id),
)
return cur.fetchone() is not None
|