from __future__ import annotations import json import re from typing import cast from s2s_client.core import logging as app_logging # --------------------------------------------------------------------------- # APIPILOT — Reasoning engine integration (two passes) # --------------------------------------------------------------------------- # Deduplication/start/end markers used by _classify_context_block to avoid # injecting a block already supplied by the client or found in base_instructions. APIPILOT_PASS1_LIST_MARKER = "<<>>" APIPILOT_PASS2_RESULT_MARKER = "<<]{1,3}", re.IGNORECASE | re.DOTALL, ) # 2026-09-25: observed inverted marker (incorrect LLM syntax): # >>>APIPILOT_CALL:NAME=tool,ARGUMENTS=json<<< _APIPILOT_CALL_MARKER_INVERTED_RE = re.compile( r"[>]{1,3}\s*APIPILOT_CALL:NAME=([A-Za-z0-9_.-]+),ARGUMENTS=(.*?)[<]{1,3}", re.IGNORECASE | re.DOTALL, ) def _apipilot_parse_call_marker(answer_text: str, *, defaults: dict | None = None) -> tuple[str | None, dict | None]: """Parse the mandatory CALL marker emitted by the LLM in pass 1. Returns (tool_name, arguments_dict), or (None, None) when no marker exists. Supports: (1) ARGUMENTS={"key":"val"}, valid JSON. (2) ARGUMENTS={}, a valid empty object. (3) ARGUMENTS= without braces, observed on realtime WebSocket sessions. (4) key=val,key2=val2 strings through a manual parsing fallback. (5) After parsing, fill missing required arguments from defaults such as phone and episode_id. """ if not isinstance(answer_text, str) or not answer_text.strip(): app_logging.logger.info("[APIPILOT-CALL-NAME] USE=PARSE_EMPTY_INPUT name=None markerFound=N") return None, None m = _APIPILOT_CALL_MARKER_RE.search(answer_text.strip()) inverted = False if not m: m = _APIPILOT_CALL_MARKER_INVERTED_RE.search(answer_text.strip()) inverted = bool(m) if not m: app_logging.logger.info( "[APIPILOT-CALL-NAME] USE=PARSE_NO_MATCH name=None markerFound=N textHead=%r", answer_text.strip()[:80] ) return None, None name_raw = m.group(1).strip() args_raw = (m.group(2) if m.lastindex and m.lastindex >= 2 and isinstance(m.group(2), str) else "").strip() if not name_raw: app_logging.logger.info( "[APIPILOT-CALL-NAME] USE=PARSE_EMPTY_NAME name=None markerFound=Y argsRawHead=%r inverted=%s", args_raw[:60], ("Y" if inverted else "N"), ) return None, None app_logging.logger.info( "[APIPILOT-CALL-NAME] USE=PARSE_MARKER_OK name=%r argsRawLen=%d markerFound=Y inverted=%s", name_raw, len(args_raw), ("Y" if inverted else "N"), ) args: dict = {} # Parse args_raw into a dict, tolerating common formatting variations. if args_raw == "" or args_raw in {"{}", "null", "None"}: args = {} else: # Attempt 1: standard JSON, with or without braces. _candidate = args_raw try: args = json.loads(_candidate) except Exception: try: # Attempt 2: supply missing outer braces. args = json.loads("{" + _candidate + "}") except Exception: try: # Attempt 3: replace single quotes with double quotes. args = json.loads(_candidate.replace("'", '"')) except Exception: try: # Attempt 4: supply braces and replace single quotes. args = json.loads("{" + _candidate.replace("'", '"') + "}") except Exception: # Manual fallback 5: equals signs and commas. # Example: args_raw="phone=+56,name=Mario" → dict. args = {} for piece in args_raw.split(","): kv = piece.strip() if not kv or "=" not in kv: continue k, v = kv.split("=", 1) k = k.strip().strip('"').strip("'") v = v.strip().strip('"').strip("'") if k: args[k] = v if not isinstance(args, dict): args = {} # Stage 2: fill missing arguments from defaults (phone, etc.). if isinstance(defaults, dict) and defaults: for dk, dv in defaults.items(): if dk not in args or args[dk] in (None, "", []): if dv not in (None, "", []): args[dk] = dv app_logging.logger.info( "[APIPILOT-CALL-NAME] USE=PARSE_RETURN_FINAL name=%r finalArgsKeys=%s defaultsApplied=%s", name_raw, sorted(list(args.keys())), ("Y" if isinstance(defaults, dict) and defaults else "N"), ) return name_raw, args def _resolve_apipilot_reasoning_context_block( tools_list: list[dict] | None, *, call_result: dict | None = None, context_defaults: dict | None = None, ) -> str: """Build the ApiPilot context block for the reasoning prompt. With call_result=None, pass 1 supplies the tool list and marker syntax the model must emit when a description matches. With call_result={'name':str, 'text':str, 'error':bool}, pass 2 injects the execution result and requests a final spoken/text response for the user. The optional context_defaults dict adds known default values to pass 1. The model must use these exact values for required arguments rather than inventing them. Usually contains {"phone": "+56..."} from inject_phone. """ # Pass 2: the tool result is already available. if isinstance(call_result, dict) and call_result.get("name"): name = call_result.get("name") or "?" text = (call_result.get("text") if isinstance(call_result.get("text"), str) else "") or "" is_err = bool(call_result.get("error")) status_label = "ERROR" if is_err else "OK" title = f"<<>>" trailer = ( "Now reinterpret THE RESULT TEXT (above) in clear, concise natural language, " "exclusively in the user's language (language field; Spanish by default). " "Do not mention the internal tool or technical details. " "Respond ONLY with the final content intended for the user." ) return ( f"{APIPILOT_PASS2_RESULT_MARKER}PASS2>>>\n" "### Invoked ApiPilot tool result\n" f"{title}\n" f"\n```tool-output\n" f"{text.rstrip() if text.strip() else '(empty output)'}\n" f"```\n\n" f"{trailer}\n" ) # Pass 1: available tools and marker rules. tools = tools_list if isinstance(tools_list, list) else [] # Limit to 100 tools, although the guide specifies at most 50 for the demo. tools_cap = tools[:100] if not tools_cap: return "" entries: list[str] = [] for t in tools_cap: if not isinstance(t, dict): continue name = t.get("name") if isinstance(t.get("name"), str) else "" if not name: continue desc = cast(str, t.get("description")) if isinstance(t.get("description"), str) else "" params = t.get("parameters") if isinstance(t.get("parameters"), dict) else None requires_auth = bool(t.get("requires_auth")) required_fields: list[str] = [] if isinstance(params, dict) and isinstance(params.get("required"), list): required_fields = [str(x) for x in cast(list, params.get("required")) if isinstance(x, str)] # Limit descriptions to about 500 characters to control prompt size. desc_cut = desc[:500].rstrip() if len(desc) > 500: desc_cut += "…" snippet = f"- name: {name}\n requires_auth: {('true' if requires_auth else 'false')}" if required_fields: snippet += f"\n required_parameters: [{', '.join(required_fields[:40])}]" if desc_cut: snippet += f"\n description: {desc_cut}" entries.append(snippet) if not entries: return "" rule = ( f"{APIPILOT_PASS1_LIST_MARKER}\n" "### Available dynamic tools (ApiPilot AquaPortal)\n" "\n" "Each tool has a `name` field (a unique identifier: do NOT modify it, " "use camelCase, or translate it) and a `description` describing its use case.\n" "\n" "MANDATORY EXECUTION RULES:\n" "1. If the USER REQUEST (transcript) clearly matches the `description` of " "EXACTLY one tool below, emit ONLY one line as the FIRST line of YOUR RESPONSE, " "with this exact syntax:\n" ' <<>>\n' "2. ARGUMENTS={…} MUST be valid JSON (double quotes, commas, no comments). " "Include ONLY fields listed in `required_parameters:`, if present.\n" "3. Add NO text after the marker line: no explanations or previews. The engine " "will execute the tool and supply its result in a second call.\n" "4. If NO description matches, emit NO marker and respond normally in Spanish.\n" "5. If multiple descriptions match, choose the most specific one and use the marker.\n" "6. 🔴 EXCLUSIVE RESPONSE MODE (HIGHEST priority, mandatory): when using a CALL " "marker, YOUR ENTIRE RESPONSE MUST CONTAIN ONLY THE MARKER LINE. " "Write NOTHING BEFORE it: no 'Claro que sí', 'Voy a revisar', " "'Voy a generar tu briefing', greeting, or transition. " "Write NOTHING AFTER it: no comments, previews, or explanations. " "Exactly one line containing the marker. Additional text prevents the engine " "from recognizing the marker and executing the tool.\n" "8. INCORRECT examples (NEVER do this):\n" " 'Claro que sí, ahora ejecuto'\n" " 'Voy a generar tu briefing ejecutivo personal.'\n" " 'Claro que sí.\\n<<>>'\n" " '<<>>\\nAhora proceso el resultado.'\n" "9. 🔴 TTS ANTI-LEAK RULES (mandatory, HIGHEST audio priority):\n" " <<>> IS AN INTERNAL SYSTEM SIGNAL and MUST NEVER be spoken,\n" " read aloud, or synthesized by TTS. When emitting the marker:\n" " - INTERNAL REASONING (text/transcript) MUST contain the exact marker line.\n" " - FINAL TTS AUDIO heard by the user MUST contain ONLY this placeholder:\n" ' "Voy a consultar tus datos, un momento por favor."\n' " - NEVER speak any of these technical terms in final TTS audio:\n" " APIPILOT, CALL, ARGUMENTS, tool, herramienta, ejecutar, function, function call,\n" " marcador, marker, even to explain or describe the call.\n" "\n" ) # 2026-09-25 known defaults: inject values already resolved from the URL or # profile, such as phoneNumber, so tool arguments use those exact values. if isinstance(context_defaults, dict) and context_defaults: defaults_lines = [] for dk, dv in context_defaults.items(): if dv in (None, "", []): continue if isinstance(dv, (dict, list)): try: dv_repr = json.dumps(dv, ensure_ascii=False) except Exception: dv_repr = str(dv) else: dv_repr = str(dv) defaults_lines.append(f'- {dk} = "{dv_repr}"') if defaults_lines: rule += ( "KNOWN DEFAULT VALUES (use these exact values in ARGUMENTS={…} when " "the corresponding tool lists these names in required_parameters; " "do not invent, translate, or normalize them):\n" f"{chr(10).join(defaults_lines)}\n" "\n" ) rule += f"Tool list (name / description):\n{chr(10).join(entries)}\n<<>>" return rule def _sanitize_pass2_internal_context(text: str) -> str: """Lightly sanitize the pass-2 block injected into the upstream model. Unlike _sanitize_text_strip_apipilot_markers, preserve the pass-2 wrapper: the model needs it to recognize tool data. Limit sanitation to filesystem path leaks, fragmented marker tokens, and contextual phone leaks. """ if not isinstance(text, str) or not text.strip(): return text # Block 3: Unix/Windows/TRAE path leaks, matching the user-output sanitizer. import re as _re_pass2 text = _re_pass2.sub( r"(?i)(?:/[A-Za-z0-9_.\-@~]+){2,}/(?:\.trae[^\s\"']*|workspaceStorage[^\s\"']*|User[^\s\"']*|home[^\s\"']*)", " ", text, ) text = _re_pass2.sub(r"(?i)(?:[A-Z]:)?[\\/](?:Users|home|root|tmp|var|mnt|opt|srv)[\\/][^\s\"']{5,}", " ", text) text = _re_pass2.sub(r"(?i)(?:\.trae-server|\.trae[^\s\"']{0,40}|workspaceStorage[0-9a-f]+)", " ", text) text = _re_pass2.sub(r"\s{2,}", " ", text) # Block 4: fragmented marker leaks, tier 4. tier4_out_parts: list[str] = [] for _line in text.splitlines(keepends=True): _orig = _line _low = _line.lower() if "apipilot" in _low and ( ("call" in _low and "name=" in _low) or ("arguments=" in _low) or ("result" in _low and "pass" in _low) ): if "<<<" in _line or ">>>" in _line: _line = _re_pass2.sub( r"[<]{1,3}\s*APIPILOT_(?:CALL|RESULT)[^<>\n]{0,500}[>]{1,3}", " ", _line, flags=_re_pass2.IGNORECASE, ) if "apipilot" in _line.lower(): _line = _re_pass2.sub( r"(?i)\bAPIPILOT(?:_CALL|_RESULT)?\b", " ", _line, ) _line = _re_pass2.sub( r"(?i)\b(?:NAME|ARGUMENTS|STATUS|REV|PASS2)\s*[:=]", " ", _line, ) tier4_out_parts.append(_line if _line.strip() else _orig) text = "".join(tier4_out_parts) text = _re_pass2.sub(r"\s{2,}", " ", text) return text.strip() if isinstance(text, str) else text def _sanitize_text_strip_apipilot_markers( text: str, *, placeholder: str | None = None, ) -> str: """Remove all internal ApiPilot markers from user-visible chat and TTS text. Replace pass-1 CALL markers with a Spanish placeholder. Silently remove pass-2 RESULT markers, their heading, tool-output block and instruction trailer. These server-only signals must never reach the user. Args: text: Text to sanitize. placeholder: Optional replacement; defaults to the Spanish UI/voice message. """ if not isinstance(text, str) or not text: return text if placeholder is None: placeholder = "Voy a consultar tus datos, un momento por favor." # ================================================================ # Block 1: silently remove pass-2 RESULT markers and wrappers from TTS/UI. # ================================================================ # (1a) Full pass-2 wrapper first: marker, heading, status, output, trailer. pass2_full_block_pattern = re.compile( r"[ \t]*(?:\r?\n|\r)?[ \t]*" r"<<>>[ \t]*(?:\r?\n|\r)+" r"###[ \t]+(?:Resultado de la herramienta ApiPilot invocada|Invoked ApiPilot tool result)[ \t]*(?:\r?\n|\r)+" r"<<>>[ \t]*(?:\r?\n|\r)*" r"(?:\r?\n|\r)?" r"```tool-output[ \t]*(?:\r?\n|\r)?" r"(?P.*?)" r"(?:\r?\n|\r)?```[ \t]*(?:\r?\n|\r){0,3}" r"(?:" r"(?:Ahora debes reinterpretar EL TEXTO RESULTADO|Now reinterpret THE RESULT TEXT)[\s\S]{0,1200}?" r"(?:\r?\n|\r){0,2}" r")?" r"[ \t]*", re.IGNORECASE | re.DOTALL, ) text = pass2_full_block_pattern.sub("", text) # (1b) Standalone PASS2 markers when the wrapper is incomplete or fragmented. pass2_marker_pattern = re.compile( r"[ \t]*(?:\r?\n|\r)?[ \t]*<<>>[ \t]*(?:\r?\n|\r)?[ \t]*", re.IGNORECASE, ) text = pass2_marker_pattern.sub("", text) # (1c) Residual Markdown heading after the marker has been removed. pass2_title_pattern = re.compile( r"[ \t]*(?:\r?\n|\r)?[ \t]*###[ \t]+(?:Resultado de la herramienta ApiPilot invocada|Invoked ApiPilot tool result)[ \t]*(?:\r?\n|\r)?[ \t]*", re.IGNORECASE, ) text = pass2_title_pattern.sub("", text) # (1d) Standalone tool-output blocks without a PASS2 wrapper. pass2_codeblock_pattern = re.compile( r"```tool-output[ \t]*(?:\r?\n|\r)?.*?(?:\r?\n|\r)?```", re.IGNORECASE | re.DOTALL, ) text = pass2_codeblock_pattern.sub("", text) # (1e) Residual instruction trailers missed by the full-wrapper pattern. pass2_trailer_pattern = re.compile( r"(?:Ahora debes reinterpretar EL TEXTO RESULTADO[\s\S]{0,1200}?contenido final que le quieres transmitir al usuario|Now reinterpret THE RESULT TEXT[\s\S]{0,1200}?final content intended for the user)\.", re.IGNORECASE | re.DOTALL, ) text = pass2_trailer_pattern.sub("", text) # Broader fallback also handles trailers truncated before their closing text. pass2_trailer_fallback = re.compile( r"(?:Ahora debes reinterpretar[\s\S]{0,1500}?(?:al usuario\.|$)|Now reinterpret THE RESULT TEXT[\s\S]{0,1500}?(?:for the user\.|$))", re.IGNORECASE | re.DOTALL, ) text = pass2_trailer_fallback.sub("", text) # Handle residual middle fragments after the trailer's opening was removed. pass2_trailer_broad = re.compile( r"(?:en lenguaje natural[\s\S]{0,800}?contenido final que le quieres transmitir al usuario|in clear, concise natural language[\s\S]{0,800}?final content intended for the user)\.?", re.IGNORECASE | re.DOTALL, ) text = pass2_trailer_broad.sub("", text) # ================================================================ # Block 2: pass-1 CALL marker replacement with a Spanish placeholder. # ================================================================ marker_full_pattern = re.compile( r"<<>>", re.IGNORECASE | re.DOTALL, ) line_wrapped_pattern = re.compile( r"[ \t]*(?:\r?\n|\r)?[ \t]*<<>>[ \t]*(?:\r?\n|\r)?[ \t]*", re.IGNORECASE | re.DOTALL, ) def _replacement_full(_m: re.Match) -> str: """Replace standalone or inline markers with the spoken placeholder.""" return placeholder text = line_wrapped_pattern.sub(_replacement_full, text) text = marker_full_pattern.sub(_replacement_full, text) # Remove truncated CALL markers such as <<\n]{0,500}?)?>>*", re.IGNORECASE, ) text = trunc_call_pattern.sub(_replacement_full, text) # Make closing brackets optional to remove RESULT markers truncated by TTS, # such as <<>>. trunc_result_pattern = re.compile( r"<<?>?>?", re.IGNORECASE, ) text = trunc_result_pattern.sub("", text) # Remove remaining APIPILOT_RESULT fragments after partial sanitation. trunc_result_residual = re.compile( r"<{0,3}APIPILOT_RESULT(?::[^\s<]{0,800})?<{0,3}", re.IGNORECASE, ) text = trunc_result_residual.sub("", text) # Collapse whitespace left by pass-2 block removal. text = re.sub(r"[ \t]{2,}", " ", text) text = re.sub(r"\n{3,}", "\n\n", text) # Remove residual trailer fragments and empty parentheses before TTS. text = re.sub(r"\s*\((?:arriba|above)\)\s*", " ", text, flags=re.IGNORECASE) text = re.sub(r"\(\s*\)", "", text) # ================================================================ # Block 3: remove filesystem paths and host information leaked into TTS. # Match long absolute Unix/Windows/mixed paths and known workspace tokens. # ================================================================ # (3a) Absolute Unix paths and nested paths with multiple separators. unix_path_pattern = re.compile( r"\b(?:/(?:home|tmp|var|root|opt|mnt|media|srv|proc|sys|etc|usr|Users|trae)[/\w@.+\-:#$%^&()[\]{}!~=,]*){2,}[A-Za-z0-9_./\-+@#$%^&()[\]{}!~=,:]*\.(?:txt|md|log|json|csv|pdf|doc|docx|py|js|ts|html|css|sh|yaml|yml|env|conf|cfg|ini|mp3|wav|ogg|png|jpg|jpeg|gif)\b" r"|" # Broad fallback for long absolute Unix paths under known directories, # including paths without extensions; avoid short URLs such as /api/v1. r"(?:^|\s+|\(|[,.;:!?])(/(?:home|tmp|var|root|opt|mnt|media|srv|proc|sys|etc|usr|Users|trae)[/A-Za-z0-9_\-+@#$%^&()[\]{}!~=,.]{10,})(?=$|\s|\)|[,.;:!?])", re.IGNORECASE, ) text = unix_path_pattern.sub("", text) # (3b) Windows paths with single/double backslashes and mixed separators. win_path_pattern = re.compile( r"(?:^|\s|\(|[,.;:])" r"([A-Za-z]:[\\/]" r"|(?:[\\/]{1,2}(?:home|Users|Windows|Program|Program\sFiles|ProgramData|Users|trae|AppData)[\\/])" r"|(?:[\\/]{1,2}\.(?:trae|trae-server|vscode|config|cache)[\\/])" r")" r"[A-Za-z0-9_@.+\-:#$%^&()[\]{}!~=,\\/\s]{10,}" r"(?:\.(?:txt|md|log|json|csv|pdf|doc|docx|py|js|ts|html|css|sh|yaml|yml|env|conf|cfg|ini|mp3|wav|ogg|png|jpg|jpeg|gif))?" r"(?=$|\s|\)|[,.;:!?])", re.IGNORECASE, ) text = win_path_pattern.sub("", text) # (3c) TRAE workspace tokens, including fragments without complete paths. trae_leak_tokens = re.compile( r"\b(?:workspaceStorage|long-text|\.trae-server|\.trae|ckg_server|ai-agent)\b[^\s()'\"`,.]{0,80}", re.IGNORECASE, ) text = trae_leak_tokens.sub("", text) # (3d) Remove long tokens containing mixed separators or many path segments. mixed_path_pattern = re.compile( r"(?=\S{0,200})" r"(?=\S*(?:[/\\].*){5,})" r"(?:\S{10,})", re.IGNORECASE, ) def _mixed_safe(m: re.Match[str]) -> str: s = m.group(0) has_slash = "/" in s has_back = "\\" in s if (has_slash and has_back) or s.count("/") + s.count("\\") >= 6: return "" return s text = mixed_path_pattern.sub(_mixed_safe, text) # ================================================================ # Block 4: remove fragmented APIPILOT CALL text from TTS. The upstream # pipeline (start_pipeline_rag / Qwen3-TTS) can remove <<< >>> and read: # "APIPILOT CALL name get_user_briefing arguments phone +56912345678" # The bracketed patterns cannot catch this, so use contextual fallback tiers. # ================================================================ # Tier 1: full normalized form without braces, through the sentence ending. call_normalized_full = re.compile( r"(?:^|\s+|\(|[,.;:!?\"\'])\s*" r"(?:" r"APIPILOT[\s_\-]+CALL\s+(?:marker\s+)?(?:de\s+)?(?:señal\s+)?(?:herramienta\s+)?(?:funci[oó]n\s+)?" r"|marcador\s+APIPILOT(?:\s+CALL)?\s*" r"|señal\s+de\s+llamada\s+APIPILOT\s*" r"|llamada\s+(?:a\s+)?(?:la\s+)?(?:herramienta|funci[oó]n)\s+APIPILOT\s*" r")" r"(?:\s*(?:con\s+)?(?:identificador|ID|name|nombre)\s*[:=]?\s*[A-Za-z0-9_.-]+)?" r"(?:\s*(?:y\s+)?(?:con\s+)?(?:argumentos|ARGUMENTS|params|par[áa]metros)\s*[:=]?\s*(?:\{[^{}]{0,800}\}|[\"'][^\"']{0,200}[\"']|[^\s.!,?;]{3,100}(?:\s*[,:]\s*[^\s.!,?;]{3,100})*\s*)?)?" r"(?:\s*(?:phone|tel[ée]fono|correo|email)\s*[:=]?\s*[+0-9A-Za-z._@\-]{6,50})?" r"(?=$|\s+|\)|[,.;:!?\"\'])", re.IGNORECASE, ) text = call_normalized_full.sub(placeholder, text) # Tier 2: shortened forms with APIPILOT_CALL or adjacent APIPILOT and CALL. apipilot_call_token = re.compile( r"(?:^|\s|\(|[,.;:!?\"\'])\s*" r"APIPILOT(?:[\s_\-]+CALL|_CALL|\s+llamada|\s+marker)" r"(?:[^.!,?;\n]{0,600})?" r"(?=$|\s|\)|[,.;:!?\"\'])", re.IGNORECASE, ) text = apipilot_call_token.sub(placeholder, text) # Tier 3: standalone service tokens in call/name/argument contexts. service_token_context = re.compile( r"(?:" r"ARGUMENTS\s*[:=]\s*(?:\{[^{}]{0,800}\}|[^\s,.;!?]{3,100})" r"|\barguments\s+(?:de\s+)?(?:la\s+)?(?:llamada|herramienta|funci[oó]n)\b[^\n.]{0,200}" r")", re.IGNORECASE, ) text = service_token_context.sub("", text) # Tier 4: remove sentences containing APIPILOT with technical call context. def _apipilot_sentence(m: re.Match[str]) -> str: sentence = m.group(0) lowered = sentence.lower() # Preserve a sentence already containing the Spanish CALL placeholder, # including any natural text surrounding it. if "voy a consultar tus datos" in lowered: return sentence ctx_kw = ( "name", "argument", "arg", "herramienta", "tool", "funci", "phone", "tel", "+56", "+44", "marker", "call", "llamada", "señal", "marcador", ) if any(k in lowered for k in ctx_kw) and "apipilot" in lowered: return "" return sentence catchall_sentence = re.compile( r"[^.?!;\n]{0,600}APIPILOT[^.?!;\n]{0,600}[.?!;\n]{0,1}", re.IGNORECASE, ) text = catchall_sentence.sub(_apipilot_sentence, text) # Tier 5: replace any remaining APIPILOT CALL tokens, even without context. final_apipilot_leak = re.compile( r"\bAPIPILOT[\s_\-]+CALL\b|\bAPIPILOT_CALL\b", re.IGNORECASE, ) text = final_apipilot_leak.sub(placeholder, text) # Tier 6: remove residual fake/real phone numbers from internal call text. # Match compact and spaced forms to tolerate TTS normalization. fake_phone_pattern = re.compile( r"(" # Compact forms without spaces: r"\+56912345678" r"|\+1234567890" r"|\+447432477433" r"|56912345678" r"|1234567890" r"|447432477433" # Spaced forms with punctuation: r"|\+?\s*56\s*[\s\-()]?9\s*[\s\-()]?1\s*[\s\-()]?2\s*[\s\-()]?3\s*[\s\-()]?4\s*[\s\-()]?5\s*[\s\-()]?6\s*[\s\-()]?7\s*[\s\-()]?8" r"|\+?\s*1\s*[\s\-()]?2\s*[\s\-()]?3\s*[\s\-()]?4\s*[\s\-()]?5\s*[\s\-()]?6\s*[\s\-()]?7\s*[\s\-()]?8\s*[\s\-()]?9\s*[\s\-()]?0" r"|\+?\s*44\s*[\s\-()]?7\s*[\s\-()]?4\s*[\s\-()]?3\s*[\s\-()]?2\s*[\s\-()]?4\s*[\s\-()]?7\s*[\s\-()]?7\s*[\s\-()]?4\s*[\s\-()]?3\s*[\s\-()]?3" # Generic labeled forms (English or Spanish phone labels): r"|phone\s*[:=]\s*[+()\-\s]*[0-9][+()\-\s0-9]{6,}" r"|tel(?:éfono|efono)?\s*[:=]\s*[+()\-\s]*[0-9][+()\-\s0-9]{6,}" r")", re.IGNORECASE, ) # Strip numbers only from lines with technical call context or the Spanish # placeholder, indicating that the number belongs to a replaced CALL marker. lines = text.splitlines(keepends=True) _phone_ctx = ( "apipilot", "call", "arguments", "marker", "herramienta", "funcion", "función", "ejecutar", "llamada", "marcador", "señal", "voy a consultar tus datos", ) def _phone_line_safe(line: str) -> str: lowered = line.lower() has_ctx = any(c in lowered for c in _phone_ctx) if has_ctx: # Remove phone numbers in this line (strip also surrounding space) return fake_phone_pattern.sub(" ", line) return line text = "".join(_phone_line_safe(line) for line in lines) # Collapse whitespace left by marker and path removal. text = re.sub(r"[ \t]{2,}", " ", text) text = re.sub(r"\n{3,}", "\n\n", text) return text.strip()