File size: 7,047 Bytes
2695555 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | """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()
# -- arquivo ---------------------------------------------------------
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) # não perde o que ainda dá pra recuperar à mão
return []
def _gravar(self, conversas):
memoria_lib.gravar_texto_atomico(
self.caminho, json.dumps({"versao": 1, "conversas": conversas}, ensure_ascii=False))
# -- operações -------------------------------------------------------
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]
|