diff --git a/.env.example b/.env.example index e89ddad..f4b3218 100644 --- a/.env.example +++ b/.env.example @@ -4,6 +4,7 @@ MKA_FFMPEG_EXECUTABLE=ffmpeg MKA_PROTOCOL_MODEL=qwen3.8:27b MKA_OLLAMA_ENDPOINT=http://127.0.0.1:11434 MKA_DATA_ROOT=data/meetings +MKA_GLOSSARY_DATABASE=data/database/glossary.sqlite3 MKA_WHISPER_THREADS=auto MKA_DIARIZATION_MODE=auto MKA_DIARIZATION_RUNTIME=native diff --git a/PROJECT_KNOWLEDGE.md b/PROJECT_KNOWLEDGE.md index 1e32d92..df8ca65 100644 --- a/PROJECT_KNOWLEDGE.md +++ b/PROJECT_KNOWLEDGE.md @@ -37,9 +37,17 @@ Meeting Assistant owns user interaction and product workflow: - structured Meeting Context editing - participant management - versioned YAML import/export for reusable People lists, using replace semantics +- versioned JSON import/export for the complete user-configurable run-input + form; source media is represented only by optional filename metadata and must + be selected again after import - optional explicit speaker mapping +- post-diarization speaker review and protocol-only regeneration from existing + run artifacts - pipeline launch and progress display - protocol review and editing +- a global SQLite terminology glossary whose active canonical core terms and + recognition aliases are rendered through Meeting Context into direct + protocol prompts - export and presentation of protocol versions Processing logic must not be duplicated in the application. diff --git a/README.md b/README.md index 54e1cf4..f7fa2f7 100644 --- a/README.md +++ b/README.md @@ -42,6 +42,12 @@ compatibility path. Diarization is optional and produces anonymous speaker labels. A label identifies a participant only when the user explicitly confirms the mapping; automatic speaker-name inference is not allowed. +After a diarized run, the result view lists detected `SPEAKER_XX` labels with +short transcript excerpts. Confirmed mappings regenerate only the protocol +from the existing diarized transcript; audio preparation, Whisper and Pyannote +are not rerun. `Unmapped / Unknown` remains valid, and the original anonymous +diarized transcript is preserved. + ## Product Outputs The product direction includes: @@ -108,6 +114,41 @@ Optional machine-specific settings include: - `MKA_DIARIZATION_CONTAINER_IMAGE` (required for container diarization) - `MKA_DIARIZATION_CONTAINER_ARGS` (default: no extra arguments), encoded as a JSON array of strings so ordering and leading dashes are preserved exactly +- `MKA_GLOSSARY_DATABASE` (default: `data/database/glossary.sqlite3`), the local + SQLite file used by the global terminology glossary + +## Terminology glossary + +The Streamlit **Terminology glossary** section manages recurring product, +material, organization, acronym, and technical names. Store canonical core +terms such as `Secugrid HS`, not every compound such as `Secugrid HS Düse`. +Aliases help the protocol model recognize transcript variants while retaining +the surrounding wording. Only active entries are added to Meeting Context as +authoritative terminology for direct protocol generation; inactive entries +remain stored but are omitted. The production database starts empty. + +The database defaults to `data/database/glossary.sqlite3`. It is created and +bootstrapped automatically and can be moved with `MKA_GLOSSARY_DATABASE`. +Glossary integration does not rewrite raw or diarized transcript artifacts. + +## Input configuration import and export + +Use **Export inputs** to save the current Meeting Assistant run-input form as a +small, versioned JSON file and **Import inputs** to restore it later. The file +contains meeting metadata, participant records, language, date, audio +normalization, and diarization choices. It contains form configuration only: +generated prompts, protocols, run artifacts, and source-media contents are not +included. + +The original source filename may be retained as a reminder, but importing does +not restore an upload. Select the audio file explicitly before starting the new +run. + +Protocol generation with `qwen3.8:27b` explicitly requests a 32,768-token +Ollama context with thinking disabled; Ollama's machine default may otherwise +be only 4,096. Keep normal prompt input at approximately 29,000 tokens or less. +Although a 31,038-token synthetic prompt passed, larger input is not assumed +safe merely because the model advertises a 262,144-token native context. For example, a compatible AMD ROCm workstation can configure validated container access without adding controls to the Streamlit UI: diff --git a/src/mka/application/config.py b/src/mka/application/config.py index 2e8a094..631935a 100644 --- a/src/mka/application/config.py +++ b/src/mka/application/config.py @@ -18,6 +18,7 @@ class AppSettings: data_root: Path whisper_model: Path | None + glossary_database: Path = Path("data/database/glossary.sqlite3") whisper_executable: str = "whisper-cli" ffmpeg_executable: str = "ffmpeg" protocol_model: str = "qwen3.8:27b" @@ -40,9 +41,12 @@ class AppSettings: return cls( data_root=Path(os.getenv("MKA_DATA_ROOT", "data/meetings")), whisper_model=Path(whisper_model) if whisper_model else None, + glossary_database=Path( + os.getenv("MKA_GLOSSARY_DATABASE", "data/database/glossary.sqlite3") + ), whisper_executable=os.getenv("MKA_WHISPER_EXECUTABLE", "whisper-cli"), ffmpeg_executable=os.getenv("MKA_FFMPEG_EXECUTABLE", "ffmpeg"), - protocol_model=os.getenv("MKA_PROTOCOL_MODEL", "qwen3.8:27B"), + protocol_model=os.getenv("MKA_PROTOCOL_MODEL", "qwen3.8:27b"), ollama_endpoint=os.getenv("MKA_OLLAMA_ENDPOINT", "http://127.0.0.1:11434"), language=os.getenv("MKA_LANGUAGE", "de"), threads=parsed_threads, diff --git a/src/mka/application/glossary.py b/src/mka/application/glossary.py new file mode 100644 index 0000000..efd5f6b --- /dev/null +++ b/src/mka/application/glossary.py @@ -0,0 +1,294 @@ +"""SQLite-backed global terminology glossary.""" + +from __future__ import annotations + +import sqlite3 +from collections.abc import Iterable, Iterator +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path + +GLOSSARY_CATEGORIES = ( + "product", + "material", + "organization", + "technical_term", + "acronym", + "other", +) + + +class GlossaryConflictError(ValueError): + """Raised when a canonical term or alias conflicts with existing terminology.""" + + +@dataclass(frozen=True) +class GlossaryEntry: + id: int + canonical_term: str + category: str + description: str | None + is_active: bool + aliases: tuple[str, ...] + created_at: str + updated_at: str + + +class GlossaryRepository: + """Small data-access boundary for the local glossary database.""" + + def __init__(self, database_path: Path) -> None: + self.database_path = Path(database_path) + + def initialize(self) -> None: + """Create the database and current schema when absent.""" + self.database_path.parent.mkdir(parents=True, exist_ok=True) + with self._connect() as connection: + connection.executescript( + """ + CREATE TABLE IF NOT EXISTS glossary_entries ( + id INTEGER PRIMARY KEY, + canonical_term TEXT NOT NULL COLLATE NOCASE UNIQUE, + category TEXT NOT NULL CHECK (category IN ( + 'product', 'material', 'organization', + 'technical_term', 'acronym', 'other' + )), + description TEXT, + is_active INTEGER NOT NULL DEFAULT 1 CHECK (is_active IN (0, 1)), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS glossary_aliases ( + id INTEGER PRIMARY KEY, + entry_id INTEGER NOT NULL REFERENCES glossary_entries(id) + ON DELETE CASCADE, + alias TEXT NOT NULL COLLATE NOCASE UNIQUE, + created_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_glossary_aliases_entry_id + ON glossary_aliases(entry_id); + PRAGMA user_version = 1; + """ + ) + + def create( + self, + canonical_term: str, + category: str, + *, + aliases: Iterable[str] = (), + description: str | None = None, + is_active: bool = True, + ) -> GlossaryEntry: + canonical, normalized_aliases = self._validate_values(canonical_term, category, aliases) + now = _timestamp() + try: + with self._connect() as connection: + self._ensure_terms_available(connection, canonical, normalized_aliases) + cursor = connection.execute( + """INSERT INTO glossary_entries + (canonical_term, category, description, is_active, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?)""", + (canonical, category, _optional_text(description), is_active, now, now), + ) + entry_id = int(cursor.lastrowid) + connection.executemany( + "INSERT INTO glossary_aliases (entry_id, alias, created_at) VALUES (?, ?, ?)", + ((entry_id, alias, now) for alias in normalized_aliases), + ) + except sqlite3.IntegrityError as exc: + raise GlossaryConflictError("Canonical term or alias already exists.") from exc + return self.get(entry_id) + + def get(self, entry_id: int) -> GlossaryEntry: + with self._connect() as connection: + row = connection.execute( + "SELECT * FROM glossary_entries WHERE id = ?", (entry_id,) + ).fetchone() + if row is None: + raise KeyError(f"Unknown glossary entry: {entry_id}") + return self._to_entry(connection, row) + + def list(self, search: str = "", *, active_only: bool = False) -> list[GlossaryEntry]: + clauses: list[str] = [] + parameters: list[object] = [] + if active_only: + clauses.append("entry.is_active = 1") + if search.strip(): + clauses.append( + "(entry.canonical_term LIKE ? COLLATE NOCASE " + "OR entry.category LIKE ? COLLATE NOCASE " + "OR entry.description LIKE ? COLLATE NOCASE " + "OR alias.alias LIKE ? COLLATE NOCASE)" + ) + pattern = f"%{search.strip()}%" + parameters.extend([pattern] * 4) + where = f"WHERE {' AND '.join(clauses)}" if clauses else "" + with self._connect() as connection: + rows = connection.execute( + f"""SELECT DISTINCT entry.* FROM glossary_entries AS entry + LEFT JOIN glossary_aliases AS alias ON alias.entry_id = entry.id + {where} ORDER BY entry.canonical_term COLLATE NOCASE""", # noqa: S608 + parameters, + ).fetchall() + return [self._to_entry(connection, row) for row in rows] + + def update( + self, + entry_id: int, + canonical_term: str, + category: str, + *, + aliases: Iterable[str] = (), + description: str | None = None, + is_active: bool = True, + ) -> GlossaryEntry: + canonical, normalized_aliases = self._validate_values(canonical_term, category, aliases) + try: + with self._connect() as connection: + exists = connection.execute( + "SELECT 1 FROM glossary_entries WHERE id = ?", (entry_id,) + ).fetchone() + if exists is None: + raise KeyError(f"Unknown glossary entry: {entry_id}") + self._ensure_terms_available( + connection, canonical, normalized_aliases, excluding_entry_id=entry_id + ) + now = _timestamp() + connection.execute( + """UPDATE glossary_entries SET canonical_term = ?, category = ?, + description = ?, is_active = ?, updated_at = ? WHERE id = ?""", + ( + canonical, + category, + _optional_text(description), + is_active, + now, + entry_id, + ), + ) + connection.execute("DELETE FROM glossary_aliases WHERE entry_id = ?", (entry_id,)) + connection.executemany( + "INSERT INTO glossary_aliases (entry_id, alias, created_at) VALUES (?, ?, ?)", + ((entry_id, alias, now) for alias in normalized_aliases), + ) + except sqlite3.IntegrityError as exc: + raise GlossaryConflictError("Canonical term or alias already exists.") from exc + return self.get(entry_id) + + def set_active(self, entry_id: int, is_active: bool) -> GlossaryEntry: + entry = self.get(entry_id) + return self.update( + entry.id, + entry.canonical_term, + entry.category, + aliases=entry.aliases, + description=entry.description, + is_active=is_active, + ) + + def delete(self, entry_id: int) -> None: + with self._connect() as connection: + cursor = connection.execute("DELETE FROM glossary_entries WHERE id = ?", (entry_id,)) + if cursor.rowcount == 0: + raise KeyError(f"Unknown glossary entry: {entry_id}") + + @contextmanager + def _connect(self) -> Iterator[sqlite3.Connection]: + connection = sqlite3.connect(self.database_path, timeout=5) + try: + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA foreign_keys = ON") + connection.execute("PRAGMA journal_mode = WAL") + connection.execute("PRAGMA busy_timeout = 5000") + with connection: + yield connection + finally: + connection.close() + + @staticmethod + def _validate_values( + canonical_term: str, category: str, aliases: Iterable[str] + ) -> tuple[str, tuple[str, ...]]: + canonical = canonical_term.strip() + if not canonical: + raise ValueError("Canonical term is required.") + if category not in GLOSSARY_CATEGORIES: + raise ValueError(f"Unsupported glossary category: {category}") + normalized_aliases = tuple( + dict.fromkeys(alias.strip() for alias in aliases if alias.strip()) + ) + folded = [alias.casefold() for alias in normalized_aliases] + if len(folded) != len(set(folded)) or canonical.casefold() in folded: + raise GlossaryConflictError( + "Aliases must be unique and differ from the canonical term." + ) + return canonical, normalized_aliases + + @staticmethod + def _ensure_terms_available( + connection: sqlite3.Connection, + canonical: str, + aliases: tuple[str, ...], + *, + excluding_entry_id: int | None = None, + ) -> None: + terms = (canonical, *aliases) + placeholders = ", ".join("?" for _ in terms) + exclusion = "AND entry_id != ?" if excluding_entry_id is not None else "" + alias_parameters: list[object] = [*terms] + if excluding_entry_id is not None: + alias_parameters.append(excluding_entry_id) + alias_conflict = connection.execute( + f"SELECT 1 FROM glossary_aliases WHERE alias IN ({placeholders}) {exclusion} LIMIT 1", # noqa: S608 + alias_parameters, + ).fetchone() + entry_exclusion = "AND id != ?" if excluding_entry_id is not None else "" + entry_parameters: list[object] = [*terms] + if excluding_entry_id is not None: + entry_parameters.append(excluding_entry_id) + canonical_conflict = connection.execute( + f"SELECT 1 FROM glossary_entries WHERE canonical_term IN ({placeholders}) " # noqa: S608 + f"{entry_exclusion} LIMIT 1", + entry_parameters, + ).fetchone() + if alias_conflict or canonical_conflict: + raise GlossaryConflictError("Canonical term or alias already exists.") + + @staticmethod + def _to_entry(connection: sqlite3.Connection, row: sqlite3.Row) -> GlossaryEntry: + aliases = connection.execute( + "SELECT alias FROM glossary_aliases WHERE entry_id = ? ORDER BY alias COLLATE NOCASE", + (row["id"],), + ).fetchall() + return GlossaryEntry( + id=row["id"], + canonical_term=row["canonical_term"], + category=row["category"], + description=row["description"], + is_active=bool(row["is_active"]), + aliases=tuple(alias["alias"] for alias in aliases), + created_at=row["created_at"], + updated_at=row["updated_at"], + ) + + +def render_glossary_terms(entries: Iterable[GlossaryEntry]) -> list[str]: + """Render concise canonical terms with recognition aliases for Meeting Context.""" + rendered = [] + for entry in entries: + item = entry.canonical_term + if entry.aliases: + item += f" (aliases: {', '.join(entry.aliases)})" + rendered.append(item) + return rendered + + +def _timestamp() -> str: + return datetime.now(UTC).isoformat(timespec="seconds") + + +def _optional_text(value: str | None) -> str | None: + stripped = value.strip() if value else "" + return stripped or None diff --git a/src/mka/application/meeting_service.py b/src/mka/application/meeting_service.py index 44d69b4..bb506f0 100644 --- a/src/mka/application/meeting_service.py +++ b/src/mka/application/meeting_service.py @@ -12,7 +12,10 @@ from pathlib import Path from typing import Any, Protocol from uuid import uuid4 +import yaml + from mka.application.config import AppSettings +from mka.application.glossary import GlossaryRepository, render_glossary_terms STAGES = ("preparing", "transcription", "diarization", "protocol_generation") @@ -31,6 +34,14 @@ class MeetingLabPort(Protocol): progress_sink: Callable[[Any], None], ) -> Any: ... + def regenerate_protocol( + self, + run_dir: Path, + meeting_context: Any, + progress_sink: Callable[[Any], None], + **options: Any, + ) -> Any: ... + @dataclass(frozen=True) class MeetingDetails: @@ -73,6 +84,20 @@ class ProcessingOutcome: protocol_path: Path | None failed_stage: str | None = None error_message: str | None = None + speaker_attribution_available: bool | None = None + + +@dataclass(frozen=True) +class SpeakerReview: + speaker_label: str + excerpts: tuple[str, ...] + + +@dataclass(frozen=True) +class SpeakerMappingReview: + speakers: tuple[SpeakerReview, ...] + participants: tuple[tuple[str, str], ...] + current_mappings: dict[str, str] def stable_id(value: str) -> str: @@ -84,9 +109,16 @@ def stable_id(value: str) -> str: class MeetingProcessingService: """Translate UI input into Meeting Lab calls and persisted artifacts.""" - def __init__(self, settings: AppSettings, meeting_lab: MeetingLabPort) -> None: + def __init__( + self, + settings: AppSettings, + meeting_lab: MeetingLabPort, + glossary: GlossaryRepository | None = None, + ) -> None: self.settings = settings self.meeting_lab = meeting_lab + self.glossary = glossary or GlossaryRepository(settings.glossary_database) + self.glossary.initialize() def build_context( self, @@ -154,6 +186,7 @@ class MeetingProcessingService: "mentioned_people_are_not_participants": True, }, } + self._apply_glossary(data) return self.meeting_lab.create_context(data) def preserve_upload(self, meeting_id: str, filename: str, source: Any) -> Path: @@ -253,8 +286,164 @@ class MeetingProcessingService: run_dir=result.run_dir, original_protocol=protocol_path.read_text(encoding="utf-8"), protocol_path=protocol_path, + speaker_attribution_available=self._speaker_attribution_available(result.run_dir), ) + def load_speaker_mapping_review( + self, + run_dir: Path, + *, + excerpts_per_speaker: int = 3, + minimum_excerpt_characters: int = 20, + ) -> SpeakerMappingReview | None: + """Load detected labels and small deterministic identification excerpts.""" + run_dir = Path(run_dir) + transcript_path = run_dir / "diarization" / "transcript_diarized.json" + context_path = run_dir / "context" / "meeting_context.yaml" + if not transcript_path.is_file() or not context_path.is_file(): + return None + transcript = json.loads(transcript_path.read_text(encoding="utf-8-sig")) + context_data = yaml.safe_load(context_path.read_text(encoding="utf-8")) + if not isinstance(transcript, dict) or not isinstance(context_data, dict): + raise ValueError("Existing run contains malformed speaker review artifacts.") + segments = transcript.get("segments") + if not isinstance(segments, list): + raise ValueError("Diarized transcript has no segments list.") + + texts_by_speaker: dict[str, list[str]] = {} + short_by_speaker: dict[str, list[str]] = {} + for segment in segments: + if not isinstance(segment, dict): + continue + label = segment.get("speaker_id") + text = segment.get("text") + if ( + not isinstance(label, str) + or not label.startswith("SPEAKER_") + or label == "SPEAKER_UNASSIGNED" + or not isinstance(text, str) + or not text.strip() + ): + continue + normalized = " ".join(text.split()) + target = ( + texts_by_speaker + if len(normalized) >= minimum_excerpt_characters + else short_by_speaker + ) + target.setdefault(label, []).append(normalized) + + labels = sorted(set(texts_by_speaker) | set(short_by_speaker)) + speakers = [] + for label in labels: + candidates = texts_by_speaker.get(label) or short_by_speaker.get(label, []) + speakers.append(SpeakerReview(label, tuple(candidates[:excerpts_per_speaker]))) + participants = tuple( + (participant["participant_id"], participant["display_name"]) + for participant in context_data.get("participants", []) + if isinstance(participant, dict) + and isinstance(participant.get("participant_id"), str) + and isinstance(participant.get("display_name"), str) + ) + mappings = context_data.get("speaker_mappings") + return SpeakerMappingReview( + speakers=tuple(speakers), + participants=participants, + current_mappings=dict(mappings) if isinstance(mappings, dict) else {}, + ) + + def regenerate_protocol( + self, + run_dir: Path, + speaker_mappings: dict[str, str], + progress_sink: Callable[[AppProgressEvent], None] | None = None, + ) -> ProcessingOutcome: + """Persist confirmed mappings and regenerate only the direct protocol.""" + review = self.load_speaker_mapping_review(run_dir) + if review is None: + raise ValueError("This run has no diarization artifacts to map.") + detected = {speaker.speaker_label for speaker in review.speakers} + participant_ids = {participant_id for participant_id, _ in review.participants} + unknown_labels = sorted(set(speaker_mappings) - detected) + if unknown_labels: + raise ValueError(f"Unknown diarization speaker label: {unknown_labels[0]}") + unknown_participants = sorted(set(speaker_mappings.values()) - participant_ids) + if unknown_participants: + raise ValueError(f"Unknown participant ID: {unknown_participants[0]}") + assigned_participants = list(speaker_mappings.values()) + if len(assigned_participants) != len(set(assigned_participants)): + raise ValueError("A participant may be assigned to only one speaker label.") + + context_path = Path(run_dir) / "context" / "meeting_context.yaml" + context_data = yaml.safe_load(context_path.read_text(encoding="utf-8")) + if not isinstance(context_data, dict): + raise ValueError("Existing run contains malformed Meeting Context.") + context_data["speaker_mappings"] = dict(sorted(speaker_mappings.items())) + self._apply_glossary(context_data) + context = self.meeting_lab.create_context(context_data) + + def relay(event: Any) -> None: + if progress_sink is not None: + progress_sink( + AppProgressEvent( + stage=event.stage, + status=event.status, + elapsed_seconds=event.elapsed_seconds, + progress=event.progress, + message=event.message, + ) + ) + + result = self.meeting_lab.regenerate_protocol( + Path(run_dir), + context, + relay, + model=self.settings.protocol_model, + ollama_endpoint=self.settings.ollama_endpoint, + protocol_num_ctx=self.settings.protocol_num_ctx, + protocol_safe_input_token_budget=(self.settings.protocol_safe_input_token_budget), + ) + protocol_path = Path(result.protocol_path) + return ProcessingOutcome( + succeeded=True, + run_dir=Path(result.run_dir), + original_protocol=protocol_path.read_text(encoding="utf-8"), + protocol_path=protocol_path, + speaker_attribution_available=self._speaker_attribution_available(result.run_dir), + ) + + def _apply_glossary(self, context_data: dict[str, Any]) -> None: + """Merge active terminology into context without discarding other entities.""" + glossary_terms = render_glossary_terms(self.glossary.list(active_only=True)) + known_entities = context_data.setdefault("known_entities", {}) + if glossary_terms: + known_entities["Authoritative terminology"] = glossary_terms + else: + known_entities.pop("Authoritative terminology", None) + rules = context_data.setdefault("context_rules", {}) + rules["glossary_canonical_spelling"] = ( + "Use canonical glossary spellings only when the meeting clearly refers to " + "those terms; do not invent matches or replace unrelated words." + ) + rules["glossary_core_terms"] = ( + "Preserve surrounding context and use canonical core terms inside compounds " + "where appropriate." + ) + + @staticmethod + def _speaker_attribution_available(run_dir: Path | None) -> bool | None: + if run_dir is None: + return None + metadata_path = Path(run_dir) / "protocol" / "runtime_metadata.json" + if not metadata_path.is_file(): + return None + try: + metadata = json.loads(metadata_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None + value = metadata.get("speaker_attribution_available") + return value if isinstance(value, bool) else None + @staticmethod def _read_failure(run_dir: Path | None) -> dict[str, str]: if run_dir is None: diff --git a/src/mka/application/run_inputs.py b/src/mka/application/run_inputs.py new file mode 100644 index 0000000..a6cd278 --- /dev/null +++ b/src/mka/application/run_inputs.py @@ -0,0 +1,218 @@ +"""Versioned JSON import/export for the user-facing run input form.""" + +from __future__ import annotations + +import json +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import date +from pathlib import Path +from typing import Any + +from mka.application.meeting_service import ParticipantInput, stable_id +from mka.application.people_yaml import ATTENDANCE_VALUES, PARTICIPANT_ID_PATTERN + +RUN_INPUT_SCHEMA_VERSION = 1 +SUPPORTED_LANGUAGES = frozenset({"de", "en"}) + + +class RunInputJsonError(ValueError): + """Raised when a run-input JSON document is malformed or unsupported.""" + + +@dataclass(frozen=True) +class RunInputState: + """All user-configurable values needed before starting a processing run.""" + + title: str + description: str + language: str + meeting_date: date | None + participants: tuple[ParticipantInput, ...] + audio_normalization: bool + diarization_enabled: bool + source_file_name: str | None = None + + @classmethod + def defaults(cls) -> RunInputState: + return cls( + title="", + description="", + language="de", + meeting_date=date.today(), + participants=(ParticipantInput(participant_id="", display_name=""),), + audio_normalization=True, + diarization_enabled=False, + ) + + +def export_run_inputs(state: RunInputState) -> str: + """Serialize form state as deterministic, human-readable JSON.""" + source_file_name = _source_file_name(state.source_file_name) + document = { + "schema_version": RUN_INPUT_SCHEMA_VERSION, + "meeting": { + "title": state.title, + "description": state.description, + "language": state.language, + "date": state.meeting_date.isoformat() if state.meeting_date else None, + "participants": [ + { + "participant_id": person.participant_id, + "display_name": person.display_name, + "role": person.role, + "organization": person.organization, + "attendance_status": person.attendance_status, + } + for person in state.participants + ], + }, + "processing": { + "audio_normalization": state.audio_normalization, + "diarization_enabled": state.diarization_enabled, + }, + "source_file_name": source_file_name, + } + return json.dumps(document, ensure_ascii=False, indent=2) + "\n" + + +def import_run_inputs(content: str | bytes) -> RunInputState: + """Parse supported fields, applying current defaults to omitted optional fields.""" + try: + if isinstance(content, bytes): + content = content.decode("utf-8") + document = json.loads(content) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise RunInputJsonError(f"Malformed input JSON: {exc}") from exc + if not isinstance(document, dict): + raise RunInputJsonError("Input JSON must contain a top-level object.") + version = document.get("schema_version") + if type(version) is not int or version != RUN_INPUT_SCHEMA_VERSION: + raise RunInputJsonError( + f"Unsupported input schema_version {version!r}; expected {RUN_INPUT_SCHEMA_VERSION}." + ) + + defaults = RunInputState.defaults() + meeting = _optional_mapping(document, "meeting") + processing = _optional_mapping(document, "processing") + title = _optional_text(meeting, "title", defaults.title) + description = _optional_text(meeting, "description", defaults.description) + language = _optional_text(meeting, "language", defaults.language) + if language not in SUPPORTED_LANGUAGES: + raise RunInputJsonError(f"Unsupported meeting language: {language!r}.") + meeting_date = _meeting_date(meeting, defaults.meeting_date) + participants = _participants(meeting, defaults.participants) + audio_normalization = _optional_bool( + processing, "audio_normalization", defaults.audio_normalization + ) + diarization_enabled = _optional_bool( + processing, "diarization_enabled", defaults.diarization_enabled + ) + source_file_name = _source_file_name(document.get("source_file_name")) + return RunInputState( + title=title, + description=description, + language=language, + meeting_date=meeting_date, + participants=participants, + audio_normalization=audio_normalization, + diarization_enabled=diarization_enabled, + source_file_name=source_file_name, + ) + + +def run_input_filename(title: str) -> str: + """Build a stable download filename without filesystem-specific characters.""" + suffix = stable_id(title) if title.strip() else "untitled" + return f"meeting-inputs-{suffix}.json" + + +def _optional_mapping(document: dict[str, Any], field: str) -> dict[str, Any]: + value = document.get(field, {}) + if not isinstance(value, dict): + raise RunInputJsonError(f"Field {field!r} must be an object.") + return value + + +def _optional_text(document: dict[str, Any], field: str, default: str) -> str: + value = document.get(field, default) + if not isinstance(value, str): + raise RunInputJsonError(f"Field {field!r} must be text.") + return value + + +def _optional_bool(document: dict[str, Any], field: str, default: bool) -> bool: + value = document.get(field, default) + if type(value) is not bool: + raise RunInputJsonError(f"Field {field!r} must be true or false.") + return value + + +def _meeting_date(document: dict[str, Any], default: date | None) -> date | None: + if "date" not in document: + return default + value = document["date"] + if value is None: + return None + if not isinstance(value, str): + raise RunInputJsonError("Field 'date' must be an ISO date or null.") + try: + return date.fromisoformat(value) + except ValueError as exc: + raise RunInputJsonError("Field 'date' must be a valid ISO date or null.") from exc + + +def _participants( + document: dict[str, Any], default: Sequence[ParticipantInput] +) -> tuple[ParticipantInput, ...]: + if "participants" not in document: + return tuple(default) + entries = document["participants"] + if not isinstance(entries, list): + raise RunInputJsonError("Field 'participants' must be a list.") + people: list[ParticipantInput] = [] + seen_ids: set[str] = set() + for index, entry in enumerate(entries, start=1): + if not isinstance(entry, dict): + raise RunInputJsonError(f"Participant {index} must be an object.") + participant_id = _optional_text(entry, "participant_id", "") + display_name = _optional_text(entry, "display_name", "") + role = _optional_text(entry, "role", "") + organization = _optional_text(entry, "organization", "") + attendance = _optional_text(entry, "attendance_status", "present") + if bool(participant_id) != bool(display_name): + raise RunInputJsonError( + f"Participant {index} must provide both participant_id and display_name." + ) + if participant_id and PARTICIPANT_ID_PATTERN.fullmatch(participant_id) is None: + raise RunInputJsonError( + f"Participant {index} has invalid participant_id {participant_id!r}." + ) + if participant_id in seen_ids: + raise RunInputJsonError(f"Duplicate participant_id: {participant_id!r}.") + if participant_id: + seen_ids.add(participant_id) + if attendance not in ATTENDANCE_VALUES: + raise RunInputJsonError( + f"Participant {index} has invalid attendance_status {attendance!r}." + ) + people.append( + ParticipantInput( + participant_id=participant_id, + display_name=display_name, + role=role, + organization=organization, + attendance_status=attendance, + ) + ) + return tuple(people) + + +def _source_file_name(value: Any) -> str | None: + if value is None: + return None + if not isinstance(value, str) or not value.strip(): + raise RunInputJsonError("Field 'source_file_name' must be non-empty text or null.") + if Path(value).name != value: + raise RunInputJsonError("Field 'source_file_name' must be a filename, not a path.") + return value diff --git a/src/mka/integrations/meeting_lab.py b/src/mka/integrations/meeting_lab.py index 498f10c..5167ad5 100644 --- a/src/mka/integrations/meeting_lab.py +++ b/src/mka/integrations/meeting_lab.py @@ -18,6 +18,7 @@ class MeetingLabGateway: from src.meeting_lab.models.meeting_context import create_meeting_context from src.meeting_lab.orchestration.mvp import ( MvpMeetingConfig, + regenerate_mvp_protocol, run_mvp_meeting, ) except ImportError as exc: @@ -28,6 +29,7 @@ class MeetingLabGateway: self._config_type = MvpMeetingConfig self._create_context = create_meeting_context self._run_mvp_meeting = run_mvp_meeting + self._regenerate_mvp_protocol = regenerate_mvp_protocol def create_context(self, data: dict[str, Any]) -> Any: """Validate structured context through Meeting Lab's domain boundary.""" @@ -49,3 +51,18 @@ class MeetingLabGateway: meeting_context=meeting_context, progress_sink=progress_sink, ) + + def regenerate_protocol( + self, + run_dir: Any, + meeting_context: Any, + progress_sink: Callable[[Any], None], + **options: Any, + ) -> Any: + """Regenerate protocol artifacts without rerunning media processing.""" + return self._regenerate_mvp_protocol( + run_dir, + meeting_context=meeting_context, + progress_sink=progress_sink, + **options, + ) diff --git a/src/mka/ui/streamlit_app.py b/src/mka/ui/streamlit_app.py index c89bcb0..648f4bf 100644 --- a/src/mka/ui/streamlit_app.py +++ b/src/mka/ui/streamlit_app.py @@ -9,6 +9,11 @@ from uuid import uuid4 import streamlit as st from mka.application.config import AppSettings, ConfigurationError +from mka.application.glossary import ( + GLOSSARY_CATEGORIES, + GlossaryConflictError, + GlossaryRepository, +) from mka.application.meeting_service import ( STAGES, AppProgressEvent, @@ -23,6 +28,13 @@ from mka.application.people_yaml import ( export_people_yaml, import_people_yaml, ) +from mka.application.run_inputs import ( + RunInputJsonError, + RunInputState, + export_run_inputs, + import_run_inputs, + run_input_filename, +) from mka.integrations.meeting_lab import ( MeetingLabGateway, MeetingLabUnavailableError, @@ -36,6 +48,87 @@ STAGE_LABELS = { } +def _parse_aliases(value: str) -> tuple[str, ...]: + """Parse one alias per line while tolerating comma-separated input.""" + return tuple( + alias.strip() for line in value.splitlines() for alias in line.split(",") if alias.strip() + ) + + +def _render_glossary(repository: GlossaryRepository) -> None: + """Render simple global glossary CRUD controls.""" + with st.expander("Terminology glossary", expanded=False): + st.caption( + "Store canonical core terms and recognition aliases. Compound phrases are " + "composed from meeting context during protocol generation." + ) + with st.form("glossary_add"): + columns = st.columns(2) + canonical = columns[0].text_input("Canonical term") + category = columns[1].selectbox("Category", GLOSSARY_CATEGORIES) + aliases = st.text_area("Aliases", help="One per line or comma-separated.") + description = st.text_area("Optional description") + if st.form_submit_button("Add glossary entry"): + try: + repository.create( + canonical, + category, + aliases=_parse_aliases(aliases), + description=description, + ) + except (GlossaryConflictError, ValueError) as exc: + st.error(str(exc)) + else: + st.success("Glossary entry added.") + st.rerun() + + search = st.text_input("Search glossary") + active_only = st.checkbox("Show active entries only") + entries = repository.list(search, active_only=active_only) + st.caption(f"{len(entries)} glossary entries") + for entry in entries: + status = "active" if entry.is_active else "inactive" + with ( + st.expander(f"{entry.canonical_term} · {entry.category} · {status}"), + st.form(f"glossary_edit_{entry.id}"), + ): + columns = st.columns(2) + edited_canonical = columns[0].text_input("Canonical term", entry.canonical_term) + edited_category = columns[1].selectbox( + "Category", + GLOSSARY_CATEGORIES, + index=GLOSSARY_CATEGORIES.index(entry.category), + ) + edited_aliases = st.text_area("Aliases", "\n".join(entry.aliases)) + edited_description = st.text_area("Optional description", entry.description or "") + edited_active = st.checkbox("Active", value=entry.is_active) + delete_confirmed = st.checkbox( + "Permanently delete this entry", + help="Prefer clearing Active for normal use.", + ) + action_columns = st.columns(2) + save = action_columns[0].form_submit_button("Save changes") + delete = action_columns[1].form_submit_button( + "Delete", disabled=not delete_confirmed + ) + try: + if save: + repository.update( + entry.id, + edited_canonical, + edited_category, + aliases=_parse_aliases(edited_aliases), + description=edited_description, + is_active=edited_active, + ) + st.rerun() + if delete: + repository.delete(entry.id) + st.rerun() + except (GlossaryConflictError, ValueError) as exc: + st.error(str(exc)) + + def _new_participant() -> dict[str, str]: return { "row_id": uuid4().hex, @@ -51,6 +144,51 @@ def _initialize_state() -> None: st.session_state.setdefault("participants", [_new_participant()]) st.session_state.setdefault("outcome", None) st.session_state.setdefault("edited_protocol", "") + st.session_state.setdefault("source_media_widget_generation", 0) + _apply_pending_run_inputs() + _apply_pending_edited_protocol() + + +def _queue_run_inputs(value: RunInputState) -> None: + """Defer form restoration until the beginning of the next Streamlit run.""" + st.session_state["pending_run_inputs"] = value + + +def _apply_pending_run_inputs() -> None: + """Restore imported values before any corresponding widget is instantiated.""" + value = st.session_state.pop("pending_run_inputs", None) + if value is None: + return + st.session_state.update( + { + "meeting_title": value.title, + "meeting_description": value.description, + "meeting_language": value.language, + "meeting_has_date": value.meeting_date is not None, + "meeting_date": value.meeting_date or date.today(), + "participants": _people_to_rows(list(value.participants)), + "audio_normalization": value.audio_normalization, + "diarization_enabled": value.diarization_enabled, + "imported_source_file_name": value.source_file_name, + "run_input_import_message": ( + "Meeting inputs imported. Select the source media before processing." + ), + "source_media_widget_generation": ( + st.session_state.get("source_media_widget_generation", 0) + 1 + ), + } + ) + + +def _apply_pending_edited_protocol() -> None: + """Apply a deferred widget value before the widget is instantiated.""" + if "pending_edited_protocol" in st.session_state: + st.session_state["edited_protocol"] = st.session_state.pop("pending_edited_protocol") + + +def _queue_edited_protocol(value: str) -> None: + """Defer an edited-protocol widget update until the next Streamlit run.""" + st.session_state["pending_edited_protocol"] = value def _people_to_rows(people: list[ParticipantInput]) -> list[dict[str, str]]: @@ -68,6 +206,27 @@ def _people_to_rows(people: list[ParticipantInput]) -> list[dict[str, str]]: ] +def _render_run_input_import() -> None: + """Render import controls before the widgets whose state they restore.""" + st.subheader("Input configuration") + imported_file = st.file_uploader( + "Import inputs", + type=["json"], + key="run_input_import_file", + help="Restore form values only; source media is never included.", + ) + if st.button("Import input configuration", disabled=imported_file is None): + try: + imported = import_run_inputs(imported_file.getvalue()) + except RunInputJsonError as exc: + st.error(str(exc)) + else: + _queue_run_inputs(imported) + st.rerun() + if message := st.session_state.pop("run_input_import_message", None): + st.success(message) + + def _render_participants() -> list[ParticipantInput]: st.subheader("People") st.caption("Record whether each relevant person attended or was only mentioned.") @@ -196,6 +355,13 @@ def _render_result() -> None: st.success("Processing completed. Review the generated protocol before use.") st.caption(f"Run artifacts: {outcome.run_dir}") + if outcome.speaker_attribution_available is False: + st.warning( + "Speaker attribution was unavailable for this protocol because the " + "safe-budget fallback removed diarization labels from the prompt." + ) + if message := st.session_state.pop("speaker_mapping_message", None): + st.success(message) with st.expander("Original generated protocol", expanded=False): st.markdown(outcome.original_protocol or "") edited = st.text_area( @@ -208,6 +374,60 @@ def _render_result() -> None: path = MeetingProcessingService.save_edited_protocol(outcome.run_dir, edited) st.success(f"Saved edited protocol to {path}") + try: + service = MeetingProcessingService(AppSettings.from_environment(), MeetingLabGateway()) + review = service.load_speaker_mapping_review(outcome.run_dir) + except (MeetingLabUnavailableError, OSError, ValueError) as exc: + st.warning(f"Speaker mapping is unavailable: {exc}") + return + if review is None or not review.speakers: + return + + st.subheader("Identify diarized speakers") + st.caption( + "Confirm identities explicitly. Unmapped speakers remain anonymous; " + "the diarized source transcript is not modified." + ) + participant_names = dict(review.participants) + options = [None, *participant_names] + selections: dict[str, str] = {} + for speaker in review.speakers: + current = review.current_mappings.get(speaker.speaker_label) + selected = st.selectbox( + speaker.speaker_label, + options=options, + index=options.index(current) if current in options else 0, + format_func=lambda value, names=participant_names: ( + "Unmapped / Unknown" if value is None else names[value] + ), + key=f"speaker_mapping_{outcome.run_dir.name}_{speaker.speaker_label}", + ) + if selected is not None: + selections[speaker.speaker_label] = selected + for excerpt in speaker.excerpts: + st.caption(f"“{excerpt}”") + + duplicate_assignments = len(selections.values()) != len(set(selections.values())) + if duplicate_assignments: + st.error("Each participant can be assigned to only one detected speaker.") + if st.button( + "Regenerate protocol with confirmed speakers", + disabled=duplicate_assignments, + type="primary", + ): + try: + with st.spinner("Regenerating protocol without rerunning audio processing..."): + regenerated = service.regenerate_protocol(outcome.run_dir, selections) + except (OSError, RuntimeError, ValueError) as exc: + st.error(f"Protocol regeneration failed: {exc}") + else: + st.session_state.outcome = regenerated + _queue_edited_protocol(regenerated.original_protocol or "") + st.session_state.speaker_mapping_message = ( + "Speaker mappings saved and protocol regenerated." + ) + st.rerun() + def main() -> None: st.set_page_config(page_title="Meeting Assistant", layout="wide") @@ -215,15 +435,36 @@ def main() -> None: st.title("Meeting Assistant") st.caption("Create meeting context, run Meeting Lab, and review the protocol.") + settings = AppSettings.from_environment() + glossary = GlossaryRepository(settings.glossary_database) + glossary.initialize() + _render_glossary(glossary) + + _render_run_input_import() + st.header("Meeting and audio") - audio = st.file_uploader("Audio recording", type=["wav", "flac", "m4a"]) - title = st.text_input("Meeting title") - description = st.text_area("Description / context", height=100) + audio = st.file_uploader( + "Audio recording", + type=["wav", "flac", "m4a"], + key=f"source_media_{st.session_state.source_media_widget_generation}", + ) + imported_source_name = st.session_state.get("imported_source_file_name") + if audio is None and imported_source_name: + st.caption( + f"Previous source filename: {imported_source_name}. " + "Select the media file again before processing." + ) + title = st.text_input("Meeting title", key="meeting_title") + description = st.text_area("Description / context", height=100, key="meeting_description") metadata_columns = st.columns(3) - language = metadata_columns[0].selectbox("Meeting language", options=["de", "en"]) - has_date = metadata_columns[1].checkbox("Meeting date is known", value=True) + language = metadata_columns[0].selectbox( + "Meeting language", options=["de", "en"], key="meeting_language" + ) + has_date = metadata_columns[1].checkbox( + "Meeting date is known", value=True, key="meeting_has_date" + ) selected_date: date | None = ( - metadata_columns[2].date_input("Meeting date") if has_date else None + metadata_columns[2].date_input("Meeting date", key="meeting_date") if has_date else None ) participants = _render_participants() @@ -232,6 +473,7 @@ def main() -> None: audio_normalization = st.checkbox( "Audio normalization", value=True, + key="audio_normalization", help=( "Normalize speech loudness during preparation. WAV, FLAC, and M4A " "are always converted to the canonical Meeting Lab audio format, " @@ -240,9 +482,28 @@ def main() -> None: ) diarization_enabled = st.checkbox( "Enable speaker diarization", + key="diarization_enabled", help="Speaker labels remain anonymous; identities are never inferred.", ) + current_inputs = RunInputState( + title=title, + description=description, + language=language, + meeting_date=selected_date, + participants=tuple(participants), + audio_normalization=audio_normalization, + diarization_enabled=diarization_enabled, + source_file_name=(audio.name if audio is not None else imported_source_name), + ) + st.download_button( + "Export inputs", + data=export_run_inputs(current_inputs).encode("utf-8"), + file_name=run_input_filename(title), + mime="application/json", + help="Download the current form configuration without source media or run results.", + ) + if st.button("Start processing", type="primary", disabled=audio is None): if not title.strip(): st.error("Meeting title is required.") @@ -252,9 +513,8 @@ def main() -> None: ): st.error("Every person row needs a name and participant ID.") else: - settings = AppSettings.from_environment() try: - service = MeetingProcessingService(settings, MeetingLabGateway()) + service = MeetingProcessingService(settings, MeetingLabGateway(), glossary) meeting = MeetingDetails( title=title, language=language, diff --git a/tests/test_application_config.py b/tests/test_application_config.py index b44c0a0..c5def52 100644 --- a/tests/test_application_config.py +++ b/tests/test_application_config.py @@ -39,6 +39,14 @@ def test_environment_defaults_to_no_diarization_container_args(monkeypatch) -> N settings = AppSettings.from_environment() assert settings.diarization_container_args == () + assert settings.glossary_database == Path("data/database/glossary.sqlite3") + + +def test_environment_configures_glossary_database(monkeypatch, tmp_path: Path) -> None: + database = tmp_path / "terms.sqlite3" + monkeypatch.setenv("MKA_GLOSSARY_DATABASE", str(database)) + + assert AppSettings.from_environment().glossary_database == database def test_environment_parses_multiple_ordered_container_args(monkeypatch) -> None: diff --git a/tests/test_glossary.py b/tests/test_glossary.py new file mode 100644 index 0000000..1e4f5d0 --- /dev/null +++ b/tests/test_glossary.py @@ -0,0 +1,78 @@ +import sqlite3 +from pathlib import Path + +import pytest + +from mka.application.glossary import GlossaryConflictError, GlossaryRepository + + +def repository(tmp_path: Path) -> GlossaryRepository: + result = GlossaryRepository(tmp_path / "database" / "glossary.sqlite3") + result.initialize() + return result + + +def test_initialization_creates_empty_versioned_database(tmp_path: Path) -> None: + glossary = repository(tmp_path) + + assert glossary.database_path.is_file() + assert glossary.list() == [] + with sqlite3.connect(glossary.database_path) as connection: + assert connection.execute("PRAGMA user_version").fetchone()[0] == 1 + tables = { + row[0] + for row in connection.execute("SELECT name FROM sqlite_master WHERE type = 'table'") + } + assert {"glossary_entries", "glossary_aliases"} <= tables + + +def test_create_read_update_and_deactivate_with_multiple_aliases(tmp_path: Path) -> None: + glossary = repository(tmp_path) + created = glossary.create( + "Secugrid HS", + "product", + aliases=("Sikirgut", "Secugrid H S"), + description="Canonical core product name", + ) + + assert glossary.get(created.id).aliases == ("Secugrid H S", "Sikirgut") + assert glossary.list("sikir")[0].canonical_term == "Secugrid HS" + + updated = glossary.update( + created.id, + "Secugrid HS", + "technical_term", + aliases=("Sekugrid HS",), + description="Updated", + is_active=False, + ) + + assert updated.category == "technical_term" + assert updated.aliases == ("Sekugrid HS",) + assert not updated.is_active + assert glossary.list(active_only=True) == [] + + +def test_terms_and_aliases_are_unique_case_insensitively_across_entries( + tmp_path: Path, +) -> None: + glossary = repository(tmp_path) + glossary.create("PBAT", "acronym", aliases=("Polybutylene adipate terephthalate",)) + + with pytest.raises(GlossaryConflictError): + glossary.create("pbat", "material") + with pytest.raises(GlossaryConflictError): + glossary.create("Other", "other", aliases=("PBAT",)) + with pytest.raises(GlossaryConflictError): + glossary.create("Polybutylene adipate terephthalate", "material") + + +def test_delete_removes_entry_and_aliases(tmp_path: Path) -> None: + glossary = repository(tmp_path) + entry = glossary.create("Luminy", "product", aliases=("Lumini",)) + + glossary.delete(entry.id) + + assert glossary.list() == [] + with sqlite3.connect(glossary.database_path) as connection: + assert connection.execute("SELECT COUNT(*) FROM glossary_aliases").fetchone()[0] == 0 diff --git a/tests/test_meeting_service.py b/tests/test_meeting_service.py index f1829aa..106f928 100644 --- a/tests/test_meeting_service.py +++ b/tests/test_meeting_service.py @@ -5,6 +5,8 @@ from pathlib import Path from types import SimpleNamespace from typing import Any +import pytest + from mka.application.config import AppSettings from mka.application.meeting_service import ( MeetingDetails, @@ -29,6 +31,7 @@ class FakeMeetingLab: self.context_data: dict[str, Any] | None = None self.config_values: dict[str, Any] | None = None self.fail = False + self.regeneration: dict[str, Any] | None = None def create_context(self, data: dict[str, Any]) -> FakeContext: self.context_data = data @@ -99,6 +102,36 @@ class FakeMeetingLab: protocol.write_text("# Generated protocol\n", encoding="utf-8") return SimpleNamespace(exit_code=0, run_dir=self.run_dir, protocol_path=protocol) + def regenerate_protocol( + self, + run_dir: Path, + meeting_context: Any, + progress_sink: Any, + **options: Any, + ) -> Any: + self.regeneration = { + "run_dir": run_dir, + "meeting_context": meeting_context, + "options": options, + } + progress_sink( + SimpleNamespace( + stage="protocol_generation", + status="started", + elapsed_seconds=0.0, + progress=None, + message=None, + ) + ) + protocol = Path(run_dir) / "protocol.md" + protocol.write_text("# Regenerated protocol\n", encoding="utf-8") + protocol_dir = Path(run_dir) / "protocol" + protocol_dir.mkdir(exist_ok=True) + (protocol_dir / "runtime_metadata.json").write_text( + json.dumps({"speaker_attribution_available": True}), encoding="utf-8" + ) + return SimpleNamespace(exit_code=0, run_dir=run_dir, protocol_path=protocol) + def make_service(tmp_path: Path) -> tuple[MeetingProcessingService, FakeMeetingLab]: model = tmp_path / "model.bin" @@ -107,6 +140,7 @@ def make_service(tmp_path: Path) -> tuple[MeetingProcessingService, FakeMeetingL settings = AppSettings( data_root=tmp_path / "meetings", whisper_model=model, + glossary_database=tmp_path / "glossary.sqlite3", whisper_executable="/opt/whisper-cli", protocol_model="test:model", diarization_mode="gpu", @@ -139,6 +173,60 @@ def participants() -> list[ParticipantInput]: ] +def write_speaker_review_artifacts(run_dir: Path) -> Path: + diarization_dir = run_dir / "diarization" + context_dir = run_dir / "context" + diarization_dir.mkdir(parents=True) + context_dir.mkdir() + transcript_path = diarization_dir / "transcript_diarized.json" + transcript_path.write_text( + json.dumps( + { + "speaker_labels_anonymous": True, + "segments": [ + { + "speaker_id": "SPEAKER_01", + "text": "I will prepare all raw materials before Wednesday.", + }, + {"speaker_id": "SPEAKER_00", "text": "Yes."}, + { + "speaker_id": "SPEAKER_00", + "text": "We will run the production trial on Wednesday.", + }, + { + "speaker_id": "SPEAKER_00", + "text": "The trial requires the complete production team.", + }, + {"speaker_id": None, "text": "Unassigned text."}, + ], + } + ), + encoding="utf-8", + ) + (context_dir / "meeting_context.yaml").write_text( + json.dumps( + { + "schema_version": "1", + "meeting": { + "meeting_id": "speaker-review", + "title": "Speaker review", + "language": "en", + }, + "participants": [ + {"participant_id": "martin", "display_name": "Martin"}, + {"participant_id": "anna", "display_name": "Anna"}, + ], + "speaker_mappings": {}, + "mentioned_people": [], + "organization": {"departments": []}, + "known_entities": {}, + } + ), + encoding="utf-8", + ) + return transcript_path + + def test_build_context_uses_actual_v1_shape(tmp_path: Path) -> None: service, gateway = make_service(tmp_path) @@ -171,6 +259,40 @@ def test_build_context_preserves_explicit_speaker_mapping(tmp_path: Path) -> Non assert gateway.context_data["speaker_mappings"] == {"SPEAKER_00": "martin"} +def test_build_context_includes_only_active_authoritative_glossary_terms( + tmp_path: Path, +) -> None: + service, gateway = make_service(tmp_path) + service.glossary.create("Secugrid HS", "product", aliases=("Sikirgut", "Secugrid H S")) + inactive = service.glossary.create("Old Name", "other") + service.glossary.set_active(inactive.id, False) + + service.build_context(meeting(), participants()) + + assert gateway.context_data is not None + assert gateway.context_data["known_entities"] == { + "Authoritative terminology": ["Secugrid HS (aliases: Secugrid H S, Sikirgut)"] + } + rules = gateway.context_data["context_rules"] + assert "do not invent matches" in rules["glossary_canonical_spelling"] + assert "inside compounds" in rules["glossary_core_terms"] + assert "Old Name" not in str(gateway.context_data) + + +def test_glossary_is_rendered_into_meeting_lab_protocol_context(tmp_path: Path) -> None: + meeting_context = pytest.importorskip("src.meeting_lab.models.meeting_context") + service, gateway = make_service(tmp_path) + service.glossary.create("PBAT", "acronym", aliases=("P B A T",)) + context = service.build_context(meeting(), participants()) + + real_context = meeting_context.create_meeting_context(context.data) + prompt_context = meeting_context.render_meeting_context_for_prompt(real_context) + + assert "Authoritative terminology: PBAT (aliases: P B A T)" in prompt_context + assert "Use canonical glossary spellings" in prompt_context + assert "use canonical core terms inside compounds" in prompt_context + + def test_build_context_translates_mentioned_only_person(tmp_path: Path) -> None: service, gateway = make_service(tmp_path) people = participants() + [ @@ -331,6 +453,96 @@ def test_failure_reports_stage_and_preserves_run_dir(tmp_path: Path) -> None: assert (gateway.run_dir / "run_metadata.json").is_file() +def test_speaker_review_lists_detected_labels_participants_and_excerpts( + tmp_path: Path, +) -> None: + service, gateway = make_service(tmp_path) + source = write_speaker_review_artifacts(gateway.run_dir) + + review = service.load_speaker_mapping_review(gateway.run_dir, excerpts_per_speaker=2) + + assert review is not None + assert [speaker.speaker_label for speaker in review.speakers] == [ + "SPEAKER_00", + "SPEAKER_01", + ] + assert review.speakers[0].excerpts == ( + "We will run the production trial on Wednesday.", + "The trial requires the complete production team.", + ) + assert review.participants == (("martin", "Martin"), ("anna", "Anna")) + assert review.current_mappings == {} + assert "SPEAKER_00" in source.read_text(encoding="utf-8") + + +def test_protocol_only_regeneration_persists_mappings_without_rewriting_transcript( + tmp_path: Path, +) -> None: + service, gateway = make_service(tmp_path) + service.glossary.create("ENLYZE", "organization", aliases=("Enlyse",)) + source = write_speaker_review_artifacts(gateway.run_dir) + original_source = source.read_bytes() + events = [] + + outcome = service.regenerate_protocol( + gateway.run_dir, + {"SPEAKER_00": "martin"}, + progress_sink=events.append, + ) + + assert outcome.succeeded + assert outcome.original_protocol == "# Regenerated protocol\n" + assert outcome.speaker_attribution_available is True + assert gateway.context_data is not None + assert gateway.context_data["speaker_mappings"] == {"SPEAKER_00": "martin"} + assert gateway.context_data["known_entities"] == { + "Authoritative terminology": ["ENLYZE (aliases: Enlyse)"] + } + assert gateway.regeneration is not None + assert gateway.regeneration["meeting_context"].data["speaker_mappings"] == { + "SPEAKER_00": "martin" + } + assert gateway.regeneration["options"]["protocol_num_ctx"] == 32_768 + assert events[0].stage == "protocol_generation" + assert source.read_bytes() == original_source + + +def test_protocol_regeneration_allows_unmapped_and_rejects_duplicate_participant( + tmp_path: Path, +) -> None: + service, gateway = make_service(tmp_path) + write_speaker_review_artifacts(gateway.run_dir) + + outcome = service.regenerate_protocol(gateway.run_dir, {}) + + assert outcome.succeeded + assert gateway.context_data is not None + assert gateway.context_data["speaker_mappings"] == {} + + with pytest.raises(ValueError, match="only one speaker"): + service.regenerate_protocol( + gateway.run_dir, + {"SPEAKER_00": "martin", "SPEAKER_01": "martin"}, + ) + + +def test_fallback_attribution_loss_is_read_from_runtime_metadata(tmp_path: Path) -> None: + run_dir = tmp_path / "run" + protocol_dir = run_dir / "protocol" + protocol_dir.mkdir(parents=True) + (protocol_dir / "runtime_metadata.json").write_text( + json.dumps( + { + "speaker_attribution_available": False, + "speaker_attribution_loss_reason": "plain_transcript_fallback", + } + ), + encoding="utf-8", + ) + + assert MeetingProcessingService._speaker_attribution_available(run_dir) is False + + def test_uploaded_source_is_preserved_in_meeting_directory(tmp_path: Path) -> None: service, _ = make_service(tmp_path) source = SimpleNamespace(getbuffer=lambda: b"source audio") diff --git a/tests/test_run_inputs.py b/tests/test_run_inputs.py new file mode 100644 index 0000000..6704b5c --- /dev/null +++ b/tests/test_run_inputs.py @@ -0,0 +1,136 @@ +import json +from datetime import date + +import pytest + +from mka.application.meeting_service import ParticipantInput +from mka.application.run_inputs import ( + RunInputJsonError, + RunInputState, + export_run_inputs, + import_run_inputs, + run_input_filename, +) + + +def populated_state() -> RunInputState: + return RunInputState( + title="F&E Technik Abteilungs Jour Fixe", + description="Review the production trial.", + language="de", + meeting_date=date(2026, 8, 25), + participants=( + ParticipantInput( + participant_id="martin", + display_name="Martin Tazl", + role="Project lead", + organization="Engineering", + ), + ParticipantInput( + participant_id="alex", + display_name="Alexander Funk", + attendance_status="mentioned_only", + ), + ), + audio_normalization=False, + diarization_enabled=True, + source_file_name="2026-08-25_jour_fixe.wav", + ) + + +def test_current_form_state_serializes_with_schema_and_all_supported_values() -> None: + document = json.loads(export_run_inputs(populated_state())) + + assert document["schema_version"] == 1 + assert document["meeting"]["title"] == "F&E Technik Abteilungs Jour Fixe" + assert document["meeting"]["description"] == "Review the production trial." + assert document["meeting"]["language"] == "de" + assert document["meeting"]["date"] == "2026-08-25" + assert document["meeting"]["participants"][1] == { + "participant_id": "alex", + "display_name": "Alexander Funk", + "role": "", + "organization": "", + "attendance_status": "mentioned_only", + } + assert document["processing"] == { + "audio_normalization": False, + "diarization_enabled": True, + } + + +def test_export_import_round_trip_restores_form_and_participants() -> None: + original = populated_state() + + restored = import_run_inputs(export_run_inputs(original)) + + assert restored == original + assert restored.participants[0].participant_id == "martin" + assert restored.participants[0].display_name == "Martin Tazl" + + +def test_missing_optional_fields_use_current_defaults() -> None: + restored = import_run_inputs('{"schema_version": 1}') + + defaults = RunInputState.defaults() + assert restored == defaults + + +def test_unknown_safe_fields_are_ignored() -> None: + restored = import_run_inputs( + json.dumps( + { + "schema_version": 1, + "meeting": {"title": "Known", "future_field": {"value": 1}}, + "processing": {"future_toggle": True}, + "future_section": [1, 2, 3], + } + ) + ) + + assert restored.title == "Known" + + +@pytest.mark.parametrize("content", ["not-json", "[]", b"\xff"]) +def test_malformed_json_is_rejected(content: str | bytes) -> None: + with pytest.raises(RunInputJsonError, match="Malformed|top-level"): + import_run_inputs(content) + + +def test_unsupported_future_schema_is_rejected() -> None: + with pytest.raises(RunInputJsonError, match="Unsupported.*schema_version"): + import_run_inputs('{"schema_version": 2}') + + +def test_media_contents_are_never_serialized() -> None: + exported = export_run_inputs(populated_state()) + + assert "2026-08-25_jour_fixe.wav" in exported + assert "audio_bytes" not in exported + assert "base64" not in exported + + +def test_source_filename_must_not_be_a_machine_specific_path() -> None: + with pytest.raises(RunInputJsonError, match="filename, not a path"): + import_run_inputs('{"schema_version": 1, "source_file_name": "/tmp/meeting.wav"}') + + +def test_invalid_or_duplicate_participants_are_not_silently_reinterpreted() -> None: + document = { + "schema_version": 1, + "meeting": { + "participants": [ + {"participant_id": "martin", "display_name": "Martin"}, + {"participant_id": "martin", "display_name": "Someone else"}, + ] + }, + } + + with pytest.raises(RunInputJsonError, match="Duplicate participant_id"): + import_run_inputs(json.dumps(document)) + + +def test_export_filename_is_readable_and_machine_independent() -> None: + assert run_input_filename("F&E Technik Jour Fixe") == ( + "meeting-inputs-f-e-technik-jour-fixe.json" + ) diff --git a/tests/test_streamlit_app.py b/tests/test_streamlit_app.py new file mode 100644 index 0000000..fee8def --- /dev/null +++ b/tests/test_streamlit_app.py @@ -0,0 +1,68 @@ +from datetime import date + +from mka.application.meeting_service import ParticipantInput +from mka.application.run_inputs import RunInputState +from mka.ui import streamlit_app + + +def test_alias_input_accepts_lines_and_commas() -> None: + assert streamlit_app._parse_aliases("Sikirgut\nSekugrid HS, Secugrid H S") == ( + "Sikirgut", + "Sekugrid HS", + "Secugrid H S", + ) + + +def test_regenerated_protocol_widget_value_is_deferred_until_next_run( + monkeypatch, +) -> None: + state = {"edited_protocol": "old protocol"} + monkeypatch.setattr(streamlit_app.st, "session_state", state) + + streamlit_app._queue_edited_protocol("regenerated protocol") + + assert state["edited_protocol"] == "old protocol" + assert state["pending_edited_protocol"] == "regenerated protocol" + + streamlit_app._apply_pending_edited_protocol() + + assert state["edited_protocol"] == "regenerated protocol" + assert "pending_edited_protocol" not in state + + +def test_imported_inputs_are_applied_via_pending_state_before_widgets( + monkeypatch, +) -> None: + existing_upload = object() + state = {"source_media_0": existing_upload, "source_media_widget_generation": 0} + monkeypatch.setattr(streamlit_app.st, "session_state", state) + imported = RunInputState( + title="Imported meeting", + description="Imported context", + language="en", + meeting_date=date(2026, 8, 25), + participants=(ParticipantInput("martin", "Martin"),), + audio_normalization=False, + diarization_enabled=True, + source_file_name="meeting.wav", + ) + + streamlit_app._queue_run_inputs(imported) + + assert "meeting_title" not in state + assert state["pending_run_inputs"] == imported + + streamlit_app._apply_pending_run_inputs() + + assert state["meeting_title"] == "Imported meeting" + assert state["meeting_description"] == "Imported context" + assert state["meeting_language"] == "en" + assert state["meeting_has_date"] is True + assert state["participants"][0]["participant_id"] == "martin" + assert state["audio_normalization"] is False + assert state["diarization_enabled"] is True + assert state["imported_source_file_name"] == "meeting.wav" + assert state["source_media_widget_generation"] == 1 + assert "source_media_1" not in state + assert "pending_run_inputs" not in state + assert state["source_media_0"] is existing_upload