# openTARS — nucleo/registro.py # Log de auditoria, estado global, utilidades de texto e a saída pro usuário (terminal ou janela). # # Parte do núcleo: o tars.py executa este arquivo DENTRO do próprio # namespace (ver _carregar_nucleo), então tudo aqui é tars. e as # partes enxergam umas às outras sem import. Os testes que trocam # tars. por uma falsa continuam valendo pra todas as partes. # flake8: noqa # ============================================================ # LOG DE AUDITORIA # ============================================================ # Toda ferramenta executada (abrir app, rodar comando, fechar # processo etc.) fica registrada aqui — nome, argumentos e resultado, # com timestamp. É o que permite, depois de qualquer coisa inesperada # acontecer, entender o que foi executado e por quê, em vez de # depender só da memória de quem estava olhando o terminal na hora. DIRETORIO_LOG = Path(os.environ.get("TARS_DIR_LOG", str(Path.home() / ".tars_log"))) ARQUIVO_LOG = DIRETORIO_LOG / "tars.log" TAMANHO_MAX_LOG_BYTES = 2 * 1024 * 1024 BACKUPS_LOG = 3 LIMITE_LOG_CAMPO_CHARS = 300 def _configurar_log(): logger = logging.getLogger("tars") logger.setLevel(logging.INFO) logger.propagate = False try: DIRETORIO_LOG.mkdir(parents=True, exist_ok=True, mode=0o700) try: os.chmod(DIRETORIO_LOG, 0o700) # o log guarda comandos e caminhos: só o dono lê except OSError: pass handler = RotatingFileHandler( ARQUIVO_LOG, maxBytes=TAMANHO_MAX_LOG_BYTES, backupCount=BACKUPS_LOG, encoding="utf-8", ) handler.setFormatter( logging.Formatter("%(asctime)s %(levelname)s %(message)s") ) logger.addHandler(handler) except Exception as e: # Auditoria é best-effort: se não der pra escrever o log (disco # cheio, permissão negada etc.), o TARS continua funcionando # normalmente sem ela, só avisa uma vez. Print puro aqui (sem # o helper de cor c()) de propósito: essa função roda antes # dele existir na ordem de definição do módulo. print(f"[openTARS] {t('msg.log_desativado', erro=e)}") logger.addHandler(logging.NullHandler()) return logger LOG = _configurar_log() def _resumir_para_log(valor): """Converte um valor qualquer (geralmente os argumentos de uma ferramenta) numa string curta o bastante pro log — sem isso, um texto digitado enorme ou um resultado de comando de terminal grande inflaria o arquivo de log rapidamente.""" try: if isinstance(valor, dict) and any(k in valor for k in ("text", "texto", "password", "senha")): # o que a pessoa digita pode ser senha: o log guarda só o tamanho valor = {k: (f"<{len(str(v))} chars>" if k in ("text", "texto", "password", "senha") else v) for k, v in valor.items()} texto = json.dumps(valor, ensure_ascii=False) except Exception: texto = str(valor) if len(texto) > LIMITE_LOG_CAMPO_CHARS: texto = texto[:LIMITE_LOG_CAMPO_CHARS] + "…" return texto # ============================================================ # ESTADO # ============================================================ modo_modelo = "AUTO" modelo_manual = None modelo_atual = None modelos_do_tars = set() # tudo que o openTARS carregou nesta sessão: ao fechar, só isso é descarregado sessoes = {} lock_modelo = threading.Lock() # Sessão HTTP única e reutilizada: evita reabrir conexão TCP/TLS # a cada chamada ao Ollama (isso importa bastante no loop de tools, # que pode chamar o Ollama várias vezes por turno). _SESSION = requests.Session() # Cache de aplicativos instalados (desktop entries e flatpak). _cache_desktop = {"dados": None, "ts": 0.0} _cache_flatpak = {"dados": None, "ts": 0.0} # ============================================================ # UTILIDADES # ============================================================ def _tabela_sem_acentos(): """Letras latinas acentuadas -> sem acento (á->a, ñ->n, ö->o, ß->ss...), montada uma vez só: str.translate com ela é bem mais rápido que decompor Unicode a cada chamada, e normalizar() roda muito (uma vez por processo do sistema, por app instalado, por palavra-chave...).""" tabela = {} for codigo in range(0xC0, 0x250): letra = chr(codigo) base = "".join(x for x in unicodedata.normalize("NFKD", letra) if not unicodedata.combining(x)) if base != letra and base.isascii() and base.isalpha(): tabela[codigo] = base.lower() tabela.update({ord("ß"): "ss", ord("æ"): "ae", ord("œ"): "oe", ord("ø"): "o", ord("đ"): "d", ord("ł"): "l", ord("ı"): "i"}) return tabela _TABELA_NORMALIZACAO = _tabela_sem_acentos() # Separa palavras (Unicode: funciona com qualquer alfabeto). _RE_TOKENS = re.compile(r"[\W_]+") def normalizar(texto): return texto.lower().strip().translate(_TABELA_NORMALIZACAO) _RE_SEPARADORES_NOME = re.compile(r"[-_.:/]+") def normalizar_nome_app(texto): """Pra comparar nomes de apps: "Counter-Strike 2" vira "counter strike 2", e bate com o "counter strike" que o usuário escreveu.""" return re.sub(r"\s+", " ", _RE_SEPARADORES_NOME.sub(" ", normalizar(texto or ""))).strip() # ------------------------------------------------------------ # Vocabulário dos idiomas: palavras-chave, comandos, verbos de busca... # Montado uma vez a partir dos arquivos de idioma (já normalizado e em # conjuntos, pra busca instantânea) e refeito só quando o idioma muda. # ------------------------------------------------------------ _CHAVES_COMANDOS = ( "cmd.modelos", "cmd.modelo", "cmd.encerrar", "cmd.limpar", "cmd.idioma", "cmd.auto", "cmd.auto_argumento", "cmd.sair", ) _vocab_atual = None def _preparar_expressoes(expressoes): """(palavras soltas, trechos de várias palavras), normalizados e já quebrados em palavras como o pedido é: "trouve-moi" vira o trecho " trouve moi " e "à l'écran", " a l ecran ". Trecho é procurado com espaço dos dois lados (palavras inteiras): "the screen" não bate em "the screenshots".""" palavras, trechos = set(), [] for expr in expressoes: partes = [x for x in _RE_TOKENS.split(normalizar(str(expr))) if x] if len(partes) > 1: trechos.append(" " + " ".join(partes) + " ") elif partes: palavras.add(partes[0]) return frozenset(palavras), tuple(dict.fromkeys(trechos)) def _normalizados(itens): return tuple(dict.fromkeys(n for n in (normalizar(str(x)) for x in itens) if n)) # Palavras das listas de abrir/fechar/buscar que não são verbos ("feche a # ABA do youtube"): não contam como uma etapa do pedido. _NAO_SAO_VERBOS = frozenset({"aba", "abas", "tab", "tabs", "pestana", "pestanas", "onglet", "onglets", "reiter", "tabs", "site", "pagina", "page", "website"}) def _montar_vocabulario(): entrada = i18n.idiomas_de_entrada() v = { "detectar": { grupo: _preparar_expressoes(list(neutras) + i18n.juntar(f"detectar.{grupo}", entrada)) for grupo, neutras in _PALAVRAS_NEUTRAS.items() }, "busca_verbos": frozenset(_normalizados(i18n.juntar("busca.verbos", entrada))), # Verbos de ação (pra achar pedidos com várias etapas): os da lista # própria + os das palavras-chave de abrir/fechar/buscar. "verbos_acao": frozenset( palavra for frase in _normalizados( i18n.juntar("detectar.verbos", entrada) + i18n.juntar("detectar.abrir", entrada) + i18n.juntar("detectar.fechar", entrada) + i18n.juntar("detectar.busca", entrada) + i18n.juntar("busca.verbos", entrada)) for palavra in frase.split()[:1] if palavra not in _NAO_SAO_VERBOS ), "busca_conectores": _normalizados(i18n.juntar("busca.conectores", entrada)), "busca_preposicoes": _normalizados(i18n.juntar("busca.preposicoes", entrada)), "navegadores": frozenset(_MARCAS_NAVEGADORES) | frozenset(_normalizados(i18n.juntar("palavras.navegador", entrada))), # Comandos são explícitos: valem em qualquer idioma. "comandos": {chave: frozenset(_normalizados(i18n.em_todos(chave))) for chave in _CHAVES_COMANDOS}, "respostas_sim": frozenset(_normalizados(i18n.em_todos("cmd.sim"))), "respostas_sempre": frozenset(_normalizados(i18n.em_todos("cmd.sempre"))), # 5.1: "sempre nesta sessão" "modelo_ignorar": frozenset(_normalizados(i18n.em_todos("modelo.ignorar"))), } superlativos = {} for chave, resolvedor in _CRITERIOS_MODELO.items(): for frase in _normalizados(i18n.em_todos(chave)): superlativos.setdefault(frase, resolvedor) v["superlativos"] = superlativos # Nomes traduzidos de apps: Name[pt_BR]= no .desktop e catálogos gettext. grupos = [tuple(i18n.juntar("_locale", [codigo]) or [codigo]) for codigo in entrada] for local in i18n.locales_do_sistema(): grupos.append((local, local.split("_")[0]) if "_" in local else (local,)) grupos = tuple(dict.fromkeys(grupos)) v["grupos_locale"] = grupos v["sufixos_desktop"] = tuple(dict.fromkeys([""] + [f"[{x}]" for grupo in grupos for x in grupo])) return v def _vocab(): global _vocab_atual v = _vocab_atual if v is None: v = _vocab_atual = _montar_vocabulario() return v def _ao_mudar_idioma(_codigo): """Idioma novo: palavras-chave novas e nomes de apps relidos.""" global _vocab_atual _vocab_atual = None _cache_arquivos_desktop.clear() _cache_desktop["dados"] = None i18n.ao_mudar_idioma(_ao_mudar_idioma) # ------------------------------------------------------------ # Interface: cores ANSI (funcionam no terminal do Zorin/qualquer # terminal Linux moderno). Usadas só no que é impresso pro humano — # nunca dentro do conteúdo de "mensagem" devolvido nas ferramentas, # porque esse texto vai para o JSON que a IA lê. # ------------------------------------------------------------ class Cor: RESET = "\033[0m" NEGRITO = "\033[1m" CINZA = "\033[90m" VERMELHO = "\033[31m" VERDE = "\033[32m" AMARELO = "\033[33m" AZUL = "\033[34m" MAGENTA = "\033[35m" CIANO = "\033[36m" ITALICO = "\033[3m" def c(texto, cor): return f"{cor}{texto}{Cor.RESET}" def imprimir_caixa(titulo, linhas, cor_titulo=Cor.CIANO): largura = ( max(LARGURA_MINIMA_CAIXA, len(titulo) + 4, *(len(l) + 4 for l in linhas)) if linhas else max(LARGURA_MINIMA_CAIXA, len(titulo) + 4) ) print() print(c("╭" + "─" * largura + "╮", Cor.CINZA)) print(c("│ ", Cor.CINZA) + c(titulo.ljust(largura - 2), cor_titulo) + c(" │", Cor.CINZA)) for linha in linhas: print(c("│ ", Cor.CINZA) + linha.ljust(largura - 2)[:largura - 2] + c(" │", Cor.CINZA)) print(c("╰" + "─" * largura + "╯", Cor.CINZA)) def executar_subprocesso(cmd, timeout=TIMEOUT_SUBPROCESSO_PADRAO): try: resultado = subprocess.run( cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", # nomes em latin-1 não podem virar "falhou" timeout=timeout, ) return { "codigo": resultado.returncode, "stdout": resultado.stdout.strip(), "stderr": resultado.stderr.strip(), } except subprocess.TimeoutExpired: return { "codigo": -1, "stdout": "", "stderr": "timeout", } except Exception as e: return { "codigo": -1, "stdout": "", "stderr": str(e), } # ============================================================ # SAÍDA PRO USUÁRIO (terminal ou janela) # ============================================================ # # O que o openTARS mostra enquanto trabalha (ferramenta rodando, modelo # carregando, raciocínio e resposta chegando) sai como EVENTO. No # terminal vira texto colorido; a janela registra um ouvinte e recebe o # evento já estruturado, em vez de ler e interpretar o texto impresso # (que muda com o idioma escolhido). _ouvinte_saida = None def definir_ouvinte_saida(funcao): """A interface gráfica registra aqui uma função(tipo, dados). None volta a imprimir no terminal.""" global _ouvinte_saida _ouvinte_saida = funcao def emitir(tipo, **dados): ouvinte = _ouvinte_saida if ouvinte is not None: try: ouvinte(tipo, dados) except Exception: pass return renderizar = _TERMINAL.get(tipo) if renderizar is not None: renderizar(dados) def nome_ferramenta(nome): """"open_application" -> "abrindo app" (no idioma escolhido).""" chave = f"ferramenta.{nome}" texto = t(chave) return nome if texto == chave else texto def nome_tarefa(tarefa): chave = f"tarefa.{tarefa}" texto = t(chave) return tarefa if texto == chave else texto def _segundos(valor): return f"{valor:.1f}" # ------------------------------------------------------------ # Terminal no estilo das CLIs de IA: "●" abre cada passo, "⎿" pendura o resultado embaixo, e um símbolo girando # mostra que o openTARS está trabalhando. Só vale sem janela (a janela recebe os eventos, não este texto). # ------------------------------------------------------------ _RECUO = " " _PENDURA = " ⎿ " class _Girando: """O símbolo de "trabalhando" na última linha do terminal. Ele só existe num terminal de verdade (ligar() chamado pelo chat), redesenha a linha a cada 90 ms e some antes de qualquer outro texto (_SaidaComGiro).""" QUADROS = "·✢✳✶✻✽✻✶✳✢" def __init__(self): self.ligado = False self.saida = None self.trava = threading.Lock() self.texto = "" self.desde = 0.0 self.visivel = False self._fio = None self._rodando = threading.Event() def ligar(self, saida): self.saida, self.ligado = saida, True def desligar(self): self.parar() self.ligado, self.saida = False, None def girar(self, texto): if not self.ligado: return with self.trava: if texto != self.texto or not self._rodando.is_set(): self.desde = time.time() self.texto = texto if not self._rodando.is_set(): self._rodando.set() self._fio = threading.Thread(target=self._laco, daemon=True) self._fio.start() def parar(self): if not self._rodando.is_set(): return self._rodando.clear() if self._fio is not None and self._fio is not threading.current_thread(): self._fio.join(0.5) with self.trava: self.apagar() def apagar(self): """Limpa a linha do símbolo (com a trava na mão).""" if self.visivel and self.saida is not None: self.saida.write("\r\033[K") self.saida.flush() self.visivel = False def _laco(self): i = 0 while self._rodando.is_set(): with self.trava: if not self._rodando.is_set(): break quadro = self.QUADROS[i % len(self.QUADROS)] segundos = int(time.time() - self.desde) linha = (f"\r\033[K{Cor.MAGENTA}{quadro}{Cor.RESET} {self.texto}… " f"{Cor.CINZA}({_tempo_curto(segundos)} · {t('term.parar')}){Cor.RESET}") self.saida.write(linha) self.saida.flush() self.visivel = True i += 1 time.sleep(0.09) _girando = _Girando() def girar(texto): _girando.girar(texto) def parar_girando(): _girando.parar() def _curto(modelo): """"ProjectOpenTARS/Gargantua:latest" -> "Gargantua": o nome inteiro não cabe na linha do símbolo.""" nome = str(modelo or "").split("/")[-1] return nome[:-len(":latest")] if nome.endswith(":latest") else nome def _recuar(texto): """Quebras de linha do texto que chega aos pedaços ficam alinhadas embaixo do "●".""" return texto.replace("\n", "\n" + _RECUO) def _term_ferramenta(d): parar_girando() nome = nome_ferramenta(d["nome"]) resumo = c(f" ({d['resumo']})", Cor.CINZA) if d.get("resumo") else "" print("\n" + c("● ", Cor.CIANO) + c(nome, Cor.NEGRITO) + resumo, flush=True) girar(nome[:1].upper() + nome[1:]) def _term_ferramenta_fim(d): parar_girando() if d.get("sucesso"): print(c(_PENDURA, Cor.CINZA) + c("✓", Cor.VERDE) + c(f" {_segundos(d['segundos'])}s", Cor.CINZA)) else: print(c(_PENDURA, Cor.CINZA) + c(f"✗ {t('msg.falhou')}", Cor.VERMELHO) + c(f" {_segundos(d['segundos'])}s", Cor.CINZA)) def _term_tarefa(d): girar(f"{t('msg.pensando').capitalize()} · {_curto(d['modelo'])}") def _term_carregando(d): if d.get("segundo_plano"): parar_girando() print(c(_PENDURA + t("msg.carregando_fundo", modelo=d["modelo"]).rstrip("."), Cor.CINZA)) else: girar(t("msg.carregando_titulo", modelo=_curto(d["modelo"]))) def _term_carregado(d): if d.get("segundo_plano") or not _girando.ligado: parar_girando() print(c(_PENDURA + t("msg.carregado", modelo=d["modelo"], segundos=_segundos(d["segundos"])), Cor.CINZA)) else: girar(f"{t('msg.pensando').capitalize()} · {_curto(d['modelo'])}") def _term_descarregando(d): parar_girando() print(c(_PENDURA + t("msg.descarregando", modelo=d["modelo"]).rstrip("."), Cor.CINZA)) def _term_pensamento(d): if d.get("inicio"): parar_girando() print("\n" + c(f"✻ {t('msg.pensando').capitalize()}…", Cor.CINZA)) print(_RECUO, end="") print(Cor.CINZA + Cor.ITALICO + _recuar(d["texto"]) + Cor.RESET, end="", flush=True) def _term_resposta(d): if d.get("inicio"): parar_girando() if d.get("apos_pensamento"): print() print("\n" + c("● ", Cor.VERDE), end="") _MD.reiniciar() texto = _MD.alimentar(d["texto"]) if _girando.ligado else d["texto"] # 5.3: formatado só num terminal de verdade print(_recuar(texto), end="", flush=True) def _term_tempo(d): parar_girando() modo = "auto" if d.get("modo") == "AUTO" else t("term.modo_fixo") print(c(f"{_PENDURA}{_curto(d['modelo'])} · {modo} · {_segundos(d['segundos'])}s", Cor.CINZA)) def _term_plano(d): parar_girando() print("\n" + c("● ", Cor.CIANO) + c(t("msg.plano").rstrip(":"), Cor.NEGRITO)) for i, etapa in enumerate(d.get("etapas") or [], 1): print(c(_PENDURA if i == 1 else " " * len(_PENDURA), Cor.CINZA) + f"{i}. {etapa}") def _term_folego(d): parar_girando() print(c(_PENDURA + t("msg.folego", etapas=d.get("etapas"), acoes=d.get("acoes")), Cor.CINZA)) def _term_analisando(d): girar(t("msg.analisando", modelo=_curto(d["modelo"])).rstrip(".").rstrip("…")) _TERMINAL = { "plano": _term_plano, "ferramenta": _term_ferramenta, "ferramenta_fim": _term_ferramenta_fim, "tarefa": _term_tarefa, "carregando": _term_carregando, "carregado": _term_carregado, "descarregando": _term_descarregando, "pensamento": _term_pensamento, "resposta": _term_resposta, "fim_stream": lambda d: print(_recuar(_MD.terminar()) if _girando.ligado else "", flush=True), "tempo": _term_tempo, "folego": _term_folego, "analisando": _term_analisando, }