Spaces:
Sleeping
Sleeping
File size: 6,546 Bytes
cc13fc9 bfb5cd4 cc13fc9 | 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 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | 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_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 add_manual_link(user_id: str, title: str, url: str) -> None:
"""Append a manually-added link to the user's pinned 'My Resources' page.
Upserts a single resource page titled 'My Resources' — the page_type is
'resource' so it shows in the Resources tab, never in the lesson sidebar.
"""
title = (title or "").strip()[:120] or url
url = (url or "").strip()
if not url:
return
new_link = {"label": title, "url": url}
with get_cursor() as cur:
# Find existing pinned resource page for this user
cur.execute(
"SELECT id::text, metadata FROM pages "
"WHERE user_id = %s AND metadata->>'page_type' = 'resource' "
"AND title = 'My Resources' ORDER BY created_at ASC LIMIT 1",
(user_id,),
)
row = cur.fetchone()
if row:
meta = dict(row["metadata"]) if row["metadata"] else {}
links = list(meta.get("links") or [])
links.append(new_link)
meta["links"] = links
cur.execute(
"UPDATE pages SET metadata = %s WHERE id = %s",
(json.dumps(meta), row["id"]),
)
else:
meta = {
"category": "Resources",
"summary": "Manually saved online resources",
"page_type": "resource",
"links": [new_link],
"books": [],
}
cur.execute(
"""INSERT INTO pages (user_id, title, date, raw_text, annotations, metadata)
VALUES (%s, %s, %s, %s, %s, %s)""",
(user_id, "My Resources", date.today(),
"Manually saved resources", json.dumps({}), json.dumps(meta)),
)
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
|