1206 lines
43 KiB
Python
1206 lines
43 KiB
Python
"""Named, versioned journal generation guidelines. Not model temperature.
|
||
|
||
Four independent dimensions are selected by ID. Instruction text lives in
|
||
generation_guidelines, seeded from JSON. This module selects, validates and
|
||
composes. It does not own prompt wording and has no numeric 0–100 path.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import uuid
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
|
||
from db import get_db, row_to_dict
|
||
from journal_policy import PolicyError
|
||
from placeholders import CONTEXT_PATTERN
|
||
|
||
PURPOSE_JOURNAL = "journal_generate"
|
||
USER_SLOTS = ("transformation", "detail", "voice", "narrative")
|
||
ALLOWED_SLOTS = USER_SLOTS
|
||
LEGACY_SOURCE_MODE_SLOT = "source_mode"
|
||
SELECTION_KEYS = (
|
||
"transformation_id",
|
||
"detail_id",
|
||
"voice_id",
|
||
"narrative_id",
|
||
)
|
||
SLOT_TO_ID_KEY = {
|
||
"transformation": "transformation_id",
|
||
"detail": "detail_id",
|
||
"voice": "voice_id",
|
||
"narrative": "narrative_id",
|
||
}
|
||
ID_KEY_TO_SLOT = {value: key for key, value in SLOT_TO_ID_KEY.items()}
|
||
SLOT_INSTRUCTION_KEYS = {
|
||
"transformation": "transformation_instructions",
|
||
"detail": "detail_instructions",
|
||
"voice": "voice_instructions",
|
||
"narrative": "narrative_instructions",
|
||
}
|
||
REQUIRED_JOURNAL_PLACEHOLDERS = (
|
||
"transformation_instructions",
|
||
"detail_instructions",
|
||
"voice_instructions",
|
||
"narrative_instructions",
|
||
"writing_profile",
|
||
"style_examples",
|
||
"reconstruction",
|
||
"existing_text",
|
||
)
|
||
RETIRED_JOURNAL_PLACEHOLDERS = (
|
||
"source_mode_instructions",
|
||
"editorial_mode",
|
||
"editorial_instructions",
|
||
)
|
||
STATUS_DRAFT = "draft"
|
||
STATUS_ACTIVE = "active"
|
||
STATUS_ARCHIVED = "archived"
|
||
STATUSES = (STATUS_DRAFT, STATUS_ACTIVE, STATUS_ARCHIVED)
|
||
MAX_INSTRUCTION_CHARS = 1000
|
||
MAX_LABEL_CHARS = 80
|
||
MAX_SUMMARY_CHARS = 160
|
||
KEY_RE = re.compile(r"^[a-z][a-z0-9_]{0,40}$")
|
||
SEED_PATH = Path(__file__).resolve().parent / "config" / "generation_instructions.seed.json"
|
||
STYLE_CONTEXT_KEYS = (
|
||
"include_core",
|
||
"include_facet",
|
||
"include_traits",
|
||
"include_style_examples",
|
||
)
|
||
EMPTY_STYLE_CONTEXT = {key: False for key in STYLE_CONTEXT_KEYS}
|
||
FULL_STYLE_CONTEXT = {key: True for key in STYLE_CONTEXT_KEYS}
|
||
LEGACY_VOICE_STYLE_CONTEXT = {
|
||
"neutral": dict(EMPTY_STYLE_CONTEXT),
|
||
"light": {**EMPTY_STYLE_CONTEXT, "include_core": True},
|
||
"noticeable": {
|
||
"include_core": True,
|
||
"include_facet": True,
|
||
"include_traits": True,
|
||
"include_style_examples": False,
|
||
},
|
||
"clear": {
|
||
"include_core": True,
|
||
"include_facet": True,
|
||
"include_traits": True,
|
||
"include_style_examples": False,
|
||
},
|
||
"with_examples": dict(FULL_STYLE_CONTEXT),
|
||
"legacy_neutral": dict(FULL_STYLE_CONTEXT),
|
||
"legacy_light": dict(FULL_STYLE_CONTEXT),
|
||
"legacy_noticeable": dict(FULL_STYLE_CONTEXT),
|
||
"legacy_clear": dict(FULL_STYLE_CONTEXT),
|
||
}
|
||
LEGACY_COMPARISON_KEYS = frozenset(
|
||
{
|
||
"legacy_neutral",
|
||
"legacy_light",
|
||
"legacy_noticeable",
|
||
"legacy_clear",
|
||
}
|
||
)
|
||
STYLE_EXAMPLES_STANDING_SENTENCE = (
|
||
"STYLE_EXAMPLES dienen ausschließlich als Stilreferenz; ihre Inhalte gehören nicht zum heutigen Tag. "
|
||
)
|
||
PUBLIC_FIELDS = (
|
||
"id",
|
||
"purpose",
|
||
"slot",
|
||
"guideline_key",
|
||
"label",
|
||
"summary",
|
||
"sort_order",
|
||
"status",
|
||
"revision",
|
||
"cloned_from",
|
||
"is_default",
|
||
"is_system_seed",
|
||
"seed_revision",
|
||
"used_at",
|
||
"created",
|
||
"updated",
|
||
)
|
||
|
||
|
||
class GenerationPolicyError(PolicyError):
|
||
def __init__(
|
||
self,
|
||
message: str,
|
||
*,
|
||
code: str = "invalid_generation_selection",
|
||
status_code: int = 400,
|
||
):
|
||
super().__init__(code, message, status_code)
|
||
|
||
|
||
class CatalogError(GenerationPolicyError):
|
||
def __init__(self, message: str, *, code: str = "generation_policy_invalid", status_code: int = 409):
|
||
super().__init__(message, code=code, status_code=status_code)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CompiledPolicy:
|
||
ids: dict[str, str]
|
||
keys: dict[str, str]
|
||
labels: dict[str, str]
|
||
summaries: dict[str, str]
|
||
revisions: dict[str, int]
|
||
instructions: dict[str, str]
|
||
seed_revision: str
|
||
style_context: dict[str, bool]
|
||
cloned_from: dict[str, str]
|
||
|
||
|
||
def load_seed_document() -> dict:
|
||
return json.loads(SEED_PATH.read_text(encoding="utf-8"))
|
||
|
||
|
||
def _as_bool(raw) -> bool:
|
||
return bool(int(raw)) if not isinstance(raw, bool) else raw
|
||
|
||
|
||
def default_style_context_for_key(guideline_key: str, *, slot: str) -> dict[str, bool]:
|
||
if slot != "voice":
|
||
return {}
|
||
mapped = LEGACY_VOICE_STYLE_CONTEXT.get((guideline_key or "").strip())
|
||
if mapped:
|
||
return dict(mapped)
|
||
return dict(FULL_STYLE_CONTEXT)
|
||
|
||
|
||
def normalize_style_context(raw, *, slot: str, required: bool = False) -> dict[str, bool]:
|
||
if slot != "voice":
|
||
if raw in (None, "", {}, []):
|
||
return {}
|
||
raise CatalogError(
|
||
"Stilkontext gehört nur zur Dimension Persönliche Stimme.",
|
||
code="invalid_generation_guideline",
|
||
status_code=400,
|
||
)
|
||
if raw in (None, "", {}):
|
||
if required:
|
||
raise CatalogError(
|
||
"Die Stilanwendung braucht eine Stilkontext-Konfiguration.",
|
||
code="invalid_generation_guideline",
|
||
status_code=400,
|
||
)
|
||
return dict(EMPTY_STYLE_CONTEXT)
|
||
if not isinstance(raw, dict):
|
||
raise CatalogError(
|
||
"style_context muss ein Objekt mit den vier Stilquellen sein.",
|
||
code="invalid_generation_guideline",
|
||
status_code=400,
|
||
)
|
||
unknown = sorted(set(raw) - set(STYLE_CONTEXT_KEYS))
|
||
if unknown:
|
||
raise CatalogError(
|
||
"Unbekannte Stilquellen: " + ", ".join(unknown),
|
||
code="invalid_generation_guideline",
|
||
status_code=400,
|
||
)
|
||
missing = [key for key in STYLE_CONTEXT_KEYS if key not in raw]
|
||
if missing and required:
|
||
raise CatalogError(
|
||
"style_context ist unvollständig: " + ", ".join(missing),
|
||
code="invalid_generation_guideline",
|
||
status_code=400,
|
||
)
|
||
parsed: dict[str, bool] = dict(EMPTY_STYLE_CONTEXT)
|
||
for key in STYLE_CONTEXT_KEYS:
|
||
if key not in raw:
|
||
continue
|
||
value = raw[key]
|
||
if isinstance(value, bool):
|
||
parsed[key] = value
|
||
elif value in (0, 1):
|
||
parsed[key] = bool(value)
|
||
else:
|
||
raise CatalogError(
|
||
f"{key} muss wahr oder falsch sein, keine Gewichtung.",
|
||
code="invalid_generation_guideline",
|
||
status_code=400,
|
||
)
|
||
return parsed
|
||
|
||
|
||
def parse_style_context_json(raw, *, slot: str, guideline_key: str = "") -> dict[str, bool]:
|
||
if slot != "voice":
|
||
return {}
|
||
text = (raw or "").strip()
|
||
if not text or text in ("{}", "null"):
|
||
return default_style_context_for_key(guideline_key, slot=slot)
|
||
try:
|
||
data = json.loads(text)
|
||
except json.JSONDecodeError:
|
||
return default_style_context_for_key(guideline_key, slot=slot)
|
||
if not isinstance(data, dict) or not any(key in data for key in STYLE_CONTEXT_KEYS):
|
||
return default_style_context_for_key(guideline_key, slot=slot)
|
||
try:
|
||
return normalize_style_context(data, slot=slot, required=False)
|
||
except CatalogError:
|
||
return default_style_context_for_key(guideline_key, slot=slot)
|
||
|
||
|
||
def style_context_json(context: dict[str, bool] | None, *, slot: str) -> str:
|
||
if slot != "voice":
|
||
return "{}"
|
||
payload = normalize_style_context(context or EMPTY_STYLE_CONTEXT, slot=slot, required=False)
|
||
return json.dumps(payload, ensure_ascii=False, sort_keys=True)
|
||
|
||
|
||
def _limits() -> tuple[int, int, int]:
|
||
seed = load_seed_document()
|
||
return (
|
||
int(seed.get("max_instruction_chars") or MAX_INSTRUCTION_CHARS),
|
||
int(seed.get("max_label_chars") or MAX_LABEL_CHARS),
|
||
int(seed.get("max_summary_chars") or MAX_SUMMARY_CHARS),
|
||
)
|
||
|
||
|
||
def _clean_text(raw, field: str, *, max_chars: int, allow_empty: bool = False) -> str:
|
||
text = (raw or "").strip()
|
||
if not text and not allow_empty:
|
||
raise CatalogError(f"{field} darf nicht leer sein.", code="invalid_generation_guideline", status_code=400)
|
||
if len(text) > max_chars:
|
||
raise CatalogError(f"{field} darf höchstens {max_chars} Zeichen haben.", code="invalid_generation_guideline", status_code=400)
|
||
return text
|
||
|
||
|
||
def _clean_key(raw) -> str:
|
||
key = (raw or "").strip()
|
||
if not KEY_RE.match(key):
|
||
raise CatalogError(
|
||
"guideline_key muss aus Kleinbuchstaben, Ziffern und Unterstrich bestehen.",
|
||
code="invalid_generation_guideline",
|
||
status_code=400,
|
||
)
|
||
return key
|
||
|
||
|
||
def _row(item: dict | None, *, include_instruction: bool = False) -> dict | None:
|
||
if not item:
|
||
return None
|
||
payload = {field: item.get(field) for field in PUBLIC_FIELDS}
|
||
payload["is_default"] = _as_bool(item.get("is_default", 0))
|
||
payload["is_system_seed"] = _as_bool(item.get("is_system_seed", 0))
|
||
payload["revision"] = int(item.get("revision") or 1)
|
||
payload["sort_order"] = int(item.get("sort_order") or 0)
|
||
payload["guideline_key"] = item.get("guideline_key") or ""
|
||
payload["style_context"] = parse_style_context_json(
|
||
item.get("style_context_json"),
|
||
slot=item.get("slot") or "",
|
||
guideline_key=payload["guideline_key"],
|
||
)
|
||
if include_instruction:
|
||
payload["instruction"] = item.get("instruction") or ""
|
||
payload["seed_id"] = item.get("seed_id") or ""
|
||
return payload
|
||
|
||
|
||
def _fetch(guideline_id: str) -> dict | None:
|
||
with get_db() as conn:
|
||
return row_to_dict(
|
||
conn.execute("SELECT * FROM generation_guidelines WHERE id = ?", (guideline_id,)).fetchone()
|
||
)
|
||
|
||
|
||
def get_guideline(guideline_id: str, *, include_instruction: bool = True) -> dict:
|
||
row = _fetch(guideline_id)
|
||
if not row:
|
||
raise CatalogError("Ausprägung nicht gefunden.", code="guideline_missing", status_code=404)
|
||
return _row(row, include_instruction=include_instruction)
|
||
|
||
|
||
def list_guidelines(
|
||
purpose: str = PURPOSE_JOURNAL,
|
||
*,
|
||
slot: str | None = None,
|
||
statuses: tuple[str, ...] | None = None,
|
||
include_instruction: bool = False,
|
||
) -> list[dict]:
|
||
query = "SELECT * FROM generation_guidelines WHERE purpose = ?"
|
||
params: list = [purpose]
|
||
if slot:
|
||
query += " AND slot = ?"
|
||
params.append(slot)
|
||
if statuses:
|
||
query += " AND status IN (" + ",".join("?" for _ in statuses) + ")"
|
||
params.extend(statuses)
|
||
query += " ORDER BY slot, sort_order, revision, label"
|
||
with get_db() as conn:
|
||
rows = [row_to_dict(row) for row in conn.execute(query, params).fetchall()]
|
||
return [_row(row, include_instruction=include_instruction) for row in rows]
|
||
|
||
|
||
def overview_payload(purpose: str = PURPOSE_JOURNAL) -> dict:
|
||
seed = load_seed_document()
|
||
items = list_guidelines(purpose)
|
||
slots = {slot: [item for item in items if item["slot"] == slot] for slot in ALLOWED_SLOTS}
|
||
return {
|
||
"purpose": purpose,
|
||
"seed_revision": seed.get("seed_revision") or "",
|
||
"max_instruction_chars": int(seed.get("max_instruction_chars") or MAX_INSTRUCTION_CHARS),
|
||
"max_label_chars": int(seed.get("max_label_chars") or MAX_LABEL_CHARS),
|
||
"max_summary_chars": int(seed.get("max_summary_chars") or MAX_SUMMARY_CHARS),
|
||
"slots": slots,
|
||
}
|
||
|
||
|
||
def active_user_options(purpose: str = PURPOSE_JOURNAL) -> dict[str, list[dict]]:
|
||
items = list_guidelines(purpose, statuses=(STATUS_ACTIVE,))
|
||
return {
|
||
slot: [
|
||
{
|
||
"id": item["id"],
|
||
"key": item["guideline_key"],
|
||
"label": item["label"],
|
||
"summary": item["summary"],
|
||
"revision": item["revision"],
|
||
"is_default": item["is_default"],
|
||
"style_context": item.get("style_context") or {},
|
||
}
|
||
for item in items
|
||
if item["slot"] == slot
|
||
]
|
||
for slot in USER_SLOTS
|
||
}
|
||
|
||
|
||
def default_selection_ids(purpose: str = PURPOSE_JOURNAL) -> dict[str, str]:
|
||
items = list_guidelines(purpose, statuses=(STATUS_ACTIVE,))
|
||
chosen = {}
|
||
for slot in USER_SLOTS:
|
||
slot_items = [item for item in items if item["slot"] == slot]
|
||
if not slot_items:
|
||
raise CatalogError(f"{slot} hat keine aktive Ausprägung.")
|
||
preferred = next((item for item in slot_items if item["is_default"]), slot_items[0])
|
||
chosen[SLOT_TO_ID_KEY[slot]] = preferred["id"]
|
||
return chosen
|
||
|
||
|
||
def _require_active_for_run(row: dict, slot: str) -> dict:
|
||
if not row:
|
||
raise GenerationPolicyError(f"{slot} ist unbekannt.")
|
||
if row.get("slot") != slot:
|
||
raise GenerationPolicyError(f"{slot} verweist auf die falsche Dimension.")
|
||
if row.get("status") != STATUS_ACTIVE:
|
||
raise GenerationPolicyError(
|
||
f"{slot} ist nicht für neue Läufe verfügbar.",
|
||
code="invalid_generation_selection",
|
||
)
|
||
if not (row.get("instruction") or "").strip():
|
||
raise CatalogError(f"{slot} hat keine Anweisung.")
|
||
return row
|
||
|
||
|
||
def validate_selection(raw, *, purpose: str = PURPOSE_JOURNAL) -> dict[str, dict]:
|
||
if not isinstance(raw, dict):
|
||
raise GenerationPolicyError("generation_selection muss die vier Ausprägungs-IDs enthalten.")
|
||
unknown = sorted(set(raw) - set(SELECTION_KEYS))
|
||
if unknown:
|
||
raise GenerationPolicyError("Unbekannte Auswahl: " + ", ".join(unknown))
|
||
missing = [key for key in SELECTION_KEYS if not (raw.get(key) or "").strip()]
|
||
if missing:
|
||
raise GenerationPolicyError("generation_selection ist unvollständig: " + ", ".join(missing))
|
||
chosen = {}
|
||
for key in SELECTION_KEYS:
|
||
slot = ID_KEY_TO_SLOT[key]
|
||
row = _fetch(str(raw[key]).strip())
|
||
chosen[slot] = _require_active_for_run(row, slot)
|
||
if (row.get("purpose") or purpose) != purpose:
|
||
raise GenerationPolicyError(f"{slot} gehört nicht zu diesem Zweck.")
|
||
return chosen
|
||
|
||
|
||
def compile_selection(
|
||
selection: dict[str, str],
|
||
*,
|
||
purpose: str = PURPOSE_JOURNAL,
|
||
) -> CompiledPolicy:
|
||
chosen = validate_selection(selection, purpose=purpose)
|
||
instructions = {
|
||
SLOT_INSTRUCTION_KEYS[slot]: chosen[slot]["instruction"] for slot in USER_SLOTS
|
||
}
|
||
revision = ""
|
||
for row in chosen.values():
|
||
revision = row.get("seed_revision") or revision
|
||
return CompiledPolicy(
|
||
ids={slot: chosen[slot]["id"] for slot in USER_SLOTS},
|
||
keys={slot: chosen[slot]["guideline_key"] for slot in USER_SLOTS},
|
||
labels={slot: chosen[slot]["label"] for slot in USER_SLOTS},
|
||
summaries={slot: chosen[slot].get("summary") or "" for slot in USER_SLOTS},
|
||
revisions={slot: int(chosen[slot].get("revision") or 1) for slot in USER_SLOTS},
|
||
instructions=instructions,
|
||
seed_revision=revision,
|
||
style_context=parse_style_context_json(
|
||
chosen["voice"].get("style_context_json"),
|
||
slot="voice",
|
||
guideline_key=chosen["voice"].get("guideline_key") or "",
|
||
),
|
||
cloned_from={slot: chosen[slot].get("cloned_from") or "" for slot in USER_SLOTS},
|
||
)
|
||
|
||
|
||
def policy_trace(
|
||
compiled: CompiledPolicy,
|
||
*,
|
||
source: str,
|
||
remembered: bool,
|
||
) -> dict:
|
||
return {
|
||
"source": source,
|
||
"remembered": remembered,
|
||
"ids": dict(compiled.ids),
|
||
"keys": dict(compiled.keys),
|
||
"labels": dict(compiled.labels),
|
||
"revisions": dict(compiled.revisions),
|
||
"seed_revision": compiled.seed_revision,
|
||
"style_context": dict(compiled.style_context),
|
||
"cloned_from": dict(compiled.cloned_from),
|
||
}
|
||
|
||
|
||
def draft_snapshot(compiled: CompiledPolicy, *, prompt: dict | None = None, model: str = "") -> dict:
|
||
"""Persistable run snapshot. No prompt bodies or instruction texts."""
|
||
return {
|
||
"transformation": {
|
||
"id": compiled.ids["transformation"],
|
||
"key": compiled.keys["transformation"],
|
||
"label": compiled.labels["transformation"],
|
||
"revision": compiled.revisions["transformation"],
|
||
},
|
||
"detail": {
|
||
"id": compiled.ids["detail"],
|
||
"key": compiled.keys["detail"],
|
||
"label": compiled.labels["detail"],
|
||
"revision": compiled.revisions["detail"],
|
||
},
|
||
"voice": {
|
||
"id": compiled.ids["voice"],
|
||
"key": compiled.keys["voice"],
|
||
"label": compiled.labels["voice"],
|
||
"revision": compiled.revisions["voice"],
|
||
"cloned_from": compiled.cloned_from.get("voice") or "",
|
||
"style_context": dict(compiled.style_context),
|
||
},
|
||
"narrative": {
|
||
"id": compiled.ids["narrative"],
|
||
"key": compiled.keys["narrative"],
|
||
"label": compiled.labels["narrative"],
|
||
"revision": compiled.revisions["narrative"],
|
||
},
|
||
"prompt_slug": (prompt or {}).get("slug") or "",
|
||
"prompt_revision": (prompt or {}).get("seed_revision") or "",
|
||
"model": model or "",
|
||
}
|
||
|
||
|
||
def snapshot_summary(snapshot: dict | None) -> str:
|
||
data = snapshot or {}
|
||
labels = [
|
||
(data.get("transformation") or {}).get("label") or "",
|
||
(data.get("detail") or {}).get("label") or "",
|
||
(data.get("voice") or {}).get("label") or "",
|
||
(data.get("narrative") or {}).get("label") or "",
|
||
]
|
||
labels = [item for item in labels if item]
|
||
if not labels:
|
||
return ""
|
||
line = " · ".join(item[:1].upper() + item[1:] if item else item for item in labels)
|
||
return f"Erzeugt mit:\n{line}"
|
||
|
||
|
||
def assert_journal_prompt_contract(template: str) -> None:
|
||
keys = CONTEXT_PATTERN.findall(template or "")
|
||
retired = [key for key in RETIRED_JOURNAL_PLACEHOLDERS if key in keys]
|
||
if retired:
|
||
shown = ", ".join(f"{{{{{key}}}}}" for key in retired)
|
||
raise CatalogError(
|
||
"Dieser Journal-Prompt verwendet den veralteten Platzhalter "
|
||
+ shown
|
||
+ ". Die Unterscheidung zwischen Fließtext- und Stichpunktmodus ist entfallen. "
|
||
"Bitte den Prompt unter Admin → Prompts aktualisieren.",
|
||
code="prompt_contract_incompatible",
|
||
)
|
||
missing = [key for key in REQUIRED_JOURNAL_PLACEHOLDERS if key not in keys]
|
||
if missing:
|
||
shown = ", ".join(f"{{{{{key}}}}}" for key in missing)
|
||
raise CatalogError(
|
||
"Dieser Journal-Prompt erfüllt den aktuellen Vertrag nicht. Es fehlen: "
|
||
+ shown
|
||
+ ". Bitte den Prompt unter Admin → Prompts aktualisieren.",
|
||
code="prompt_contract_incompatible",
|
||
)
|
||
|
||
|
||
def assert_template_resolved(rendered: str) -> None:
|
||
leftover = CONTEXT_PATTERN.findall(rendered or "")
|
||
if leftover:
|
||
raise CatalogError(
|
||
"Promptplatzhalter konnten nicht aufgelöst werden: "
|
||
+ ", ".join(f"{{{{{key}}}}}" for key in leftover),
|
||
code="unresolved_placeholder",
|
||
)
|
||
|
||
|
||
def omit_orphaned_style_example_rule(template: str) -> str:
|
||
"""Drop the standing STYLE_EXAMPLES sentence when no example block is sent."""
|
||
text = template or ""
|
||
return text.replace(STYLE_EXAMPLES_STANDING_SENTENCE, "").replace(
|
||
"STYLE_EXAMPLES dienen ausschließlich als Stilreferenz; ihre Inhalte gehören nicht zum heutigen Tag.",
|
||
"",
|
||
)
|
||
|
||
|
||
def prompt_without_orphaned_style_example_rule(prompt: dict, *, include_style_examples: bool) -> dict:
|
||
if include_style_examples:
|
||
return prompt
|
||
template = prompt.get("template") or ""
|
||
cleaned = omit_orphaned_style_example_rule(template)
|
||
if cleaned == template:
|
||
return prompt
|
||
return {**prompt, "template": cleaned}
|
||
|
||
|
||
def load_selection(profile_id: str) -> dict[str, str] | None:
|
||
with get_db() as conn:
|
||
row = row_to_dict(
|
||
conn.execute(
|
||
"""
|
||
SELECT transformation_id, detail_id, voice_id, narrative_id, updated
|
||
FROM journal_generation_selection
|
||
WHERE profile_id = ?
|
||
""",
|
||
(profile_id,),
|
||
).fetchone()
|
||
)
|
||
if not row:
|
||
return None
|
||
values = {key: row[key] for key in SELECTION_KEYS}
|
||
values["updated"] = row.get("updated") or ""
|
||
return values
|
||
|
||
|
||
def save_selection(profile_id: str, selection: dict[str, str]) -> dict[str, str]:
|
||
chosen = validate_selection(selection)
|
||
checked = {SLOT_TO_ID_KEY[slot]: chosen[slot]["id"] for slot in USER_SLOTS}
|
||
with get_db() as conn:
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO journal_generation_selection (
|
||
profile_id, transformation_id, detail_id, voice_id, narrative_id, updated
|
||
)
|
||
VALUES (?, ?, ?, ?, ?, datetime('now'))
|
||
ON CONFLICT(profile_id) DO UPDATE SET
|
||
transformation_id = excluded.transformation_id,
|
||
detail_id = excluded.detail_id,
|
||
voice_id = excluded.voice_id,
|
||
narrative_id = excluded.narrative_id,
|
||
updated = datetime('now')
|
||
""",
|
||
(
|
||
profile_id,
|
||
checked["transformation_id"],
|
||
checked["detail_id"],
|
||
checked["voice_id"],
|
||
checked["narrative_id"],
|
||
),
|
||
)
|
||
stored = load_selection(profile_id) or {**checked, "updated": ""}
|
||
return stored
|
||
|
||
|
||
def get_or_create_selection(profile_id: str) -> dict[str, str]:
|
||
existing = load_selection(profile_id)
|
||
if existing is not None:
|
||
return existing
|
||
return save_selection(profile_id, default_selection_ids())
|
||
|
||
|
||
def settings_payload(profile_id: str) -> dict:
|
||
stored = get_or_create_selection(profile_id)
|
||
selection = {key: stored[key] for key in SELECTION_KEYS}
|
||
return {
|
||
"selection": selection,
|
||
"updated": stored.get("updated") or "",
|
||
"options": active_user_options(),
|
||
"defaults": default_selection_ids(),
|
||
}
|
||
|
||
|
||
def resolve_run_selection(
|
||
profile_id: str,
|
||
snapshot: dict | None,
|
||
remember: bool,
|
||
) -> tuple[dict[str, str], dict]:
|
||
if snapshot is None:
|
||
stored = get_or_create_selection(profile_id)
|
||
values = {key: stored[key] for key in SELECTION_KEYS}
|
||
validate_selection(values)
|
||
return values, {"source": "profile", "remembered": False}
|
||
values = {key: str(snapshot.get(key) or "").strip() for key in SELECTION_KEYS}
|
||
validate_selection(values)
|
||
remembered = False
|
||
if remember:
|
||
save_selection(profile_id, values)
|
||
remembered = True
|
||
return values, {"source": "request", "remembered": remembered}
|
||
|
||
|
||
def mark_guidelines_used(ids: list[str]) -> None:
|
||
clean = [item for item in ids if item]
|
||
if not clean:
|
||
return
|
||
with get_db() as conn:
|
||
conn.executemany(
|
||
"""
|
||
UPDATE generation_guidelines
|
||
SET used_at = COALESCE(used_at, datetime('now'))
|
||
WHERE id = ?
|
||
""",
|
||
[(item,) for item in clean],
|
||
)
|
||
|
||
|
||
def _write_fields(raw: dict, *, slot: str) -> dict:
|
||
max_instruction, max_label, max_summary = _limits()
|
||
style_context = normalize_style_context(
|
||
raw.get("style_context"),
|
||
slot=slot,
|
||
required=False,
|
||
)
|
||
return {
|
||
"guideline_key": _clean_key(raw.get("guideline_key") or raw.get("key")),
|
||
"label": _clean_text(raw.get("label"), "label", max_chars=max_label),
|
||
"summary": _clean_text(raw.get("summary"), "summary", max_chars=max_summary, allow_empty=True),
|
||
"instruction": _clean_text(raw.get("instruction"), "instruction", max_chars=max_instruction),
|
||
"sort_order": int(raw.get("sort_order") or 0),
|
||
"style_context": style_context,
|
||
"style_context_json": style_context_json(style_context, slot=slot),
|
||
}
|
||
|
||
|
||
def create_guideline(purpose: str, slot: str, body: dict) -> dict:
|
||
if slot not in ALLOWED_SLOTS:
|
||
raise CatalogError("Unbekannter Slot.", code="invalid_generation_guideline", status_code=400)
|
||
fields = _write_fields(body, slot=slot)
|
||
guideline_id = str(uuid.uuid4())
|
||
with get_db() as conn:
|
||
count = conn.execute(
|
||
"SELECT COALESCE(MAX(sort_order), -1) AS n FROM generation_guidelines WHERE purpose = ? AND slot = ?",
|
||
(purpose, slot),
|
||
).fetchone()["n"]
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO generation_guidelines (
|
||
id, purpose, slot, guideline_key, label, summary, instruction, style_context_json, sort_order,
|
||
status, revision, cloned_from, is_default, is_system_seed, seed_id, seed_revision,
|
||
created, updated
|
||
)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, NULL, 0, 0, '', '', datetime('now'), datetime('now'))
|
||
""",
|
||
(
|
||
guideline_id,
|
||
purpose,
|
||
slot,
|
||
fields["guideline_key"],
|
||
fields["label"],
|
||
fields["summary"],
|
||
fields["instruction"],
|
||
fields["style_context_json"],
|
||
fields["sort_order"] if body.get("sort_order") is not None else int(count) + 1,
|
||
STATUS_DRAFT,
|
||
),
|
||
)
|
||
return get_guideline(guideline_id)
|
||
|
||
|
||
def clone_guideline(guideline_id: str) -> dict:
|
||
current = _fetch(guideline_id)
|
||
if not current:
|
||
raise CatalogError("Ausprägung nicht gefunden.", code="guideline_missing", status_code=404)
|
||
if current.get("slot") == LEGACY_SOURCE_MODE_SLOT:
|
||
raise CatalogError(
|
||
"Der Quellenmodus ist kein aktiver Slot mehr.",
|
||
code="invalid_generation_guideline",
|
||
status_code=400,
|
||
)
|
||
new_id = str(uuid.uuid4())
|
||
with get_db() as conn:
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO generation_guidelines (
|
||
id, purpose, slot, guideline_key, label, summary, instruction, style_context_json, sort_order,
|
||
status, revision, cloned_from, is_default, is_system_seed, seed_id, seed_revision,
|
||
created, updated
|
||
)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0, '', ?, datetime('now'), datetime('now'))
|
||
""",
|
||
(
|
||
new_id,
|
||
current["purpose"],
|
||
current["slot"],
|
||
current["guideline_key"],
|
||
current["label"],
|
||
current["summary"],
|
||
current["instruction"],
|
||
current.get("style_context_json") or style_context_json(
|
||
parse_style_context_json(
|
||
current.get("style_context_json"),
|
||
slot=current.get("slot") or "",
|
||
guideline_key=current.get("guideline_key") or "",
|
||
),
|
||
slot=current.get("slot") or "",
|
||
),
|
||
int(current.get("sort_order") or 0),
|
||
STATUS_DRAFT,
|
||
int(current.get("revision") or 1) + 1,
|
||
current["id"],
|
||
current.get("seed_revision") or "",
|
||
),
|
||
)
|
||
return get_guideline(new_id)
|
||
|
||
|
||
def update_guideline(guideline_id: str, body: dict) -> dict:
|
||
current = _fetch(guideline_id)
|
||
if not current:
|
||
raise CatalogError("Ausprägung nicht gefunden.", code="guideline_missing", status_code=404)
|
||
if current.get("status") != STATUS_DRAFT:
|
||
raise CatalogError(
|
||
"Aktive oder archivierte Ausprägungen sind unveränderlich. Bitte klonen.",
|
||
code="guideline_immutable",
|
||
status_code=409,
|
||
)
|
||
merged = {**current, **body, "guideline_key": body.get("guideline_key") or current["guideline_key"]}
|
||
if "style_context" not in body:
|
||
merged["style_context"] = parse_style_context_json(
|
||
current.get("style_context_json"),
|
||
slot=current.get("slot") or "",
|
||
guideline_key=current.get("guideline_key") or "",
|
||
)
|
||
fields = _write_fields(merged, slot=current.get("slot") or "")
|
||
with get_db() as conn:
|
||
conn.execute(
|
||
"""
|
||
UPDATE generation_guidelines
|
||
SET guideline_key = ?, label = ?, summary = ?, instruction = ?, style_context_json = ?,
|
||
sort_order = ?, updated = datetime('now')
|
||
WHERE id = ?
|
||
""",
|
||
(
|
||
fields["guideline_key"],
|
||
fields["label"],
|
||
fields["summary"],
|
||
fields["instruction"],
|
||
fields["style_context_json"],
|
||
fields["sort_order"],
|
||
guideline_id,
|
||
),
|
||
)
|
||
return get_guideline(guideline_id)
|
||
|
||
|
||
def publish_guideline(guideline_id: str) -> dict:
|
||
current = _fetch(guideline_id)
|
||
if not current:
|
||
raise CatalogError("Ausprägung nicht gefunden.", code="guideline_missing", status_code=404)
|
||
if current.get("status") == STATUS_ARCHIVED:
|
||
raise CatalogError("Archivierte Ausprägungen können nicht veröffentlicht werden.", code="guideline_archived", status_code=409)
|
||
max_instruction, max_label, max_summary = _limits()
|
||
_clean_text(current.get("label"), "label", max_chars=max_label)
|
||
_clean_text(current.get("summary"), "summary", max_chars=max_summary, allow_empty=True)
|
||
_clean_text(current.get("instruction"), "instruction", max_chars=max_instruction)
|
||
if current.get("slot") == "voice":
|
||
normalize_style_context(
|
||
parse_style_context_json(
|
||
current.get("style_context_json"),
|
||
slot="voice",
|
||
guideline_key=current.get("guideline_key") or "",
|
||
),
|
||
slot="voice",
|
||
required=True,
|
||
)
|
||
with get_db() as conn:
|
||
conn.execute(
|
||
"UPDATE generation_guidelines SET status = ?, updated = datetime('now') WHERE id = ?",
|
||
(STATUS_ACTIVE, guideline_id),
|
||
)
|
||
return get_guideline(guideline_id, include_instruction=False)
|
||
|
||
|
||
def archive_guideline(guideline_id: str) -> dict:
|
||
current = _fetch(guideline_id)
|
||
if not current:
|
||
raise CatalogError("Ausprägung nicht gefunden.", code="guideline_missing", status_code=404)
|
||
if current.get("status") == STATUS_DRAFT:
|
||
raise CatalogError("Drafts werden gelöscht, nicht archiviert.", code="invalid_generation_guideline", status_code=400)
|
||
with get_db() as conn:
|
||
remaining = conn.execute(
|
||
"""
|
||
SELECT COUNT(*) AS n FROM generation_guidelines
|
||
WHERE purpose = ? AND slot = ? AND status = ? AND id != ?
|
||
""",
|
||
(current["purpose"], current["slot"], STATUS_ACTIVE, guideline_id),
|
||
).fetchone()["n"]
|
||
if int(remaining or 0) < 1:
|
||
raise CatalogError(
|
||
"Die letzte aktive Ausprägung dieser Dimension kann nicht archiviert werden.",
|
||
code="guideline_last_active",
|
||
status_code=409,
|
||
)
|
||
with get_db() as conn:
|
||
conn.execute(
|
||
"""
|
||
UPDATE generation_guidelines
|
||
SET status = ?, is_default = 0, updated = datetime('now')
|
||
WHERE id = ?
|
||
""",
|
||
(STATUS_ARCHIVED, guideline_id),
|
||
)
|
||
return get_guideline(guideline_id, include_instruction=False)
|
||
|
||
|
||
def set_default_guideline(guideline_id: str) -> dict:
|
||
current = _fetch(guideline_id)
|
||
if not current:
|
||
raise CatalogError("Ausprägung nicht gefunden.", code="guideline_missing", status_code=404)
|
||
if current.get("status") != STATUS_ACTIVE:
|
||
raise CatalogError("Nur aktive Ausprägungen können Standard sein.", code="invalid_generation_guideline", status_code=400)
|
||
if current["slot"] not in USER_SLOTS:
|
||
raise CatalogError("Nur die vier Gestaltungsdimensionen haben einen Nutzerstandard.", code="invalid_generation_guideline", status_code=400)
|
||
with get_db() as conn:
|
||
conn.execute(
|
||
"""
|
||
UPDATE generation_guidelines
|
||
SET is_default = CASE WHEN id = ? THEN 1 ELSE 0 END, updated = datetime('now')
|
||
WHERE purpose = ? AND slot = ?
|
||
""",
|
||
(guideline_id, current["purpose"], current["slot"]),
|
||
)
|
||
return get_guideline(guideline_id, include_instruction=False)
|
||
|
||
|
||
def delete_guideline(guideline_id: str) -> None:
|
||
current = _fetch(guideline_id)
|
||
if not current:
|
||
raise CatalogError("Ausprägung nicht gefunden.", code="guideline_missing", status_code=404)
|
||
if current.get("status") != STATUS_DRAFT:
|
||
raise CatalogError("Nur unverwendete Drafts können gelöscht werden.", code="guideline_immutable", status_code=409)
|
||
if current.get("used_at"):
|
||
raise CatalogError("Verwendete Drafts können nicht gelöscht werden.", code="guideline_in_use", status_code=409)
|
||
with get_db() as conn:
|
||
conn.execute("DELETE FROM generation_guidelines WHERE id = ?", (guideline_id,))
|
||
|
||
|
||
def reset_seed_drafts(purpose: str = PURPOSE_JOURNAL) -> dict:
|
||
"""Create new drafts from the current seed. Never overwrite published variants."""
|
||
seed = load_seed_document()
|
||
created = []
|
||
slots = seed.get("slots") or {}
|
||
for slot in ALLOWED_SLOTS:
|
||
for index, item in enumerate((slots.get(slot) or {}).get("variants") or []):
|
||
created.append(
|
||
create_guideline(
|
||
purpose,
|
||
slot,
|
||
{
|
||
"guideline_key": item.get("guideline_key") or item.get("variant_key"),
|
||
"label": item.get("label") or "",
|
||
"summary": item.get("summary") or "",
|
||
"instruction": item.get("instruction") or "",
|
||
"sort_order": index,
|
||
"style_context": item.get("style_context") or {},
|
||
},
|
||
)
|
||
)
|
||
overview = overview_payload(purpose)
|
||
overview["reset_drafts"] = created
|
||
return overview
|
||
|
||
|
||
def preview_selection(selection: dict[str, str], *, purpose: str = PURPOSE_JOURNAL) -> dict:
|
||
from engine import load_active_prompt, preview_prompt
|
||
|
||
compiled = compile_selection(selection, purpose=purpose)
|
||
prompt = load_active_prompt("mvp.journal_generate")
|
||
assert_journal_prompt_contract(prompt.get("template") or "")
|
||
include_examples = bool(compiled.style_context.get("include_style_examples"))
|
||
prompt = prompt_without_orphaned_style_example_rule(
|
||
prompt,
|
||
include_style_examples=include_examples,
|
||
)
|
||
rendered = preview_prompt(
|
||
prompt,
|
||
{
|
||
**compiled.instructions,
|
||
"writing_profile": "(nicht enthalten)",
|
||
"style_examples": "(Stilbeispiele)" if include_examples else "",
|
||
"reconstruction": "(nicht enthalten)",
|
||
"existing_text": "",
|
||
},
|
||
)
|
||
assert_template_resolved(rendered.get("rendered") or "")
|
||
return {
|
||
**compiled.instructions,
|
||
"rendered": rendered.get("rendered") or "",
|
||
"prompt_slug": prompt.get("slug") or "",
|
||
"prompt_revision": prompt.get("seed_revision") or "",
|
||
"selection": {
|
||
"ids": dict(compiled.ids),
|
||
"keys": dict(compiled.keys),
|
||
"labels": dict(compiled.labels),
|
||
"revisions": dict(compiled.revisions),
|
||
"cloned_from": dict(compiled.cloned_from),
|
||
"style_context": dict(compiled.style_context),
|
||
},
|
||
"seed_revision": compiled.seed_revision,
|
||
}
|
||
|
||
|
||
def _seed_rows(seed: dict, purpose: str) -> list[tuple]:
|
||
revision = seed.get("seed_revision") or ""
|
||
rows = []
|
||
slots = seed.get("slots") or {}
|
||
for slot in ALLOWED_SLOTS:
|
||
variants = (slots.get(slot) or {}).get("variants") or []
|
||
for index, item in enumerate(variants):
|
||
context = normalize_style_context(item.get("style_context") or {}, slot=slot, required=False)
|
||
rows.append(
|
||
(
|
||
item.get("id") or str(uuid.uuid4()),
|
||
purpose,
|
||
slot,
|
||
item.get("guideline_key") or item.get("variant_key"),
|
||
item.get("label") or "",
|
||
item.get("summary") or "",
|
||
item.get("instruction") or "",
|
||
style_context_json(context, slot=slot),
|
||
index,
|
||
STATUS_ACTIVE,
|
||
1,
|
||
1 if item.get("is_default") else 0,
|
||
1,
|
||
item.get("id") or "",
|
||
revision,
|
||
)
|
||
)
|
||
return rows
|
||
|
||
|
||
def _semantic_signature(item: dict | None = None, *, seed_row: tuple | None = None) -> tuple:
|
||
if seed_row is not None:
|
||
slot = seed_row[2] or ""
|
||
key = seed_row[3] or ""
|
||
return (
|
||
key,
|
||
seed_row[4] or "",
|
||
seed_row[5] or "",
|
||
seed_row[6] or "",
|
||
style_context_json(
|
||
parse_style_context_json(seed_row[7], slot=slot, guideline_key=key),
|
||
slot=slot,
|
||
),
|
||
)
|
||
slot = (item or {}).get("slot") or ""
|
||
key = (item or {}).get("guideline_key") or ""
|
||
return (
|
||
key,
|
||
(item or {}).get("label") or "",
|
||
(item or {}).get("summary") or "",
|
||
(item or {}).get("instruction") or "",
|
||
style_context_json(
|
||
parse_style_context_json((item or {}).get("style_context_json"), slot=slot, guideline_key=key),
|
||
slot=slot,
|
||
),
|
||
)
|
||
|
||
|
||
def _successor_id(base_id: str, revision: int) -> str:
|
||
root = re.sub(r"-r\d+$", "", base_id or "")
|
||
return f"{root}-r{int(revision)}"
|
||
|
||
|
||
def _catalog_maps(conn, purpose: str) -> tuple[dict, dict]:
|
||
rows = [row_to_dict(row) for row in conn.execute(
|
||
"SELECT * FROM generation_guidelines WHERE purpose = ?",
|
||
(purpose,),
|
||
).fetchall()]
|
||
by_id = {row["id"]: row for row in rows}
|
||
by_seed_id = {row["seed_id"]: row for row in rows if row.get("seed_id")}
|
||
return by_id, by_seed_id
|
||
|
||
|
||
def _find_existing_seed_row(by_id: dict, by_seed_id: dict, seed_row: tuple, all_rows: list[dict]) -> dict | None:
|
||
current = by_id.get(seed_row[0]) or by_seed_id.get(seed_row[13])
|
||
if current:
|
||
return current
|
||
key = seed_row[3] or ""
|
||
slot = seed_row[2] or ""
|
||
if key not in LEGACY_COMPARISON_KEYS:
|
||
return None
|
||
for item in all_rows:
|
||
if item.get("slot") == slot and item.get("guideline_key") == key:
|
||
return item
|
||
return None
|
||
|
||
|
||
def _find_matching_successor(all_rows: list[dict], predecessor_id: str, seed_row: tuple) -> dict | None:
|
||
wanted = _semantic_signature(seed_row=seed_row)
|
||
for item in all_rows:
|
||
if (item.get("cloned_from") or "") != predecessor_id:
|
||
continue
|
||
if _semantic_signature(item) == wanted:
|
||
return item
|
||
return None
|
||
|
||
|
||
def _insert_seed_guideline(
|
||
conn,
|
||
seed_row: tuple,
|
||
*,
|
||
guideline_id: str | None = None,
|
||
cloned_from: str | None = None,
|
||
revision: int | None = None,
|
||
is_default: int | None = None,
|
||
) -> None:
|
||
row_id = guideline_id or seed_row[0]
|
||
seed_id = row_id if guideline_id else seed_row[13]
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO generation_guidelines (
|
||
id, purpose, slot, guideline_key, label, summary, instruction, style_context_json, sort_order,
|
||
status, revision, cloned_from, is_default, is_system_seed, seed_id, seed_revision,
|
||
created, updated
|
||
)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'), datetime('now'))
|
||
""",
|
||
(
|
||
row_id,
|
||
seed_row[1],
|
||
seed_row[2],
|
||
seed_row[3],
|
||
seed_row[4],
|
||
seed_row[5],
|
||
seed_row[6],
|
||
seed_row[7],
|
||
seed_row[8],
|
||
STATUS_ACTIVE,
|
||
int(revision if revision is not None else seed_row[10] or 1),
|
||
cloned_from,
|
||
int(is_default if is_default is not None else seed_row[11] or 0),
|
||
1,
|
||
seed_id,
|
||
seed_row[14],
|
||
),
|
||
)
|
||
|
||
|
||
def _assign_slot_default(conn, purpose: str, slot: str, guideline_id: str) -> None:
|
||
conn.execute(
|
||
"""
|
||
UPDATE generation_guidelines
|
||
SET is_default = CASE WHEN id = ? THEN 1 ELSE 0 END, updated = datetime('now')
|
||
WHERE purpose = ? AND slot = ? AND status = ?
|
||
""",
|
||
(guideline_id, purpose, slot, STATUS_ACTIVE),
|
||
)
|
||
|
||
|
||
def seed_generation_instructions(conn, seed: dict | None = None) -> None:
|
||
"""Insert missing seed guidelines. Never rewrite existing IDs, clones, or archives.
|
||
|
||
A semantic change in the seed creates a new successor ID. New defaults may
|
||
point at that successor. Stored selections keep their previous IDs.
|
||
"""
|
||
seed = seed if seed is not None else load_seed_document()
|
||
purpose = seed.get("purpose") or PURPOSE_JOURNAL
|
||
conn.execute(
|
||
"""
|
||
UPDATE generation_guidelines
|
||
SET status = ?, is_default = 0, updated = datetime('now')
|
||
WHERE purpose = ? AND slot = ? AND status != ?
|
||
""",
|
||
(STATUS_ARCHIVED, purpose, LEGACY_SOURCE_MODE_SLOT, STATUS_ARCHIVED),
|
||
)
|
||
for row in _seed_rows(seed, purpose):
|
||
by_id, by_seed_id = _catalog_maps(conn, purpose)
|
||
all_rows = list(by_id.values())
|
||
current = _find_existing_seed_row(by_id, by_seed_id, row, all_rows)
|
||
if current:
|
||
if not _as_bool(current.get("is_system_seed", 0)):
|
||
continue
|
||
if current.get("status") != STATUS_ACTIVE:
|
||
continue
|
||
if _semantic_signature(current) == _semantic_signature(seed_row=row):
|
||
continue
|
||
successor = _find_matching_successor(all_rows, current["id"], row)
|
||
if successor:
|
||
if row[11] and successor.get("status") == STATUS_ACTIVE:
|
||
_assign_slot_default(conn, purpose, row[2], successor["id"])
|
||
continue
|
||
next_revision = int(current.get("revision") or 1) + 1
|
||
taken = set(by_id)
|
||
successor_id = _successor_id(current["id"], next_revision)
|
||
while successor_id in taken:
|
||
next_revision += 1
|
||
successor_id = _successor_id(current["id"], next_revision)
|
||
_insert_seed_guideline(
|
||
conn,
|
||
row,
|
||
guideline_id=successor_id,
|
||
cloned_from=current["id"],
|
||
revision=next_revision,
|
||
is_default=int(row[11] or 0),
|
||
)
|
||
if row[11]:
|
||
_assign_slot_default(conn, purpose, row[2], successor_id)
|
||
continue
|
||
_insert_seed_guideline(conn, row)
|
||
if row[11]:
|
||
_assign_slot_default(conn, purpose, row[2], row[0])
|
||
|
||
|
||
def backfill_missing_settings(conn) -> None:
|
||
seed_generation_instructions(conn)
|
||
defaults = {}
|
||
for slot, key in SLOT_TO_ID_KEY.items():
|
||
row = conn.execute(
|
||
"""
|
||
SELECT id FROM generation_guidelines
|
||
WHERE purpose = ? AND slot = ? AND status = ? AND is_default = 1
|
||
ORDER BY sort_order
|
||
LIMIT 1
|
||
""",
|
||
(PURPOSE_JOURNAL, slot, STATUS_ACTIVE),
|
||
).fetchone()
|
||
if not row:
|
||
row = conn.execute(
|
||
"""
|
||
SELECT id FROM generation_guidelines
|
||
WHERE purpose = ? AND slot = ? AND status = ?
|
||
ORDER BY sort_order
|
||
LIMIT 1
|
||
""",
|
||
(PURPOSE_JOURNAL, slot, STATUS_ACTIVE),
|
||
).fetchone()
|
||
if row:
|
||
defaults[key] = row["id"]
|
||
if len(defaults) != 4:
|
||
return
|
||
conn.execute(
|
||
"""
|
||
INSERT OR IGNORE INTO journal_generation_selection (
|
||
profile_id, transformation_id, detail_id, voice_id, narrative_id, updated
|
||
)
|
||
SELECT id, ?, ?, ?, ?, datetime('now') FROM profiles
|
||
""",
|
||
(
|
||
defaults["transformation_id"],
|
||
defaults["detail_id"],
|
||
defaults["voice_id"],
|
||
defaults["narrative_id"],
|
||
),
|
||
)
|