From ffe36e782b064a31b7a695ac1372f1329ac79153 Mon Sep 17 00:00:00 2001 From: Elvis Claros Castro Date: Mon, 17 Aug 2026 15:14:11 -0300 Subject: Renombrar el proyecto a claude-logbook MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit El nombre en español no se encontraba, y la zona de "claude + sessions" ya está poblada por cosas distintas con las que se confundía: gapmiss/claude-sessions plugin de Obsidian kyupid/claude-sessions monitor de sesiones en curso (PyPI) claude-session-manager wrapper que preserva contexto (PyPI) LearnedGeek/claude-recall búsqueda con SQLite FTS5 (PyPI) "logbook" cubre las dos mitades — las conversaciones y las memorias — sin caer en "session manager", y encaja con lo que el proyecto produce: una bitácora que se abre en cualquier lado. claude_sesiones/ → claude_logbook/ y todas las referencias ~/.cache/claude-sesiones → ~/.cache/claude-logbook el resumen de PyPI pasa a inglés, como el README principal El comando viejo sigue funcionando: pyproject declara claude-sesiones como segundo entry point del mismo main, para no rehacer la memoria muscular. La interfaz sigue en español; solo cambia cómo se llama e instala el paquete. De paso, memory.py faltaba en la tabla de módulos de los dos READMEs. Claude-Session: https://claude.ai/code/session_01RmtZ9qBemrc9TncwVTG6ED --- claude_sesiones/__init__.py | 8 - claude_sesiones/__main__.py | 6 - claude_sesiones/cli.py | 455 -------------- claude_sesiones/memory.py | 272 -------- claude_sesiones/sessions.py | 409 ------------ claude_sesiones/template.html | 1394 ----------------------------------------- claude_sesiones/terminal.py | 372 ----------- claude_sesiones/webpage.py | 74 --- 8 files changed, 2990 deletions(-) delete mode 100644 claude_sesiones/__init__.py delete mode 100644 claude_sesiones/__main__.py delete mode 100644 claude_sesiones/cli.py delete mode 100644 claude_sesiones/memory.py delete mode 100644 claude_sesiones/sessions.py delete mode 100644 claude_sesiones/template.html delete mode 100644 claude_sesiones/terminal.py delete mode 100644 claude_sesiones/webpage.py (limited to 'claude_sesiones') diff --git a/claude_sesiones/__init__.py b/claude_sesiones/__init__.py deleted file mode 100644 index fa839b1..0000000 --- a/claude_sesiones/__init__.py +++ /dev/null @@ -1,8 +0,0 @@ -"""Explorador de las sesiones que Claude Code guarda en ~/.claude/projects/. - -Sin dependencias: solo la biblioteca estándar. -""" - -__version__ = "1.0.0" - -__all__ = ["__version__"] diff --git a/claude_sesiones/__main__.py b/claude_sesiones/__main__.py deleted file mode 100644 index dbdd066..0000000 --- a/claude_sesiones/__main__.py +++ /dev/null @@ -1,6 +0,0 @@ -import sys - -from .cli import main - -if __name__ == "__main__": - sys.exit(main()) diff --git a/claude_sesiones/cli.py b/claude_sesiones/cli.py deleted file mode 100644 index 4dc5ab2..0000000 --- a/claude_sesiones/cli.py +++ /dev/null @@ -1,455 +0,0 @@ -"""Interfaz de línea de comandos.""" - -import argparse -import io -import os -import sys -import textwrap -import time -import webbrowser - -from . import __version__ -from .sessions import ( - SessionError, apply_filters, default_root, drop_from_cache, latest_activity, - load_sessions, pick, public_records, session_path, -) -from .terminal import ( - Style, clip, fmt_date, fmt_size, plural, print_audit, print_chat, - print_memories, print_memory, print_table, resume_cmd, -) -from . import memory as mem -from . import webpage - -DEFAULT_HTML = "sesiones.html" - -# Una sesión escrita hace menos de esto puede estar abierta en otra terminal. -RECENT_SECONDS = 300 - -EPILOG = """\ -ejemplos: - claude-sesiones tabla de todas las sesiones - claude-sesiones docker filtra por título, ruta o rama - claude-sesiones -s 3 lee el chat nº 3 de la tabla - claude-sesiones -s 5d10f1ee lo mismo, por prefijo de UUID - claude-sesiones -g "port already" busca dentro de las conversaciones - claude-sesiones -r 3 comando para reanudar la nº 3 - eval "$(claude-sesiones -r 3)" reanudarla directamente - claude-sesiones --html --open genera sesiones.html y lo abre - -el nº es la posición en la tabla que estás viendo, así que si filtraste -hay que repetir el filtro para leer esa fila: - - claude-sesiones docker muestra 3 resultados - claude-sesiones docker -s 2 lee el 2º de esos tres - -borrado (irreversible; pregunta antes, salvo con -y): - claude-sesiones --delete-empty --dry-run qué borraría - claude-sesiones --delete-empty borra las vacías - claude-sesiones -D 101 -D e0a4300e borra sesiones puntuales - claude-sesiones -p /tmp --delete-empty solo las vacías de ese proyecto - -memoria de los proyectos (-m cambia de sesiones a memorias y reusa los mismos -verbos: filtro, -s para leer, -D para borrar): - claude-sesiones -m tabla de memorias - claude-sesiones -m docker busca en nombre, descripción y cuerpo - claude-sesiones -m -s 3 lee la memoria nº 3 - claude-sesiones -m -s deadlock lo mismo, por nombre - claude-sesiones -m --check audita índices, enlaces y orígenes - claude-sesiones -m -D 3 la borra y la saca de MEMORY.md -""" - - -def build_parser(): - ap = argparse.ArgumentParser( - prog="claude-sesiones", - description="Explorador de sesiones de Claude Code para la terminal.", - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=textwrap.dedent(EPILOG), - ) - ap.add_argument("query", nargs="*", help="texto a buscar en título, ruta o rama") - ap.add_argument("-s", "--show", metavar="REF", - help="muestra el chat: índice de la tabla o prefijo de UUID") - ap.add_argument("-r", "--resume", metavar="REF", - help="imprime el comando para reanudar esa sesión") - ap.add_argument("-g", "--grep", metavar="TEXTO", - help="filtra por contenido de las conversaciones") - ap.add_argument("-p", "--project", metavar="RUTA", - help="filtra por ruta del proyecto") - ap.add_argument("-n", "--limit", type=int, metavar="N", - help="muestra solo las N más recientes") - ap.add_argument("-E", "--hide-empty", action="store_true", - help="oculta las sesiones sin mensajes") - ap.add_argument("--no-tools", action="store_true", - help="en el chat, oculta las llamadas a herramientas") - ap.add_argument("--no-pager", action="store_true", - help="no usa $PAGER para el chat") - ap.add_argument("--no-color", action="store_true", help="salida sin color") - - recuerdos = ap.add_argument_group("memoria de los proyectos") - recuerdos.add_argument("-m", "--memory", action="store_true", - help="trabaja sobre las memorias en vez de las sesiones") - recuerdos.add_argument("--type", metavar="TIPO", choices=mem.TYPES, - help="filtra por tipo: " + " | ".join(mem.TYPES)) - recuerdos.add_argument("--check", action="store_true", - help="audita índices, enlaces y sesiones de origen") - - salida = ap.add_argument_group("exportar") - salida.add_argument("--json", action="store_true", - help="vuelca todas las sesiones en JSON") - salida.add_argument("--html", nargs="?", const=DEFAULT_HTML, metavar="ARCHIVO", - help=f"genera una página autocontenida (por defecto {DEFAULT_HTML})") - salida.add_argument("--template", metavar="ARCHIVO", - help="usa otro template para --html") - salida.add_argument("--open", action="store_true", - help="abre en el navegador lo que genere --html") - - borrar = ap.add_argument_group("borrado") - borrar.add_argument("-D", "--delete", metavar="REF", nargs="+", - help="borra esas sesiones (índice o prefijo de UUID)") - borrar.add_argument("--delete-empty", action="store_true", - help="borra todas las sesiones sin mensajes") - borrar.add_argument("-y", "--yes", action="store_true", - help="no pregunta antes de borrar") - borrar.add_argument("--dry-run", action="store_true", - help="muestra qué se borraría y no toca nada") - - ap.add_argument("--no-cache", action="store_true", - help="ignora el caché y re-parsea todo") - ap.add_argument("--version", action="version", - version=f"claude-sesiones {__version__}") - return ap - - -def filtered(sessions, args): - return apply_filters( - sessions, - project=args.project, - grep=args.grep, - query=" ".join(args.query) if args.query else None, - hide_empty=args.hide_empty, - ) - - -# ──────────────────────────────── borrado ──────────────────────────────── - -def confirm(question): - """Pregunta s/N. Sin terminal no hay confirmación posible: devuelve False.""" - try: - tty = open("/dev/tty") - except OSError: - return False - try: - sys.stderr.write(question) - sys.stderr.flush() - return tty.readline().strip().lower() in ("s", "si", "sí", "y", "yes") - except (OSError, KeyboardInterrupt): - return False - finally: - tty.close() - - -def delete_sessions(targets, args, st): - """Borra las sesiones dadas. Devuelve el código de salida.""" - if not targets: - print("No hay sesiones que borrar con ese criterio.", file=sys.stderr) - return 0 - - print(f"{st.bold}Se van a borrar " - f"{plural(len(targets), 'sesión', 'sesiones')}:{st.reset}\n") - - total_kb = 0 - recent = [] - for s in targets: - path = session_path(s) - total_kb += s["k"] - title = s["t"] or "sesión abierta sin mensajes" - flag = "" - try: - if time.time() - os.stat(path).st_mtime < RECENT_SECONDS: - recent.append(s) - flag = f" {st.copper}← modificada hace menos de 5 min{st.reset}" - except OSError: - flag = f" {st.copper}← ya no existe{st.reset}" - print(f" {st.faint}{s['id'][:8]}{st.reset} {clip(title, 52):<52} " - f"{st.grey}{clip(s['p'], 34):<34}{st.reset} " - f"{fmt_date(s['l'])} {st.faint}{s['k']:>7.1f} KB{st.reset}{flag}") - - print(f"\n{st.faint}{fmt_size(total_kb)} en total{st.reset}") - - if recent: - verbo = "se escribió" if len(recent) == 1 else "se escribieron" - print(f"\n{st.copper}Ojo: {len(recent)} de estas {verbo} hace menos de " - f"5 minutos. Si es una sesión abierta ahora mismo, Claude Code la " - f"sigue usando y va a volver a escribirla al cerrarse.{st.reset}") - - if args.dry_run: - print(f"\n{st.faint}--dry-run: no se tocó nada.{st.reset}") - return 0 - - if not args.yes: - print(f"\n{st.copper}Esto no se puede deshacer.{st.reset}") - if not confirm("¿Confirmás? [s/N] "): - print("Cancelado.", file=sys.stderr) - return 1 - - done, failed, paths = 0, 0, [] - for s in targets: - path = session_path(s) - try: - os.remove(path) - paths.append(path) - done += 1 - except OSError as e: - print(f"error: {s['id'][:8]}: {e}", file=sys.stderr) - failed += 1 - - drop_from_cache(paths) - - print(f"\n{plural(done, 'sesión borrada', 'sesiones borradas')}.") - return 1 if failed else 0 - - -def delete_targets(pool, args): - """Las sesiones que pidió borrar, sin repetidas y en el orden pedido.""" - targets, seen = [], set() - - if args.delete_empty: - for s in pool: - if s["e"]: - targets.append(s) - seen.add(s["id"]) - - for ref in args.delete or []: - s = pick(pool, ref) - if s["id"] not in seen: - seen.add(s["id"]) - targets.append(s) - - return targets - - -# ──────────────────────────────── comandos ──────────────────────────────── - -def cmd_json(sessions): - import json - payload = webpage.build_payload( - public_records(sessions), - mem.public_records(mem.load_memories(sessions))) - json.dump(payload, sys.stdout, ensure_ascii=False, separators=(",", ":")) - sys.stdout.write("\n") - return 0 - - -def cmd_html(sessions, args): - out = args.html - memories = mem.public_records(mem.load_memories(sessions)) - stats = webpage.write(public_records(sessions), out, memories=memories, - template=webpage.template_text(args.template)) - print(f"{stats['sesiones']} sesiones · {stats['proyectos']} proyectos · " - f"{stats['mensajes']} mensajes · {stats['bloques']} bloques " - f"de transcripción · {stats['memorias']} memorias → {out}", - file=sys.stderr) - if args.open: - webbrowser.open("file://" + os.path.abspath(out)) - return 0 - - -def cmd_show(sessions, args, st): - s = pick(filtered(sessions, args), args.show) - buf = io.StringIO() - print_chat(s, st, buf, show_tools=not args.no_tools) - text = buf.getvalue() - if args.no_pager: - sys.stdout.write(text) - else: - from .terminal import pager - pager(text) - return 0 - - -def cmd_table(sessions, args, st): - shown = filtered(sessions, args) - if not shown: - print("Ninguna sesión coincide con ese filtro.", file=sys.stderr) - return 1 - if args.limit: - shown = shown[: args.limit] - - print_table(shown, st, latest_activity(sessions), sys.stdout) - - total, projects = len(sessions), len({s["p"] for s in sessions}) - tail = (f"{len(shown)} de {total} sesiones" if len(shown) != total - else f"{plural(total, 'sesión', 'sesiones')} · " - f"{plural(projects, 'proyecto', 'proyectos')}") - print(f"\n{st.faint}{tail} · -s para leer una{st.reset}") - return 0 - - -def mem_filtered(memories, args): - return mem.apply_filters(memories, project=args.project, - query=" ".join(args.query) or None, - kind=args.type) - - -def cmd_mem_show(memories, args, st): - m = mem.pick(mem_filtered(memories, args), args.show) - buf = io.StringIO() - print_memory(m, st, buf, path=mem.memory_path(m)) - text = buf.getvalue() - if args.no_pager: - sys.stdout.write(text) - else: - from .terminal import pager - pager(text) - return 0 - - -def cmd_mem_check(memories, sessions, st): - report = mem.audit(memories, sessions) - total = print_audit(report, st, sys.stdout) - if total: - print(f"{st.faint}{plural(total, 'cosa para mirar', 'cosas para mirar')} " - f"en {plural(len(memories), 'memoria', 'memorias')}.{st.reset}") - return 1 - print(f"{st.amber}Todo en orden: " - f"{plural(len(memories), 'memoria', 'memorias')}, " - f"índices y enlaces consistentes.{st.reset}") - return 0 - - -def delete_memories(targets, args, st): - if not targets: - print("No hay memorias que borrar con ese criterio.", file=sys.stderr) - return 0 - - print(f"{st.bold}Se van a borrar " - f"{plural(len(targets), 'memoria', 'memorias')}:{st.reset}\n") - total_kb = 0 - for m in targets: - total_kb += m["k"] - print(f" {st.ink}{clip(m['name'], 38):<38}{st.reset} " - f"{clip(m['ty'], 9):<9} " - f"{st.grey}{clip(m['p'], 34):<34}{st.reset} " - f"{st.faint}{fmt_size(m['k']):>9}{st.reset}") - if m["desc"]: - print(f" {st.faint}{clip(m['desc'], 86)}{st.reset}") - - print(f"\n{st.faint}{fmt_size(total_kb)} en total · " - f"también se quita su línea de MEMORY.md{st.reset}") - - if args.dry_run: - print(f"\n{st.faint}--dry-run: no se tocó nada.{st.reset}") - return 0 - - if not args.yes: - print(f"\n{st.copper}Esto no se puede deshacer.{st.reset}") - if not confirm("¿Confirmás? [s/N] "): - print("Cancelado.", file=sys.stderr) - return 1 - - done = failed = unlisted = 0 - for m in targets: - try: - if mem.delete(m): - unlisted += 1 - done += 1 - except OSError as e: - print(f"error: {m['name']}: {e}", file=sys.stderr) - failed += 1 - - extra = f", {unlisted} sacadas del índice" if unlisted else "" - print(f"\n{plural(done, 'memoria borrada', 'memorias borradas')}{extra}.") - return 1 if failed else 0 - - -def cmd_mem_table(memories, args, st, total): - if not memories: - print("Ninguna memoria coincide con ese filtro.", file=sys.stderr) - return 1 - shown = memories[: args.limit] if args.limit else memories - - print_memories(shown, st, latest_activity(shown), sys.stdout) - - projects = len({m["p"] for m in memories}) - tail = (f"{len(shown)} de {total} memorias" if len(shown) != total - else f"{plural(total, 'memoria', 'memorias')} · " - f"{plural(projects, 'proyecto', 'proyectos')}") - print(f"\n{st.faint}{tail} · -m -s para leer una{st.reset}") - return 0 - - -def run_memory(sessions, args, st): - memories = mem.load_memories(sessions) - if not memories: - print("Ningún proyecto tiene memorias todavía.", file=sys.stderr) - return 1 - - if args.check: - return cmd_mem_check(memories, sessions, st) - - if args.delete: - pool = mem_filtered(memories, args) - targets, seen = [], set() - for ref in args.delete: - m = mem.pick(pool, ref) - if m["name"] not in seen: - seen.add(m["name"]) - targets.append(m) - return delete_memories(targets, args, st) - - if args.show: - return cmd_mem_show(memories, args, st) - - return cmd_mem_table(mem_filtered(memories, args), args, st, len(memories)) - - -def run(args): - root = default_root() - if not os.path.isdir(root): - print(f"error: no existe {root} — ¿usaste Claude Code en esta máquina?", - file=sys.stderr) - return 2 - - sessions = load_sessions(root=root, use_cache=not args.no_cache) - if not sessions: - print("No hay ninguna sesión registrada todavía.", file=sys.stderr) - return 1 - - st = Style.from_stream(sys.stdout, args.no_color) - - if args.memory: - return run_memory(sessions, args, st) - - if args.json: - return cmd_json(sessions) - if args.html: - return cmd_html(sessions, args) - - if args.delete or args.delete_empty: - targets = delete_targets(filtered(sessions, args), args) - return delete_sessions(targets, args, st) - - if args.resume: - print(resume_cmd(pick(filtered(sessions, args), args.resume))) - return 0 - - if args.show: - return cmd_show(sessions, args, st) - - return cmd_table(sessions, args, st) - - -def main(argv=None): - args = build_parser().parse_args(argv) - try: - return run(args) - except SessionError as e: - print(f"error: {e}", file=sys.stderr) - return 2 - except (BrokenPipeError, KeyboardInterrupt): - # El pipe ya está cerrado: silenciamos el flush de salida al terminar. - try: - sys.stdout.close() - except Exception: - pass - return 130 diff --git a/claude_sesiones/memory.py b/claude_sesiones/memory.py deleted file mode 100644 index ccb2e65..0000000 --- a/claude_sesiones/memory.py +++ /dev/null @@ -1,272 +0,0 @@ -"""Memorias de proyecto: los .md que Claude Code deja en /memory/. - -Cada proyecto puede acumular recuerdos en - - ~/.claude/projects//memory/.md - -Un archivo por recuerdo, con frontmatter YAML (`name`, `description`, -`metadata.type`, `metadata.originSessionId`) y cuerpo markdown. Al lado vive -`MEMORY.md`, el índice: una línea por memoria, y es lo único que se carga en -contexto al arrancar una sesión. Una memoria que no figura ahí sigue en disco -pero deja de recordarse, así que la diferencia entre ambos vale la pena mirarla. - -Esquema del registro que devuelve `read_memory`, con las mismas claves cortas -que `sessions` porque también viaja embebido en el HTML: - - name nombre del frontmatter (o el del archivo si falta) - file nombre del archivo, con extensión - p cwd del proyecto - desc descripción del frontmatter - ty tipo declarado: project | user | feedback | reference - src uuid de la sesión que la creó, si lo declara - body cuerpo markdown, sin el frontmatter - ln enlaces [[...]] que aparecen en el cuerpo - k tamaño en KB - l mtime del archivo (ISO 8601) - ix True si figura en MEMORY.md - hix True si el proyecto tiene MEMORY.md - -`project_dir` es interno y `public_records()` lo saca antes de serializar. -""" - -import glob -import os -import re -from datetime import datetime, timezone - -from .sessions import SessionError, default_root - -INDEX_NAME = "MEMORY.md" - -INTERNAL_KEYS = ("project_dir",) - -TYPES = ("project", "user", "feedback", "reference") - -FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n?", re.S) -LINK_RE = re.compile(r"\[\[([^\]\n]+)\]\]") -# En el índice cada línea es "- [Título](archivo.md) — pista". -INDEX_LINK_RE = re.compile(r"\(([^)\n]+)\.md\)") - - -def memory_dir(project_dir, root=None): - return os.path.join(root or default_root(), project_dir, "memory") - - -def memory_path(m, root=None): - return os.path.join(memory_dir(m["project_dir"], root), m["file"]) - - -def index_path(project_dir, root=None): - return os.path.join(memory_dir(project_dir, root), INDEX_NAME) - - -# ──────────────────────────────── parseo ──────────────────────────────── - -def _field(front, key): - """Valor de una clave del frontmatter. Plano: alcanza para lo que escribe - Claude Code, que anida `type` y `originSessionId` pero sin repetirlas.""" - hit = re.search(r"^\s*%s:\s*(.+?)\s*$" % re.escape(key), front, re.M) - if not hit: - return None - value = hit.group(1).strip() - # YAML de una línea: si viene entrecomillado, las comillas internas están - # escapadas y hay que devolverlas como estaban. - for quote in ('"', "'"): - if len(value) >= 2 and value[0] == quote and value[-1] == quote: - value = value[1:-1] - if quote == '"': - value = value.replace('\\"', '"').replace("\\\\", "\\") - break - return value or None - - -def read_memory(path, project_dir): - with open(path, "r", encoding="utf-8", errors="ignore") as f: - raw = f.read() - - match = FRONTMATTER_RE.match(raw) - front, body = (match.group(1), raw[match.end():]) if match else ("", raw) - stat = os.stat(path) - filename = os.path.basename(path) - - return { - "name": _field(front, "name") or filename[:-3], - "file": filename, - "project_dir": project_dir, - "desc": _field(front, "description") or "", - "ty": _field(front, "type") or "—", - "src": _field(front, "originSessionId"), - "body": body.strip(), - "ln": sorted(set(LINK_RE.findall(body))), - "k": round(stat.st_size / 1024, 1), - "l": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(), - } - - -def read_index(project_dir, root=None): - """Nombres (sin .md) que el MEMORY.md del proyecto enlaza.""" - try: - with open(index_path(project_dir, root), "r", - encoding="utf-8", errors="ignore") as f: - return set(INDEX_LINK_RE.findall(f.read())) - except OSError: - return set() - - -# ──────────────────────────────── carga ──────────────────────────────── - -def load_memories(sessions, root=None): - """Lee las memorias de todos los proyectos. - - La ruta real del proyecto sale de las sesiones: el nombre del directorio - codifica "/" y "." los dos como "-" y no se puede invertir. - """ - root = root or default_root() - cwd_by_dir = {} - for s in sessions: - cwd_by_dir.setdefault(s.get("project_dir"), s.get("p")) - - memories = [] - for d in sorted(glob.glob(os.path.join(root, "*", "memory"))): - project_dir = os.path.basename(os.path.dirname(d)) - files = sorted(f for f in glob.glob(os.path.join(d, "*.md")) - if os.path.basename(f) != INDEX_NAME) - if not files: - continue # un memory/ vacío no es un proyecto con memoria - - has_index = os.path.exists(os.path.join(d, INDEX_NAME)) - listed = read_index(project_dir, root) if has_index else set() - - for path in files: - try: - m = read_memory(path, project_dir) - except OSError: - continue - m["p"] = cwd_by_dir.get(project_dir) or project_dir - m["hix"] = has_index - m["ix"] = m["file"][:-3] in listed - memories.append(m) - - memories.sort(key=lambda m: m["l"], reverse=True) - return memories - - -def public_records(memories): - """Copia sin las claves internas, lista para serializar.""" - out = [] - for m in memories: - clean = dict(m) - for key in INTERNAL_KEYS: - clean.pop(key, None) - out.append(clean) - return out - - -# ──────────────────────────────── filtros ──────────────────────────────── - -def apply_filters(memories, project=None, query=None, kind=None): - out = memories - - if project: - needle = os.path.expanduser(project).rstrip("/").lower() - out = [m for m in out if needle in m["p"].lower()] - - if kind: - out = [m for m in out if m["ty"].lower() == kind.lower()] - - if query: - needle = query.lower() - out = [m for m in out - if needle in m["name"].lower() - or needle in m["desc"].lower() - or needle in m["p"].lower() - or needle in m["body"].lower()] - - return out - - -def pick(memories, ref): - """Resuelve un índice de la tabla (1-based) o un prefijo del nombre.""" - if ref.isdigit(): - i = int(ref) - if 1 <= i <= len(memories): - return memories[i - 1] - raise SessionError( - f"el índice {i} está fuera de rango (hay {len(memories)} memorias)") - - needle = ref.lower() - hits = [m for m in memories if m["name"].lower().startswith(needle)] - if not hits: - hits = [m for m in memories if needle in m["name"].lower()] - if len(hits) == 1: - return hits[0] - if not hits: - raise SessionError(f"ninguna memoria coincide con '{ref}'") - nombres = ", ".join(m["name"] for m in hits[:4]) - raise SessionError( - f"'{ref}' es ambiguo, coincide con {len(hits)}: {nombres}" - + (", …" if len(hits) > 4 else "")) - - -# ──────────────────────────────── auditoría ──────────────────────────────── - -def audit(memories, sessions, root=None): - """Inconsistencias entre archivos, índices, enlaces y sesiones de origen.""" - known = {m["name"] for m in memories} | {m["file"][:-3] for m in memories} - session_ids = {s["id"] for s in sessions} - - report = { - "sin_indice": [m for m in memories if not m["hix"]], - "sin_listar": [m for m in memories if m["hix"] and not m["ix"]], - "enlaces_rotos": [(m, link) for m in memories - for link in m["ln"] if link not in known], - "origen_perdido": [m for m in memories - if m["src"] and m["src"] not in session_ids], - "indice_fantasma": [], - } - - for project_dir in sorted({m["project_dir"] for m in memories if m["hix"]}): - real = {m["file"][:-3] for m in memories - if m["project_dir"] == project_dir} - for missing in sorted(read_index(project_dir, root) - real): - report["indice_fantasma"].append((project_dir, missing)) - - return report - - -def audit_total(report): - return sum(len(v) for v in report.values()) - - -# ──────────────────────────────── borrado ──────────────────────────────── - -def unindex(m, root=None): - """Saca del MEMORY.md la línea que apunta a esta memoria. - - Devuelve True si el índice cambió. No es un error que no cambie: la memoria - podía no estar listada. - """ - path = index_path(m["project_dir"], root) - try: - with open(path, "r", encoding="utf-8", errors="ignore") as f: - lines = f.readlines() - except OSError: - return False - - needle = "(%s)" % m["file"] - kept = [ln for ln in lines if needle not in ln] - if len(kept) == len(lines): - return False - - try: - with open(path, "w", encoding="utf-8") as f: - f.writelines(kept) - except OSError: - return False - return True - - -def delete(m, root=None): - """Borra el archivo y lo saca del índice. Devuelve si se desindexó.""" - os.remove(memory_path(m, root)) - return unindex(m, root) diff --git a/claude_sesiones/sessions.py b/claude_sesiones/sessions.py deleted file mode 100644 index 0a00f8e..0000000 --- a/claude_sesiones/sessions.py +++ /dev/null @@ -1,409 +0,0 @@ -"""Parseo de los .jsonl que Claude Code deja en ~/.claude/projects/. - -Cada conversación es un archivo JSON Lines: una línea por evento. De ahí sale un -registro por sesión con claves de una letra, porque ese mismo registro viaja -embebido dentro del HTML y los nombres largos se pagan una vez por sesión. - -Esquema del registro que devuelve `read_session`: - - id uuid de la sesión (el nombre del archivo) - p cwd del proyecto - b rama de git - t título - ai True si el título lo generó Claude, False si es el primer mensaje - n True si parece un `claude -p` no interactivo - e True si la sesión no tiene ningún mensaje - i True si `p` se dedujo de otra sesión del mismo proyecto - f/l timestamp del primer y del último evento (ISO 8601) - d duración en minutos - u/a cantidad de mensajes tuyos / de Claude - k tamaño del .jsonl en KB - v versión de Claude Code - c transcripción: [{"r": "u" | "a" | "t", "x": texto}] - -`project_dir` y `mtime` son internos y no salen del módulo: `public_records()` -los saca antes de que el registro se serialice. -""" - -import glob -import json -import os -import re -from datetime import datetime, timezone - -# Sube si cambia el esquema del registro: invalida los cachés viejos en vez de -# leer registros con la forma anterior. -CACHE_VERSION = 2 - -EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc) - -INTERNAL_KEYS = ("project_dir", "mtime") - - -class SessionError(Exception): - """Error de uso que la CLI convierte en un mensaje y un código de salida.""" - - -# ──────────────────────────────── ubicaciones ──────────────────────────────── - -def default_root(): - """~/.claude/projects, o el equivalente si CLAUDE_CONFIG_DIR está seteada.""" - base = os.environ.get("CLAUDE_CONFIG_DIR") or os.path.join( - os.path.expanduser("~"), ".claude") - return os.path.join(base, "projects") - - -def default_cache_path(): - base = os.environ.get("XDG_CACHE_HOME") or os.path.expanduser("~/.cache") - return os.path.join(base, "claude-sesiones", "cache.json") - - -def session_path(s, root=None): - """Ruta del .jsonl. El nombre del archivo es el UUID y el del directorio - padre es lo que guardamos en project_dir, así que es reconstruible.""" - return os.path.join(root or default_root(), - s["project_dir"], s["id"] + ".jsonl") - - -# ─────────────────────────── parseo de los .jsonl ─────────────────────────── - -TAG_RE = re.compile(r"<[^>]+>") -REMINDER_RE = re.compile(r".*?", re.S) - -# Un mensaje que empieza con alguno de estos no es texto del usuario: es un -# bloque que genera la propia CLI al ejecutar un comando local. -SKIP_PREFIXES = ( - "= 3 else None - - -def tool_summary(block): - """Una línea del estilo 'Bash: git status' para una llamada a herramienta.""" - name = block.get("name") or "tool" - args = block.get("input") or {} - if not isinstance(args, dict): - return name - val = args.get(TOOL_KEY.get(name, "")) - if val is None: - val = next((v for v in args.values() if isinstance(v, str)), None) - if not isinstance(val, str): - return name - val = re.sub(r"\s+", " ", val).strip() - if len(val) > TOOL_ARG_MAX: - val = val[:TOOL_ARG_MAX] + "…" - return f"{name}: {val}" if val else name - - -def blocks_of(message): - content = message.get("content") - if isinstance(content, str): - return [{"type": "text", "text": content}] - return content if isinstance(content, list) else [] - - -def parse_ts(ts): - """ISO 8601 → datetime con zona, o None si no se puede leer.""" - if not ts: - return None - try: - return datetime.fromisoformat(ts.replace("Z", "+00:00")) - except (ValueError, AttributeError): - return None - - -def read_session(path): - """Parsea un .jsonl entero y devuelve el registro de esa sesión.""" - session_id = os.path.basename(path)[:-6] # sin .jsonl - first_ts = last_ts = cwd = git_branch = version = None - ai_title = fallback_title = None - user_msgs = assistant_msgs = 0 - convo = [] - - with open(path, "r", encoding="utf-8", errors="ignore") as f: - for line in f: - line = line.strip() - if not line: - continue - try: - obj = json.loads(line) - except json.JSONDecodeError: - continue # línea truncada por una sesión que sigue escribiendo - if not isinstance(obj, dict): - continue - - kind = obj.get("type") - - if kind == "ai-title": - if obj.get("aiTitle"): - ai_title = obj["aiTitle"] # nos quedamos con el más reciente - continue - - ts = obj.get("timestamp") - if ts: - if first_ts is None: - first_ts = ts - last_ts = ts - if cwd is None and obj.get("cwd"): - cwd = obj["cwd"] - if git_branch is None and obj.get("gitBranch"): - git_branch = obj["gitBranch"] - if obj.get("version"): - version = obj["version"] - - if kind not in ("user", "assistant") or obj.get("isSidechain"): - continue - - message = obj.get("message") - if not isinstance(message, dict): - continue - - if kind == "user": - if obj.get("isMeta"): - continue - for b in blocks_of(message): - if not isinstance(b, dict): - continue - if b.get("type") == "text": - text = clean_text(b.get("text")) - if text: - user_msgs += 1 - if fallback_title is None: - fallback_title = text[:TITLE_MAX] - convo.append({"r": "u", "x": text}) - elif b.get("type") == "image": - convo.append({"r": "u", "x": "[imagen adjunta]"}) - else: - counted = False - for b in blocks_of(message): - if not isinstance(b, dict): - continue - if b.get("type") == "text": - text = (b.get("text") or "").strip() - if text: - convo.append({"r": "a", "x": text}) - counted = True - elif b.get("type") == "tool_use": - convo.append({"r": "t", "x": tool_summary(b)}) - if counted: - assistant_msgs += 1 - - st = os.stat(path) - ft, lt = parse_ts(first_ts), parse_ts(last_ts) - - # Un único mensaje enorme y ninguna ida y vuelta es la firma de un - # `claude -p` con algo piped por stdin (p. ej. un git diff para redactar el - # mensaje de commit), no de una conversación. - noninteractive = ( - user_msgs == 1 and not ai_title and bool(convo) - and len(convo[0]["x"]) > NONINTERACTIVE_CHARS - ) - - return { - "id": session_id, - "project_dir": os.path.basename(os.path.dirname(path)), - "p": cwd, - "b": git_branch, - "t": ai_title or fallback_title, - "ai": bool(ai_title), - "n": noninteractive, - "e": not convo, - "f": first_ts, - "l": last_ts, - "d": round((lt - ft).total_seconds() / 60) if ft and lt else None, - "u": user_msgs, - "a": assistant_msgs, - "k": round(st.st_size / 1024, 1), - "v": version, - "c": convo, - "mtime": datetime.fromtimestamp(st.st_mtime, tz=timezone.utc).isoformat(), - } - - -# ──────────────────────────────── caché ──────────────────────────────── - -def _load_cache(path): - """Entradas del caché, o {} si no existe, está roto o quedó viejo.""" - try: - with open(path, encoding="utf-8") as f: - blob = json.load(f) - except (OSError, ValueError, UnicodeDecodeError): - return {} - if not isinstance(blob, dict) or blob.get("v") != CACHE_VERSION: - return {} - entries = blob.get("entries") - return entries if isinstance(entries, dict) else {} - - -def _save_cache(path, entries): - """Escribe el caché de forma atómica. Si falla, no pasa nada.""" - try: - os.makedirs(os.path.dirname(path), exist_ok=True) - # El pid en el temporal evita que dos corridas simultáneas se pisen. - tmp = f"{path}.{os.getpid()}.tmp" - with open(tmp, "w", encoding="utf-8") as f: - json.dump({"v": CACHE_VERSION, "entries": entries}, f, - ensure_ascii=False, separators=(",", ":")) - os.replace(tmp, path) - except OSError: - pass # el caché es una optimización, no una condición de uso - - -def drop_from_cache(paths, cache_path=None): - """Saca del caché las sesiones borradas para que no reaparezcan.""" - cache_path = cache_path or default_cache_path() - entries = _load_cache(cache_path) - if not entries: - return - if any(entries.pop(p, None) is not None for p in list(paths)): - _save_cache(cache_path, entries) - - -# ──────────────────────────────── carga ──────────────────────────────── - -def _fill_gaps(sessions): - """Completa lo que falta después de parsear todos los archivos. - - Algunas sesiones (un /resume cancelado) nunca registran cwd. El nombre del - directorio no se puede invertir de forma fiable porque "/" y "." se - codifican los dos como "-", así que tomamos la ruta prestada de otra sesión - del mismo proyecto y lo dejamos marcado en `i`. - """ - known = {} - for s in sessions: - if s["p"]: - known.setdefault(s["project_dir"], s["p"]) - - for s in sessions: - s["i"] = not s["p"] - if not s["p"]: - s["p"] = known.get(s["project_dir"], s["project_dir"]) - if not s["l"]: - s["l"] = s["mtime"] - if not s["f"]: - s["f"] = s["mtime"] - - -def load_sessions(root=None, cache_path=None, use_cache=True): - """Parsea todas las sesiones, reusando del caché las que no cambiaron.""" - root = root or default_root() - cache_path = cache_path or default_cache_path() - paths = sorted(glob.glob(os.path.join(root, "*", "*.jsonl"))) - - cache = _load_cache(cache_path) if use_cache else {} - - sessions, fresh, reparsed = [], {}, False - for path in paths: - try: - st = os.stat(path) - except OSError: - continue - stamp = f"{st.st_mtime_ns}:{st.st_size}" - hit = cache.get(path) - if (isinstance(hit, dict) and hit.get("stamp") == stamp - and isinstance(hit.get("rec"), dict)): - rec = hit["rec"] - else: - try: - rec = read_session(path) - except OSError: - continue - reparsed = True - fresh[path] = {"stamp": stamp, "rec": rec} - sessions.append(rec) - - # Antes de `_fill_gaps`, a propósito: al caché va el registro tal como salió - # del archivo, sin los campos deducidos a partir de las otras sesiones. - if use_cache and (reparsed or len(fresh) != len(cache)): - _save_cache(cache_path, fresh) - - _fill_gaps(sessions) - sessions.sort(key=lambda s: parse_ts(s["l"]) or EPOCH, reverse=True) - return sessions - - -def latest_activity(sessions): - """El instante más reciente de los datos: el "ahora" contra el que se - calculan las fechas relativas, para que no dependan del reloj de quien mira.""" - stamps = [parse_ts(s["l"]) for s in sessions] - return max([t for t in stamps if t], default=EPOCH) - - -def public_records(sessions): - """Copias sin las claves internas, listas para serializar.""" - return [{k: v for k, v in s.items() if k not in INTERNAL_KEYS} - for s in sessions] - - -# ──────────────────────────────── filtros ──────────────────────────────── - -def apply_filters(sessions, project=None, grep=None, query=None, - hide_empty=False): - out = sessions - - if project: - needle = os.path.expanduser(project).rstrip("/").lower() - out = [s for s in out if needle in s["p"].lower()] - - if grep: - needle = grep.lower() - out = [s for s in out - if any(needle in m["x"].lower() for m in s["c"])] - - if query: - needle = query.lower() - out = [s for s in out - if needle in (s["t"] or "").lower() - or needle in s["p"].lower() - or needle in (s["b"] or "").lower() - or s["id"].startswith(needle)] - - if hide_empty: - out = [s for s in out if not s["e"]] - - return out - - -def pick(sessions, ref): - """Resuelve un índice de la tabla (1-based) o un prefijo de UUID.""" - if ref.isdigit(): - i = int(ref) - if 1 <= i <= len(sessions): - return sessions[i - 1] - raise SessionError( - f"el índice {i} está fuera de rango (hay {len(sessions)})") - - hits = [s for s in sessions if s["id"].startswith(ref.lower())] - if len(hits) == 1: - return hits[0] - if not hits: - raise SessionError(f"ninguna sesión empieza con '{ref}'") - raise SessionError(f"'{ref}' es ambiguo, coincide con {len(hits)} sesiones") diff --git a/claude_sesiones/template.html b/claude_sesiones/template.html deleted file mode 100644 index e56cc30..0000000 --- a/claude_sesiones/template.html +++ /dev/null @@ -1,1394 +0,0 @@ - - - - - - - -Sesiones de Claude Code - - - -