"""Shared cleanup for free-text (non-JSON) model output. Models wrap prose in ```` blocks, markdown code fences, literal ``\\n`` sequences, "Word count: 123" trailers, or TeX ``\\(...\\)`` delimiters that the frontend renderers don't recognise. Every route that returns model prose to the client should pass it through this module so the UI gets one predictable shape: * ``clean_llm_text`` — for fields the frontend renders as Markdown+LaTeX (chat answers, explanations, hints, lesson sections). Keeps Markdown, fixes delimiters and escapes. * ``to_plain_text`` — for fields the frontend renders verbatim (titles, answer options, short labels, notification copy). Strips Markdown syntax entirely while keeping ``$...$`` math untouched. * ``sanitize_tree`` — walks a parsed JSON payload and cleans every string, with per-key control over which fields are plain text. """ from __future__ import annotations import re from typing import Any, Callable, Iterable, Mapping, Optional from services.llm_json import TEX_N_COMMANDS __all__ = [ "clean_llm_text", "to_plain_text", "normalize_math_delimiters", "sanitize_tree", "PLAIN_TEXT_KEYS", ] _THINK_BLOCK_RE = re.compile(r"<\s*think(?:ing)?\b[^>]*>[\s\S]*?<\s*/\s*think(?:ing)?\s*>", re.IGNORECASE) _THINK_TAG_RE = re.compile(r"<\s*/?\s*think(?:ing)?\b[^>]*>", re.IGNORECASE) _ESCAPED_THINK_RE = re.compile(r"<\s*(/?)\s*think(?:ing)?\b[^&]*>", re.IGNORECASE) _WHOLE_FENCE_RE = re.compile(r"^\s*```[a-zA-Z0-9_-]*\s*\n([\s\S]*?)\n?```\s*$") _WORD_COUNT_RE = re.compile(r"\s*[*_(\[]*\s*word\s*count\s*[:=]?\s*[*_]*\s*\d+\s*(?:words?)?\s*[*_)\]]*\s*$", re.IGNORECASE) _LEADING_LABEL_RE = re.compile( r"^\s*(?:(?:final\s+)?answer|response|message|explanation|hint|output)\s*:\s*", re.IGNORECASE, ) _DISPLAY_TEX_RE = re.compile(r"\\\[([\s\S]*?)\\\]") _INLINE_TEX_RE = re.compile(r"\\\(([\s\S]*?)\\\)") _MATH_SEGMENT_RE = re.compile(r"(\$\$[\s\S]*?\$\$|\$[^$\n]+\$)") _CODE_SEGMENT_RE = re.compile(r"(```[\s\S]*?```|`[^`\n]+`)") _TRAILING_WS_RE = re.compile(r"[ \t]+\n") _MULTI_BLANK_RE = re.compile(r"\n{3,}") # Keys whose values the UI renders as plain strings (no Markdown pass). PLAIN_TEXT_KEYS: frozenset = frozenset({ "title", "lessontitle", "headline", "label", "name", "topic", "subject", "options", "choices", "correctanswer", "correct_answer", "answer", "correctanswertext", "videotitle", "videochannel", "suggested_action", "suggestedaction", "tone", "type", }) def _strip_think_blocks(text: str) -> str: cleaned = _ESCAPED_THINK_RE.sub(lambda m: f"<{m.group(1)}think>", text) previous = None while previous != cleaned: previous = cleaned cleaned = _THINK_BLOCK_RE.sub("", cleaned) # An unclosed at the start swallows everything up to the first blank line. lowered = cleaned.lower() open_idx = lowered.find(" str: match = _WHOLE_FENCE_RE.match(text) return match.group(1) if match else text def normalize_math_delimiters(text: str) -> str: """Rewrite ``\\(x\\)`` → ``$x$`` and ``\\[x\\]`` → ``$$x$$`` so remark-math parses them.""" if not text or "\\" not in text: return text text = _DISPLAY_TEX_RE.sub(lambda m: f"$${m.group(1).strip()}$$", text) text = _INLINE_TEX_RE.sub(lambda m: f"${m.group(1).strip()}$", text) return text def _outside_code(text: str, fn: Callable[[str], str]) -> str: parts = _CODE_SEGMENT_RE.split(text) return "".join(part if i % 2 else fn(part) for i, part in enumerate(parts)) def _outside_math(text: str, fn: Callable[[str], str]) -> str: parts = _MATH_SEGMENT_RE.split(text) return "".join(part if i % 2 else fn(part) for i, part in enumerate(parts)) def _fix_escapes(segment: str) -> str: # JSON-escaped newlines/tabs that survived parsing as literal text. # A letter run after \n or \t can be a TeX command (\neq, \times): keep those. segment = segment.replace("\\r\\n", "\n") segment = re.sub( r"\\n([A-Za-z]*)", lambda m: m.group(0) if "n" + m.group(1) in TEX_N_COMMANDS else "\n" + m.group(1), segment, ) segment = re.sub(r"\\t(?![A-Za-z])", " ", segment) # Double-escaped TeX from over-escaped JSON strings: \\frac → \frac segment = re.sub(r"\\\\(?=[A-Za-z])", r"\\", segment) # Escaped dollars in front of TeX commands block the math delimiter. segment = re.sub(r"\\\$(?=\\)", "$", segment) return segment def clean_llm_text(text: Any, *, max_chars: Optional[int] = None) -> str: """Normalise model prose for Markdown+LaTeX rendering on the client.""" if text is None: return "" if not isinstance(text, str): text = str(text) cleaned = text.replace("\r\n", "\n").replace("\r", "\n") cleaned = _strip_think_blocks(cleaned) cleaned = _WORD_COUNT_RE.sub("", cleaned) cleaned = _unwrap_fence(cleaned.strip()) cleaned = _outside_code(cleaned, _fix_escapes) cleaned = _outside_code(cleaned, normalize_math_delimiters) cleaned = _WORD_COUNT_RE.sub("", cleaned) cleaned = _TRAILING_WS_RE.sub("\n", cleaned) cleaned = _MULTI_BLANK_RE.sub("\n\n", cleaned).strip() # Model echoing the label it was asked for ("Message: ..."). if cleaned.count("\n") <= 2: cleaned = _LEADING_LABEL_RE.sub("", cleaned, count=1) # Whole-answer wrapped in matching quotes. if len(cleaned) >= 2 and cleaned[0] == cleaned[-1] and cleaned[0] in "\"“”'": cleaned = cleaned[1:-1].strip() if max_chars is not None and len(cleaned) > max_chars: cleaned = cleaned[:max_chars].rstrip() return cleaned _MD_HEADING_RE = re.compile(r"^\s{0,3}#{1,6}\s+", re.MULTILINE) _MD_BULLET_RE = re.compile(r"^\s{0,3}(?:[-*+•]|\d+[.)])\s+", re.MULTILINE) _MD_QUOTE_RE = re.compile(r"^\s{0,3}>\s?", re.MULTILINE) _MD_BOLD_RE = re.compile(r"(\*\*|__)(?=\S)([\s\S]*?\S)\1") _MD_ITALIC_RE = re.compile(r"(? str: segment = _MD_HR_RE.sub("", segment) segment = _MD_HEADING_RE.sub("", segment) segment = _MD_QUOTE_RE.sub("", segment) segment = _MD_LINK_RE.sub(r"\1", segment) previous = None while previous != segment: previous = segment segment = _MD_BOLD_RE.sub(r"\2", segment) segment = _MD_ITALIC_RE.sub(r"\2", segment) segment = _MD_STRIKE_RE.sub(r"\1", segment) segment = _MD_CODE_RE.sub(r"\1", segment) return segment def to_plain_text( text: Any, *, single_line: bool = False, strip_bullets: bool = True, strip_option_label: bool = False, max_chars: Optional[int] = None, ) -> str: """Strip Markdown syntax for fields the UI shows verbatim; math stays delimited.""" cleaned = clean_llm_text(text) if not cleaned: return "" if strip_bullets: cleaned = _MD_BULLET_RE.sub("", cleaned) cleaned = _outside_math(cleaned, _strip_markdown_segment) if strip_option_label: cleaned = _LEADING_OPTION_LABEL_RE.sub("", cleaned, count=1) if single_line: cleaned = re.sub(r"\s*\n+\s*", " ", cleaned) cleaned = re.sub(r"[ \t]{2,}", " ", cleaned).strip() if max_chars is not None and len(cleaned) > max_chars: cleaned = cleaned[:max_chars].rstrip() return cleaned def sanitize_tree( value: Any, *, plain_keys: Iterable[str] = PLAIN_TEXT_KEYS, skip_keys: Iterable[str] = (), _key: Optional[str] = None, _plain: bool = False, ) -> Any: """Recursively clean every string in a parsed model payload. Strings under a key in ``plain_keys`` (case-insensitive) go through ``to_plain_text``; everything else through ``clean_llm_text``. Keys in ``skip_keys`` (ids, urls, codes) are returned untouched. """ plain = {k.lower() for k in plain_keys} skip = {k.lower() for k in skip_keys} | {"id", "url", "embedurl", "thumbnailurl", "videoid", "code", "competencycode", "competency_code", "storage_path", "storagepath", "source_file"} key_l = (_key or "").lower() if key_l in skip: return value if isinstance(value, str): return to_plain_text(value, single_line=True) if _plain or key_l in plain else clean_llm_text(value) if isinstance(value, Mapping): return { k: sanitize_tree(v, plain_keys=plain_keys, skip_keys=skip_keys, _key=str(k), _plain=False) for k, v in value.items() } if isinstance(value, list): # A list under a plain key (e.g. "options") is a list of plain strings. inherit_plain = _plain or key_l in plain return [ sanitize_tree(v, plain_keys=plain_keys, skip_keys=skip_keys, _key=_key, _plain=inherit_plain) for v in value ] return value