feat(keywords): Custom Keywords backend — glossary-driven transcript + summary correction (draft)

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:

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)

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"(?<!\w)(?:{body})(?!\w)", re.IGNORECASE)
+    return pattern, mapping
+
+
+def apply_aliases(text: str, entries: list[dict]) -> 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"
+    )

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}
 

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()