"""conversa.py — a lógica da conversa, sem microfone, sem tela e sem som. Tudo aqui é puro (entra texto/evento, sai decisão), de propósito: é o que o teste mede. O que toca hardware (escuta.py, hud.py, __main__.py) só chama isto. Maquina ouvindo → pensando → falando → (interrompido) → ouvindo Fatiador texto em streaming → frases prontas para falar resumo resposta longa → (fala curta, o que vai para o painel) comando "esconde o painel" → "esconder" confirmacao ferramenta → "rodando os testes" """ from __future__ import annotations import re import unicodedata from typing import List, Optional, Tuple def _sem_acento(s: str) -> str: return "".join(ch for ch in unicodedata.normalize("NFD", s) if unicodedata.category(ch) != "Mn") def normalizar(s: str) -> str: """minúsculo, sem acento, sem pontuação — para comparar o que o Whisper deu.""" s = _sem_acento(str(s or "").lower()) return re.sub(r"\s+", " ", re.sub(r"[^a-z0-9 ]+", " ", s)).strip() # ── a máquina de estados ───────────────────────────────────────────────── ESTADOS = ("dormindo", "ouvindo", "pensando", "falando", "executando", "interrompido") # evento → {estado de origem → estado novo}. O que não está aqui é ignorado # (um evento fora de hora não pode quebrar a conversa). TRANSICOES = { "ativar": {"dormindo": "ouvindo", "ouvindo": "ouvindo"}, "dormir": {"ouvindo": "dormindo", "interrompido": "dormindo"}, "ouviu": {"ouvindo": "pensando", "interrompido": "pensando", "dormindo": "pensando"}, "ferramenta": {"pensando": "executando", "executando": "executando", "falando": "falando"}, "resposta": {"pensando": "falando", "executando": "falando", "ouvindo": "falando"}, "fim_fala": {"falando": "ouvindo", "interrompido": "ouvindo"}, "sem_fala": {"pensando": "ouvindo", "executando": "ouvindo"}, "cortou": {"falando": "interrompido"}, "escutar": {"interrompido": "ouvindo"}, "parar": {"pensando": "ouvindo", "executando": "ouvindo", "falando": "ouvindo", "interrompido": "ouvindo"}, } class Maquina: """O estado da conversa. `mandar(evento)` devolve o estado novo.""" def __init__(self, inicial: str = "ouvindo") -> None: self.estado = inicial self.historico: List[str] = [inicial] def mandar(self, evento: str) -> str: novo = TRANSICOES.get(evento, {}).get(self.estado) if novo and novo != self.estado: self.estado = novo self.historico.append(novo) return self.estado @property def pode_cortar(self) -> bool: """Falar por cima só corta se ele está falando.""" return self.estado == "falando" # ── frase a frase ──────────────────────────────────────────────────────── _FIM = re.compile(r"(?<=[.!?…])\s+|(?<=\n)") _ABREV = re.compile(r"\b(sr|sra|dr|dra|etc|ex|obs|vs|p\.ex)\.$", re.I) class Fatiador: """Recebe texto aos pedaços (streaming) e devolve frases completas. Frase curta demais não sai sozinha (a voz picotada soa pior que a espera): gruda na seguinte. A primeira frase sai mais cedo (mínimo menor), porque é ela que decide quanto tempo a pessoa espera até ouvir algo.""" def __init__(self, minimo: int = 24, minimo_primeira: int = 12) -> None: self.buf = "" self.minimo = minimo self.minimo_primeira = minimo_primeira self.saiu = 0 def _alvo(self) -> int: return self.minimo_primeira if self.saiu == 0 else self.minimo def alimentar(self, pedaco: str) -> List[str]: self.buf += pedaco or "" return self._tirar(final=False) def fechar(self) -> List[str]: return self._tirar(final=True) def _tirar(self, final: bool) -> List[str]: saida: List[str] = [] partes = _FIM.split(self.buf) resto = partes.pop() if partes else "" acumulado = "" for p in partes: acumulado = (acumulado + " " + p).strip() if acumulado else p.strip() if _ABREV.search(acumulado): continue if len(acumulado) >= self._alvo(): saida.append(acumulado) self.saiu += 1 acumulado = "" self.buf = (acumulado + " " + resto).strip() if acumulado else resto if final and self.buf.strip(): saida.append(self.buf.strip()) self.saiu += 1 self.buf = "" return saida def dividir_frases(texto: str) -> List[str]: f = Fatiador() return f.alimentar(texto) + f.fechar() # ── o que vai para o ouvido e o que vai para o painel ──────────────────── def limpar_para_voz(texto: str) -> str: t = str(texto or "") t = re.sub(r"```[\s\S]*?```", " ", t) t = re.sub(r"\s+(em|no|na|de|do|da|para)\s+`[^`]*[/\\.][^`]*`", "", t) t = re.sub(r"`([^`]*)`", lambda m: "" if re.search(r"[/\\.]", m.group(1)) else m.group(1), t) t = re.sub(r"https?://\S+", "", t) t = re.sub(r"(^|\s)[~.]?/[\w.\-/]+", " ", t) t = re.sub(r"^\s*(#+|[-*•]|\d+[.)])\s+", "", t, flags=re.M) t = re.sub(r"[*_>|]", "", t) return re.sub(r"\s+", " ", t).strip() def resumo(texto: str, frases: int = 60, teto: int = 3000) -> Tuple[str, str]: """(fala, painel). A voz diz a resposta COMPLETA (sem código, links nem caminhos); o painel guarda o texto inteiro. Só código e lista enorme ficam fora da fala — e aí a prosa em volta já diz o que foi feito.""" painel = str(texto or "").strip() sem_codigo = re.sub(r"```[\s\S]*?```", " ", painel) # lista ENORME (mais de 8 itens seguidos) não é lida item a item linhas, saida, run = sem_codigo.splitlines(), [], [] for l in linhas + [""]: if re.match(r"^\s*(?:[-*•]|\d+[.)])\s+", l): run.append(l) continue if run: saida += run if len(run) <= 8 else [f"São {len(run)} itens, a lista completa está no painel."] run = [] saida.append(l) limpo = limpar_para_voz("\n".join(saida)) partes = [p.strip() for p in re.findall(r"[^.!?…]+[.!?…]*", limpo) if p.strip()] fala = "" for p in partes[:frases]: if len(fala) + len(p) > teto: break fala = (fala + " " + p).strip() if not fala: fala = limpo[:teto].strip() if not fala and re.search(r"```", painel): fala = "Mandei o código para o painel." return fala, painel # ── comandos de voz que o painel e o modo resolvem sozinhos ────────────── _COMANDOS = [ ("esconder", r"(esconde|esconder|some com|fecha|oculta) (o )?(painel|hud)"), ("mostrar", r"(mostra|mostrar|abre|exibe|volta com) (o )?(painel|hud)"), ("parar", r"(para|pare|calma|chega|cala a boca|silencio|cancela)"), ("olhar", r"(olha aqui|olha isso|olha a tela|veja a tela|ve a tela|olha pra tela|ve isso)"), ("resumo_dia", r"(resumo do dia|como foi o dia|resumo de hoje|briefing|o que mudou hoje)"), ] def comando(texto: str) -> Optional[str]: n = normalizar(texto) if not n or len(n.split()) > 6: # comando é curto; frase longa é pedido return None for nome, rx in _COMANDOS: if re.fullmatch(rx + r"( por favor| ai| agora)?", n): return nome return None # ── confirmações curtas de ação ────────────────────────────────────────── def confirmacao(ferramenta: str, args: Optional[dict] = None) -> str: """O que o Primo diz ao começar uma ferramenta. Vazio = não diz nada.""" f = str(ferramenta or "") cmd = str((args or {}).get("command") or "") if f == "run_shell": if re.search(r"\b(test|jest|pytest|vitest)\b", cmd): return "Rodando os testes." if re.search(r"\bgit\s+(push|commit)\b", cmd): return "Mandando pro git." if re.search(r"\b(build|tsc)\b", cmd): return "Fazendo o build." return "Rodando." if f in ("desktop_open_app",): return "Abrindo." if f in ("write_file", "edit_file", "apply_patch"): return "Editando." if f in ("web_search", "pesquisar_referencia"): return "Pesquisando." return "" def rotulo_ferramenta(ferramenta: str, args: Optional[dict] = None) -> str: """O texto do painel: ferramenta + alvo curto.""" a = args or {} alvo = a.get("path") or a.get("command") or a.get("query") or a.get("url") or "" alvo = re.sub(r"\s+", " ", str(alvo)).strip() return (str(ferramenta) + (" · " + alvo[:48] if alvo else ""))[:70] def frase_final(texto: str) -> str: """Termina a fala de ação com um 'feito' quando a resposta não tem fala.""" return texto if texto.strip() else "Feito."