File size: 14,661 Bytes
17640d9 aafbcb7 17640d9 aafbcb7 17640d9 aafbcb7 17640d9 aafbcb7 17640d9 aafbcb7 17640d9 aafbcb7 17640d9 aafbcb7 17640d9 aafbcb7 17640d9 aafbcb7 df58f94 17640d9 aafbcb7 17640d9 aafbcb7 17640d9 aafbcb7 17640d9 aafbcb7 17640d9 aafbcb7 17640d9 aafbcb7 491673e aafbcb7 fe9cfaa aafbcb7 fe9cfaa aafbcb7 | 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 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 | import sqlite3
import json
from config import DB_PATH
def init_db():
with sqlite3.connect(DB_PATH) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS captures (
id INTEGER PRIMARY KEY AUTOINCREMENT,
type TEXT NOT NULL,
raw TEXT,
source_url TEXT,
file_path TEXT,
summary TEXT,
tags TEXT DEFAULT '[]',
intent TEXT,
created_at TEXT DEFAULT (datetime('now', 'localtime')),
last_surfaced_at TEXT,
reviewed INTEGER DEFAULT 0
)
""")
# migrate existing DB if intent column is missing
cols = [r[1] for r in conn.execute("PRAGMA table_info(captures)").fetchall()]
conn.execute("""
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
capture_id INTEGER,
event TEXT,
value TEXT,
created_at TEXT DEFAULT (datetime('now', 'localtime'))
)
""")
if "intent" not in cols:
conn.execute("ALTER TABLE captures ADD COLUMN intent TEXT")
if "last_surfaced_at" not in cols:
conn.execute("ALTER TABLE captures ADD COLUMN last_surfaced_at TEXT")
if "embedding" not in cols:
conn.execute("ALTER TABLE captures ADD COLUMN embedding TEXT")
if "related_ids" not in cols:
conn.execute("ALTER TABLE captures ADD COLUMN related_ids TEXT DEFAULT '[]'")
if "review_due_at" not in cols:
conn.execute("ALTER TABLE captures ADD COLUMN review_due_at TEXT")
if "review_interval" not in cols:
conn.execute("ALTER TABLE captures ADD COLUMN review_interval INTEGER DEFAULT 1")
if "review_count" not in cols:
conn.execute("ALTER TABLE captures ADD COLUMN review_count INTEGER DEFAULT 0")
if "recall_question" not in cols:
conn.execute("ALTER TABLE captures ADD COLUMN recall_question TEXT")
if "title" not in cols:
conn.execute("ALTER TABLE captures ADD COLUMN title TEXT")
if "your_take" not in cols:
conn.execute("ALTER TABLE captures ADD COLUMN your_take TEXT")
if "claims" not in cols:
conn.execute("ALTER TABLE captures ADD COLUMN claims TEXT DEFAULT '[]'")
if "source_content_path" not in cols:
conn.execute("ALTER TABLE captures ADD COLUMN source_content_path TEXT")
def save_capture(type, raw=None, source_url=None, file_path=None, your_take=None):
with sqlite3.connect(DB_PATH) as conn:
cur = conn.execute(
"INSERT INTO captures (type, raw, source_url, file_path, your_take) VALUES (?, ?, ?, ?, ?)",
(type, raw, source_url, file_path, your_take or None),
)
return cur.lastrowid
def update_capture(capture_id, summary, tags, intent=None, embedding=None, related_ids=None, recall_question=None, claims=None, source_content_path=None, title=None):
with sqlite3.connect(DB_PATH) as conn:
conn.execute(
"UPDATE captures SET summary=?, tags=?, intent=?, embedding=?, related_ids=?, recall_question=?, claims=?, source_content_path=?, title=? WHERE id=?",
(
summary,
json.dumps(tags),
intent,
json.dumps(embedding) if embedding else None,
json.dumps(related_ids or []),
recall_question,
json.dumps(claims or []),
source_content_path,
title,
capture_id,
),
)
def delete_capture(capture_id: int):
with sqlite3.connect(DB_PATH) as conn:
conn.execute("DELETE FROM captures WHERE id=?", (capture_id,))
def get_captures_by_ids(ids: list) -> list:
if not ids:
return []
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
placeholders = ",".join("?" * len(ids))
rows = conn.execute(
f"SELECT * FROM captures WHERE id IN ({placeholders})", ids
).fetchall()
result = []
for r in rows:
d = dict(r)
d["tags"] = json.loads(d["tags"] or "[]")
d["claims"] = json.loads(d.get("claims") or "[]")
d.pop("embedding", None)
d.pop("related_ids", None)
result.append(d)
return result
def get_all_embeddings():
"""Return [(id, embedding)] for all captures that have embeddings."""
with sqlite3.connect(DB_PATH) as conn:
rows = conn.execute(
"SELECT id, embedding FROM captures WHERE embedding IS NOT NULL"
).fetchall()
return [(r[0], json.loads(r[1])) for r in rows]
def get_captures_by_intent(intent: str, limit=50):
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
rows = conn.execute(
"SELECT * FROM captures WHERE intent=? ORDER BY created_at DESC LIMIT ?", (intent, limit)
).fetchall()
result = []
for r in rows:
d = dict(r)
d["tags"] = json.loads(d["tags"] or "[]")
d["claims"] = json.loads(d.get("claims") or "[]")
d.pop("embedding", None)
result.append(d)
return result
def get_capture_by_id(capture_id: int):
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
row = conn.execute("SELECT * FROM captures WHERE id=?", (capture_id,)).fetchone()
if not row:
return None
d = dict(row)
d["tags"] = json.loads(d["tags"] or "[]")
d["related_ids"] = json.loads(d.get("related_ids") or "[]")
d["claims"] = json.loads(d.get("claims") or "[]")
d.pop("embedding", None)
return d
def get_captures(limit=50):
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
rows = conn.execute(
"SELECT * FROM captures ORDER BY created_at DESC LIMIT ?", (limit,)
).fetchall()
result = []
for r in rows:
d = dict(r)
d["tags"] = json.loads(d["tags"] or "[]")
d["related_ids"] = json.loads(d.get("related_ids") or "[]")
d["claims"] = json.loads(d.get("claims") or "[]")
result.append(d)
return result
def get_surfaceable(include_ephemeral=False):
"""Return all captures eligible for surfacing."""
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
where = "intent IS NOT NULL AND summary IS NOT NULL AND reviewed = 0"
if not include_ephemeral:
where += " AND intent != 'ephemeral'"
rows = conn.execute(f"""
SELECT * FROM captures
WHERE {where}
ORDER BY created_at DESC
""").fetchall()
result = []
for r in rows:
d = dict(r)
d["tags"] = json.loads(d["tags"] or "[]")
d["claims"] = json.loads(d.get("claims") or "[]")
result.append(d)
return result
def mark_surfaced(capture_id):
with sqlite3.connect(DB_PATH) as conn:
conn.execute(
"UPDATE captures SET last_surfaced_at = datetime('now', 'localtime') WHERE id = ?",
(capture_id,),
)
def mark_done(capture_id):
with sqlite3.connect(DB_PATH) as conn:
conn.execute(
"UPDATE captures SET reviewed = 1 WHERE id = ?",
(capture_id,),
)
def patch_capture(capture_id: int, intent: str = None, tags: list = None):
with sqlite3.connect(DB_PATH) as conn:
if intent is not None:
conn.execute("UPDATE captures SET intent=? WHERE id=?", (intent, capture_id))
if tags is not None:
conn.execute("UPDATE captures SET tags=? WHERE id=?", (json.dumps(tags), capture_id))
def get_review_queue(limit=10) -> list:
"""Return learn captures with review_due_at <= now, ordered by due date."""
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
rows = conn.execute("""
SELECT * FROM captures
WHERE intent = 'learn'
AND summary IS NOT NULL
AND recall_question IS NOT NULL
AND (
review_due_at IS NULL
OR review_due_at <= datetime('now', 'localtime')
)
ORDER BY COALESCE(review_due_at, created_at) ASC
LIMIT ?
""", (limit,)).fetchall()
result = []
for r in rows:
d = dict(r)
d["tags"] = json.loads(d["tags"] or "[]")
d["related_ids"] = json.loads(d.get("related_ids") or "[]")
d["claims"] = json.loads(d.get("claims") or "[]")
d.pop("embedding", None)
result.append(d)
return result
def record_review(capture_id: int, rating: str):
"""SM-2 simplified: rating is 'got_it' or 'again'."""
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
row = conn.execute(
"SELECT review_interval, review_count FROM captures WHERE id=?", (capture_id,)
).fetchone()
if not row:
return
interval = row["review_interval"] or 1
count = row["review_count"] or 0
if rating == "got_it":
new_interval = max(1, round(interval * 2.5))
new_count = count + 1
else:
new_interval = 1
new_count = count
conn.execute("""
UPDATE captures
SET review_interval = ?,
review_count = ?,
review_due_at = datetime('now', 'localtime', ? || ' days'),
last_surfaced_at = datetime('now', 'localtime')
WHERE id = ?
""", (new_interval, new_count, str(new_interval), capture_id))
# persist review event for streak tracking
conn.execute(
"INSERT INTO events (capture_id, event, value, created_at) VALUES (?, 'review', ?, datetime('now','localtime'))",
(capture_id, rating)
)
def get_review_history(days: int = 7) -> list:
"""Return list of dates (YYYY-MM-DD) in the past `days` that had at least one review."""
with sqlite3.connect(DB_PATH) as conn:
rows = conn.execute("""
SELECT DISTINCT date(created_at) as day
FROM events
WHERE event = 'review'
AND created_at >= datetime('now', 'localtime', ? || ' days')
ORDER BY day DESC
""", (f"-{days}",)).fetchall()
return [r[0] for r in rows]
def get_review_streak() -> int:
"""Count consecutive days ending today that had at least one review."""
with sqlite3.connect(DB_PATH) as conn:
rows = conn.execute("""
SELECT DISTINCT date(created_at) as day
FROM events
WHERE event = 'review'
ORDER BY day DESC
LIMIT 60
""").fetchall()
dates = [r[0] for r in rows]
if not dates:
return 0
from datetime import date, timedelta
today = date.today()
streak = 0
for i in range(len(dates)):
expected = (today - timedelta(days=i)).isoformat()
if i < len(dates) and dates[i] == expected:
streak += 1
else:
break
return streak
def search_captures(q: str, limit=20) -> list:
pattern = f"%{q}%"
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
rows = conn.execute("""
SELECT * FROM captures
WHERE summary IS NOT NULL AND (
summary LIKE ? OR tags LIKE ? OR raw LIKE ?
)
ORDER BY created_at DESC
LIMIT ?
""", (pattern, pattern, pattern, limit)).fetchall()
result = []
for r in rows:
d = dict(r)
d["tags"] = json.loads(d["tags"] or "[]")
d["related_ids"] = json.loads(d.get("related_ids") or "[]")
d["claims"] = json.loads(d.get("claims") or "[]")
d.pop("embedding", None)
result.append(d)
return result
def get_brief(limit=50, date: str = None) -> list:
"""Captures for digest view. date='YYYY-MM-DD' fetches ALL captures for that day, else recent unreviewed."""
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
if date:
rows = conn.execute("""
SELECT * FROM captures
WHERE summary IS NOT NULL
AND summary NOT LIKE '⚠%'
AND date(created_at) = ?
ORDER BY created_at DESC
""", (date,)).fetchall()
else:
rows = conn.execute("""
SELECT * FROM captures
WHERE summary IS NOT NULL AND summary NOT LIKE '⚠%' AND reviewed = 0
ORDER BY created_at DESC
LIMIT ?
""", (limit,)).fetchall()
result = []
for r in rows:
d = dict(r)
d["tags"] = json.loads(d["tags"] or "[]")
d["related_ids"] = json.loads(d.get("related_ids") or "[]")
d["claims"] = json.loads(d.get("claims") or "[]")
d.pop("embedding", None)
result.append(d)
return result
def get_brief_week() -> list:
"""All captures from the last 7 days, newest first."""
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
rows = conn.execute("""
SELECT * FROM captures
WHERE summary IS NOT NULL
AND date(created_at) >= date('now', '-6 days', 'localtime')
ORDER BY created_at DESC
""").fetchall()
result = []
for r in rows:
d = dict(r)
d["tags"] = json.loads(d["tags"] or "[]")
d["related_ids"] = json.loads(d.get("related_ids") or "[]")
d["claims"] = json.loads(d.get("claims") or "[]")
d.pop("embedding", None)
result.append(d)
return result
def get_brief_dates(limit=30) -> list:
"""Return distinct dates that have captures, newest first."""
with sqlite3.connect(DB_PATH) as conn:
rows = conn.execute("""
SELECT date(created_at) as day, COUNT(*) as count
FROM captures
WHERE summary IS NOT NULL
GROUP BY day
ORDER BY day DESC
LIMIT ?
""", (limit,)).fetchall()
return [{"date": r[0], "count": r[1]} for r in rows]
|