bin/sources.py

""" sources.py - everywhere a history can be imported from, in one table. WHY ONE TABLE ------------- This started as two files named after a particular product. That worked and it was wrong: anyone who downloaded the package read a competitor's name in a filename, so a write-up that had been carefully de-identified for readers was re-identified for downloaders. Hiding the name would have been worse - dishonest and still there. The fix is to stop needing it. A source is a ROW here: an id, a label, a detector, a reader and a mapper. Adding support for another tool is a table entry. The architecture stops producing accusations, which is a better outcome than being careful about wording. TRANSCRIPTS ARE ROW ONE, AND THE DEFAULT ---------------------------------------- Claude Code writes ~/.claude/projects/*.jsonl for everybody - whatever memory tool they run, or none at all. That is the universal source. It names nobody, it needs no other software present, and it serves the much larger group who never installed a memory tool and simply want their last two months back. THE REASON ANY OF THIS EXISTS ----------------------------- The people most likely to want this tool are the ones already running the other one. If importing their history means reading a README, finding a path and running a script with an env var, most will not - and then leaving costs them two months of memory. That turns "your privacy or your history" into a real choice, which is exactly the choice this release exists to remove. You do not have to choose between your privacy and your history. """ from __future__ import annotations import os import sqlite3 import sys from datetime import datetime, timezone from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) # ── detectors ─────────────────────────────────────────────────────────────────── def _detect_transcripts() -> dict | None: root = Path.home() / ".claude" / "projects" if not root.is_dir(): return None files = [f for d in root.iterdir() if d.is_dir() for f in d.glob("*.jsonl")] if not files: return None size = sum(f.stat().st_size for f in files) oldest = min(f.stat().st_mtime for f in files) return {"path": root, "size": size, "items": len(files), "since": datetime.fromtimestamp(oldest, timezone.utc), "counts": {"transcripts": len(files)}} # A legacy store is identified by its TABLES, not by its filename or folder - a # directory called ".something" proves nothing, and guessing from a name is how a # tool ends up touching a file it did not understand. _LEGACY_TABLES = {"observations", "sdk_sessions"} _LEGACY_EXTRA = {"session_summaries", "user_prompts"} _LEGACY_PATHS = ( Path.home() / ".claude-mem" / "claude-mem.db", Path.home() / ".claude-mem" / "claude-mem.sqlite", ) # Directory names the old tool created beside its database. These are DATA - strings # matched against a disk - and they live here, in the source table, for the same # reason the store paths do: a literal name an importer must match is allowed, and # everywhere else in the package is not. leftovers.py reads this rather than carrying # the names itself. LEGACY_SIBLINGS = { "logs": "its log files", "observer-sessions": "its saved observer sessions", "chroma": "its search index", "archive": "its own archives", "handoff": "session handoffs (plain text from your conversations)", "backups": "its backups", } def dedicated_store_dirs() -> tuple[Path, ...]: """The directories a legacy store is known to live in ALONE. This matters for cleanup, not for import. When the database sits in a folder that belongs to nothing else, the files beside it can be attributed to the same tool. When the user points --path at a database sitting in their home directory or Downloads, the folder is theirs, and a directory in it named "archive" or "backups" is THEIR archive - not a leftover. Identifying it by name alone would be a guess, and this package does not act on guesses. """ return tuple({p.parent.resolve() for p in _LEGACY_PATHS}) def _open_ro(db: Path) -> sqlite3.Connection | None: try: return sqlite3.connect(f"file:{db}?mode=ro", uri=True) except Exception: return None def _detect_legacy_store(explicit: Path | None = None) -> dict | None: cands = [explicit] if explicit else [] env = os.environ.get("RECALL_IMPORT_DB") if env and not explicit: cands.append(Path(env).expanduser()) cands += list(_LEGACY_PATHS) for db in cands: if not db or not Path(db).is_file(): continue con = _open_ro(Path(db)) if not con: continue try: names = {r[0] for r in con.execute( "SELECT name FROM sqlite_master WHERE type='table'")} if not _LEGACY_TABLES.issubset(names): continue counts = {} for t in sorted(_LEGACY_TABLES | (_LEGACY_EXTRA & names)): try: counts[t] = con.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0] except Exception: counts[t] = 0 st = Path(db).stat() return {"path": Path(db), "size": st.st_size, "items": sum(counts.values()), "since": datetime.fromtimestamp(st.st_mtime, timezone.utc), "counts": counts} finally: con.close() return None # ── readers: turn a detected source into rows the store understands ───────────── def _read_transcripts(found: dict, project: str) -> dict: import capture return capture.ingest_all(project) if hasattr(capture, "ingest_all") \ else _read_transcripts_fallback(project) def _read_transcripts_fallback(project: str) -> dict: import subprocess, json as _json r = subprocess.run([sys.executable, str(Path(__file__).resolve().parent / "capture.py"), "--all", "--project", project], capture_output=True, text=True) try: return _json.loads(r.stdout[r.stdout.index("{"):]) except Exception: return {} def _read_legacy_store(found: dict, project: str) -> dict: import legacy_store return legacy_store.import_from(found["path"]) # ── the table ─────────────────────────────────────────────────────────────────── SOURCES = [ { "id": "transcripts", "label": "Claude Code transcripts", "note": "the universal source - present for everyone, needs no other tool", "primary": True, "removable": False, # never offered for cleanup; they are not ours "detect": _detect_transcripts, "read": _read_transcripts, }, { "id": "legacy-store", "label": "a previous memory tool's database", "note": "detected by its tables, not its name", "primary": False, "removable": True, "detect": _detect_legacy_store, "read": _read_legacy_store, }, ] def detect_all(explicit: Path | None = None) -> list[dict]: out = [] for s in SOURCES: try: found = s["detect"](explicit) if s["id"] == "legacy-store" else s["detect"]() except TypeError: found = s["detect"]() except Exception: found = None if found: out.append({**s, "found": found}) return out def describe(entry: dict) -> str: f = entry["found"] lines = [f" {entry['label']}", f" at {f['path']}", f" size {f['size']/1e6:.1f} MB", f" since {f['since'].strftime('%d %B %Y')}"] for k, v in f["counts"].items(): lines.append(f" {k:20} {v:,}") return "\n".join(lines)