my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	#!/usr/bin/env python3
#2	"""
#3	Mnemosyne Temporal Triples Backfill Script
#4	==========================================
#5
#6	Generates temporal triples (occurred_on, has_source) for all existing
#7	working_memory and episodic_memory entries that lack them.
#8
#9	Usage:
#10	python scripts/backfill_temporal_triples.py [--dry-run]
#11
#12	This is a one-time migration script for Mnemosyne v1.13.0.
#13	"""
#14
#15	import os
#16	import sqlite3
#17	import argparse
#18	from pathlib import Path
#19	from datetime import datetime
#20	from typing import Tuple
#21
#22
#23	def get_db_path() -> Path:
#24	"""Resolve Mnemosyne database path."""
#25	default_dir = os.environ.get("MNEMOSYNE_DATA_DIR") or (
#26	Path.home() / ".hermes" / "mnemosyne" / "data"
#27	)
#28	return Path(default_dir) / "mnemosyne.db"
#29
#30
#31	def count_missing_triples(conn: sqlite3.Connection) -> Tuple[int, int]:
#32	"""Count working and episodic memories lacking temporal triples."""
#33	cursor = conn.cursor()
#34
#35	# Count working memories without occurred_on triples
#36	cursor.execute("""
#37	SELECT COUNT(*) FROM working_memory wm
#38	WHERE NOT EXISTS (
#39	SELECT 1 FROM triples t
#40	WHERE t.subject = wm.id AND t.predicate = 'occurred_on'
#41	)
#42	""")
#43	working_missing = cursor.fetchone()[0]
#44
#45	# Count episodic memories without occurred_on triples
#46	cursor.execute("""
#47	SELECT COUNT(*) FROM episodic_memory em
#48	WHERE NOT EXISTS (
#49	SELECT 1 FROM triples t
#50	WHERE t.subject = em.id AND t.predicate = 'occurred_on'
#51	)
#52	""")
#53	episodic_missing = cursor.fetchone()[0]
#54
#55	return working_missing, episodic_missing
#56
#57
#58	def backfill_working_memory(conn: sqlite3.Connection, dry_run: bool = False) -> int:
#59	"""Generate temporal triples for all working_memory entries."""
#60	cursor = conn.cursor()
#61
#62	cursor.execute("""
#63	SELECT id, timestamp, source FROM working_memory
#64	WHERE NOT EXISTS (
#65	SELECT 1 FROM triples t
#66	WHERE t.subject = working_memory.id AND t.predicate = 'occurred_on'
#67	)
#68	""")
#69
#70	rows = cursor.fetchall()
#71	inserted = 0
#72
#73	for memory_id, timestamp, source in rows:
#74	date_str = timestamp[:10] if timestamp else datetime.now().isoformat()[:10]
#75
#76	if not dry_run:
#77	cursor.execute("""
#78	INSERT INTO triples (subject, predicate, object, valid_from, source, confidence)
#79	VALUES (?, ?, ?, ?, ?, ?)
#80	""", (memory_id, 'occurred_on', date_str, date_str, 'backfill', 1.0))
#81
#82	if source and source not in ('conversation', 'user', 'assistant'):
#83	cursor.execute("""
#84	INSERT INTO triples (subject, predicate, object, valid_from, source, confidence)
#85	VALUES (?, ?, ?, ?, ?, ?)
#86	""", (memory_id, 'has_source', source, date_str, 'backfill', 1.0))
#87
#88	inserted += 1
#89
#90	if inserted % 500 == 0:
#91	print(f" Processed {inserted} working memories...")
#92	if not dry_run:
#93	conn.commit()
#94
#95	if not dry_run:
#96	conn.commit()
#97
#98	return inserted
#99
#100
#101	def backfill_episodic_memory(conn: sqlite3.Connection, dry_run: bool = False) -> int:
#102	"""Generate temporal triples for all episodic_memory entries."""
#103	cursor = conn.cursor()
#104
#105	cursor.execute("""
#106	SELECT id, timestamp, source FROM episodic_memory
#107	WHERE NOT EXISTS (
#108	SELECT 1 FROM triples t
#109	WHERE t.subject = episodic_memory.id AND t.predicate = 'occurred_on'
#110	)
#111	""")
#112
#113	rows = cursor.fetchall()
#114	inserted = 0
#115
#116	for memory_id, timestamp, source in rows:
#117	date_str = timestamp[:10] if timestamp else datetime.now().isoformat()[:10]
#118
#119	if not dry_run:
#120	cursor.execute("""
#121	INSERT INTO triples (subject, predicate, object, valid_from, source, confidence)
#122	VALUES (?, ?, ?, ?, ?, ?)
#123	""", (memory_id, 'occurred_on', date_str, date_str, 'backfill', 1.0))
#124
#125	if source and source not in ('conversation', 'user', 'assistant'):
#126	cursor.execute("""
#127	INSERT INTO triples (subject, predicate, object, valid_from, source, confidence)
#128	VALUES (?, ?, ?, ?, ?, ?)
#129	""", (memory_id, 'has_source', source, date_str, 'backfill', 1.0))
#130
#131	inserted += 1
#132
#133	if not dry_run:
#134	conn.commit()
#135
#136	return inserted
#137
#138
#139	def main():
#140	parser = argparse.ArgumentParser(description='Backfill temporal triples for Mnemosyne')
#141	parser.add_argument('--dry-run', action='store_true', help='Show what would be done without writing')
#142	parser.add_argument('--db-path', type=str, help='Path to mnemosyne.db (default: ~/.hermes/mnemosyne/data/mnemosyne.db)')
#143	args = parser.parse_args()
#144
#145	db_path = Path(args.db_path) if args.db_path else get_db_path()
#146
#147	print(f"Mnemosyne Temporal Triples Backfill")
#148	print(f"Database: {db_path}")
#149	print(f"Mode: {'DRY RUN' if args.dry_run else 'LIVE'}")
#150	print()
#151
#152	if not db_path.exists():
#153	print(f"ERROR: Database not found at {db_path}")
#154	return 1
#155
#156	conn = sqlite3.connect(str(db_path))
#157
#158	try:
#159	# Check current state
#160	working_missing, episodic_missing = count_missing_triples(conn)
#161	total_missing = working_missing + episodic_missing
#162
#163	print(f"Missing temporal triples:")
#164	print(f" Working memory: {working_missing}")
#165	print(f" Episodic memory: {episodic_missing}")
#166	print(f" Total: {total_missing}")
#167	print()
#168
#169	if total_missing == 0:
#170	print("All memories already have temporal triples. Nothing to do.")
#171	return 0
#172
#173	if args.dry_run:
#174	print("DRY RUN — no changes will be made.")
#175	print(f"Would insert {total_missing} occurred_on triples.")
#176	return 0
#177
#178	# Confirm
#179	print("This will insert temporal triples for all existing memories.")
#180	print("Type 'yes' to continue: ", end='')
#181
#182	# In non-interactive mode, just proceed
#183	print("(proceeding in non-interactive mode)")
#184	print()
#185
#186	# Backfill working memory
#187	print("Backfilling working_memory...")
#188	working_inserted = backfill_working_memory(conn, dry_run=args.dry_run)
#189	print(f" Inserted {working_inserted} triples for working memory")
#190
#191	# Backfill episodic memory
#192	print("Backfilling episodic_memory...")
#193	episodic_inserted = backfill_episodic_memory(conn, dry_run=args.dry_run)
#194	print(f" Inserted {episodic_inserted} triples for episodic memory")
#195
#196	print()
#197	print(f"Done. Total triples inserted: {working_inserted + episodic_inserted}")
#198
#199	finally:
#200	conn.close()
#201
#202	return 0
#203
#204
#205	if __name__ == '__main__':
#206	exit(main())
#207

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public