From cbebb93bfa91ffe6429488ab2668a8ed6b63b025 Mon Sep 17 00:00:00 2001 From: Optic00 Date: Wed, 22 Jul 2026 10:49:25 +0200 Subject: [PATCH] =?UTF-8?q?feat(keywords):=20Custom=20Keywords=20backend?= =?UTF-8?q?=20=E2=80=94=20glossary-driven=20transcript=20+=20summary=20cor?= =?UTF-8?q?rection?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Backend-only slice of Custom Keywords (RFC-adjacent, issue #365): a user glossary of preferred spellings + aliases that (a) persists in config.json, (b) heals the transcript at transcription time and on reprocess, and (c) injects a reference block into the summariser prompts. Empty glossary is a no-op (zero behavior change). - src/keywords.py: parse/normalize (defensive against malformed config), boundary-aware + diarised-safe alias replacement, reference-block builder. - src/config.py: get/set_custom_keywords (validated, replace-not-merge). - src/summarizer.py: reference block injected into permissive/markdown/ template-report/title/query prompts. - simple_recorder.py: guarded transcript heal in the pipeline + live-fallback + reprocess; get/set-custom-keywords CLI. Never heals the silence sentinel. - Unit tests incl. silence-sentinel safety, diarised timestamped labels, and malformed-config hardening. Test fixtures use fictional placeholders only. UI/IPC bridge + e2e spec land in a follow-up PR (re-integrated against #344). Refs #365. --- simple_recorder.py | 64 ++++++++ src/config.py | 11 ++ src/keywords.py | 184 +++++++++++++++++++++++ src/summarizer.py | 23 ++- tests/test_config.py | 28 ++++ tests/test_keywords.py | 236 ++++++++++++++++++++++++++++++ tests/test_keywords_cli.py | 23 +++ tests/test_keywords_pipeline.py | 145 ++++++++++++++++++ tests/test_reprocess_keywords.py | 100 +++++++++++++ tests/test_summarizer_keywords.py | 31 ++++ 10 files changed, 840 insertions(+), 5 deletions(-) create mode 100644 src/keywords.py create mode 100644 tests/test_keywords.py create mode 100644 tests/test_keywords_cli.py create mode 100644 tests/test_keywords_pipeline.py create mode 100644 tests/test_reprocess_keywords.py create mode 100644 tests/test_summarizer_keywords.py diff --git a/simple_recorder.py b/simple_recorder.py index 01d5ee54..dff6e9f8 100644 --- a/simple_recorder.py +++ b/simple_recorder.py @@ -461,6 +461,23 @@ async def transcribe_audio(self, audio_file: str, session_name: str = "Recording is_diarised = transcript_result.get("is_diarised", False) diarised_text = transcript_result.get("diarised_text") + # Custom Keywords: correct known mis-transcriptions before persisting + + # summarizing. No-op when no keywords are configured. + # + # NEVER heal the silence sentinel: the downstream silence / live-rescue + # decision compares the transcript against the exact sentinel string, so + # an alias like "speech -> Speechly" would mutate "No speech detected in + # audio", defeat that check, and get a fake-empty transcript summarised + # and saved (audio possibly deleted). Skip healing when the text IS the + # sentinel; heal only genuine content. + from src import keywords as _kw + _kw_entries = config.get_custom_keywords() + if _kw_entries: + if transcript_text and transcript_text.strip() != _SILENCE_SENTINEL: + transcript_text = _kw.apply_to_transcript(transcript_text, _kw_entries) + if diarised_text and diarised_text.strip() != _SILENCE_SENTINEL: + diarised_text = _kw.apply_to_transcript(diarised_text, _kw_entries) + output_language = self._resolve_output_language( configured_language, detected_language, transcript_text=diarised_text or transcript_text ) @@ -1040,6 +1057,18 @@ def _heartbeat_sink(done, total): len(live_transcript_text), ) is_live_transcript = True + # Custom Keywords: heal the live fallback text too. The batch path's + # healing (in transcribe_audio) never touched this text, so without + # this the transcription-time correction guarantee is violated and + # language resolution + the saved .txt below would run on uncorrected + # text. Guard the sentinel defensively (live text is genuine content). + from src.config import get_config as _kw_get_config + from src import keywords as _kw + _kw_entries = _kw_get_config().get_custom_keywords() + if _kw_entries and live_transcript_text.strip() != _SILENCE_SENTINEL: + live_transcript_text = _kw.apply_to_transcript( + live_transcript_text, _kw_entries + ) # Always (re)write _transcript.txt with the live text via the shared # formatter so the on-disk file matches the markdown/summary the user # sees and uses the canonical filename/header (#207, review-2 @@ -1596,6 +1625,28 @@ def set_keep_recordings_cmd(enabled: bool): print(json.dumps({"success": False, "error": "Failed to persist setting"})) +@cli.command(name='get-custom-keywords') +def get_custom_keywords_cmd(): + """Print the custom keywords as textarea-formatted text.""" + from src.config import get_config + from src import keywords + text = keywords.format_keywords_text(get_config().get_custom_keywords()) + print(json.dumps({"success": True, "text": text})) + + +@cli.command(name='set-custom-keywords') +@click.argument('text') +def set_custom_keywords_cmd(text): + """Parse + persist custom keywords from raw textarea text.""" + from src.config import get_config + from src import keywords + entries = keywords.parse_keywords_text(text) + ok = get_config().set_custom_keywords(entries) + # Exit 0 regardless: the JSON on stdout IS the result (mirrors save-template). + print(json.dumps({"success": ok} if ok + else {"success": False, "error": "Failed to save config"})) + + @cli.command(name='get-auto-summarize') def get_auto_summarize_cmd(): """Get whether notes are generated automatically after transcription.""" @@ -2756,6 +2807,19 @@ def reprocess(summary_file, regenerate_title): print("ERROR: No transcript found in summary file") sys.exit(1) + # Custom Keywords retroactive heal (A2): correct the stored transcript + # before re-summarizing. No re-ASR; pure text. No-op when unconfigured. + from src.config import get_config as _get_config + from src import keywords as _kw + _kw_entries = _get_config().get_custom_keywords() + if _kw_entries: + transcript = _kw.apply_to_transcript(transcript, _kw_entries) + existing_data['transcript'] = transcript + if existing_data.get('diarised_text'): + existing_data['diarised_text'] = _kw.apply_to_transcript( + existing_data['diarised_text'], _kw_entries + ) + session_name = existing_data.get('session_info', {}).get('name', 'Reprocessed') duration_minutes = existing_data.get('session_info', {}).get('duration_minutes', 10) if duration_minutes is None: diff --git a/src/config.py b/src/config.py index 7a8262af..2bab1ea4 100644 --- a/src/config.py +++ b/src/config.py @@ -887,6 +887,17 @@ def list_supported_models(self) -> Dict[str, Dict[str, str]]: """Get all supported models with their metadata.""" return self.SUPPORTED_MODELS.copy() + def get_custom_keywords(self) -> list: + """Normalized custom-keyword entries: [{'preferred', 'aliases'}].""" + from src import keywords + return keywords.normalize_keywords(self._config.get("custom_keywords", []) or []) + + def set_custom_keywords(self, entries: list) -> bool: + """Validate + persist the whole list (replace, not merge).""" + from src import keywords + self._config["custom_keywords"] = keywords.normalize_keywords(entries or []) + return self._save() + def get_notifications_enabled(self) -> bool: """Get whether desktop notifications are enabled.""" return self._config.get("notifications_enabled", True) diff --git a/src/keywords.py b/src/keywords.py new file mode 100644 index 00000000..989e8443 --- /dev/null +++ b/src/keywords.py @@ -0,0 +1,184 @@ +"""Pure model for the Custom Keywords feature: parsing, validation, alias +replacement, and the summary reference block. No I/O, no config access.""" +from __future__ import annotations + +import re + +MAX_ENTRIES = 200 +MAX_ALIASES = 10 +MAX_LEN = 80 + +# Control chars (incl. newlines/tabs) are stripped from persisted values so a +# hand-edited/corrupted config.json can't smuggle in extra lines or break the +# alias matcher. Collapsed to single spaces to avoid merging adjacent tokens. +_CONTROL_RE = re.compile(r"[\x00-\x1f\x7f]") + + +def _clean_str(value) -> str: + """Coerce a persisted value to a clean single-line string. + + Non-strings (numbers, None, dicts from a corrupted config) become '' and + are dropped by the caller. Control/newline characters are neutralised, + surrounding whitespace trimmed, and length capped at MAX_LEN. + """ + if not isinstance(value, str): + return "" + cleaned = _CONTROL_RE.sub(" ", value) + cleaned = re.sub(r"\s+", " ", cleaned).strip() + return cleaned[:MAX_LEN] + + +def parse_keywords_text(text: str) -> list[dict]: + """Textarea text -> entries. One entry per line; `Preferred: a, b` for aliases.""" + entries: list[dict] = [] + for line in (text or "").splitlines(): + line = line.strip() + if not line: + continue + if ":" in line: + preferred, _, alias_str = line.partition(":") + aliases = [a.strip() for a in alias_str.split(",") if a.strip()] + else: + preferred, aliases = line, [] + preferred = preferred.strip() + if preferred: + entries.append({"preferred": preferred, "aliases": aliases}) + return entries + + +def format_keywords_text(entries: list[dict]) -> str: + """Entries -> textarea text (inverse of parse_keywords_text).""" + lines: list[str] = [] + for e in entries: + pref = (e.get("preferred") or "").strip() + if not pref: + continue + aliases = [a.strip() for a in (e.get("aliases") or []) if a.strip()] + lines.append(f"{pref}: {', '.join(aliases)}" if aliases else pref) + return "\n".join(lines) + + +def normalize_keywords(entries: list[dict]) -> list[dict]: + """Validate + de-dupe per spec. Idempotent.""" + out: list[dict] = [] + pref_index: dict[str, int] = {} # lower(preferred) -> index in out + seen_alias: set[str] = set() # global lowercased aliases + for e in (entries or []): + # Defensive against a corrupted/hand-edited config.json: a non-dict + # entry (str, int, None, ...) has no .get and must be skipped. + if not isinstance(e, dict): + continue + pref = _clean_str(e.get("preferred")) # non-str preferred -> '' -> dropped + if not pref: + continue + pl = pref.lower() + # last wins: a duplicate preferred releases the superseded entry's aliases + # so the replacement can reclaim them before its own alias loop runs. + if pl in pref_index: + for a in out[pref_index[pl]]["aliases"]: + seen_alias.discard(a.lower()) + # A string aliases value is treated as a SINGLE alias, never iterated as + # characters (the char-iteration bug would silently corrupt matching). + # Any other non-list shape is dropped. + raw_aliases = e.get("aliases") + if isinstance(raw_aliases, str): + raw_aliases = [raw_aliases] + elif not isinstance(raw_aliases, (list, tuple)): + raw_aliases = [] + aliases: list[str] = [] + for a in raw_aliases: + a = _clean_str(a) # non-str alias -> '' -> skipped below + al = a.lower() + if not a or al in seen_alias: + continue + aliases.append(a) + seen_alias.add(al) + if len(aliases) >= MAX_ALIASES: + break + if pl in pref_index: + out[pref_index[pl]] = {"preferred": pref, "aliases": aliases} # last wins + else: + if len(out) >= MAX_ENTRIES: + continue + pref_index[pl] = len(out) + out.append({"preferred": pref, "aliases": aliases}) + # Drop any alias that equals a preferred term (would be a no-op / loop). + prefs = {o["preferred"].lower() for o in out} + for o in out: + o["aliases"] = [a for a in o["aliases"] if a.lower() not in prefs] + return out + + +# Protect EVERY leading `[...]` label on a diarised line, not just the first. +# The real production format is `[MM:SS] [You] text` / `[H:MM:SS] [Others] text` +# (transcriber._format_timestamp + the speaker tag), so both the timestamp and +# the [You]/[Others] speaker label must be skipped by alias replacement - +# otherwise an alias like `you` would rewrite the speaker label itself. +_LABEL_RE = re.compile(r"^((?:\[[^\]]*\]\s*)+)(.*)$") + + +def _build_matcher(entries: list[dict]): + """Return (compiled_pattern, lower_alias -> preferred) or (None, {}).""" + mapping: dict[str, str] = {} + for e in entries or []: + pref = e.get("preferred") + if not pref: + continue + for a in (e.get("aliases") or []): + if a: + mapping.setdefault(a.lower(), pref) + if not mapping: + return None, {} + ordered = sorted(mapping.keys(), key=len, reverse=True) # longest first + body = "|".join(re.escape(a) for a in ordered) + pattern = re.compile(rf"(? str: + """Replace aliases with their preferred term in plain text.""" + if not text or not entries: + return text + pattern, mapping = _build_matcher(entries) + if pattern is None: + return text + # Defensive: re.IGNORECASE can match text whose .lower() (Unicode case + # folding, e.g. 'İ' -> 'i̇') is not a mapping key -> leave it unchanged + # rather than raising KeyError mid-transcription. + return pattern.sub(lambda m: mapping.get(m.group(0).lower(), m.group(0)), text) + + +def apply_aliases_diarised(text: str, entries: list[dict]) -> str: + """Like apply_aliases but skips the leading `[Label]` on each line.""" + if not text or not entries: + return text + out = [] + for line in text.split("\n"): + m = _LABEL_RE.match(line) + if m: + out.append(m.group(1) + apply_aliases(m.group(2), entries)) + else: + out.append(apply_aliases(line, entries)) + return "\n".join(out) + + +def apply_to_transcript(text: str, entries: list[dict]) -> str: + """Auto-route: diarised-aware when speaker labels are present.""" + if not text or not entries: + return text + if "[You]" in text or "[Others]" in text: + return apply_aliases_diarised(text, entries) + return apply_aliases(text, entries) + + +def reference_block(entries: list[dict]) -> str: + """Labelled glossary block for summary prompts. Empty config -> ''.""" + terms = [e["preferred"] for e in (entries or []) if e.get("preferred")] + if not terms: + return "" + bullets = "\n".join(f"- {t}" for t in terms) + return ( + "REFERENCE TERMS - exact spellings of names that may appear in the " + "transcript. Use these spellings; do NOT add content from this list.\n" + f"{bullets}\n\n" + ) diff --git a/src/summarizer.py b/src/summarizer.py index cb4383c6..5e4593d5 100644 --- a/src/summarizer.py +++ b/src/summarizer.py @@ -1039,6 +1039,11 @@ def _bedrock_chat(self, prompt: str, timeout_seconds: int = 300) -> str: raise raise RuntimeError("Bedrock chat failed after all retries") + def _keyword_reference_block(self) -> str: + from .config import get_config + from . import keywords + return keywords.reference_block(get_config().get_custom_keywords()) + def _create_permissive_prompt(self, transcript: str, language: str = "en", notes: str = None) -> str: """ Create an enhanced prompt with discussion_areas and improved extraction. @@ -1075,7 +1080,9 @@ def _create_permissive_prompt(self, transcript: str, language: str = "en", notes """ - return f"""{diarisation_note}{notes_context}You are a helpful meeting assistant. Summarise this meeting transcript into discussion areas, key points and any next steps mentioned. Only base your summary on what was explicitly discussed in the transcript. + ref = self._keyword_reference_block() + + return f"""{ref}{diarisation_note}{notes_context}You are a helpful meeting assistant. Summarise this meeting transcript into discussion areas, key points and any next steps mentioned. Only base your summary on what was explicitly discussed in the transcript. IMPORTANT: Do not infer or assume information that wasn't directly mentioned. @@ -1391,7 +1398,9 @@ def _create_markdown_prompt(self, transcript: str, language: str = "en", notes: if notes and notes.strip(): notes_context = f"USER NOTES (written during the meeting):\n{notes.strip()}\n\n" - return f"""{diarisation_note}{notes_context}Summarise this meeting transcript as markdown. Output ONLY the markdown below with no preamble, commentary, or explanation. Start directly with ## Summary. + ref = self._keyword_reference_block() + + return f"""{ref}{diarisation_note}{notes_context}Summarise this meeting transcript as markdown. Output ONLY the markdown below with no preamble, commentary, or explanation. Start directly with ## Summary. ## Summary A 1-3 sentence overview of the main topics and outcomes, written directly. Do not refer to "the transcript", "the meeting", or "the recording", and do not open with phrases like "The transcript discusses" or "In this meeting". @@ -1435,8 +1444,9 @@ def _create_template_report_prompt(self, transcript: str, template_prompt: str, diarisation_note = "" if "[You]" in transcript and "[Others]" in transcript: diarisation_note = "NOTE: [You] is the recorder, [Others] are remote participants.\n\n" + ref = self._keyword_reference_block() return ( - f"{diarisation_note}{notes_context}{template_prompt.strip()}\n\n" + f"{ref}{diarisation_note}{notes_context}{template_prompt.strip()}\n\n" "Base the report only on what was explicitly discussed; do not infer. " "Output the report as markdown with no preamble." f"{language_instruction}\n\nTRANSCRIPT:\n{transcript}" @@ -1704,7 +1714,9 @@ def generate_title(self, summary: str, transcript: str, language: str = "en") -> else: lang_instruction = "" - prompt = f"""Generate a short, descriptive title for this meeting based on the summary below. + ref = self._keyword_reference_block() + + prompt = f"""{ref}Generate a short, descriptive title for this meeting based on the summary below. RULES: 1. Maximum 6 words @@ -1802,7 +1814,8 @@ def _build_query_prompt(self, transcript: str, question: str, language: str = "e query_lang_instruction = f"\nRespond in {language_name}." if language_name != "Unknown" else "" else: query_lang_instruction = "" - return f"""Answer the following question based on the meeting content below (summary, key topics, and transcript). + ref = self._keyword_reference_block() + return f"""{ref}Answer the following question based on the meeting content below (summary, key topics, and transcript). Be concise and direct. If the answer requires inference from what was discussed, that's fine. Only say you don't know if the topic truly wasn't discussed at all.{query_lang_instruction} diff --git a/tests/test_config.py b/tests/test_config.py index 21e16df0..cc5c06d7 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,4 +1,5 @@ import json +import os import tempfile import unittest from pathlib import Path @@ -618,5 +619,32 @@ def test_mlx_to_gguf_is_exact_reverse_of_mlx_equivalents(self): self.assertEqual(len(Config._MLX_TO_GGUF), len(Config._MLX_EQUIVALENTS)) +class CustomKeywordsConfigTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.mkdtemp() + os.environ["STENOAI_USER_DATA_DIR"] = self._tmp + self.cfg = Config() + + def tearDown(self): + os.environ.pop("STENOAI_USER_DATA_DIR", None) + + def test_default_empty(self): + self.assertEqual(self.cfg.get_custom_keywords(), []) + + def test_set_get_roundtrip_and_persist(self): + ok = self.cfg.set_custom_keywords([{"preferred": "NexGen Suite", "aliases": ["NexGan Suite"]}]) + self.assertTrue(ok) + self.assertEqual( + self.cfg.get_custom_keywords(), + [{"preferred": "NexGen Suite", "aliases": ["NexGan Suite"]}], + ) + # persisted: a fresh Config reads it back + self.assertEqual(Config().get_custom_keywords()[0]["preferred"], "NexGen Suite") + + def test_set_normalizes(self): + self.cfg.set_custom_keywords([{"preferred": " ", "aliases": ["x"]}]) + self.assertEqual(self.cfg.get_custom_keywords(), []) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_keywords.py b/tests/test_keywords.py new file mode 100644 index 00000000..4058cbb7 --- /dev/null +++ b/tests/test_keywords.py @@ -0,0 +1,236 @@ +import unittest +from src import keywords + + +class ParseFormatTests(unittest.TestCase): + def test_no_colon_line(self): + self.assertEqual( + keywords.parse_keywords_text("DataFlow GmbH"), + [{"preferred": "DataFlow GmbH", "aliases": []}], + ) + + def test_aliases_split_on_first_colon(self): + self.assertEqual( + keywords.parse_keywords_text("NexGen Suite: NexGan Suite, NexGin Suite"), + [{"preferred": "NexGen Suite", "aliases": ["NexGan Suite", "NexGin Suite"]}], + ) + + def test_trims_and_drops_blank_lines_and_aliases(self): + self.assertEqual( + keywords.parse_keywords_text(" BrightLedger : bright ledger , \n\n"), + [{"preferred": "BrightLedger", "aliases": ["bright ledger"]}], + ) + + def test_round_trip(self): + text = "NexGen Suite: NexGan Suite, NexGin Suite\nBrightLedger: bright ledger\nDataFlow GmbH" + self.assertEqual( + keywords.format_keywords_text(keywords.parse_keywords_text(text)), + text, + ) + + +class NormalizeTests(unittest.TestCase): + def test_drops_empty_preferred(self): + self.assertEqual(keywords.normalize_keywords([{"preferred": " ", "aliases": ["x"]}]), []) + + def test_preferred_dedupe_case_insensitive_last_wins(self): + out = keywords.normalize_keywords([ + {"preferred": "BrightLedger", "aliases": ["bright ledger"]}, + {"preferred": "brightledger", "aliases": ["ledger"]}, + ]) + self.assertEqual(out, [{"preferred": "brightledger", "aliases": ["ledger"]}]) + + def test_global_alias_dedupe(self): + out = keywords.normalize_keywords([ + {"preferred": "A", "aliases": ["shared"]}, + {"preferred": "B", "aliases": ["shared", "bee"]}, + ]) + self.assertEqual(out, [ + {"preferred": "A", "aliases": ["shared"]}, + {"preferred": "B", "aliases": ["bee"]}, + ]) + + def test_alias_equal_to_preferred_dropped(self): + out = keywords.normalize_keywords([ + {"preferred": "Steno", "aliases": []}, + {"preferred": "X", "aliases": ["steno", "ex"]}, + ]) + self.assertEqual(out[1]["aliases"], ["ex"]) + + def test_caps(self): + many = [{"preferred": f"T{i}", "aliases": []} for i in range(300)] + self.assertEqual(len(keywords.normalize_keywords(many)), 200) + long = keywords.normalize_keywords([{"preferred": "p" * 200, "aliases": ["a" * 200]}]) + self.assertEqual(len(long[0]["preferred"]), 80) + self.assertEqual(len(long[0]["aliases"][0]), 80) + + def test_alias_cap(self): + out = keywords.normalize_keywords([ + {"preferred": "P", "aliases": [f"a{i}" for i in range(15)]}, + ]) + self.assertEqual(len(out[0]["aliases"]), 10) + + def test_last_wins_reclaims_alias(self): + out = keywords.normalize_keywords([ + {"preferred": "A", "aliases": ["x"]}, + {"preferred": "a", "aliases": ["x"]}, + ]) + self.assertEqual(out, [{"preferred": "a", "aliases": ["x"]}]) + + +ENTRIES = [ + {"preferred": "NexGen Suite", "aliases": ["NexGan Suite", "bright ledger", "C++"]}, +] + + +class ApplyAliasesTests(unittest.TestCase): + def test_empty_entries_is_identity(self): + self.assertEqual(keywords.apply_aliases("anything", []), "anything") + + def test_replaces_case_insensitive_whole_token(self): + self.assertEqual( + keywords.apply_aliases("we discussed nexgan suite today", ENTRIES), + "we discussed NexGen Suite today", + ) + + def test_does_not_match_inside_word(self): + # "bright ledger" must not fire inside "rebright ledgery" style runs + self.assertEqual(keywords.apply_aliases("xbright ledgerx", ENTRIES), "xbright ledgerx") + + def test_punctuation_alias_bounded(self): + # alias "C++" matches standalone but NOT inside C++17 + self.assertEqual(keywords.apply_aliases("use C++ now", ENTRIES), "use NexGen Suite now") + self.assertEqual(keywords.apply_aliases("C++17 release", ENTRIES), "C++17 release") + + def test_longest_alias_first(self): + entries = [ + {"preferred": "FOO", "aliases": ["ab"]}, + {"preferred": "BAR", "aliases": ["abc"]}, + ] + self.assertEqual(keywords.apply_aliases("abc", entries), "BAR") + + def test_diarised_label_protected(self): + entries = [{"preferred": "PREF", "aliases": ["you"]}] + out = keywords.apply_aliases_diarised("[You] you said hi", entries) + self.assertEqual(out, "[You] PREF said hi") + + def test_apply_to_transcript_autoroutes(self): + entries = [{"preferred": "PREF", "aliases": ["others"]}] + self.assertEqual( + keywords.apply_to_transcript("[Others] the others left", entries), + "[Others] the PREF left", + ) + + +class ApplyAliasesUnicodeTests(unittest.TestCase): + def test_unicode_case_fold_does_not_raise(self): + # 'İSTANBUL'.lower() -> 'i̇stanbul' (combining dot), not a mapping key. + # Must not raise KeyError; returns a string (matched text left unchanged + # is acceptable for the fold-asymmetry edge case). + entries = [{"preferred": "Istanbul City", "aliases": ["istanbul"]}] + result = keywords.apply_aliases("İSTANBUL", entries) + self.assertIsInstance(result, str) + + def test_unicode_letter_is_word_boundary(self): + # alias "foo" must NOT match inside äfoo / fooä (non-ASCII letter = \w) + entries = [{"preferred": "FOO_PREF", "aliases": ["foo"]}] + self.assertEqual(keywords.apply_aliases("äfoo", entries), "äfoo") + self.assertEqual(keywords.apply_aliases("fooä", entries), "fooä") + # but standalone still fires + self.assertEqual(keywords.apply_aliases("a foo b", entries), "a FOO_PREF b") + + def test_cpp_boundaries_hold_after_unicode_switch(self): + self.assertEqual(keywords.apply_aliases("use C++ now", ENTRIES), "use NexGen Suite now") + self.assertEqual(keywords.apply_aliases("C++17 release", ENTRIES), "C++17 release") + + +class DiarisedTimestampLabelTests(unittest.TestCase): + """FIX 2: the REAL production diarised line is `[MM:SS] [You] text` / + `[H:MM:SS] [Others] text` (transcriber._format_timestamp + speaker tag). + Both the timestamp AND the [You]/[Others] speaker label must be protected + from alias replacement - an alias `you`/`others` must not rewrite the label. + """ + + def test_speaker_label_protected_with_timestamp(self): + entries = [{"preferred": "Speaker Alpha", "aliases": ["you"]}] + line = "[00:05] [You] you said we should ship" + self.assertEqual( + keywords.apply_to_transcript(line, entries), + "[00:05] [You] Speaker Alpha said we should ship", + ) + + def test_others_label_protected_with_hour_timestamp(self): + entries = [{"preferred": "Team", "aliases": ["others"]}] + line = "[1:02:33] [Others] the others agreed" + self.assertEqual( + keywords.apply_to_transcript(line, entries), + "[1:02:33] [Others] the Team agreed", + ) + + def test_multiline_timestamped_diarised_block(self): + entries = [{"preferred": "NexGen Suite", "aliases": ["NexGan Suite"]}] + text = "[00:01] [You] we shipped NexGan Suite\n\n[00:09] [Others] great, NexGan Suite rocks" + self.assertEqual( + keywords.apply_to_transcript(text, entries), + "[00:01] [You] we shipped NexGen Suite\n\n[00:09] [Others] great, NexGen Suite rocks", + ) + + +class NormalizeMalformedConfigTests(unittest.TestCase): + """FIX 4: normalize_keywords must survive a hand-edited / corrupted + config.json without crashing or silently corrupting the alias table.""" + + def test_non_dict_entries_dropped(self): + out = keywords.normalize_keywords( + ["oops", 5, None, ["x"], {"preferred": "OK", "aliases": []}] + ) + self.assertEqual(out, [{"preferred": "OK", "aliases": []}]) + + def test_numeric_preferred_dropped(self): + self.assertEqual( + keywords.normalize_keywords([{"preferred": 42, "aliases": ["x"]}]), [] + ) + + def test_string_aliases_treated_as_single_alias_not_chars(self): + # The bug: `for a in "foo"` yields 'f','o','o'. Must be one alias "foo". + out = keywords.normalize_keywords([{"preferred": "P", "aliases": "foo"}]) + self.assertEqual(out, [{"preferred": "P", "aliases": ["foo"]}]) + + def test_non_list_aliases_dropped(self): + out = keywords.normalize_keywords([{"preferred": "P", "aliases": 123}]) + self.assertEqual(out, [{"preferred": "P", "aliases": []}]) + + def test_numeric_alias_elements_dropped(self): + out = keywords.normalize_keywords( + [{"preferred": "P", "aliases": [1, "ok", None]}] + ) + self.assertEqual(out, [{"preferred": "P", "aliases": ["ok"]}]) + + def test_control_and_newline_chars_stripped(self): + out = keywords.normalize_keywords( + [{"preferred": "A\nB\tC", "aliases": ["x\x00y"]}] + ) + self.assertEqual(out, [{"preferred": "A B C", "aliases": ["x y"]}]) + + def test_no_crash_on_fully_garbage_input(self): + # Must not raise, just yield []. + self.assertEqual(keywords.normalize_keywords([1, "a", None, 3.5]), []) + + +class ReferenceBlockTests(unittest.TestCase): + def test_empty_is_empty_string(self): + self.assertEqual(keywords.reference_block([]), "") + + def test_lists_preferred_terms(self): + block = keywords.reference_block([ + {"preferred": "NexGen Suite", "aliases": ["x"]}, + {"preferred": "BrightLedger", "aliases": []}, + ]) + self.assertIn("REFERENCE TERMS", block) + self.assertIn("- NexGen Suite", block) + self.assertIn("- BrightLedger", block) + self.assertTrue(block.endswith("\n\n")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_keywords_cli.py b/tests/test_keywords_cli.py new file mode 100644 index 00000000..e1d18549 --- /dev/null +++ b/tests/test_keywords_cli.py @@ -0,0 +1,23 @@ +import os +import tempfile +import unittest +from click.testing import CliRunner +from simple_recorder import cli + + +class KeywordsCliTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.mkdtemp() + os.environ["STENOAI_USER_DATA_DIR"] = self._tmp + + def tearDown(self): + os.environ.pop("STENOAI_USER_DATA_DIR", None) + + def test_set_then_get_roundtrip(self): + runner = CliRunner() + r1 = runner.invoke(cli, ["set-custom-keywords", "NexGen Suite: NexGan Suite"]) + self.assertEqual(r1.exit_code, 0, r1.output) + self.assertIn('"success": true', r1.output) + r2 = runner.invoke(cli, ["get-custom-keywords"]) + self.assertEqual(r2.exit_code, 0, r2.output) + self.assertIn("NexGen Suite: NexGan Suite", r2.output) diff --git a/tests/test_keywords_pipeline.py b/tests/test_keywords_pipeline.py new file mode 100644 index 00000000..45c10087 --- /dev/null +++ b/tests/test_keywords_pipeline.py @@ -0,0 +1,145 @@ +"""Pipeline-level regression tests for Custom Keywords healing in +``simple_recorder``: the silence-sentinel guard (data safety) and the +live-transcript fallback healing. +""" + +import asyncio +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from simple_recorder import MeetingPipeline, _parse_meeting_markdown +from src.transcriber import SILENCE_SENTINEL + + +def _reset_config(): + from src import config as _config + _config._config_instance = None + + +class _ConfigTempDirMixin(unittest.TestCase): + def setUp(self): + self._orig_dd = os.environ.get("STENOAI_USER_DATA_DIR") + self._tmp = tempfile.mkdtemp() + os.environ["STENOAI_USER_DATA_DIR"] = self._tmp + _reset_config() + + def tearDown(self): + if self._orig_dd is not None: + os.environ["STENOAI_USER_DATA_DIR"] = self._orig_dd + else: + os.environ.pop("STENOAI_USER_DATA_DIR", None) + _reset_config() + + +class SilenceSentinelHealingGuardTests(_ConfigTempDirMixin): + """FIX 1 (data safety): a configured alias that would rewrite the silence + sentinel must NOT mutate it - otherwise the downstream silence / live-rescue + check (an exact-string compare) breaks and a fake-empty transcript gets + summarised/saved with the audio possibly deleted.""" + + def _make_audio(self): + p = Path(self._tmp) / "meeting.wav" + p.write_bytes(b"\x00" * 2048) + return str(p) + + def _run_transcribe(self, transcript_result): + pipeline = MeetingPipeline() + # Pre-seed the transcriber so transcribe_audio never builds a real one. + with patch.object( + pipeline, "transcriber", create=True + ) as fake_transcriber: + fake_transcriber.transcribe_diarised.return_value = transcript_result + return asyncio.run(pipeline.transcribe_audio(self._make_audio(), "Note")) + + def test_sentinel_is_not_healed(self): + # Alias "speech" WOULD match inside "No speech detected in audio". + from src.config import get_config + get_config().set_custom_keywords( + [{"preferred": "Speechly", "aliases": ["speech"]}] + ) + out = self._run_transcribe({ + "text": SILENCE_SENTINEL, + "diarised_text": None, + "is_diarised": False, + "duration_seconds": None, + "detected_language": None, + }) + # The sentinel survives verbatim so the caller's silence check still fires. + self.assertEqual(out["transcript_text"], SILENCE_SENTINEL) + + def test_genuine_content_is_still_healed(self): + from src.config import get_config + get_config().set_custom_keywords( + [{"preferred": "Speechly", "aliases": ["speech"]}] + ) + out = self._run_transcribe({ + "text": "we discussed speech recognition today", + "diarised_text": None, + "is_diarised": False, + "duration_seconds": 5, + "detected_language": "en", + }) + self.assertEqual( + out["transcript_text"], "we discussed Speechly recognition today" + ) + + +class LiveTranscriptFallbackHealingTests(_ConfigTempDirMixin): + """FIX 3: when batch transcription returns the sentinel and we fall back to + the live transcript, that live text must be healed too (the batch-path + healing never touched it). Driven end to end through process-streaming with + auto-summarize OFF, so the transcript-only note is written with zero Ollama + calls.""" + + def test_live_fallback_text_is_healed(self): + from click.testing import CliRunner + import simple_recorder as sr + from src.config import get_config + + cfg = get_config() + cfg.set_custom_keywords( + [{"preferred": "NexGen Suite", "aliases": ["NexGan Suite"]}] + ) + cfg.set_auto_summarize_enabled(False) + + audio_file = str(Path(self._tmp) / "meeting.wav") + Path(audio_file).write_bytes(b"\x00" * 2048) + live_file = Path(self._tmp) / "live.txt" + live_file.write_text("we shipped NexGan Suite this sprint", encoding="utf-8") + + async def _fake_transcribe(self_pipeline, af, name="Recording"): + # Batch returned only silence -> triggers the live fallback. + return { + "transcript_text": SILENCE_SENTINEL, + "diarised_text": None, + "is_diarised": False, + "duration_seconds": 3, + "detected_language": None, + "transcription_failed": False, + } + + with patch.object(MeetingPipeline, "transcribe_audio", new=_fake_transcribe): + result = CliRunner().invoke( + sr.cli, + ["process-streaming", "--name", "Note", + "--live-transcript", str(live_file), audio_file], + ) + if result.exception is not None and not isinstance( + result.exception, SystemExit + ): + raise result.exception + + summary_path = Path(self._tmp) / "output" / "meeting_summary.md" + self.assertTrue(summary_path.exists(), result.output) + data = _parse_meeting_markdown(summary_path) + self.assertIn("NexGen Suite", data["transcript"]) + self.assertNotIn("NexGan Suite", data["transcript"]) + # Sanity: it really took the live-fallback branch. + self.assertTrue(data["session_info"].get("is_live_transcript")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_reprocess_keywords.py b/tests/test_reprocess_keywords.py new file mode 100644 index 00000000..69b6762f --- /dev/null +++ b/tests/test_reprocess_keywords.py @@ -0,0 +1,100 @@ +import json +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from src import keywords + + +class A1SeamTests(unittest.TestCase): + """The A1 seam is a thin call into keywords.apply_to_transcript; we assert + the helper does the right thing on the two derived strings the seam heals.""" + + def test_plain_and_diarised_healed(self): + entries = [{"preferred": "NexGen Suite", "aliases": ["NexGan Suite"]}] + plain = keywords.apply_to_transcript("about NexGan Suite", entries) + diarised = keywords.apply_to_transcript("[Others] NexGan Suite rocks", entries) + self.assertEqual(plain, "about NexGen Suite") + self.assertEqual(diarised, "[Others] NexGen Suite rocks") + + +class A2ReprocessTests(unittest.TestCase): + """reprocess must retroactively heal the stored transcript (JSON: transcript + + diarised_text; MD: ## Transcript) using configured Custom Keywords, and the + MD rewrite must preserve the folders frontmatter.""" + + def setUp(self): + self._orig_dd = os.environ.get("STENOAI_USER_DATA_DIR") + self._tmp = tempfile.mkdtemp() + os.environ["STENOAI_USER_DATA_DIR"] = self._tmp + # Fresh config bound to the temp data dir. + from src import config as _config + _config._config_instance = None + from src.config import get_config + get_config().set_custom_keywords( + [{"preferred": "NexGen Suite", "aliases": ["NexGan Suite"]}] + ) + + def tearDown(self): + if self._orig_dd is not None: + os.environ["STENOAI_USER_DATA_DIR"] = self._orig_dd + else: + os.environ.pop("STENOAI_USER_DATA_DIR", None) + from src import config as _config + _config._config_instance = None + + def _run_reprocess(self, summary_path): + # Patch the whole OllamaSummarizer class so __init__ never runs (no real + # Ollama probe / httpx.get to 11434), keeping the test hermetic. reprocess + # does `from src.summarizer import OllamaSummarizer` locally, so the class + # is looked up on the src.summarizer module at call time - patch it there. + import simple_recorder as sr + from click.testing import CliRunner + with patch("src.summarizer.OllamaSummarizer", autospec=True) as MockSummarizer: + MockSummarizer.return_value.summarize_transcript_streaming.return_value = ( + iter(["## Summary\nok\n"]) + ) + # reprocess snapshots the prior note as a backup (#249) and may + # regenerate the title; both read the summarizer instance. Give the + # autospec mock a concrete model_name and a no-op title so those + # paths run without touching a real model. + MockSummarizer.return_value.model_name = "mock-model" + MockSummarizer.return_value.generate_title.return_value = None + runner = CliRunner() + result = runner.invoke(sr.cli, ["reprocess", str(summary_path)]) + if result.exception is not None and not isinstance( + result.exception, SystemExit + ): + raise result.exception + return result + + def test_json_heals_both_fields(self): + p = Path(self._tmp) / "m.json" + p.write_text(json.dumps({ + "session_info": {"name": "M", "duration_minutes": 5}, + "transcript": "about NexGan Suite", + "diarised_text": "[Others] NexGan Suite rocks", + "is_diarised": True, + "folders": ["f1"], + })) + self._run_reprocess(p) + data = json.loads(p.read_text()) + self.assertEqual(data["transcript"], "about NexGen Suite") + self.assertEqual(data["diarised_text"], "[Others] NexGen Suite rocks") + + def test_md_heals_and_preserves_folders(self): + p = Path(self._tmp) / "m.md" + p.write_text( + '---\ntitle: "M"\nis_diarised: false\nfolders: ["f1"]\n---\n\n' + '## Summary\n\nold\n\n## Transcript\n\nabout NexGan Suite\n' + ) + self._run_reprocess(p) + text = p.read_text() + self.assertIn("about NexGen Suite", text) + self.assertIn('folders: ["f1"]', text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_summarizer_keywords.py b/tests/test_summarizer_keywords.py new file mode 100644 index 00000000..26e20380 --- /dev/null +++ b/tests/test_summarizer_keywords.py @@ -0,0 +1,31 @@ +import unittest +from unittest.mock import patch + +from src.summarizer import OllamaSummarizer + + +class SummarizerReferenceBlockTests(unittest.TestCase): + def _summarizer(self): + # Avoid network/Ollama init; we only call the pure prompt builders. + return OllamaSummarizer.__new__(OllamaSummarizer) + + def test_permissive_prompt_includes_block_when_configured(self): + s = self._summarizer() + with patch("src.config.get_config") as gc: + gc.return_value.get_custom_keywords.return_value = [ + {"preferred": "NexGen Suite", "aliases": []} + ] + prompt = s._create_permissive_prompt("hello transcript", language="en") + self.assertIn("REFERENCE TERMS", prompt) + self.assertIn("- NexGen Suite", prompt) + + def test_permissive_prompt_unchanged_when_empty(self): + s = self._summarizer() + with patch("src.config.get_config") as gc: + gc.return_value.get_custom_keywords.return_value = [] + prompt = s._create_permissive_prompt("hello transcript", language="en") + self.assertNotIn("REFERENCE TERMS", prompt) + + +if __name__ == "__main__": + unittest.main()