1001 lines
37 KiB
Python
1001 lines
37 KiB
Python
import json
|
||
|
||
import aiosqlite
|
||
from config import DB_PATH
|
||
|
||
CREATE_GUIDES = """
|
||
CREATE TABLE IF NOT EXISTS guides (
|
||
id TEXT PRIMARY KEY,
|
||
topic TEXT NOT NULL,
|
||
format TEXT NOT NULL,
|
||
instructions TEXT NOT NULL DEFAULT '',
|
||
status TEXT NOT NULL DEFAULT 'queued',
|
||
progress TEXT,
|
||
step INTEGER,
|
||
error_msg TEXT,
|
||
created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL
|
||
)
|
||
"""
|
||
|
||
CREATE_PROGRESS = """
|
||
CREATE TABLE IF NOT EXISTS guide_progress (
|
||
guide_id TEXT NOT NULL,
|
||
chapter TEXT NOT NULL,
|
||
created_at TEXT NOT NULL,
|
||
PRIMARY KEY (guide_id, chapter)
|
||
)
|
||
"""
|
||
|
||
CREATE_TOPICS = """
|
||
CREATE TABLE IF NOT EXISTS topics (
|
||
name TEXT PRIMARY KEY,
|
||
created_at TEXT NOT NULL
|
||
)
|
||
"""
|
||
|
||
CREATE_ELEMENTS = """
|
||
CREATE TABLE IF NOT EXISTS elements (
|
||
id TEXT PRIMARY KEY,
|
||
topic TEXT NOT NULL,
|
||
title TEXT NOT NULL,
|
||
description TEXT NOT NULL DEFAULT '',
|
||
examples TEXT NOT NULL DEFAULT '[]',
|
||
hints TEXT NOT NULL DEFAULT '[]',
|
||
created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL
|
||
)
|
||
"""
|
||
|
||
CREATE_BLOCK_TEXTE = """
|
||
CREATE TABLE IF NOT EXISTS block_texte (
|
||
topic TEXT NOT NULL,
|
||
block TEXT NOT NULL,
|
||
kind TEXT NOT NULL,
|
||
md TEXT NOT NULL,
|
||
created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, block, kind)
|
||
)
|
||
"""
|
||
|
||
CREATE_BLOCK_PROGRESS = """
|
||
CREATE TABLE IF NOT EXISTS block_progress (
|
||
topic TEXT NOT NULL,
|
||
block TEXT NOT NULL,
|
||
good_answers INTEGER NOT NULL DEFAULT 0,
|
||
streak INTEGER NOT NULL DEFAULT 0,
|
||
completed TEXT,
|
||
understood TEXT,
|
||
mastered TEXT,
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, block)
|
||
)
|
||
"""
|
||
|
||
# --- Blocks pipeline content (replaces file sidecars) ---
|
||
|
||
# Inventory: one block per (topic, title_norm). mentions = number of agents/rounds
|
||
# that named it (≥2 = consensus). status: candidate/consensus/rest/discarded.
|
||
CREATE_BLOCKS = """
|
||
CREATE TABLE IF NOT EXISTS blocks (
|
||
topic TEXT NOT NULL,
|
||
title_norm TEXT NOT NULL,
|
||
title TEXT NOT NULL,
|
||
description TEXT NOT NULL DEFAULT '',
|
||
mentions INTEGER NOT NULL DEFAULT 1,
|
||
status TEXT NOT NULL DEFAULT 'candidate',
|
||
sources TEXT NOT NULL DEFAULT '[]',
|
||
reader TEXT NOT NULL DEFAULT '[]',
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, title_norm)
|
||
)
|
||
"""
|
||
|
||
# Subblocks per block. level (einfach/mittel/schwer) + relevance (relevant/rand)
|
||
# are set later. mentions analogous to the inventory.
|
||
CREATE_SUBBLOCKS = """
|
||
CREATE TABLE IF NOT EXISTS subblocks (
|
||
topic TEXT NOT NULL,
|
||
block_norm TEXT NOT NULL,
|
||
sub_norm TEXT NOT NULL,
|
||
block TEXT NOT NULL,
|
||
sub_title TEXT NOT NULL,
|
||
mentions INTEGER NOT NULL DEFAULT 1,
|
||
level TEXT,
|
||
relevance TEXT,
|
||
facts TEXT NOT NULL DEFAULT '',
|
||
status TEXT NOT NULL DEFAULT 'candidate',
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, block_norm, sub_norm)
|
||
)
|
||
"""
|
||
|
||
# One question pattern per (block, subblock). Difficulty comes only at exam time
|
||
# from the learner tier, not from the pattern — hence no more type cross-product.
|
||
CREATE_QUESTION_PATTERN = """
|
||
CREATE TABLE IF NOT EXISTS question_pattern (
|
||
topic TEXT NOT NULL,
|
||
block_norm TEXT NOT NULL,
|
||
sub_norm TEXT NOT NULL,
|
||
block TEXT NOT NULL,
|
||
sub_title TEXT NOT NULL,
|
||
question TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, block_norm, sub_norm)
|
||
)
|
||
"""
|
||
|
||
# Crawl pages per topic: content (1=content/0=noise, from the triage) + read_done (from the research).
|
||
CREATE_RESEARCH_COVERAGE = """
|
||
CREATE TABLE IF NOT EXISTS research_coverage (
|
||
topic TEXT NOT NULL,
|
||
source TEXT NOT NULL,
|
||
read_done INTEGER NOT NULL DEFAULT 0,
|
||
content INTEGER,
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, source)
|
||
)
|
||
"""
|
||
|
||
# Step status of the pipeline (replaces file-existence resume + reset globs).
|
||
CREATE_PIPELINE_STATE = """
|
||
CREATE TABLE IF NOT EXISTS pipeline_state (
|
||
topic TEXT NOT NULL,
|
||
step TEXT NOT NULL,
|
||
status TEXT NOT NULL DEFAULT 'open',
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, step)
|
||
)
|
||
"""
|
||
|
||
# Finished guide content per (topic, format) as a JSON blob (replaces the guide JSON file).
|
||
# Shared across all guide runs of the same topic+format (as the content file was before).
|
||
CREATE_GUIDE_CONTENT = """
|
||
CREATE TABLE IF NOT EXISTS guide_content (
|
||
topic TEXT NOT NULL,
|
||
format TEXT NOT NULL,
|
||
json TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, format)
|
||
)
|
||
"""
|
||
|
||
# Source choice per topic (replaces source.json).
|
||
CREATE_SOURCE = """
|
||
CREATE TABLE IF NOT EXISTS source (
|
||
topic TEXT PRIMARY KEY,
|
||
type TEXT NOT NULL,
|
||
location TEXT NOT NULL DEFAULT '',
|
||
spec TEXT NOT NULL DEFAULT '',
|
||
updated_at TEXT NOT NULL
|
||
)
|
||
"""
|
||
|
||
# Outline (chapter → block numbers) per topic as JSON. Produced in the blocks phase,
|
||
# the guide only reads it + filters per format. Format-agnostic (all blocks).
|
||
CREATE_GUIDE_OUTLINE = """
|
||
CREATE TABLE IF NOT EXISTS guide_outline (
|
||
topic TEXT PRIMARY KEY,
|
||
json TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL
|
||
)
|
||
"""
|
||
|
||
# Generic learning artifact layer: one artifact per (block, subblock, type) (JSON in `data`).
|
||
# type: flashcard | example. sub_norm='' = block level (for future block-wide artifacts).
|
||
# Produced from the facts in the blocks phase, presented by the frontend (artifact/display split).
|
||
CREATE_SUB_ARTEFAKTE = """
|
||
CREATE TABLE IF NOT EXISTS sub_artefakte (
|
||
topic TEXT NOT NULL,
|
||
block_norm TEXT NOT NULL,
|
||
sub_norm TEXT NOT NULL DEFAULT '',
|
||
type TEXT NOT NULL,
|
||
block TEXT NOT NULL DEFAULT '',
|
||
sub_title TEXT NOT NULL DEFAULT '',
|
||
data TEXT NOT NULL DEFAULT '{}',
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, block_norm, sub_norm, type)
|
||
)
|
||
"""
|
||
|
||
_db: aiosqlite.Connection | None = None
|
||
|
||
|
||
async def get_db() -> aiosqlite.Connection:
|
||
global _db
|
||
if _db is None:
|
||
_db = await aiosqlite.connect(DB_PATH)
|
||
_db.row_factory = None
|
||
return _db
|
||
|
||
|
||
async def init_db():
|
||
db = await get_db()
|
||
# WAL survives crashes much better; busy_timeout absorbs short locks.
|
||
await db.execute("PRAGMA journal_mode=WAL")
|
||
await db.execute("PRAGMA busy_timeout=5000")
|
||
await db.execute(CREATE_GUIDES)
|
||
await db.execute(CREATE_PROGRESS)
|
||
await db.execute(CREATE_TOPICS)
|
||
await db.execute(CREATE_ELEMENTS)
|
||
await db.execute(CREATE_BLOCK_TEXTE)
|
||
await db.execute(CREATE_BLOCK_PROGRESS)
|
||
await db.execute(CREATE_BLOCKS)
|
||
await db.execute(CREATE_SUBBLOCKS)
|
||
await db.execute(CREATE_QUESTION_PATTERN)
|
||
await db.execute(CREATE_RESEARCH_COVERAGE)
|
||
await db.execute(CREATE_PIPELINE_STATE)
|
||
await db.execute(CREATE_GUIDE_CONTENT)
|
||
await db.execute(CREATE_SOURCE)
|
||
await db.execute(CREATE_GUIDE_OUTLINE)
|
||
await db.execute(CREATE_SUB_ARTEFAKTE)
|
||
try: # migration for existing DBs without the step column
|
||
await db.execute("ALTER TABLE guides ADD COLUMN step INTEGER")
|
||
except aiosqlite.OperationalError:
|
||
pass
|
||
try: # migration: research_coverage.content (content/noise from the triage)
|
||
await db.execute("ALTER TABLE research_coverage ADD COLUMN content INTEGER")
|
||
except aiosqlite.OperationalError:
|
||
pass
|
||
try: # migration for existing DBs without the understood column (mastery level)
|
||
await db.execute("ALTER TABLE block_progress ADD COLUMN understood TEXT")
|
||
except aiosqlite.OperationalError:
|
||
pass
|
||
try: # migration for existing DBs without the mastered column (master path 25)
|
||
await db.execute("ALTER TABLE block_progress ADD COLUMN mastered TEXT")
|
||
except aiosqlite.OperationalError:
|
||
pass
|
||
try: # migration for existing DBs without the streak column (persistent streak-bonus run)
|
||
await db.execute("ALTER TABLE block_progress ADD COLUMN streak INTEGER NOT NULL DEFAULT 0")
|
||
except aiosqlite.OperationalError:
|
||
pass
|
||
# Open-question anchor: base/streak BEFORE the currently open question — makes the rating
|
||
# idempotent server-side (re-rating) and drift-free across questions.
|
||
for _col, _type in (("offene_question", "TEXT"), ("offene_basis", "INTEGER"), ("offene_streak", "INTEGER")):
|
||
try:
|
||
await db.execute(f"ALTER TABLE block_progress ADD COLUMN {_col} {_type}")
|
||
except aiosqlite.OperationalError:
|
||
pass
|
||
# Migration: question_pattern without a type column (1 pattern per sub instead of a sub×type cross-product).
|
||
# PK change → rebuild the table once. Existing patterns are lost (intentional, no mapping).
|
||
cursor = await db.execute("PRAGMA table_info(question_pattern)")
|
||
if any(_r[1] == "type" for _r in await cursor.fetchall()):
|
||
await db.execute("DROP TABLE question_pattern")
|
||
await db.execute(CREATE_QUESTION_PATTERN)
|
||
try: # migration: subblocks.facts (source facts per sub, JSON blob) — extract-once grounding.
|
||
await db.execute("ALTER TABLE subblocks ADD COLUMN facts TEXT NOT NULL DEFAULT ''") # DEFAULT '' needed for NOT NULL on ADD COLUMN
|
||
except aiosqlite.OperationalError:
|
||
pass
|
||
try: # migration: blocks.reader (reader set per candidate, JSON) — exact consensus count (≥2 readers)
|
||
await db.execute("ALTER TABLE blocks ADD COLUMN reader TEXT NOT NULL DEFAULT '[]'")
|
||
except aiosqlite.OperationalError:
|
||
pass
|
||
# Migration: old vertiefungen table → block_texte (existing = long form, kind 'deepdive')
|
||
cursor = await db.execute("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'vertiefungen'")
|
||
if await cursor.fetchone():
|
||
await db.execute(
|
||
"INSERT OR IGNORE INTO block_texte (topic, block, kind, md, created_at, updated_at) "
|
||
"SELECT topic, block, 'deepdive', md, created_at, updated_at FROM vertiefungen"
|
||
)
|
||
await db.execute("DROP TABLE vertiefungen")
|
||
await db.execute(
|
||
"UPDATE guides SET status = 'error', progress = NULL, error_msg = 'Server restart' "
|
||
"WHERE status IN ('queued', 'generating')"
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def close_db():
|
||
global _db
|
||
if _db is not None:
|
||
await _db.close()
|
||
_db = None
|
||
|
||
|
||
def _row_to_dict(row, cursor):
|
||
columns = [d[0] for d in cursor.description]
|
||
return dict(zip(columns, row))
|
||
|
||
|
||
async def create_guide(guide: dict) -> dict:
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO guides (id, topic, format, instructions, status, progress, created_at, updated_at)
|
||
VALUES (:id, :topic, :format, :instructions, :status, :progress, :created_at, :updated_at)""",
|
||
guide,
|
||
)
|
||
await db.commit()
|
||
return guide
|
||
|
||
|
||
async def get_guide(guide_id: str) -> dict | None:
|
||
db = await get_db()
|
||
cursor = await db.execute("SELECT * FROM guides WHERE id = ?", (guide_id,))
|
||
row = await cursor.fetchone()
|
||
if row is None:
|
||
return None
|
||
return _row_to_dict(row, cursor)
|
||
|
||
|
||
async def list_guides() -> list[dict]:
|
||
db = await get_db()
|
||
cursor = await db.execute("SELECT * FROM guides ORDER BY created_at DESC")
|
||
rows = await cursor.fetchall()
|
||
return [_row_to_dict(row, cursor) for row in rows]
|
||
|
||
|
||
async def _update(table: str, fields: dict, where: dict) -> None:
|
||
"""UPDATE <table> SET <fields> WHERE <where> (+ commit). WHERE params are aliased (`w_<k>`)
|
||
so a field and a WHERE key of the same name don't collide — needed e.g. for a `title_norm`
|
||
rename (SET new norm WHERE old norm)."""
|
||
sets = ", ".join(f"{k} = :{k}" for k in fields)
|
||
cond = " AND ".join(f"{k} = :w_{k}" for k in where)
|
||
db = await get_db()
|
||
await db.execute(f"UPDATE {table} SET {sets} WHERE {cond}", {**fields, **{f"w_{k}": v for k, v in where.items()}})
|
||
await db.commit()
|
||
|
||
|
||
async def update_guide(guide_id: str, **fields) -> None:
|
||
await _update("guides", fields, {"id": guide_id})
|
||
|
||
|
||
async def delete_guide(guide_id: str) -> bool:
|
||
db = await get_db()
|
||
cursor = await db.execute("DELETE FROM guides WHERE id = ?", (guide_id,))
|
||
await db.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
|
||
# --- Topics ---
|
||
|
||
async def create_topic(name: str) -> None:
|
||
from datetime import datetime, timezone
|
||
db = await get_db()
|
||
await db.execute(
|
||
"INSERT OR IGNORE INTO topics (name, created_at) VALUES (?, ?)",
|
||
(name, datetime.now(timezone.utc).isoformat()),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def list_topics() -> list[str]:
|
||
db = await get_db()
|
||
cursor = await db.execute("SELECT name FROM topics ORDER BY created_at DESC")
|
||
rows = await cursor.fetchall()
|
||
return [row[0] for row in rows]
|
||
|
||
|
||
async def delete_topic(name: str) -> None:
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM topics WHERE name = ?", (name,))
|
||
await db.commit()
|
||
|
||
|
||
# --- Elements ---
|
||
|
||
def _element_row(row, cursor) -> dict:
|
||
el = _row_to_dict(row, cursor)
|
||
el["examples"] = json.loads(el["examples"] or "[]")
|
||
el["hints"] = json.loads(el["hints"] or "[]")
|
||
return el
|
||
|
||
|
||
async def create_element(element: dict) -> dict:
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO elements (id, topic, title, description, examples, hints, created_at, updated_at)
|
||
VALUES (:id, :topic, :title, :description, :examples, :hints, :created_at, :updated_at)""",
|
||
{**element, "examples": json.dumps(element["examples"], ensure_ascii=False),
|
||
"hints": json.dumps(element["hints"], ensure_ascii=False)},
|
||
)
|
||
await db.commit()
|
||
return element
|
||
|
||
|
||
async def list_elements(topic: str) -> list[dict]:
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
"SELECT * FROM elements WHERE topic = ? ORDER BY updated_at DESC", (topic,)
|
||
)
|
||
rows = await cursor.fetchall()
|
||
return [_element_row(row, cursor) for row in rows]
|
||
|
||
|
||
async def get_element(element_id: str) -> dict | None:
|
||
db = await get_db()
|
||
cursor = await db.execute("SELECT * FROM elements WHERE id = ?", (element_id,))
|
||
row = await cursor.fetchone()
|
||
if row is None:
|
||
return None
|
||
return _element_row(row, cursor)
|
||
|
||
|
||
async def update_element(element_id: str, **fields) -> None:
|
||
for key in ("examples", "hints"):
|
||
if key in fields:
|
||
fields[key] = json.dumps(fields[key], ensure_ascii=False)
|
||
await _update("elements", fields, {"id": element_id})
|
||
|
||
|
||
async def delete_element(element_id: str) -> bool:
|
||
db = await get_db()
|
||
cursor = await db.execute("DELETE FROM elements WHERE id = ?", (element_id,))
|
||
await db.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
|
||
# --- Chapter progress ---
|
||
|
||
async def list_progress_all() -> dict[str, set[str]]:
|
||
"""Complete chapter progress in one query: guide_id → chapter title."""
|
||
db = await get_db()
|
||
cursor = await db.execute("SELECT guide_id, chapter FROM guide_progress")
|
||
rows = await cursor.fetchall()
|
||
out: dict[str, set[str]] = {}
|
||
for guide_id, chapter in rows:
|
||
out.setdefault(guide_id, set()).add(chapter)
|
||
return out
|
||
|
||
|
||
async def list_progress(guide_id: str) -> list[str]:
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
"SELECT chapter FROM guide_progress WHERE guide_id = ?", (guide_id,)
|
||
)
|
||
rows = await cursor.fetchall()
|
||
return [row[0] for row in rows]
|
||
|
||
|
||
async def set_progress(guide_id: str, chapter: str, done: bool) -> None:
|
||
from datetime import datetime, timezone
|
||
db = await get_db()
|
||
if done:
|
||
await db.execute(
|
||
"INSERT OR IGNORE INTO guide_progress (guide_id, chapter, created_at) VALUES (?, ?, ?)",
|
||
(guide_id, chapter, datetime.now(timezone.utc).isoformat()),
|
||
)
|
||
else:
|
||
await db.execute(
|
||
"DELETE FROM guide_progress WHERE guide_id = ? AND chapter = ?", (guide_id, chapter)
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def delete_progress(guide_id: str) -> None:
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM guide_progress WHERE guide_id = ?", (guide_id,))
|
||
await db.commit()
|
||
|
||
|
||
# --- Block learning: deep-dives + exam progress ---
|
||
|
||
def _now() -> str:
|
||
from datetime import datetime, timezone
|
||
return datetime.now(timezone.utc).isoformat()
|
||
|
||
|
||
async def list_block_progress(topic: str) -> list[dict]:
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
"SELECT block, good_answers, streak, completed, understood, mastered FROM block_progress WHERE topic = ?", (topic,)
|
||
)
|
||
rows = await cursor.fetchall()
|
||
return [{"block": b, "good_answers": n, "streak": s, "completed": a, "understood": v, "mastered": m} for b, n, s, a, v, m in rows]
|
||
|
||
|
||
async def get_block_progress(topic: str, block: str) -> dict:
|
||
"""One block row incl. open-question anchor. Defaults if none exists yet."""
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
"SELECT good_answers, streak, completed, understood, mastered, "
|
||
"offene_question, offene_basis, offene_streak FROM block_progress "
|
||
"WHERE topic = ? AND block = ?",
|
||
(topic, block),
|
||
)
|
||
row = await cursor.fetchone()
|
||
if row is None:
|
||
return {"good_answers": 0, "streak": 0, "completed": None, "understood": None,
|
||
"mastered": None, "offene_question": None, "offene_basis": None, "offene_streak": None}
|
||
return {"good_answers": row[0], "streak": row[1], "completed": row[2], "understood": row[3],
|
||
"mastered": row[4], "offene_question": row[5], "offene_basis": row[6], "offene_streak": row[7]}
|
||
|
||
|
||
async def set_open_question(topic: str, block: str, question: str, basis: int, streak: int) -> None:
|
||
"""Freeze base + streak BEFORE the now-open question (anchor for idempotent re-rating)."""
|
||
db = await get_db()
|
||
now = _now()
|
||
await db.execute(
|
||
"""INSERT INTO block_progress (topic, block, offene_question, offene_basis, offene_streak, updated_at)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT(topic, block) DO UPDATE SET
|
||
offene_question = excluded.offene_question, offene_basis = excluded.offene_basis,
|
||
offene_streak = excluded.offene_streak, updated_at = excluded.updated_at""",
|
||
(topic, block, question, basis, streak, now),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def set_block_score_and_streak(topic: str, block: str, score: int, streak: int) -> tuple[int, int]:
|
||
"""Set score + streak atomically (clamped by the caller). Returns (score, streak)."""
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO block_progress (topic, block, good_answers, streak, updated_at)
|
||
VALUES (?, ?, ?, ?, ?)
|
||
ON CONFLICT(topic, block) DO UPDATE SET
|
||
good_answers = excluded.good_answers, streak = excluded.streak,
|
||
updated_at = excluded.updated_at""",
|
||
(topic, block, score, streak, _now()),
|
||
)
|
||
await db.commit()
|
||
return score, streak
|
||
|
||
|
||
async def delete_block_progress(topic: str, block: str) -> None:
|
||
"""Reset the progress of ONE block: delete the row (score/streak/flags/open question).
|
||
If the row is missing, get_block_progress returns defaults (0) — i.e. a full reset."""
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM block_progress WHERE topic = ? AND block = ?", (topic, block))
|
||
await db.commit()
|
||
|
||
|
||
async def set_block_completed(topic: str, block: str) -> bool:
|
||
"""Marks completed; True only the first time (drives the element task)."""
|
||
db = await get_db()
|
||
now = _now()
|
||
await db.execute(
|
||
"INSERT OR IGNORE INTO block_progress (topic, block, good_answers, updated_at) VALUES (?, ?, 0, ?)",
|
||
(topic, block, now),
|
||
)
|
||
cursor = await db.execute(
|
||
"UPDATE block_progress SET completed = ?, updated_at = ? "
|
||
"WHERE topic = ? AND block = ? AND completed IS NULL",
|
||
(now, now, topic, block),
|
||
)
|
||
await db.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
|
||
# Sub-level from the two orthogonal columns: peripheral → 4 (V), otherwise level (learning-path position):
|
||
# beginner/NULL → 1, advanced → 2, expert → 3. Old values (einfach/mittel/schwer) are
|
||
# mapped in backward-compatibly. Drives the guide view A/F/E/V + cap (unlocked subs × 25).
|
||
_LEVEL_CASE = """CASE
|
||
WHEN relevance = 'peripheral' THEN 4
|
||
WHEN level IN ('advanced', 'medium') THEN 2
|
||
WHEN level IN ('expert', 'hard') THEN 3
|
||
ELSE 1
|
||
END"""
|
||
|
||
|
||
def _empty_levels() -> dict[int, int]:
|
||
return {1: 0, 2: 0, 3: 0, 4: 0}
|
||
|
||
|
||
async def subs_per_level(topic: str, block: str) -> dict[int, int]:
|
||
"""Consensus subblocks of a block per level 1–4. Pass in the raw block title."""
|
||
from textkit import _norm_title
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
f"SELECT {_LEVEL_CASE} AS level, COUNT(*) FROM subblocks "
|
||
"WHERE topic = ? AND block_norm = ? AND status = 'consensus' GROUP BY level",
|
||
(topic, _norm_title(block)),
|
||
)
|
||
out = _empty_levels()
|
||
for level, n in await cursor.fetchall():
|
||
out[level] = n
|
||
return out
|
||
|
||
|
||
async def subs_per_level_raw(topic: str) -> dict[str, dict[int, int]]:
|
||
"""Subblocks per level, grouped by RAW block title (= guide section title)."""
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
f"SELECT block, {_LEVEL_CASE} AS level, COUNT(*) FROM subblocks "
|
||
"WHERE topic = ? AND status = 'consensus' GROUP BY block, level",
|
||
(topic,),
|
||
)
|
||
out: dict[str, dict[int, int]] = {}
|
||
for b, level, n in await cursor.fetchall():
|
||
out.setdefault(b, _empty_levels())[level] = n
|
||
return out
|
||
|
||
|
||
async def subs_per_level_all() -> dict[tuple[str, str], dict[int, int]]:
|
||
"""Subblocks per level per (topic, block_norm) — for the topic-wide levels derivation."""
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
f"SELECT topic, block_norm, {_LEVEL_CASE} AS level, COUNT(*) FROM subblocks "
|
||
"WHERE status = 'consensus' GROUP BY topic, block_norm, level"
|
||
)
|
||
out: dict[tuple[str, str], dict[int, int]] = {}
|
||
for t, bn, level, n in await cursor.fetchall():
|
||
out.setdefault((t, bn), _empty_levels())[level] = n
|
||
return out
|
||
|
||
|
||
async def subs_with_level(topic: str, block: str) -> list[dict]:
|
||
"""Consensus subblocks of a block with title, sub_norm and level 1–4."""
|
||
from textkit import _norm_title
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
f"SELECT sub_title, sub_norm, {_LEVEL_CASE} AS level FROM subblocks "
|
||
"WHERE topic = ? AND block_norm = ? AND status = 'consensus'",
|
||
(topic, _norm_title(block)),
|
||
)
|
||
return [{"title": t, "norm": sn, "level": e} for t, sn, e in await cursor.fetchall()]
|
||
|
||
|
||
async def list_block_scores_all() -> list[tuple[str, str, int]]:
|
||
"""(topic, block, good_answers) per block — raw data for the levels derivation."""
|
||
db = await get_db()
|
||
cursor = await db.execute("SELECT topic, block, good_answers FROM block_progress")
|
||
return [(t, b, n) for t, b, n in await cursor.fetchall()]
|
||
|
||
|
||
async def delete_block_data(topic: str) -> None:
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM block_texte WHERE topic = ?", (topic,))
|
||
await db.execute("DELETE FROM block_progress WHERE topic = ?", (topic,))
|
||
await db.commit()
|
||
|
||
|
||
# --- Blocks pipeline content: inventory / subblocks / question pattern / coverage / state / source ---
|
||
|
||
async def upsert_block(topic: str, title_norm: str, title: str, description: str = "",
|
||
sources: list | None = None, reader: str | None = None) -> None:
|
||
"""Insert a candidate or union the reader set. The first description is kept.
|
||
|
||
`reader` = ID of the research reader (e.g. "a5-1"). The union runs race-free in
|
||
ONE statement (json1), because several reader coroutines upsert concurrently — a
|
||
read-modify-write across `await` would lose members. `mentions` stays in sync
|
||
with `len(reader)`. `reader=None` (e.g. from `_set_inventar`) leaves the set unchanged."""
|
||
db = await get_db()
|
||
rid = reader if isinstance(reader, str) and reader else None
|
||
await db.execute(
|
||
"""INSERT INTO blocks (topic, title_norm, title, description, mentions, status, sources, reader, updated_at)
|
||
VALUES (?, ?, ?, ?, 1, 'candidate', ?, ?, ?)
|
||
ON CONFLICT(topic, title_norm) DO UPDATE SET
|
||
reader = (SELECT json_group_array(v) FROM (
|
||
SELECT value AS v FROM json_each(blocks.reader)
|
||
UNION SELECT ? WHERE ? IS NOT NULL)),
|
||
mentions = (SELECT count(*) FROM (
|
||
SELECT value AS v FROM json_each(blocks.reader)
|
||
UNION SELECT ? WHERE ? IS NOT NULL)),
|
||
sources = excluded.sources, updated_at = excluded.updated_at""",
|
||
(topic, title_norm, title, description,
|
||
json.dumps(sources or [], ensure_ascii=False),
|
||
json.dumps([rid] if rid else [], ensure_ascii=False), _now(),
|
||
rid, rid, rid, rid),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def list_blocks(topic: str, status: str | None = None) -> list[dict]:
|
||
db = await get_db()
|
||
if status is None:
|
||
cursor = await db.execute("SELECT * FROM blocks WHERE topic = ? ORDER BY rowid", (topic,))
|
||
else:
|
||
cursor = await db.execute("SELECT * FROM blocks WHERE topic = ? AND status = ? ORDER BY rowid", (topic, status))
|
||
rows = await cursor.fetchall()
|
||
out = []
|
||
for row in rows:
|
||
d = _row_to_dict(row, cursor)
|
||
d["sources"] = json.loads(d.get("sources") or "[]")
|
||
d["reader"] = json.loads(d.get("reader") or "[]")
|
||
out.append(d)
|
||
return out
|
||
|
||
|
||
async def set_block_status(topic: str, title_norm: str, status: str, title: str | None = None, description: str | None = None, neu_norm: str | None = None) -> None:
|
||
"""Set status; optionally update title/description (e.g. after a semantic merge).
|
||
`neu_norm` renames the norm key (clarification: reference title → meaningful name). Only
|
||
safe while no subblocks/facts are attached to the old `title_norm` yet."""
|
||
fields = {"status": status, "updated_at": _now()}
|
||
if title is not None:
|
||
fields["title"] = title
|
||
if description is not None:
|
||
fields["description"] = description
|
||
if neu_norm is not None and neu_norm != title_norm:
|
||
fields["title_norm"] = neu_norm
|
||
await _update("blocks", fields, {"topic": topic, "title_norm": title_norm})
|
||
|
||
|
||
async def delete_blocks(topic: str) -> None:
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM blocks WHERE topic = ?", (topic,))
|
||
await db.commit()
|
||
|
||
|
||
async def upsert_subblock(topic: str, block_norm: str, sub_norm: str, block: str, sub_title: str) -> None:
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO subblocks (topic, block_norm, sub_norm, block, sub_title, mentions, status, updated_at)
|
||
VALUES (?, ?, ?, ?, ?, 1, 'candidate', ?)
|
||
ON CONFLICT(topic, block_norm, sub_norm) DO UPDATE SET
|
||
mentions = mentions + 1, updated_at = excluded.updated_at""",
|
||
(topic, block_norm, sub_norm, block, sub_title, _now()),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def put_subblock(topic: str, block_norm: str, sub_norm: str, block: str, sub_title: str,
|
||
level: str | None = None, relevance: str | None = None,
|
||
facts: str | None = None, status: str = "consensus") -> None:
|
||
"""Insert/update WITHOUT a mention counter (mirror from the sidecar). Overwrite
|
||
level/relevance/facts only when a new value is passed (COALESCE protects existing data)."""
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO subblocks (topic, block_norm, sub_norm, block, sub_title, mentions, level, relevance, facts, status, updated_at)
|
||
VALUES (?, ?, ?, ?, ?, 1, ?, ?, COALESCE(?, ''), ?, ?)
|
||
ON CONFLICT(topic, block_norm, sub_norm) DO UPDATE SET
|
||
block = excluded.block, sub_title = excluded.sub_title,
|
||
level = COALESCE(excluded.level, subblocks.level),
|
||
relevance = COALESCE(excluded.relevance, subblocks.relevance),
|
||
facts = COALESCE(NULLIF(excluded.facts, ''), subblocks.facts),
|
||
status = excluded.status, updated_at = excluded.updated_at""",
|
||
(topic, block_norm, sub_norm, block, sub_title, level, relevance, facts, status, _now()),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def list_subblocks(topic: str, block_norm: str | None = None) -> list[dict]:
|
||
db = await get_db()
|
||
if block_norm is None:
|
||
cursor = await db.execute("SELECT * FROM subblocks WHERE topic = ? ORDER BY rowid", (topic,))
|
||
else:
|
||
cursor = await db.execute(
|
||
"SELECT * FROM subblocks WHERE topic = ? AND block_norm = ? ORDER BY rowid", (topic, block_norm)
|
||
)
|
||
rows = await cursor.fetchall()
|
||
return [_row_to_dict(row, cursor) for row in rows]
|
||
|
||
|
||
async def set_subblock_fields(topic: str, block_norm: str, sub_norm: str, **fields) -> None:
|
||
"""Set fields (level/relevance/status/sub_title) of a subblock row."""
|
||
fields["updated_at"] = _now()
|
||
await _update("subblocks", fields, {"topic": topic, "block_norm": block_norm, "sub_norm": sub_norm})
|
||
|
||
|
||
async def delete_subblocks(topic: str) -> None:
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM subblocks WHERE topic = ?", (topic,))
|
||
await db.commit()
|
||
|
||
|
||
async def upsert_question_pattern(topic: str, block_norm: str, sub_norm: str, block: str, sub_title: str, question: str) -> None:
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO question_pattern (topic, block_norm, sub_norm, block, sub_title, question, updated_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT(topic, block_norm, sub_norm) DO UPDATE SET
|
||
sub_title = excluded.sub_title, question = excluded.question, updated_at = excluded.updated_at""",
|
||
(topic, block_norm, sub_norm, block, sub_title, question, _now()),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def list_question_pattern(topic: str, block_norm: str | None = None) -> list[dict]:
|
||
db = await get_db()
|
||
if block_norm is None:
|
||
cursor = await db.execute("SELECT * FROM question_pattern WHERE topic = ? ORDER BY rowid", (topic,))
|
||
else:
|
||
cursor = await db.execute(
|
||
"SELECT * FROM question_pattern WHERE topic = ? AND block_norm = ? ORDER BY rowid", (topic, block_norm)
|
||
)
|
||
rows = await cursor.fetchall()
|
||
return [_row_to_dict(row, cursor) for row in rows]
|
||
|
||
|
||
async def delete_question_pattern(topic: str) -> None:
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM question_pattern WHERE topic = ?", (topic,))
|
||
await db.commit()
|
||
|
||
|
||
async def mark_sources_read_done(topic: str, sources: list[str]) -> None:
|
||
"""Mark the cited crawl pages as read_done (research-loop coverage)."""
|
||
if not sources:
|
||
return
|
||
db = await get_db()
|
||
now = _now()
|
||
await db.executemany(
|
||
"""INSERT INTO research_coverage (topic, source, read_done, updated_at) VALUES (?, ?, 1, ?)
|
||
ON CONFLICT(topic, source) DO UPDATE SET read_done = 1, updated_at = excluded.updated_at""",
|
||
[(topic, q, now) for q in sources],
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def list_coverage(topic: str) -> dict[str, int]:
|
||
db = await get_db()
|
||
cursor = await db.execute("SELECT source, read_done FROM research_coverage WHERE topic = ?", (topic,))
|
||
rows = await cursor.fetchall()
|
||
return {q: g for q, g in rows}
|
||
|
||
|
||
async def mark_content(topic: str, content: list[str], noise: list[str]) -> None:
|
||
"""Store the triage result per crawl page: content=1 (content) or 0 (noise)."""
|
||
db = await get_db()
|
||
now = _now()
|
||
rows = [(topic, q, 1, now) for q in content] + [(topic, q, 0, now) for q in noise]
|
||
if not rows:
|
||
return
|
||
await db.executemany(
|
||
"""INSERT INTO research_coverage (topic, source, content, updated_at) VALUES (?, ?, ?, ?)
|
||
ON CONFLICT(topic, source) DO UPDATE SET content = excluded.content, updated_at = excluded.updated_at""",
|
||
rows,
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def list_content(topic: str) -> list[str]:
|
||
"""Crawl pages the triage marked as content (content=1)."""
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
"SELECT source FROM research_coverage WHERE topic = ? AND content = 1 ORDER BY source", (topic,)
|
||
)
|
||
return [r[0] for r in await cursor.fetchall()]
|
||
|
||
|
||
async def delete_coverage(topic: str) -> None:
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM research_coverage WHERE topic = ?", (topic,))
|
||
await db.commit()
|
||
|
||
|
||
async def set_step_status(topic: str, step: str, status: str) -> None:
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO pipeline_state (topic, step, status, updated_at) VALUES (?, ?, ?, ?)
|
||
ON CONFLICT(topic, step) DO UPDATE SET status = excluded.status, updated_at = excluded.updated_at""",
|
||
(topic, step, status, _now()),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def get_step_status(topic: str, step: str) -> str:
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
"SELECT status FROM pipeline_state WHERE topic = ? AND step = ?", (topic, step)
|
||
)
|
||
row = await cursor.fetchone()
|
||
return row[0] if row else "open"
|
||
|
||
|
||
|
||
|
||
async def delete_pipeline_state(topic: str, steps: list[str] | None = None) -> None:
|
||
db = await get_db()
|
||
if steps is None:
|
||
await db.execute("DELETE FROM pipeline_state WHERE topic = ?", (topic,))
|
||
elif steps:
|
||
marks = ",".join("?" for _ in steps)
|
||
await db.execute(f"DELETE FROM pipeline_state WHERE topic = ? AND step IN ({marks})", (topic, *steps))
|
||
await db.commit()
|
||
|
||
|
||
|
||
|
||
|
||
|
||
async def delete_source(topic: str) -> None:
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM source WHERE topic = ?", (topic,))
|
||
await db.commit()
|
||
|
||
|
||
async def set_guide_content(topic: str, format: str, content_json: str) -> None:
|
||
"""Store finished guide content (JSON blob) per topic+format."""
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO guide_content (topic, format, json, updated_at) VALUES (?, ?, ?, ?)
|
||
ON CONFLICT(topic, format) DO UPDATE SET json = excluded.json, updated_at = excluded.updated_at""",
|
||
(topic, format, content_json, _now()),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def get_guide_content(topic: str, format: str) -> str | None:
|
||
db = await get_db()
|
||
cursor = await db.execute("SELECT json FROM guide_content WHERE topic = ? AND format = ?", (topic, format))
|
||
row = await cursor.fetchone()
|
||
return row[0] if row else None
|
||
|
||
|
||
async def delete_guide_content(topic: str, format: str | None = None) -> None:
|
||
db = await get_db()
|
||
if format is None:
|
||
await db.execute("DELETE FROM guide_content WHERE topic = ?", (topic,))
|
||
else:
|
||
await db.execute("DELETE FROM guide_content WHERE topic = ? AND format = ?", (topic, format))
|
||
await db.commit()
|
||
|
||
|
||
async def set_outline(topic: str, outline_json: str) -> None:
|
||
"""Store the outline (chapter→numbers, JSON) per topic — blocks artifact for the guide."""
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO guide_outline (topic, json, updated_at) VALUES (?, ?, ?)
|
||
ON CONFLICT(topic) DO UPDATE SET json = excluded.json, updated_at = excluded.updated_at""",
|
||
(topic, outline_json, _now()),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def get_outline(topic: str) -> str | None:
|
||
db = await get_db()
|
||
cursor = await db.execute("SELECT json FROM guide_outline WHERE topic = ?", (topic,))
|
||
row = await cursor.fetchone()
|
||
return row[0] if row else None
|
||
|
||
|
||
async def delete_outline(topic: str) -> None:
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM guide_outline WHERE topic = ?", (topic,))
|
||
await db.commit()
|
||
|
||
|
||
async def put_sub_artifact(topic: str, block_norm: str, sub_norm: str, type: str,
|
||
data: str, block: str = "", sub_title: str = "") -> None:
|
||
"""Store one learning artifact (flashcard/example) as JSON in `data`."""
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO sub_artefakte (topic, block_norm, sub_norm, type, block, sub_title, data, updated_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT(topic, block_norm, sub_norm, type) DO UPDATE SET
|
||
block = excluded.block, sub_title = excluded.sub_title,
|
||
data = excluded.data, updated_at = excluded.updated_at""",
|
||
(topic, block_norm, sub_norm, type, block, sub_title, data, _now()),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def get_sub_artefakte(topic: str, type: str | None = None) -> list[dict]:
|
||
db = await get_db()
|
||
if type is None:
|
||
cursor = await db.execute("SELECT * FROM sub_artefakte WHERE topic = ? ORDER BY rowid", (topic,))
|
||
else:
|
||
cursor = await db.execute(
|
||
"SELECT * FROM sub_artefakte WHERE topic = ? AND type = ? ORDER BY rowid", (topic, type)
|
||
)
|
||
rows = await cursor.fetchall()
|
||
return [_row_to_dict(row, cursor) for row in rows]
|
||
|
||
|
||
async def delete_sub_artefakte(topic: str) -> None:
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM sub_artefakte WHERE topic = ?", (topic,))
|
||
await db.commit()
|
||
|
||
|
||
async def get_block_hurdles(topic: str, block_norm: str) -> list[str]:
|
||
"""Typical misconceptions (hurdles) of all subs of a block — as a distractor pool for the quiz.
|
||
Reads from subblocks.facts (JSON blob); empty/missing ones are skipped."""
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
"SELECT facts FROM subblocks WHERE topic = ? AND block_norm = ?", (topic, block_norm)
|
||
)
|
||
rows = await cursor.fetchall()
|
||
hurdles = []
|
||
for (facts,) in rows:
|
||
if not facts:
|
||
continue
|
||
try:
|
||
fk = json.loads(facts)
|
||
except (ValueError, TypeError):
|
||
continue
|
||
h = (fk.get("hurdles") or "").strip() if isinstance(fk, dict) else ""
|
||
if h:
|
||
hurdles.append(h)
|
||
return hurdles
|
||
|
||
|
||
async def delete_topic_pipeline(topic: str) -> None:
|
||
"""Discard the blocks area of a topic (inventory/subs/pattern/coverage/state/artifacts).
|
||
NOT the topic config `source` — that is managed separately (delete_source)."""
|
||
db = await get_db()
|
||
for tab in ("blocks", "subblocks", "question_pattern", "research_coverage",
|
||
"pipeline_state", "guide_outline", "sub_artefakte"):
|
||
await db.execute(f"DELETE FROM {tab} WHERE topic = ?", (topic,))
|
||
await db.commit()
|