| """openTARS 5.0: o histórico de conversas (a lista da barra lateral). |
| |
| Só biblioteca padrão. Fica em ~/.config/opentars/historico.json (permissão 0600), neste |
| computador e só nele. Cada conversa guarda as mensagens no mesmo formato do núcleo |
| (sem o prompt de sistema e sem imagens), então abrir uma conversa antiga devolve o |
| contexto exato à IA. TARS_ARQUIVO_HISTORICO muda o arquivo (testes).""" |
|
|
| import json |
| import os |
| import threading |
| import time |
| import uuid |
| from pathlib import Path |
|
|
| import tars_memoria as memoria_lib |
| import tars_plataforma as plataforma |
|
|
| LIMITE_CONVERSAS = 200 |
| LIMITE_TITULO = 60 |
|
|
|
|
| def arquivo_historico(): |
| env = os.environ.get("TARS_ARQUIVO_HISTORICO") |
| if env: |
| return Path(env) |
| return plataforma.pastas()["config"] / "historico.json" |
|
|
|
|
| def _e_pedido_da_pessoa(m): |
| """Mensagem 'user' escrita pela pessoa (as automáticas do núcleo começam com '[').""" |
| return m.get("role") == "user" and bool((m.get("content") or "").strip()) \ |
| and not (m.get("content") or "").lstrip().startswith("[") |
|
|
|
|
| def titulo_da(mensagens): |
| for m in mensagens: |
| if _e_pedido_da_pessoa(m): |
| texto = " ".join((m.get("content") or "").split()) |
| return texto if len(texto) <= LIMITE_TITULO else texto[:LIMITE_TITULO - 1].rstrip() + "…" |
| return "" |
|
|
|
|
| def _limpar(mensagens): |
| return [{k: v for k, v in m.items() if k != "images"} for m in mensagens |
| if isinstance(m, dict) and m.get("role") != "system"] |
|
|
|
|
| LIMITE_MENSAGENS_POR_CONVERSA = 400 |
|
|
|
|
| def _juntar_mensagens(antigas, novas): |
| """A conversa em memória é podada (só as últimas ~30 mensagens), mas o histórico guarda tudo: acha onde o final |
| do que já estava salvo coincide com o começo do que chegou e emenda, em vez de jogar o começo fora.""" |
| if not antigas or not novas: |
| return novas |
| for k in range(min(len(antigas), len(novas)), 0, -1): |
| if antigas[-k:] == novas[:k]: |
| return (antigas[:-k] + novas)[-LIMITE_MENSAGENS_POR_CONVERSA:] |
| return novas |
|
|
|
|
| class Historico: |
| def __init__(self, caminho=None, agora=None): |
| self.caminho = Path(caminho) if caminho else arquivo_historico() |
| self.agora = agora or time.time |
| self._trava = threading.Lock() |
|
|
| |
| def _ler(self): |
| try: |
| dados = json.loads(self.caminho.read_text(encoding="utf-8")) |
| itens = dados.get("conversas") if isinstance(dados, dict) else None |
| return [c for c in (itens or []) if isinstance(c, dict) and c.get("id")] |
| except OSError: |
| return [] |
| except ValueError: |
| memoria_lib.guardar_quebrado(self.caminho) |
| return [] |
|
|
| def _gravar(self, conversas): |
| memoria_lib.gravar_texto_atomico( |
| self.caminho, json.dumps({"versao": 1, "conversas": conversas}, ensure_ascii=False)) |
|
|
| |
| def nova_id(self): |
| return uuid.uuid4().hex[:12] |
|
|
| def salvar(self, id_, mensagens): |
| """Grava (ou atualiza) a conversa. Sem nenhuma mensagem da pessoa, não grava nada.""" |
| mensagens = _limpar(mensagens) |
| titulo = titulo_da(mensagens) |
| if not titulo: |
| return False |
| with self._trava: |
| conversas = [c for c in self._ler() if c["id"] != id_] |
| anterior = next((c for c in self._ler() if c["id"] == id_), None) |
| if anterior: |
| mensagens = _juntar_mensagens(anterior.get("mensagens", []), mensagens) |
| conversas.append({"id": id_, "titulo": (anterior or {}).get("titulo_manual") or titulo, |
| "titulo_manual": (anterior or {}).get("titulo_manual", ""), |
| "criada": (anterior or {}).get("criada", self.agora()), |
| "atualizada": self.agora(), "mensagens": mensagens}) |
| conversas.sort(key=lambda c: c["atualizada"]) |
| self._gravar(conversas[-LIMITE_CONVERSAS:]) |
| return True |
|
|
| def listar(self, busca=""): |
| """[{'id', 'titulo', 'atualizada'}] da mais recente pra mais antiga; busca nos títulos e nas mensagens.""" |
| chave = " ".join(str(busca or "").lower().split()) |
| saida = [] |
| for c in sorted(self._ler(), key=lambda c: c.get("atualizada", 0), reverse=True): |
| if chave: |
| corpo = " ".join((m.get("content") or "") for m in c.get("mensagens", []) if isinstance(m.get("content"), str)) |
| if chave not in c.get("titulo", "").lower() and chave not in corpo.lower(): |
| continue |
| saida.append({"id": c["id"], "titulo": c.get("titulo", ""), "atualizada": c.get("atualizada", 0)}) |
| return saida |
|
|
| def abrir(self, id_): |
| c = next((c for c in self._ler() if c["id"] == id_), None) |
| return list(c.get("mensagens", [])) if c else None |
|
|
| def renomear(self, id_, titulo): |
| titulo = " ".join(str(titulo or "").split())[:LIMITE_TITULO] |
| with self._trava: |
| conversas = self._ler() |
| for c in conversas: |
| if c["id"] == id_: |
| c["titulo_manual"] = titulo |
| c["titulo"] = titulo or titulo_da(c.get("mensagens", [])) |
| self._gravar(conversas) |
| return True |
| return False |
|
|
| def apagar(self, id_): |
| with self._trava: |
| conversas = self._ler() |
| restantes = [c for c in conversas if c["id"] != id_] |
| if len(restantes) == len(conversas): |
| return False |
| self._gravar(restantes) |
| return True |
|
|
| def apagar_tudo(self): |
| with self._trava: |
| n = len(self._ler()) |
| try: |
| self.caminho.unlink() |
| except OSError: |
| pass |
| return n |
|
|
| def exportar(self, destino): |
| """Copia tudo (texto puro, JSON) pra `destino`. Devolve quantas conversas.""" |
| conversas = self._ler() |
| memoria_lib.gravar_texto_atomico(destino, json.dumps({"conversas": conversas}, ensure_ascii=False, indent=1)) |
| return len(conversas) |
|
|
|
|
| def agrupar_por_dia(itens, agora=None): |
| """[(chave, [itens])] com chave em 'hoje' | 'ontem' | 'semana' | 'antes', na ordem em que aparecem.""" |
| import datetime as dt |
| agora = dt.datetime.fromtimestamp(agora or time.time()) |
| inicio_hoje = agora.replace(hour=0, minute=0, second=0, microsecond=0) |
| grupos = {} |
| for item in itens: |
| quando = dt.datetime.fromtimestamp(item["atualizada"]) |
| dias = (inicio_hoje - quando.replace(hour=0, minute=0, second=0, microsecond=0)).days |
| chave = "hoje" if dias <= 0 else "ontem" if dias == 1 else "semana" if dias < 7 else "antes" |
| grupos.setdefault(chave, []).append(item) |
| return [(k, grupos[k]) for k in ("hoje", "ontem", "semana", "antes") if k in grupos] |
|
|
