diff options
Diffstat (limited to 'claude_sesiones')
| -rw-r--r-- | claude_sesiones/__init__.py | 8 | ||||
| -rw-r--r-- | claude_sesiones/__main__.py | 6 | ||||
| -rw-r--r-- | claude_sesiones/cli.py | 455 | ||||
| -rw-r--r-- | claude_sesiones/memory.py | 272 | ||||
| -rw-r--r-- | claude_sesiones/sessions.py | 409 | ||||
| -rw-r--r-- | claude_sesiones/template.html | 1394 | ||||
| -rw-r--r-- | claude_sesiones/terminal.py | 372 | ||||
| -rw-r--r-- | claude_sesiones/webpage.py | 74 |
8 files changed, 0 insertions, 2990 deletions
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 <nº> 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 <nº> 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 <proyecto>/memory/. - -Cada proyecto puede acumular recuerdos en - - ~/.claude/projects/<proyecto>/memory/<nombre>.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"<system-reminder>.*?</system-reminder>", 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 = ( - "<local-command-caveat", "<command-name", "<command-message", - "<command-args", "<local-command-stdout", "<system-reminder", -) - -TITLE_MAX = 160 -TOOL_ARG_MAX = 140 - -# Umbral del heurístico de `claude -p`: un único mensaje más largo que esto, sin -# ninguna ida y vuelta, es un pipe por stdin y no una conversación. -NONINTERACTIVE_CHARS = 1500 - -# Para cada herramienta, el parámetro que mejor resume qué hizo. -TOOL_KEY = { - "Bash": "command", "Read": "file_path", "Edit": "file_path", - "Write": "file_path", "NotebookEdit": "notebook_path", "Glob": "pattern", - "Grep": "pattern", "WebFetch": "url", "WebSearch": "query", - "Task": "description", "Agent": "description", "Skill": "skill", -} - - -def clean_text(s): - """Devuelve texto de usuario legible, o None si es ruido del harness.""" - if not isinstance(s, str): - return None - s = s.strip() - if not s or s.startswith(SKIP_PREFIXES): - return None - s = REMINDER_RE.sub(" ", s) - s = TAG_RE.sub(" ", s) - s = re.sub(r"\s+", " ", s).strip() - return s if len(s) >= 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"] - |