852 lines
31 KiB
Python
852 lines
31 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_BAUSTEIN_TEXTE = """
|
||
CREATE TABLE IF NOT EXISTS baustein_texte (
|
||
topic TEXT NOT NULL,
|
||
baustein TEXT NOT NULL,
|
||
art TEXT NOT NULL,
|
||
md TEXT NOT NULL,
|
||
created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, baustein, art)
|
||
)
|
||
"""
|
||
|
||
CREATE_BAUSTEIN_PROGRESS = """
|
||
CREATE TABLE IF NOT EXISTS baustein_progress (
|
||
topic TEXT NOT NULL,
|
||
baustein TEXT NOT NULL,
|
||
gute_antworten INTEGER NOT NULL DEFAULT 0,
|
||
streak INTEGER NOT NULL DEFAULT 0,
|
||
absolviert TEXT,
|
||
verstanden TEXT,
|
||
gemeistert TEXT,
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, baustein)
|
||
)
|
||
"""
|
||
|
||
# --- Bausteine-Pipeline-Inhalt (ersetzt Datei-Sidecars) ---
|
||
|
||
# Inventar: ein Baustein je (topic, titel_norm). nennungen = Anzahl Agenten/Runden,
|
||
# die ihn nannten (≥2 = Konsens). status: kandidat/konsens/rest/verworfen.
|
||
CREATE_BAUSTEINE = """
|
||
CREATE TABLE IF NOT EXISTS bausteine (
|
||
topic TEXT NOT NULL,
|
||
titel_norm TEXT NOT NULL,
|
||
titel TEXT NOT NULL,
|
||
beschreibung TEXT NOT NULL DEFAULT '',
|
||
nennungen INTEGER NOT NULL DEFAULT 1,
|
||
status TEXT NOT NULL DEFAULT 'kandidat',
|
||
quellen TEXT NOT NULL DEFAULT '[]',
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, titel_norm)
|
||
)
|
||
"""
|
||
|
||
# Subbausteine je Baustein. stufe (einfach/mittel/schwer) + relevanz (relevant/rand)
|
||
# werden später gesetzt. nennungen analog zum Inventar.
|
||
CREATE_SUBBAUSTEINE = """
|
||
CREATE TABLE IF NOT EXISTS subbausteine (
|
||
topic TEXT NOT NULL,
|
||
baustein_norm TEXT NOT NULL,
|
||
sub_norm TEXT NOT NULL,
|
||
baustein TEXT NOT NULL,
|
||
sub_titel TEXT NOT NULL,
|
||
nennungen INTEGER NOT NULL DEFAULT 1,
|
||
stufe TEXT,
|
||
relevanz TEXT,
|
||
status TEXT NOT NULL DEFAULT 'kandidat',
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, baustein_norm, sub_norm)
|
||
)
|
||
"""
|
||
|
||
# Ein Frage-Muster je (Baustein, Subbaustein, Typ).
|
||
CREATE_FRAGE_MUSTER = """
|
||
CREATE TABLE IF NOT EXISTS frage_muster (
|
||
topic TEXT NOT NULL,
|
||
baustein_norm TEXT NOT NULL,
|
||
sub_norm TEXT NOT NULL,
|
||
baustein TEXT NOT NULL,
|
||
sub_titel TEXT NOT NULL,
|
||
typ TEXT NOT NULL,
|
||
frage TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, baustein_norm, sub_norm, typ)
|
||
)
|
||
"""
|
||
|
||
# Crawl-Seiten je Thema: inhalt (1=Content/0=Noise, von der Sichtung) + gelesen (von der Recherche).
|
||
CREATE_RECHERCHE_COVERAGE = """
|
||
CREATE TABLE IF NOT EXISTS recherche_coverage (
|
||
topic TEXT NOT NULL,
|
||
quelle TEXT NOT NULL,
|
||
gelesen INTEGER NOT NULL DEFAULT 0,
|
||
inhalt INTEGER,
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, quelle)
|
||
)
|
||
"""
|
||
|
||
# Schritt-Status der Pipeline (ersetzt Datei-Existenz-Resume + Reset-Globs).
|
||
CREATE_PIPELINE_STATE = """
|
||
CREATE TABLE IF NOT EXISTS pipeline_state (
|
||
topic TEXT NOT NULL,
|
||
schritt TEXT NOT NULL,
|
||
status TEXT NOT NULL DEFAULT 'offen',
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (topic, schritt)
|
||
)
|
||
"""
|
||
|
||
# Fertiger Guide-Inhalt je (Thema, Format) als JSON-Blob (ersetzt die Guide-JSON-Datei).
|
||
# Geteilt über alle Guide-Läufe desselben Thema+Formats (wie zuvor die Content-Datei).
|
||
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)
|
||
)
|
||
"""
|
||
|
||
# Quellen-Wahl je Thema (ersetzt quelle.json).
|
||
CREATE_QUELLE = """
|
||
CREATE TABLE IF NOT EXISTS quelle (
|
||
topic TEXT PRIMARY KEY,
|
||
type TEXT NOT NULL,
|
||
ort TEXT NOT NULL DEFAULT '',
|
||
spec TEXT NOT NULL DEFAULT '',
|
||
updated_at TEXT NOT NULL
|
||
)
|
||
"""
|
||
|
||
_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 übersteht Crashes deutlich besser; busy_timeout fängt kurze Locks ab.
|
||
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_BAUSTEIN_TEXTE)
|
||
await db.execute(CREATE_BAUSTEIN_PROGRESS)
|
||
await db.execute(CREATE_BAUSTEINE)
|
||
await db.execute(CREATE_SUBBAUSTEINE)
|
||
await db.execute(CREATE_FRAGE_MUSTER)
|
||
await db.execute(CREATE_RECHERCHE_COVERAGE)
|
||
await db.execute(CREATE_PIPELINE_STATE)
|
||
await db.execute(CREATE_GUIDE_CONTENT)
|
||
await db.execute(CREATE_QUELLE)
|
||
try: # Migration für Bestands-DBs ohne step-Spalte
|
||
await db.execute("ALTER TABLE guides ADD COLUMN step INTEGER")
|
||
except aiosqlite.OperationalError:
|
||
pass
|
||
try: # Migration: recherche_coverage.inhalt (Content/Noise aus der Sichtung)
|
||
await db.execute("ALTER TABLE recherche_coverage ADD COLUMN inhalt INTEGER")
|
||
except aiosqlite.OperationalError:
|
||
pass
|
||
try: # Migration für Bestands-DBs ohne verstanden-Spalte (Mastery-Stufe)
|
||
await db.execute("ALTER TABLE baustein_progress ADD COLUMN verstanden TEXT")
|
||
except aiosqlite.OperationalError:
|
||
pass
|
||
try: # Migration für Bestands-DBs ohne gemeistert-Spalte (Meisterpfad 25)
|
||
await db.execute("ALTER TABLE baustein_progress ADD COLUMN gemeistert TEXT")
|
||
except aiosqlite.OperationalError:
|
||
pass
|
||
try: # Migration für Bestands-DBs ohne streak-Spalte (persistente Streak-Bonus-Folge)
|
||
await db.execute("ALTER TABLE baustein_progress ADD COLUMN streak INTEGER NOT NULL DEFAULT 0")
|
||
except aiosqlite.OperationalError:
|
||
pass
|
||
# Offene-Frage-Anker: Basis/Streak VOR der aktuell offenen Frage — macht die Bewertung
|
||
# serverseitig idempotent (Re-Bewertung) und driftfrei über Fragen hinweg.
|
||
for _spalte, _typ in (("offene_frage", "TEXT"), ("offene_basis", "INTEGER"), ("offene_streak", "INTEGER")):
|
||
try:
|
||
await db.execute(f"ALTER TABLE baustein_progress ADD COLUMN {_spalte} {_typ}")
|
||
except aiosqlite.OperationalError:
|
||
pass
|
||
# Migration: alte vertiefungen-Tabelle → baustein_texte (Bestand = lange Form, art '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 baustein_texte (topic, baustein, art, md, created_at, updated_at) "
|
||
"SELECT topic, baustein, '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-Neustart' "
|
||
"WHERE status IN ('queued', 'generating')"
|
||
)
|
||
# Milestone-Migration auf Score-Skala 0–30: absolviert≥6, verstanden≥12, gemeistert≥18.
|
||
# Leitet die Meilensteine aus dem aktuellen Score ab (bestehende Timestamps bleiben),
|
||
# idempotent — nach einem Lauf stabil, im Normalbetrieb deckungsgleich mit der Prüf-Route.
|
||
await db.execute("UPDATE baustein_progress SET gute_antworten = 30 WHERE gute_antworten > 30") # Alt-Scores deckeln
|
||
for _spalte, _schwelle in (("absolviert", 6), ("verstanden", 12), ("gemeistert", 18)):
|
||
await db.execute(
|
||
f"UPDATE baustein_progress SET {_spalte} = COALESCE({_spalte}, datetime('now')) "
|
||
f"WHERE gute_antworten >= {_schwelle}"
|
||
)
|
||
await db.execute(f"UPDATE baustein_progress SET {_spalte} = NULL WHERE gute_antworten < {_schwelle}")
|
||
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_guide(guide_id: str, **fields) -> None:
|
||
sets = ", ".join(f"{k} = :{k}" for k in fields)
|
||
fields["id"] = guide_id
|
||
db = await get_db()
|
||
await db.execute(f"UPDATE guides SET {sets} WHERE id = :id", fields)
|
||
await db.commit()
|
||
|
||
|
||
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
|
||
|
||
|
||
# --- Themen ---
|
||
|
||
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()
|
||
|
||
|
||
# --- Elemente ---
|
||
|
||
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)
|
||
sets = ", ".join(f"{k} = :{k}" for k in fields)
|
||
fields["id"] = element_id
|
||
db = await get_db()
|
||
await db.execute(f"UPDATE elements SET {sets} WHERE id = :id", fields)
|
||
await db.commit()
|
||
|
||
|
||
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
|
||
|
||
|
||
# --- Kapitel-Fortschritt ---
|
||
|
||
async def list_progress_all() -> dict[str, set[str]]:
|
||
"""Kompletter Kapitel-Fortschritt in einem Query: guide_id → Kapitel-Titel."""
|
||
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()
|
||
|
||
|
||
# --- Baustein-Lernen: Vertiefungen + Prüfungs-Fortschritt ---
|
||
|
||
def _now() -> str:
|
||
from datetime import datetime, timezone
|
||
return datetime.now(timezone.utc).isoformat()
|
||
|
||
|
||
async def list_baustein_progress(topic: str) -> list[dict]:
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
"SELECT baustein, gute_antworten, streak, absolviert, verstanden, gemeistert FROM baustein_progress WHERE topic = ?", (topic,)
|
||
)
|
||
rows = await cursor.fetchall()
|
||
return [{"baustein": b, "gute_antworten": n, "streak": s, "absolviert": a, "verstanden": v, "gemeistert": m} for b, n, s, a, v, m in rows]
|
||
|
||
|
||
async def get_baustein_progress(topic: str, baustein: str) -> dict:
|
||
"""Eine Baustein-Zeile inkl. Offene-Frage-Anker. Defaults, falls noch keine existiert."""
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
"SELECT gute_antworten, streak, absolviert, verstanden, gemeistert, "
|
||
"offene_frage, offene_basis, offene_streak FROM baustein_progress "
|
||
"WHERE topic = ? AND baustein = ?",
|
||
(topic, baustein),
|
||
)
|
||
row = await cursor.fetchone()
|
||
if row is None:
|
||
return {"gute_antworten": 0, "streak": 0, "absolviert": None, "verstanden": None,
|
||
"gemeistert": None, "offene_frage": None, "offene_basis": None, "offene_streak": None}
|
||
return {"gute_antworten": row[0], "streak": row[1], "absolviert": row[2], "verstanden": row[3],
|
||
"gemeistert": row[4], "offene_frage": row[5], "offene_basis": row[6], "offene_streak": row[7]}
|
||
|
||
|
||
async def set_offene_frage(topic: str, baustein: str, frage: str, basis: int, streak: int) -> None:
|
||
"""Friert Basis + Streak VOR der jetzt offenen Frage ein (Anker für idempotente Re-Bewertung)."""
|
||
db = await get_db()
|
||
now = _now()
|
||
await db.execute(
|
||
"""INSERT INTO baustein_progress (topic, baustein, offene_frage, offene_basis, offene_streak, updated_at)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT(topic, baustein) DO UPDATE SET
|
||
offene_frage = excluded.offene_frage, offene_basis = excluded.offene_basis,
|
||
offene_streak = excluded.offene_streak, updated_at = excluded.updated_at""",
|
||
(topic, baustein, frage, basis, streak, now),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def set_baustein_score(topic: str, baustein: str, score: int) -> int:
|
||
"""Setzt den Score absolut (vom Aufrufer geclampt) und liefert ihn zurück."""
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO baustein_progress (topic, baustein, gute_antworten, updated_at)
|
||
VALUES (?, ?, ?, ?)
|
||
ON CONFLICT(topic, baustein) DO UPDATE SET
|
||
gute_antworten = excluded.gute_antworten, updated_at = excluded.updated_at""",
|
||
(topic, baustein, score, _now()),
|
||
)
|
||
await db.commit()
|
||
return score
|
||
|
||
|
||
|
||
|
||
async def set_baustein_verstanden(topic: str, baustein: str) -> bool:
|
||
"""Markiert verstanden (Mastery); True nur beim ersten Mal. Sticky wie absolviert."""
|
||
db = await get_db()
|
||
now = _now()
|
||
await db.execute(
|
||
"INSERT OR IGNORE INTO baustein_progress (topic, baustein, gute_antworten, updated_at) VALUES (?, ?, 0, ?)",
|
||
(topic, baustein, now),
|
||
)
|
||
cursor = await db.execute(
|
||
"UPDATE baustein_progress SET verstanden = ?, updated_at = ? "
|
||
"WHERE topic = ? AND baustein = ? AND verstanden IS NULL",
|
||
(now, now, topic, baustein),
|
||
)
|
||
await db.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
|
||
async def set_baustein_gemeistert(topic: str, baustein: str) -> bool:
|
||
"""Markiert gemeistert (Meisterpfad, Score 25); True nur beim ersten Mal. Sticky."""
|
||
db = await get_db()
|
||
now = _now()
|
||
await db.execute(
|
||
"INSERT OR IGNORE INTO baustein_progress (topic, baustein, gute_antworten, updated_at) VALUES (?, ?, 0, ?)",
|
||
(topic, baustein, now),
|
||
)
|
||
cursor = await db.execute(
|
||
"UPDATE baustein_progress SET gemeistert = ?, updated_at = ? "
|
||
"WHERE topic = ? AND baustein = ? AND gemeistert IS NULL",
|
||
(now, now, topic, baustein),
|
||
)
|
||
await db.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
|
||
async def set_baustein_absolviert(topic: str, baustein: str) -> bool:
|
||
"""Markiert absolviert; True nur beim ersten Mal (steuert den Element-Task)."""
|
||
db = await get_db()
|
||
now = _now()
|
||
await db.execute(
|
||
"INSERT OR IGNORE INTO baustein_progress (topic, baustein, gute_antworten, updated_at) VALUES (?, ?, 0, ?)",
|
||
(topic, baustein, now),
|
||
)
|
||
cursor = await db.execute(
|
||
"UPDATE baustein_progress SET absolviert = ?, updated_at = ? "
|
||
"WHERE topic = ? AND baustein = ? AND absolviert IS NULL",
|
||
(now, now, topic, baustein),
|
||
)
|
||
await db.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
|
||
async def count_relevant_subs(topic: str, baustein: str) -> int:
|
||
"""Anzahl relevanter Konsens-Subbausteine eines Bausteins (für den cap = 2×Subs)."""
|
||
from textkit import _norm_titel
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
"SELECT COUNT(*) FROM subbausteine WHERE topic = ? AND baustein_norm = ? "
|
||
"AND status = 'konsens' AND relevanz != 'rand'",
|
||
(topic, _norm_titel(baustein)),
|
||
)
|
||
return (await cursor.fetchone())[0]
|
||
|
||
|
||
async def subs_count_alle() -> dict[tuple[str, str], int]:
|
||
"""Relevante Konsens-Subbausteine je (topic, baustein_norm) — für die Stufen-Ableitung."""
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
"SELECT topic, baustein_norm, COUNT(*) FROM subbausteine "
|
||
"WHERE status = 'konsens' AND relevanz != 'rand' GROUP BY topic, baustein_norm"
|
||
)
|
||
return {(t, bn): n for t, bn, n in await cursor.fetchall()}
|
||
|
||
|
||
async def list_baustein_scores_all() -> list[tuple[str, str, int]]:
|
||
"""(topic, baustein, gute_antworten) je Baustein — Rohdaten für die Stufen-Ableitung."""
|
||
db = await get_db()
|
||
cursor = await db.execute("SELECT topic, baustein, gute_antworten FROM baustein_progress")
|
||
return [(t, b, n) for t, b, n in await cursor.fetchall()]
|
||
|
||
|
||
async def delete_baustein_daten(topic: str) -> None:
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM baustein_texte WHERE topic = ?", (topic,))
|
||
await db.execute("DELETE FROM baustein_progress WHERE topic = ?", (topic,))
|
||
await db.commit()
|
||
|
||
|
||
# --- Bausteine-Pipeline-Inhalt: Inventar / Subbausteine / Frage-Muster / Coverage / State / Quelle ---
|
||
|
||
async def upsert_baustein(topic: str, titel_norm: str, titel: str, beschreibung: str = "", quellen: list | None = None) -> None:
|
||
"""Kandidat einfügen oder Nennungszähler erhöhen. Erst-Beschreibung bleibt erhalten."""
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO bausteine (topic, titel_norm, titel, beschreibung, nennungen, status, quellen, updated_at)
|
||
VALUES (?, ?, ?, ?, 1, 'kandidat', ?, ?)
|
||
ON CONFLICT(topic, titel_norm) DO UPDATE SET
|
||
nennungen = nennungen + 1, quellen = excluded.quellen, updated_at = excluded.updated_at""",
|
||
(topic, titel_norm, titel, beschreibung, json.dumps(quellen or [], ensure_ascii=False), _now()),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def list_bausteine(topic: str, status: str | None = None) -> list[dict]:
|
||
db = await get_db()
|
||
if status is None:
|
||
cursor = await db.execute("SELECT * FROM bausteine WHERE topic = ? ORDER BY rowid", (topic,))
|
||
else:
|
||
cursor = await db.execute("SELECT * FROM bausteine 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["quellen"] = json.loads(d.get("quellen") or "[]")
|
||
out.append(d)
|
||
return out
|
||
|
||
|
||
async def set_baustein_status(topic: str, titel_norm: str, status: str, titel: str | None = None, beschreibung: str | None = None) -> None:
|
||
"""Status setzen; optional Titel/Beschreibung aktualisieren (z.B. nach semantischem Merge)."""
|
||
db = await get_db()
|
||
fields = {"status": status, "updated_at": _now()}
|
||
if titel is not None:
|
||
fields["titel"] = titel
|
||
if beschreibung is not None:
|
||
fields["beschreibung"] = beschreibung
|
||
sets = ", ".join(f"{k} = :{k}" for k in fields)
|
||
await db.execute(
|
||
f"UPDATE bausteine SET {sets} WHERE topic = :topic AND titel_norm = :titel_norm",
|
||
{**fields, "topic": topic, "titel_norm": titel_norm},
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def delete_bausteine(topic: str) -> None:
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM bausteine WHERE topic = ?", (topic,))
|
||
await db.commit()
|
||
|
||
|
||
async def upsert_subbaustein(topic: str, baustein_norm: str, sub_norm: str, baustein: str, sub_titel: str) -> None:
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO subbausteine (topic, baustein_norm, sub_norm, baustein, sub_titel, nennungen, status, updated_at)
|
||
VALUES (?, ?, ?, ?, ?, 1, 'kandidat', ?)
|
||
ON CONFLICT(topic, baustein_norm, sub_norm) DO UPDATE SET
|
||
nennungen = nennungen + 1, updated_at = excluded.updated_at""",
|
||
(topic, baustein_norm, sub_norm, baustein, sub_titel, _now()),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def put_subbaustein(topic: str, baustein_norm: str, sub_norm: str, baustein: str, sub_titel: str,
|
||
stufe: str | None = None, relevanz: str | None = None, status: str = "konsens") -> None:
|
||
"""Insert/Update OHNE Nennungszähler (Spiegel aus dem Sidecar). stufe/relevanz nur überschreiben,
|
||
wenn ein neuer Wert übergeben wird (COALESCE schützt Bestehendes)."""
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO subbausteine (topic, baustein_norm, sub_norm, baustein, sub_titel, nennungen, stufe, relevanz, status, updated_at)
|
||
VALUES (?, ?, ?, ?, ?, 1, ?, ?, ?, ?)
|
||
ON CONFLICT(topic, baustein_norm, sub_norm) DO UPDATE SET
|
||
baustein = excluded.baustein, sub_titel = excluded.sub_titel,
|
||
stufe = COALESCE(excluded.stufe, subbausteine.stufe),
|
||
relevanz = COALESCE(excluded.relevanz, subbausteine.relevanz),
|
||
status = excluded.status, updated_at = excluded.updated_at""",
|
||
(topic, baustein_norm, sub_norm, baustein, sub_titel, stufe, relevanz, status, _now()),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def list_subbausteine(topic: str, baustein_norm: str | None = None) -> list[dict]:
|
||
db = await get_db()
|
||
if baustein_norm is None:
|
||
cursor = await db.execute("SELECT * FROM subbausteine WHERE topic = ? ORDER BY rowid", (topic,))
|
||
else:
|
||
cursor = await db.execute(
|
||
"SELECT * FROM subbausteine WHERE topic = ? AND baustein_norm = ? ORDER BY rowid", (topic, baustein_norm)
|
||
)
|
||
rows = await cursor.fetchall()
|
||
return [_row_to_dict(row, cursor) for row in rows]
|
||
|
||
|
||
async def set_subbaustein_felder(topic: str, baustein_norm: str, sub_norm: str, **fields) -> None:
|
||
"""Setzt Felder (stufe/relevanz/status/sub_titel) einer Subbaustein-Zeile."""
|
||
fields["updated_at"] = _now()
|
||
sets = ", ".join(f"{k} = :{k}" for k in fields)
|
||
db = await get_db()
|
||
await db.execute(
|
||
f"UPDATE subbausteine SET {sets} WHERE topic = :topic AND baustein_norm = :baustein_norm AND sub_norm = :sub_norm",
|
||
{**fields, "topic": topic, "baustein_norm": baustein_norm, "sub_norm": sub_norm},
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def delete_subbausteine(topic: str) -> None:
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM subbausteine WHERE topic = ?", (topic,))
|
||
await db.commit()
|
||
|
||
|
||
async def upsert_frage_muster(topic: str, baustein_norm: str, sub_norm: str, baustein: str, sub_titel: str, typ: str, frage: str) -> None:
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO frage_muster (topic, baustein_norm, sub_norm, baustein, sub_titel, typ, frage, updated_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT(topic, baustein_norm, sub_norm, typ) DO UPDATE SET
|
||
frage = excluded.frage, updated_at = excluded.updated_at""",
|
||
(topic, baustein_norm, sub_norm, baustein, sub_titel, typ, frage, _now()),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def list_frage_muster(topic: str, baustein_norm: str | None = None) -> list[dict]:
|
||
db = await get_db()
|
||
if baustein_norm is None:
|
||
cursor = await db.execute("SELECT * FROM frage_muster WHERE topic = ? ORDER BY rowid", (topic,))
|
||
else:
|
||
cursor = await db.execute(
|
||
"SELECT * FROM frage_muster WHERE topic = ? AND baustein_norm = ? ORDER BY rowid", (topic, baustein_norm)
|
||
)
|
||
rows = await cursor.fetchall()
|
||
return [_row_to_dict(row, cursor) for row in rows]
|
||
|
||
|
||
async def delete_frage_muster(topic: str) -> None:
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM frage_muster WHERE topic = ?", (topic,))
|
||
await db.commit()
|
||
|
||
|
||
async def mark_quellen_gelesen(topic: str, quellen: list[str]) -> None:
|
||
"""Markiert die zitierten Crawl-Seiten als gelesen (Recherche-Loop-Abdeckung)."""
|
||
if not quellen:
|
||
return
|
||
db = await get_db()
|
||
now = _now()
|
||
await db.executemany(
|
||
"""INSERT INTO recherche_coverage (topic, quelle, gelesen, updated_at) VALUES (?, ?, 1, ?)
|
||
ON CONFLICT(topic, quelle) DO UPDATE SET gelesen = 1, updated_at = excluded.updated_at""",
|
||
[(topic, q, now) for q in quellen],
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def list_coverage(topic: str) -> dict[str, int]:
|
||
db = await get_db()
|
||
cursor = await db.execute("SELECT quelle, gelesen FROM recherche_coverage WHERE topic = ?", (topic,))
|
||
rows = await cursor.fetchall()
|
||
return {q: g for q, g in rows}
|
||
|
||
|
||
async def mark_inhalt(topic: str, content: list[str], noise: list[str]) -> None:
|
||
"""Sichtungs-Ergebnis je Crawl-Seite ablegen: inhalt=1 (Content) bzw. 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 recherche_coverage (topic, quelle, inhalt, updated_at) VALUES (?, ?, ?, ?)
|
||
ON CONFLICT(topic, quelle) DO UPDATE SET inhalt = excluded.inhalt, updated_at = excluded.updated_at""",
|
||
rows,
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def list_content(topic: str) -> list[str]:
|
||
"""Crawl-Seiten, die die Sichtung als Content markiert hat (inhalt=1)."""
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
"SELECT quelle FROM recherche_coverage WHERE topic = ? AND inhalt = 1 ORDER BY quelle", (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 recherche_coverage WHERE topic = ?", (topic,))
|
||
await db.commit()
|
||
|
||
|
||
async def set_step_status(topic: str, schritt: str, status: str) -> None:
|
||
db = await get_db()
|
||
await db.execute(
|
||
"""INSERT INTO pipeline_state (topic, schritt, status, updated_at) VALUES (?, ?, ?, ?)
|
||
ON CONFLICT(topic, schritt) DO UPDATE SET status = excluded.status, updated_at = excluded.updated_at""",
|
||
(topic, schritt, status, _now()),
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
async def get_step_status(topic: str, schritt: str) -> str:
|
||
db = await get_db()
|
||
cursor = await db.execute(
|
||
"SELECT status FROM pipeline_state WHERE topic = ? AND schritt = ?", (topic, schritt)
|
||
)
|
||
row = await cursor.fetchone()
|
||
return row[0] if row else "offen"
|
||
|
||
|
||
|
||
|
||
async def delete_pipeline_state(topic: str, schritte: list[str] | None = None) -> None:
|
||
db = await get_db()
|
||
if schritte is None:
|
||
await db.execute("DELETE FROM pipeline_state WHERE topic = ?", (topic,))
|
||
elif schritte:
|
||
marks = ",".join("?" for _ in schritte)
|
||
await db.execute(f"DELETE FROM pipeline_state WHERE topic = ? AND schritt IN ({marks})", (topic, *schritte))
|
||
await db.commit()
|
||
|
||
|
||
|
||
|
||
|
||
|
||
async def delete_quelle(topic: str) -> None:
|
||
db = await get_db()
|
||
await db.execute("DELETE FROM quelle WHERE topic = ?", (topic,))
|
||
await db.commit()
|
||
|
||
|
||
async def set_guide_content(topic: str, format: str, content_json: str) -> None:
|
||
"""Fertigen Guide-Inhalt (JSON-Blob) je Thema+Format speichern."""
|
||
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 delete_topic_pipeline(topic: str) -> None:
|
||
"""Bausteine-Bereich eines Themas verwerfen (Inventar/Subs/Muster/Coverage/State).
|
||
NICHT die Themen-Config `quelle` — die wird separat verwaltet (delete_quelle)."""
|
||
db = await get_db()
|
||
for tab in ("bausteine", "subbausteine", "frage_muster", "recherche_coverage", "pipeline_state"):
|
||
await db.execute(f"DELETE FROM {tab} WHERE topic = ?", (topic,))
|
||
await db.commit()
|