Cleanup: move transcripts out, scripts into scripts/, drop scaffold
- transcripts/ moved to ~/shorashim-transcripts (separate from code repo) Added to .gitignore so it stays out - src/l1_aron/build_l1.py → scripts/build_l1.py - src/tools/conversation_search.py → scripts/conversation_search.py - Removed empty src/ scaffolding (l0_luchot, sanhedrin, shema_layer, shoresh_engine, urim_v_tumim — all .gitkeep stubs from an earlier layout, never used; the real code lives in shoresh-engine/, sanhedrin/, vav_pipeline/, etc.) - Removed Untitled/ and erl_crash.dump (untracked junk) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Scott KarpandClaude Opus 4.6 committed
1 parent
5a6fce4945
commit
4775122c84
10 files changed
+1
-17
No files matched your search
@@ -0,0 +1,683 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
L1 Torah Filesystem Builder — Bezalel's Chisel
|
||||
|
||||
Parses the Westminster Leningrad Codex (morphhb OSIS XML) and produces
|
||||
the read-only Torah filesystem for Shorashim.
|
||||
|
||||
This is a construction tool. It runs once. It produces JSON files.
|
||||
The consumers (Haskell L0, Erlang Shoresh Engine) will read these files.
|
||||
|
||||
Ka'asher tzivah Hashem et Moshe.
|
||||
"""
|
||||
|
||||
import json
|
||||
import hashlib
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import unicodedata
|
||||
import xml.etree.ElementTree as ET
|
||||
from pathlib import Path
|
||||
from collections import defaultdict
|
||||
|
||||
# === CONSTANTS ===
|
||||
|
||||
OSIS_NS = "http://www.bibletechnologies.net/2003/OSIS/namespace"
|
||||
|
||||
TORAH_BOOKS = [
|
||||
("Gen", "bereishit"),
|
||||
("Exod", "shemot"),
|
||||
("Lev", "vayikra"),
|
||||
("Num", "bamidbar"),
|
||||
("Deut", "devarim"),
|
||||
]
|
||||
|
||||
# Hebrew prefix letters and their morphhb codes
|
||||
PREFIX_MAP = {
|
||||
"b": "ב", # bet - in/with
|
||||
"c": "ו", # vav-consecutive/conjunctive
|
||||
"d": "ה", # he ha-yedia (definite article)
|
||||
"e": "ה", # he interrogative
|
||||
"f": "ו", # vav conjunctive (variant)
|
||||
"k": "כ", # kaf - like/as
|
||||
"l": "ל", # lamed - to/for
|
||||
"m": "מ", # mem - from
|
||||
"s": "ש", # shin - that/which (relative)
|
||||
}
|
||||
|
||||
# Unicode ranges for Hebrew cantillation marks (te'amim)
|
||||
# Range: U+0591 to U+05AF
|
||||
TEAMIM_MAP = {
|
||||
"\u0591": "etnachta",
|
||||
"\u0592": "segol_accent", # segol (accent, not vowel)
|
||||
"\u0593": "shalshelet",
|
||||
"\u0594": "zaqef_qatan",
|
||||
"\u0595": "zaqef_gadol",
|
||||
"\u0596": "tipcha",
|
||||
"\u0597": "revia",
|
||||
"\u0598": "zarqa",
|
||||
"\u0599": "pashta",
|
||||
"\u059A": "yetiv",
|
||||
"\u059B": "tevir",
|
||||
"\u059C": "geresh",
|
||||
"\u059D": "geresh_muqdam",
|
||||
"\u059E": "gershayim",
|
||||
"\u059F": "qarney_para",
|
||||
"\u05A0": "telisha_gedola",
|
||||
"\u05A1": "pazer",
|
||||
"\u05A2": "atnah_hafukh",
|
||||
"\u05A3": "munach",
|
||||
"\u05A4": "mahapakh",
|
||||
"\u05A5": "mercha",
|
||||
"\u05A6": "mercha_kefula",
|
||||
"\u05A7": "darga",
|
||||
"\u05A8": "qadma",
|
||||
"\u05A9": "telisha_qetana",
|
||||
"\u05AA": "yerah_ben_yomo",
|
||||
"\u05AB": "ole",
|
||||
"\u05AC": "iluy",
|
||||
"\u05AD": "dehi",
|
||||
"\u05AE": "zinor",
|
||||
"\u05AF": "masora_circle",
|
||||
}
|
||||
|
||||
# U+05BD (meteg) doubles as siluq (cantillation mark on final word before sof-pasuq).
|
||||
# We treat it as a cantillation mark so it shows up in te'amim extraction.
|
||||
TEAMIM_MAP["\u05BD"] = "meteg_siluq"
|
||||
|
||||
# Unicode ranges for Hebrew vowel points (nikkud)
|
||||
# Range: U+05B0 to U+05BC, U+05BF, U+05C1, U+05C2, U+05C7
|
||||
# NOTE: U+05BD (meteg/siluq) is in TEAMIM_MAP, not here
|
||||
VOWEL_POINTS = set(
|
||||
"\u05B0\u05B1\u05B2\u05B3\u05B4\u05B5\u05B6\u05B7"
|
||||
"\u05B8\u05B9\u05BA\u05BB\u05BC\u05BF\u05C1\u05C2\u05C7"
|
||||
)
|
||||
|
||||
# Sof pasuq and maqaf
|
||||
SOF_PASUQ = "\u05C3" # ׃
|
||||
MAQAF = "\u05BE" # ־
|
||||
PASEQ = "\u05C0" # ׀
|
||||
|
||||
|
||||
def strip_cantillation(text: str) -> str:
|
||||
"""Remove cantillation marks, keep vowels and consonants."""
|
||||
return "".join(c for c in text if c not in TEAMIM_MAP)
|
||||
|
||||
|
||||
def strip_vowels_and_cantillation(text: str) -> str:
|
||||
"""Extract consonantal text only — strip vowels, cantillation, and dagesh."""
|
||||
result = []
|
||||
for c in text:
|
||||
if c in TEAMIM_MAP:
|
||||
continue
|
||||
if c in VOWEL_POINTS:
|
||||
continue
|
||||
if c == "/":
|
||||
continue
|
||||
if c == MAQAF or c == SOF_PASUQ or c == PASEQ:
|
||||
continue
|
||||
result.append(c)
|
||||
return "".join(result)
|
||||
|
||||
|
||||
def extract_teamim(text: str) -> list[str]:
|
||||
"""Extract cantillation mark names from Hebrew text."""
|
||||
marks = []
|
||||
for c in text:
|
||||
if c in TEAMIM_MAP:
|
||||
marks.append(TEAMIM_MAP[c])
|
||||
return marks
|
||||
|
||||
|
||||
def parse_prefixes(lemma_str: str, morph_str: str) -> tuple[list[str], str, str]:
|
||||
"""
|
||||
Parse prefix codes from lemma and morph strings.
|
||||
|
||||
Returns: (prefix_hebrew_letters, core_lemma, core_morph)
|
||||
|
||||
lemma format: "b/7225" or "c/d/776" (prefix codes / Strong's number)
|
||||
morph format: "HR/Ncfsa" or "HC/Td/Ncbsa" (H + prefix types / core morph)
|
||||
"""
|
||||
prefixes = []
|
||||
|
||||
# Parse lemma prefixes
|
||||
lemma_parts = lemma_str.split("/")
|
||||
core_lemma = lemma_parts[-1] # Last part is always the Strong's number
|
||||
|
||||
for part in lemma_parts[:-1]:
|
||||
if part in PREFIX_MAP:
|
||||
prefixes.append(PREFIX_MAP[part])
|
||||
|
||||
# Parse morph — strip leading 'H' and split on '/'
|
||||
core_morph = morph_str
|
||||
if morph_str.startswith("H"):
|
||||
morph_inner = morph_str[1:]
|
||||
morph_parts = morph_inner.split("/")
|
||||
core_morph = morph_parts[-1] # Last part is the core morphology
|
||||
|
||||
return prefixes, core_lemma, core_morph
|
||||
|
||||
|
||||
def parse_morph_code(morph: str) -> dict:
|
||||
"""
|
||||
Parse a morphhb morphological code into a structured dict.
|
||||
|
||||
Code format (after stripping H prefix and prefixes):
|
||||
First char = part of speech:
|
||||
V=verb, N=noun, A=adjective, P=pronoun, R=preposition,
|
||||
T=particle, C=conjunction, D=adverb, S=suffix
|
||||
|
||||
For verbs: V + stem(q/N/p/P/h/H/t/etc) + conjugation(p/i/w/j/v/r/c/a)
|
||||
+ person(1/2/3) + gender(m/f/c) + number(s/p/d)
|
||||
For nouns: N + type(c/p) + gender(m/f/b) + number(s/p/d) + state(a/c/d)
|
||||
"""
|
||||
if not morph:
|
||||
return {"raw": morph}
|
||||
|
||||
result = {"raw": morph}
|
||||
|
||||
if len(morph) < 1:
|
||||
return result
|
||||
|
||||
pos = morph[0]
|
||||
pos_map = {
|
||||
"V": "verb", "N": "noun", "A": "adjective",
|
||||
"P": "pronoun", "R": "preposition", "T": "particle",
|
||||
"C": "conjunction", "D": "adverb", "S": "suffix",
|
||||
}
|
||||
result["pos"] = pos_map.get(pos, pos)
|
||||
|
||||
if pos == "V" and len(morph) >= 3:
|
||||
stem_map = {
|
||||
"q": "qal", "N": "nifal", "p": "piel", "P": "pual",
|
||||
"h": "hifil", "H": "hofal", "t": "hitpael",
|
||||
"o": "polel", "O": "polal", "r": "hitpolel",
|
||||
"m": "poel", "M": "poal", "k": "palel",
|
||||
"K": "pulal", "Q": "qal_passive", "l": "pilpel",
|
||||
"L": "polpal", "f": "hithpalpel", "D": "nithpael",
|
||||
"j": "pealal", "i": "pilel", "u": "hothpaal",
|
||||
"c": "tiphil", "v": "hishtaphel", "w": "nithpalel",
|
||||
"y": "nithpoel", "z": "hithpoel",
|
||||
}
|
||||
result["stem"] = stem_map.get(morph[1], morph[1])
|
||||
|
||||
conj_map = {
|
||||
"p": "perfect", "i": "imperfect", "w": "vav_consecutive",
|
||||
"j": "jussive", "v": "imperative", "r": "participle_active",
|
||||
"s": "participle_passive", "a": "infinitive_absolute",
|
||||
"c": "infinitive_construct",
|
||||
}
|
||||
if len(morph) >= 3:
|
||||
result["conjugation"] = conj_map.get(morph[2], morph[2])
|
||||
|
||||
elif pos == "N" and len(morph) >= 2:
|
||||
type_map = {"c": "common", "p": "proper"}
|
||||
result["noun_type"] = type_map.get(morph[1], morph[1])
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def strong_to_lemma_text(strong_num: str) -> str:
|
||||
"""
|
||||
Clean a Strong's number string for use as lemma identifier.
|
||||
Handles formats like "1254 a", "430", "7225"
|
||||
"""
|
||||
return strong_num.strip()
|
||||
|
||||
|
||||
def derive_shoresh_from_word(consonants: str, morph_info: dict) -> str | None:
|
||||
"""
|
||||
Attempt to derive a triliteral root from consonantal text and morphology.
|
||||
|
||||
This is a heuristic — not all words have triliteral roots (particles,
|
||||
pronouns, etc). Returns None when derivation fails.
|
||||
"""
|
||||
pos = morph_info.get("pos", "")
|
||||
|
||||
# Particles, prepositions, conjunctions, pronouns don't have shoreshim
|
||||
if pos in ("preposition", "particle", "conjunction", "pronoun", "suffix"):
|
||||
return None
|
||||
|
||||
# For the consonantal text, we want only Hebrew letters
|
||||
hebrew_only = re.sub(r"[^\u05D0-\u05EA]", "", consonants)
|
||||
|
||||
if len(hebrew_only) < 2:
|
||||
return None
|
||||
|
||||
stem = morph_info.get("stem", "")
|
||||
conjugation = morph_info.get("conjugation", "")
|
||||
|
||||
if pos == "verb":
|
||||
# Strip imperfect/jussive/vav-consecutive prefixes: י מ נ ת א
|
||||
# These are conjugation prefixes, not part of the root
|
||||
if conjugation in ("imperfect", "vav_consecutive", "jussive"):
|
||||
if len(hebrew_only) >= 4 and hebrew_only[0] in "ימנתא":
|
||||
hebrew_only = hebrew_only[1:]
|
||||
# Strip hifil/hofal he prefix
|
||||
if stem in ("hifil", "hofal") and len(hebrew_only) >= 4:
|
||||
if hebrew_only[0] == "ה":
|
||||
hebrew_only = hebrew_only[1:]
|
||||
# Imperfect hifil: prefix already stripped, check for yod remnant
|
||||
# Strip hitpael tav prefix (and sometimes he)
|
||||
elif stem == "hitpael" and len(hebrew_only) >= 4:
|
||||
if hebrew_only[0] == "ה" and len(hebrew_only) >= 5 and hebrew_only[1] == "ת":
|
||||
hebrew_only = hebrew_only[2:]
|
||||
elif hebrew_only[0] == "ת":
|
||||
hebrew_only = hebrew_only[1:]
|
||||
# Strip nifal nun prefix
|
||||
elif stem == "nifal" and len(hebrew_only) >= 4:
|
||||
if hebrew_only[0] == "נ":
|
||||
hebrew_only = hebrew_only[1:]
|
||||
# Strip participle mem prefix
|
||||
if conjugation in ("participle_active", "participle_passive"):
|
||||
if len(hebrew_only) >= 4 and hebrew_only[0] == "מ":
|
||||
hebrew_only = hebrew_only[1:]
|
||||
|
||||
elif pos == "noun":
|
||||
# Strip common noun prefixes: מ (place/instrument nouns), ת (abstract nouns)
|
||||
if len(hebrew_only) >= 4 and hebrew_only[0] in "מת":
|
||||
pass # Don't strip by default for nouns — too aggressive
|
||||
|
||||
# Strip common suffixes
|
||||
for suffix in ["ים", "ות", "ין", "תי", "תם", "תן", "נו"]:
|
||||
if hebrew_only.endswith(suffix) and len(hebrew_only) - len(suffix) >= 2:
|
||||
hebrew_only = hebrew_only[:-len(suffix)]
|
||||
break
|
||||
for suffix in ["ה", "ת", "ו", "י", "ך", "ם", "ן"]:
|
||||
if hebrew_only.endswith(suffix) and len(hebrew_only) - 1 >= 3:
|
||||
hebrew_only = hebrew_only[:-1]
|
||||
break
|
||||
|
||||
if len(hebrew_only) == 3:
|
||||
return hebrew_only
|
||||
elif len(hebrew_only) == 2:
|
||||
return hebrew_only
|
||||
elif len(hebrew_only) > 3:
|
||||
# Take first three as best guess
|
||||
return hebrew_only[:3]
|
||||
|
||||
return hebrew_only if hebrew_only else None
|
||||
|
||||
|
||||
def parse_verse(verse_elem, book_osis: str, book_hebrew: str) -> dict:
|
||||
"""Parse a single verse element into the L1 JSON format."""
|
||||
osis_id = verse_elem.get("osisID") # e.g., "Gen.1.1"
|
||||
parts = osis_id.split(".")
|
||||
chapter = int(parts[1])
|
||||
verse_num = int(parts[2])
|
||||
|
||||
words = []
|
||||
full_text_parts = []
|
||||
word_position = 0
|
||||
|
||||
for elem in verse_elem:
|
||||
tag = elem.tag.replace(f"{{{OSIS_NS}}}", "")
|
||||
|
||||
if tag == "w":
|
||||
word_position += 1
|
||||
word_data = parse_word(elem, word_position)
|
||||
words.append(word_data)
|
||||
# For text_hebrew, use the joined text (no / separators)
|
||||
full_text_parts.append((elem.text or "").replace("/", ""))
|
||||
|
||||
elif tag == "seg":
|
||||
seg_type = elem.get("type", "")
|
||||
if seg_type == "x-sof-pasuq":
|
||||
full_text_parts.append("׃")
|
||||
elif seg_type == "x-maqqef":
|
||||
full_text_parts.append("־")
|
||||
elif seg_type == "x-paseq":
|
||||
full_text_parts.append(" ׀ ")
|
||||
# Also check for trailing <w> elements after seg (maqaf-joined words)
|
||||
|
||||
# Handle mixed content: seg followed by w within verse
|
||||
if elem.tail and elem.tail.strip():
|
||||
pass # tail text handled by next element
|
||||
|
||||
# Reconstruct full verse text
|
||||
text_hebrew = " ".join(full_text_parts).strip()
|
||||
# Clean up spacing around maqaf
|
||||
text_hebrew = text_hebrew.replace(" ־ ", "־").replace(" ־", "־").replace("־ ", "־")
|
||||
text_hebrew = re.sub(r"\s+", " ", text_hebrew).strip()
|
||||
|
||||
ref = f"{book_hebrew.capitalize()} {chapter}:{verse_num}"
|
||||
|
||||
return {
|
||||
"ref": ref,
|
||||
"book": book_hebrew,
|
||||
"chapter": chapter,
|
||||
"verse": verse_num,
|
||||
"text_hebrew": text_hebrew,
|
||||
"words": words,
|
||||
}
|
||||
|
||||
|
||||
def parse_word(w_elem, position: int) -> dict:
|
||||
"""Parse a single <w> element into word data."""
|
||||
raw_text = w_elem.text or ""
|
||||
lemma_str = w_elem.get("lemma", "")
|
||||
morph_str = w_elem.get("morph", "")
|
||||
word_id = w_elem.get("id", "")
|
||||
|
||||
# Parse prefixes
|
||||
prefixes, core_lemma, core_morph = parse_prefixes(lemma_str, morph_str)
|
||||
|
||||
# The displayed text includes prefix separators as /
|
||||
# Remove the / for the display text but preserve for analysis
|
||||
display_text = raw_text # Keep as-is with / markers
|
||||
|
||||
# Get text without prefix separators for the full word
|
||||
joined_text = raw_text.replace("/", "")
|
||||
|
||||
# Get the word part (after last /) for the core word
|
||||
text_parts = raw_text.split("/")
|
||||
core_text = text_parts[-1] if text_parts else raw_text
|
||||
|
||||
# Extract consonantal text (strip everything but Hebrew letters)
|
||||
consonants_full = strip_vowels_and_cantillation(joined_text)
|
||||
consonants_core = strip_vowels_and_cantillation(core_text)
|
||||
|
||||
# Extract cantillation marks
|
||||
teamim = extract_teamim(joined_text)
|
||||
|
||||
# Parse morphology
|
||||
morph_info = parse_morph_code(core_morph)
|
||||
|
||||
# Derive shoresh
|
||||
shoresh = derive_shoresh_from_word(consonants_core, morph_info)
|
||||
|
||||
# Build prefix string
|
||||
prefix_str = "".join(prefixes) if prefixes else ""
|
||||
|
||||
word_data = {
|
||||
"position": position,
|
||||
"text": joined_text,
|
||||
"consonants": consonants_full,
|
||||
"lemma": core_lemma,
|
||||
"morph": morph_str,
|
||||
"teamim": teamim,
|
||||
}
|
||||
|
||||
if prefix_str:
|
||||
word_data["prefix"] = prefix_str
|
||||
|
||||
if shoresh:
|
||||
word_data["shoresh"] = shoresh
|
||||
|
||||
return word_data
|
||||
|
||||
|
||||
def parse_book(xml_path: str, book_osis: str, book_hebrew: str) -> list[dict]:
|
||||
"""Parse an entire OSIS XML book file into verses."""
|
||||
tree = ET.parse(xml_path)
|
||||
root = tree.getroot()
|
||||
|
||||
verses = []
|
||||
|
||||
# Navigate: osis > osisText > div (book) > chapter > verse
|
||||
for elem in root.iter(f"{{{OSIS_NS}}}verse"):
|
||||
osis_id = elem.get("osisID", "")
|
||||
if not osis_id.startswith(f"{book_osis}."):
|
||||
continue
|
||||
|
||||
verse_data = parse_verse(elem, book_osis, book_hebrew)
|
||||
verses.append(verse_data)
|
||||
|
||||
return verses
|
||||
|
||||
|
||||
def build_l1(morphhb_dir: str, output_dir: str):
|
||||
"""Build the complete L1 Torah filesystem."""
|
||||
wlc_dir = os.path.join(morphhb_dir, "wlc")
|
||||
torah_dir = os.path.join(output_dir, "torah")
|
||||
|
||||
total_verses = 0
|
||||
total_words = 0
|
||||
total_roots = set()
|
||||
missing_teamim = []
|
||||
failed_shoresh = []
|
||||
book_stats = {}
|
||||
|
||||
print("=" * 60)
|
||||
print("L1 TORAH FILESYSTEM BUILDER")
|
||||
print("Bereishit bara Elohim et hashamayim v'et ha'aretz.")
|
||||
print("=" * 60)
|
||||
print()
|
||||
|
||||
for book_osis, book_hebrew in TORAH_BOOKS:
|
||||
xml_path = os.path.join(wlc_dir, f"{book_osis}.xml")
|
||||
|
||||
if not os.path.exists(xml_path):
|
||||
print(f"ERROR: {xml_path} not found!")
|
||||
sys.exit(1)
|
||||
|
||||
print(f"Parsing {book_osis} ({book_hebrew})...", end=" ", flush=True)
|
||||
|
||||
verses = parse_book(xml_path, book_osis, book_hebrew)
|
||||
|
||||
# Organize by chapter
|
||||
chapters = defaultdict(list)
|
||||
for v in verses:
|
||||
chapters[v["chapter"]].append(v)
|
||||
|
||||
book_dir = os.path.join(torah_dir, book_hebrew)
|
||||
book_verse_count = 0
|
||||
book_word_count = 0
|
||||
chapter_info = {}
|
||||
|
||||
for chapter_num in sorted(chapters.keys()):
|
||||
chapter_verses = chapters[chapter_num]
|
||||
chapter_dir = os.path.join(book_dir, str(chapter_num))
|
||||
os.makedirs(chapter_dir, exist_ok=True)
|
||||
|
||||
for verse_data in chapter_verses:
|
||||
verse_num = verse_data["verse"]
|
||||
verse_path = os.path.join(chapter_dir, f"{verse_num}.json")
|
||||
|
||||
with open(verse_path, "w", encoding="utf-8") as f:
|
||||
json.dump(verse_data, f, ensure_ascii=False, indent=2)
|
||||
|
||||
book_verse_count += 1
|
||||
total_verses += 1
|
||||
|
||||
# Count words and collect stats
|
||||
for word in verse_data["words"]:
|
||||
book_word_count += 1
|
||||
total_words += 1
|
||||
|
||||
if word.get("shoresh"):
|
||||
total_roots.add(word["shoresh"])
|
||||
else:
|
||||
# Only flag content words (not particles) as failed
|
||||
morph_info = parse_morph_code(
|
||||
word["morph"].split("/")[-1].lstrip("H") if word["morph"] else ""
|
||||
)
|
||||
if morph_info.get("pos") in ("verb", "noun", "adjective"):
|
||||
failed_shoresh.append({
|
||||
"ref": verse_data["ref"],
|
||||
"word": word["text"],
|
||||
"morph": word["morph"],
|
||||
})
|
||||
|
||||
# Check for te'amim
|
||||
verse_has_teamim = any(
|
||||
len(w.get("teamim", [])) > 0 for w in verse_data["words"]
|
||||
)
|
||||
if not verse_has_teamim:
|
||||
missing_teamim.append(verse_data["ref"])
|
||||
|
||||
chapter_info[chapter_num] = len(chapter_verses)
|
||||
|
||||
# Write book index
|
||||
book_index = {
|
||||
"book": book_hebrew,
|
||||
"osis_id": book_osis,
|
||||
"chapters": len(chapters),
|
||||
"verses": book_verse_count,
|
||||
"words": book_word_count,
|
||||
"chapter_verse_counts": chapter_info,
|
||||
}
|
||||
|
||||
index_path = os.path.join(book_dir, "index.json")
|
||||
with open(index_path, "w", encoding="utf-8") as f:
|
||||
json.dump(book_index, f, ensure_ascii=False, indent=2)
|
||||
|
||||
book_stats[book_hebrew] = book_index
|
||||
|
||||
print(f"{len(chapters)} chapters, {book_verse_count} verses, {book_word_count} words")
|
||||
|
||||
# Write master index
|
||||
master_index = {
|
||||
"name": "Torah",
|
||||
"source": "Westminster Leningrad Codex (morphhb)",
|
||||
"books": list(book_stats.values()),
|
||||
"total_chapters": sum(b["chapters"] for b in book_stats.values()),
|
||||
"total_verses": total_verses,
|
||||
"total_words": total_words,
|
||||
"unique_roots": len(total_roots),
|
||||
}
|
||||
|
||||
master_path = os.path.join(torah_dir, "index.json")
|
||||
with open(master_path, "w", encoding="utf-8") as f:
|
||||
json.dump(master_index, f, ensure_ascii=False, indent=2)
|
||||
|
||||
# Generate SHA-256 integrity hash
|
||||
print("\nGenerating integrity hash...")
|
||||
sha256 = generate_integrity_hash(torah_dir)
|
||||
|
||||
hash_path = os.path.join(output_dir, "integrity.sha256")
|
||||
with open(hash_path, "w") as f:
|
||||
f.write(sha256)
|
||||
|
||||
# Print report
|
||||
print()
|
||||
print("=" * 60)
|
||||
print("L1 BUILD REPORT")
|
||||
print("=" * 60)
|
||||
print(f"Total books: {len(TORAH_BOOKS)}")
|
||||
print(f"Total chapters: {master_index['total_chapters']}")
|
||||
print(f"Total verses: {total_verses}")
|
||||
print(f"Total words: {total_words}")
|
||||
print(f"Unique shoreshim: {len(total_roots)}")
|
||||
print(f"Verses missing te'amim: {len(missing_teamim)}")
|
||||
print(f"Content words missing shoresh: {len(failed_shoresh)}")
|
||||
print(f"SHA-256: {sha256}")
|
||||
print()
|
||||
|
||||
if missing_teamim:
|
||||
print(f"Verses without te'amim ({len(missing_teamim)}):")
|
||||
for ref in missing_teamim[:20]:
|
||||
print(f" - {ref}")
|
||||
if len(missing_teamim) > 20:
|
||||
print(f" ... and {len(missing_teamim) - 20} more")
|
||||
print()
|
||||
|
||||
if failed_shoresh:
|
||||
print(f"Content words without shoresh ({len(failed_shoresh)}):")
|
||||
for entry in failed_shoresh[:20]:
|
||||
print(f" - {entry['ref']}: {entry['word']} ({entry['morph']})")
|
||||
if len(failed_shoresh) > 20:
|
||||
print(f" ... and {len(failed_shoresh) - 20} more")
|
||||
print()
|
||||
|
||||
# Write the report as JSON too
|
||||
report = {
|
||||
"total_books": len(TORAH_BOOKS),
|
||||
"total_chapters": master_index["total_chapters"],
|
||||
"total_verses": total_verses,
|
||||
"total_words": total_words,
|
||||
"unique_shoreshim": len(total_roots),
|
||||
"verses_missing_teamim": len(missing_teamim),
|
||||
"content_words_missing_shoresh": len(failed_shoresh),
|
||||
"sha256": sha256,
|
||||
"missing_teamim_refs": missing_teamim,
|
||||
"failed_shoresh_samples": failed_shoresh[:100],
|
||||
}
|
||||
|
||||
report_path = os.path.join(output_dir, "build_report.json")
|
||||
with open(report_path, "w", encoding="utf-8") as f:
|
||||
json.dump(report, f, ensure_ascii=False, indent=2)
|
||||
|
||||
print(f"Report written to: {report_path}")
|
||||
print(f"Master index: {master_path}")
|
||||
print(f"Integrity hash: {hash_path}")
|
||||
|
||||
return report
|
||||
|
||||
|
||||
def generate_integrity_hash(torah_dir: str) -> str:
|
||||
"""Generate SHA-256 hash of the entire L1 Torah directory tree."""
|
||||
sha256 = hashlib.sha256()
|
||||
|
||||
for root, dirs, files in sorted(os.walk(torah_dir)):
|
||||
dirs.sort()
|
||||
for filename in sorted(files):
|
||||
filepath = os.path.join(root, filename)
|
||||
rel_path = os.path.relpath(filepath, torah_dir)
|
||||
# Hash the relative path
|
||||
sha256.update(rel_path.encode("utf-8"))
|
||||
# Hash the file contents
|
||||
with open(filepath, "rb") as f:
|
||||
while True:
|
||||
chunk = f.read(8192)
|
||||
if not chunk:
|
||||
break
|
||||
sha256.update(chunk)
|
||||
|
||||
return sha256.hexdigest()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Paths relative to the shorashim repository
|
||||
repo_root = Path(__file__).resolve().parent.parent.parent
|
||||
morphhb_dir = repo_root / "data" / "morphhb"
|
||||
output_dir = repo_root / "l1"
|
||||
|
||||
if not morphhb_dir.exists():
|
||||
print(f"ERROR: morphhb not found at {morphhb_dir}")
|
||||
print("Run: git clone https://github.com/openscriptures/morphhb.git data/morphhb")
|
||||
sys.exit(1)
|
||||
|
||||
# Clean output directory
|
||||
if output_dir.exists():
|
||||
import shutil
|
||||
shutil.rmtree(output_dir)
|
||||
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
|
||||
report = build_l1(str(morphhb_dir), str(output_dir))
|
||||
|
||||
# Verify counts
|
||||
print()
|
||||
print("=" * 60)
|
||||
print("VERIFICATION")
|
||||
print("=" * 60)
|
||||
|
||||
expected_books = 5
|
||||
expected_chapters = 187
|
||||
# WLC/BHS versification gives 5853 (8 more than the traditional Masoretic
|
||||
# count of 5845 due to verse numbering differences in Shemot, Bamidbar, Devarim)
|
||||
expected_verses = 5853
|
||||
|
||||
checks = [
|
||||
("Books", report["total_books"], expected_books),
|
||||
("Chapters", report["total_chapters"], expected_chapters),
|
||||
("Verses", report["total_verses"], expected_verses),
|
||||
]
|
||||
|
||||
all_pass = True
|
||||
for name, actual, expected in checks:
|
||||
status = "PASS" if actual == expected else "FAIL"
|
||||
if actual != expected:
|
||||
all_pass = False
|
||||
print(f" {name}: {actual} (expected {expected}) [{status}]")
|
||||
|
||||
te_amim_status = "PASS" if report["verses_missing_teamim"] == 0 else "WARN"
|
||||
print(f" Te'amim coverage: {report['verses_missing_teamim']} missing [{te_amim_status}]")
|
||||
|
||||
print()
|
||||
if all_pass:
|
||||
print("The firmament is in place. The throne sits above it.")
|
||||
else:
|
||||
print("Verification incomplete. Review the report.")
|
||||
@@ -0,0 +1,529 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Bezalel's Memory Well — Conversation History Search
|
||||
|
||||
Converts Claude Code JSONL conversation transcripts into a searchable
|
||||
SQLite database with FTS5 full-text search. This is the bridge between
|
||||
sessions — the builder's ability to remember what was said, what broke,
|
||||
and what was decided.
|
||||
|
||||
This is a construction tool. It runs when the builder needs to search
|
||||
past conversations. It is not a running service.
|
||||
|
||||
Usage:
|
||||
# Index all transcripts
|
||||
python conversation_search.py index
|
||||
|
||||
# Search for a term
|
||||
python conversation_search.py search "shoresh derivation"
|
||||
|
||||
# Search with context (shows surrounding messages)
|
||||
python conversation_search.py search "siluq meteg" --context 3
|
||||
|
||||
# List all sessions
|
||||
python conversation_search.py sessions
|
||||
|
||||
# Re-index (rebuild from scratch)
|
||||
python conversation_search.py reindex
|
||||
"""
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
import sys
|
||||
import os
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
|
||||
# Default paths
|
||||
CLAUDE_PROJECTS_DIR = Path.home() / ".claude" / "projects"
|
||||
DB_PATH = Path.home() / ".claude" / "conversation_history.db"
|
||||
|
||||
|
||||
def create_db(db_path: Path) -> sqlite3.Connection:
|
||||
"""Create the SQLite database with FTS5 tables."""
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
conn.execute("PRAGMA journal_mode=WAL")
|
||||
conn.execute("PRAGMA synchronous=NORMAL")
|
||||
|
||||
conn.executescript("""
|
||||
CREATE TABLE IF NOT EXISTS sessions (
|
||||
session_id TEXT PRIMARY KEY,
|
||||
source_file TEXT NOT NULL,
|
||||
first_timestamp TEXT,
|
||||
last_timestamp TEXT,
|
||||
message_count INTEGER DEFAULT 0,
|
||||
user_message_count INTEGER DEFAULT 0,
|
||||
assistant_message_count INTEGER DEFAULT 0
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS messages (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
session_id TEXT NOT NULL,
|
||||
uuid TEXT UNIQUE,
|
||||
parent_uuid TEXT,
|
||||
type TEXT NOT NULL,
|
||||
role TEXT,
|
||||
timestamp TEXT,
|
||||
content TEXT,
|
||||
model TEXT,
|
||||
cwd TEXT,
|
||||
has_tool_use INTEGER DEFAULT 0,
|
||||
has_thinking INTEGER DEFAULT 0,
|
||||
FOREIGN KEY (session_id) REFERENCES sessions(session_id)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_messages_session
|
||||
ON messages(session_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_messages_timestamp
|
||||
ON messages(timestamp);
|
||||
CREATE INDEX IF NOT EXISTS idx_messages_type
|
||||
ON messages(type);
|
||||
|
||||
-- FTS5 virtual table for full-text search
|
||||
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
|
||||
content,
|
||||
session_id UNINDEXED,
|
||||
uuid UNINDEXED,
|
||||
role UNINDEXED,
|
||||
timestamp UNINDEXED,
|
||||
content='messages',
|
||||
content_rowid='id',
|
||||
tokenize='porter unicode61'
|
||||
);
|
||||
|
||||
-- Triggers to keep FTS in sync
|
||||
CREATE TRIGGER IF NOT EXISTS messages_ai AFTER INSERT ON messages BEGIN
|
||||
INSERT INTO messages_fts(rowid, content, session_id, uuid, role, timestamp)
|
||||
VALUES (new.id, new.content, new.session_id, new.uuid, new.role, new.timestamp);
|
||||
END;
|
||||
|
||||
CREATE TRIGGER IF NOT EXISTS messages_ad AFTER DELETE ON messages BEGIN
|
||||
INSERT INTO messages_fts(messages_fts, rowid, content, session_id, uuid, role, timestamp)
|
||||
VALUES ('delete', old.id, old.content, old.session_id, old.uuid, old.role, old.timestamp);
|
||||
END;
|
||||
|
||||
CREATE TRIGGER IF NOT EXISTS messages_au AFTER UPDATE ON messages BEGIN
|
||||
INSERT INTO messages_fts(messages_fts, rowid, content, session_id, uuid, role, timestamp)
|
||||
VALUES ('delete', old.id, old.content, old.session_id, old.uuid, old.role, old.timestamp);
|
||||
INSERT INTO messages_fts(rowid, content, session_id, uuid, role, timestamp)
|
||||
VALUES (new.id, new.content, new.session_id, new.uuid, new.role, new.timestamp);
|
||||
END;
|
||||
""")
|
||||
|
||||
conn.commit()
|
||||
return conn
|
||||
|
||||
|
||||
def extract_text_content(message_data: dict) -> Optional[str]:
|
||||
"""Extract readable text from a message's content field.
|
||||
|
||||
Handles both string content and array content blocks.
|
||||
Extracts text, tool_use inputs, and tool_result content.
|
||||
Skips thinking blocks (those are private to the model).
|
||||
"""
|
||||
msg = message_data.get("message", {})
|
||||
content = msg.get("content")
|
||||
|
||||
if content is None:
|
||||
return None
|
||||
|
||||
if isinstance(content, str):
|
||||
return content if content.strip() else None
|
||||
|
||||
if isinstance(content, list):
|
||||
parts = []
|
||||
for block in content:
|
||||
if isinstance(block, str):
|
||||
parts.append(block)
|
||||
elif isinstance(block, dict):
|
||||
block_type = block.get("type", "")
|
||||
|
||||
if block_type == "text":
|
||||
text = block.get("text", "")
|
||||
if text.strip():
|
||||
parts.append(text)
|
||||
|
||||
elif block_type == "tool_use":
|
||||
tool_name = block.get("name", "unknown")
|
||||
tool_input = block.get("input", {})
|
||||
# Include tool name and key inputs for searchability
|
||||
input_str = json.dumps(tool_input, ensure_ascii=False)
|
||||
# Truncate very long tool inputs
|
||||
if len(input_str) > 2000:
|
||||
input_str = input_str[:2000] + "..."
|
||||
parts.append(f"[tool:{tool_name}] {input_str}")
|
||||
|
||||
elif block_type == "tool_result":
|
||||
result_content = block.get("content", "")
|
||||
if isinstance(result_content, str) and result_content.strip():
|
||||
# Truncate very long tool results
|
||||
if len(result_content) > 3000:
|
||||
result_content = result_content[:3000] + "..."
|
||||
parts.append(f"[result] {result_content}")
|
||||
|
||||
# Skip thinking blocks — those are the model's private reasoning
|
||||
|
||||
return "\n".join(parts) if parts else None
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def index_jsonl_file(conn: sqlite3.Connection, jsonl_path: Path) -> dict:
|
||||
"""Index a single JSONL transcript file into the database.
|
||||
|
||||
Returns stats about what was indexed.
|
||||
"""
|
||||
stats = {
|
||||
"file": str(jsonl_path),
|
||||
"total_lines": 0,
|
||||
"messages_indexed": 0,
|
||||
"skipped": 0,
|
||||
"errors": 0,
|
||||
}
|
||||
|
||||
session_id = jsonl_path.stem # filename without extension
|
||||
timestamps = []
|
||||
user_count = 0
|
||||
assistant_count = 0
|
||||
|
||||
with open(jsonl_path, "r", encoding="utf-8") as f:
|
||||
for line_num, line in enumerate(f, 1):
|
||||
stats["total_lines"] += 1
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
|
||||
try:
|
||||
record = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
stats["errors"] += 1
|
||||
continue
|
||||
|
||||
record_type = record.get("type")
|
||||
|
||||
# Only index user and assistant messages
|
||||
if record_type not in ("user", "assistant"):
|
||||
stats["skipped"] += 1
|
||||
continue
|
||||
|
||||
uuid = record.get("uuid")
|
||||
if not uuid:
|
||||
stats["skipped"] += 1
|
||||
continue
|
||||
|
||||
text_content = extract_text_content(record)
|
||||
if not text_content:
|
||||
stats["skipped"] += 1
|
||||
continue
|
||||
|
||||
parent_uuid = record.get("parentUuid")
|
||||
timestamp = record.get("timestamp")
|
||||
msg = record.get("message", {})
|
||||
role = msg.get("role", record_type)
|
||||
model = msg.get("model")
|
||||
cwd = record.get("cwd")
|
||||
|
||||
# Detect tool use and thinking
|
||||
has_tool_use = 0
|
||||
has_thinking = 0
|
||||
content = msg.get("content", [])
|
||||
if isinstance(content, list):
|
||||
for block in content:
|
||||
if isinstance(block, dict):
|
||||
if block.get("type") == "tool_use":
|
||||
has_tool_use = 1
|
||||
if block.get("type") == "thinking":
|
||||
has_thinking = 1
|
||||
|
||||
if timestamp:
|
||||
timestamps.append(timestamp)
|
||||
|
||||
if role == "user":
|
||||
user_count += 1
|
||||
elif role == "assistant":
|
||||
assistant_count += 1
|
||||
|
||||
try:
|
||||
conn.execute("""
|
||||
INSERT OR REPLACE INTO messages
|
||||
(session_id, uuid, parent_uuid, type, role, timestamp,
|
||||
content, model, cwd, has_tool_use, has_thinking)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""", (session_id, uuid, parent_uuid, record_type, role,
|
||||
timestamp, text_content, model, cwd,
|
||||
has_tool_use, has_thinking))
|
||||
stats["messages_indexed"] += 1
|
||||
except sqlite3.Error as e:
|
||||
stats["errors"] += 1
|
||||
|
||||
# Update session record
|
||||
first_ts = min(timestamps) if timestamps else None
|
||||
last_ts = max(timestamps) if timestamps else None
|
||||
|
||||
conn.execute("""
|
||||
INSERT OR REPLACE INTO sessions
|
||||
(session_id, source_file, first_timestamp, last_timestamp,
|
||||
message_count, user_message_count, assistant_message_count)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||||
""", (session_id, str(jsonl_path), first_ts, last_ts,
|
||||
stats["messages_indexed"], user_count, assistant_count))
|
||||
|
||||
conn.commit()
|
||||
return stats
|
||||
|
||||
|
||||
def find_jsonl_files(projects_dir: Path) -> list[Path]:
|
||||
"""Find all JSONL conversation transcripts."""
|
||||
files = []
|
||||
if projects_dir.exists():
|
||||
for jsonl in projects_dir.rglob("*.jsonl"):
|
||||
files.append(jsonl)
|
||||
return sorted(files, key=lambda p: p.stat().st_mtime, reverse=True)
|
||||
|
||||
|
||||
def cmd_index(args):
|
||||
"""Index all JSONL transcripts into the search database."""
|
||||
projects_dir = Path(args.projects_dir)
|
||||
db_path = Path(args.db)
|
||||
|
||||
jsonl_files = find_jsonl_files(projects_dir)
|
||||
if not jsonl_files:
|
||||
print(f"No JSONL files found in {projects_dir}")
|
||||
return
|
||||
|
||||
print(f"Found {len(jsonl_files)} transcript(s)")
|
||||
|
||||
conn = create_db(db_path)
|
||||
|
||||
for jsonl_path in jsonl_files:
|
||||
size_mb = jsonl_path.stat().st_size / (1024 * 1024)
|
||||
print(f"\nIndexing: {jsonl_path.name} ({size_mb:.1f} MB)")
|
||||
stats = index_jsonl_file(conn, jsonl_path)
|
||||
print(f" Lines: {stats['total_lines']}")
|
||||
print(f" Indexed: {stats['messages_indexed']}")
|
||||
print(f" Skipped: {stats['skipped']} (non-message records)")
|
||||
if stats["errors"]:
|
||||
print(f" Errors: {stats['errors']}")
|
||||
|
||||
# Print total stats
|
||||
total = conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0]
|
||||
sessions = conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0]
|
||||
print(f"\nTotal: {total} messages across {sessions} session(s)")
|
||||
print(f"Database: {db_path}")
|
||||
|
||||
conn.close()
|
||||
|
||||
|
||||
def cmd_reindex(args):
|
||||
"""Drop and rebuild the entire database."""
|
||||
db_path = Path(args.db)
|
||||
if db_path.exists():
|
||||
os.remove(db_path)
|
||||
print(f"Removed existing database: {db_path}")
|
||||
cmd_index(args)
|
||||
|
||||
|
||||
def cmd_search(args):
|
||||
"""Search conversation history using FTS5."""
|
||||
db_path = Path(args.db)
|
||||
if not db_path.exists():
|
||||
print(f"Database not found: {db_path}")
|
||||
print("Run 'conversation_search.py index' first.")
|
||||
return
|
||||
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
conn.row_factory = sqlite3.Row
|
||||
|
||||
query = args.query
|
||||
limit = args.limit
|
||||
context = args.context
|
||||
role_filter = args.role
|
||||
|
||||
# Build the search query
|
||||
sql = """
|
||||
SELECT
|
||||
m.id,
|
||||
m.session_id,
|
||||
m.uuid,
|
||||
m.role,
|
||||
m.timestamp,
|
||||
m.content,
|
||||
m.model,
|
||||
m.has_tool_use,
|
||||
m.has_thinking,
|
||||
highlight(messages_fts, 0, '>>>', '<<<') as highlighted,
|
||||
rank
|
||||
FROM messages_fts
|
||||
JOIN messages m ON m.id = messages_fts.rowid
|
||||
WHERE messages_fts MATCH ?
|
||||
"""
|
||||
params = [query]
|
||||
|
||||
if role_filter:
|
||||
sql += " AND m.role = ?"
|
||||
params.append(role_filter)
|
||||
|
||||
sql += " ORDER BY rank LIMIT ?"
|
||||
params.append(limit)
|
||||
|
||||
results = conn.execute(sql, params).fetchall()
|
||||
|
||||
if not results:
|
||||
print(f"No results for: {query}")
|
||||
return
|
||||
|
||||
print(f"Found {len(results)} result(s) for: {query}\n")
|
||||
|
||||
for i, row in enumerate(results):
|
||||
ts = row["timestamp"]
|
||||
if ts:
|
||||
try:
|
||||
dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
|
||||
ts_display = dt.strftime("%Y-%m-%d %H:%M")
|
||||
except (ValueError, AttributeError):
|
||||
ts_display = ts[:16] if ts else "?"
|
||||
else:
|
||||
ts_display = "?"
|
||||
|
||||
role = row["role"] or "?"
|
||||
model = row["model"] or ""
|
||||
session = row["session_id"][:8]
|
||||
|
||||
print(f"--- [{i+1}] {role} | {ts_display} | session:{session}... ---")
|
||||
if model:
|
||||
print(f" model: {model}")
|
||||
|
||||
# Show highlighted content
|
||||
highlighted = row["highlighted"]
|
||||
# Trim to reasonable length for display
|
||||
lines = highlighted.split("\n")
|
||||
display_lines = []
|
||||
for line in lines:
|
||||
if ">>>" in line or len(display_lines) < 5:
|
||||
display_lines.append(line)
|
||||
if len(display_lines) >= 20:
|
||||
display_lines.append("...")
|
||||
break
|
||||
|
||||
for line in display_lines:
|
||||
# Truncate very long lines
|
||||
if len(line) > 200:
|
||||
line = line[:200] + "..."
|
||||
print(f" {line}")
|
||||
|
||||
# Show context messages if requested
|
||||
if context > 0:
|
||||
context_msgs = conn.execute("""
|
||||
SELECT role, timestamp, content
|
||||
FROM messages
|
||||
WHERE session_id = ?
|
||||
AND id BETWEEN ? - ? AND ? + ?
|
||||
AND id != ?
|
||||
ORDER BY id
|
||||
""", (row["session_id"], row["id"], context,
|
||||
row["id"], context, row["id"])).fetchall()
|
||||
|
||||
if context_msgs:
|
||||
print(f" --- context ---")
|
||||
for ctx in context_msgs:
|
||||
ctx_role = ctx["role"] or "?"
|
||||
ctx_content = ctx["content"] or ""
|
||||
if len(ctx_content) > 150:
|
||||
ctx_content = ctx_content[:150] + "..."
|
||||
print(f" [{ctx_role}] {ctx_content}")
|
||||
|
||||
print()
|
||||
|
||||
conn.close()
|
||||
|
||||
|
||||
def cmd_sessions(args):
|
||||
"""List all indexed sessions."""
|
||||
db_path = Path(args.db)
|
||||
if not db_path.exists():
|
||||
print(f"Database not found: {db_path}")
|
||||
return
|
||||
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
conn.row_factory = sqlite3.Row
|
||||
|
||||
sessions = conn.execute("""
|
||||
SELECT * FROM sessions ORDER BY first_timestamp DESC
|
||||
""").fetchall()
|
||||
|
||||
if not sessions:
|
||||
print("No sessions indexed.")
|
||||
return
|
||||
|
||||
print(f"{'Session ID':<40} {'Messages':>8} {'User':>6} {'Asst':>6} {'First':>20} {'Last':>20}")
|
||||
print("-" * 110)
|
||||
|
||||
for s in sessions:
|
||||
sid = s["session_id"][:36]
|
||||
first = s["first_timestamp"][:16] if s["first_timestamp"] else "?"
|
||||
last = s["last_timestamp"][:16] if s["last_timestamp"] else "?"
|
||||
print(f"{sid:<40} {s['message_count']:>8} {s['user_message_count']:>6} "
|
||||
f"{s['assistant_message_count']:>6} {first:>20} {last:>20}")
|
||||
|
||||
conn.close()
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Bezalel's Memory Well — Search conversation history"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--db", default=str(DB_PATH),
|
||||
help=f"Path to SQLite database (default: {DB_PATH})"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--projects-dir", default=str(CLAUDE_PROJECTS_DIR),
|
||||
help=f"Path to Claude projects directory (default: {CLAUDE_PROJECTS_DIR})"
|
||||
)
|
||||
|
||||
subparsers = parser.add_subparsers(dest="command")
|
||||
|
||||
# index
|
||||
subparsers.add_parser("index", help="Index JSONL transcripts")
|
||||
|
||||
# reindex
|
||||
subparsers.add_parser("reindex", help="Rebuild database from scratch")
|
||||
|
||||
# search
|
||||
search_parser = subparsers.add_parser("search", help="Search conversations")
|
||||
search_parser.add_argument("query", help="FTS5 search query")
|
||||
search_parser.add_argument(
|
||||
"--limit", type=int, default=10,
|
||||
help="Max results (default: 10)"
|
||||
)
|
||||
search_parser.add_argument(
|
||||
"--context", type=int, default=0,
|
||||
help="Show N surrounding messages for context"
|
||||
)
|
||||
search_parser.add_argument(
|
||||
"--role", choices=["user", "assistant"],
|
||||
help="Filter by role"
|
||||
)
|
||||
|
||||
# sessions
|
||||
subparsers.add_parser("sessions", help="List indexed sessions")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.command == "index":
|
||||
cmd_index(args)
|
||||
elif args.command == "reindex":
|
||||
cmd_reindex(args)
|
||||
elif args.command == "search":
|
||||
cmd_search(args)
|
||||
elif args.command == "sessions":
|
||||
cmd_sessions(args)
|
||||
else:
|
||||
parser.print_help()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in new issue
Block a user