my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	SuperMemory memory provider importer.
#3
#4	SuperMemory is a cloud-hosted memory API that organizes memories
#5	with container tags (not sessions). Extraction via:
#6	- client.documents.list() — get all documents
#7	- client.search.execute() — semantic search per container
#8
#9	SuperMemory has documented migration paths from Mem0 and Zep.
#10	"""
#11
#12	import json
#13	from datetime import datetime
#14	from typing import List, Dict, Optional, Any
#15
#16	from mnemosyne.core.importers.base import BaseImporter, ImporterResult
#17
#18
#19	class SuperMemoryImporter(BaseImporter):
#20	"""Import memories from SuperMemory into Mnemosyne.
#21
#22	Usage:
#23	importer = SuperMemoryImporter(
#24	api_key="sk-xxx", # SuperMemory API key
#25	container_tag="my-app", # optional: filter by container
#26	)
#27	result = importer.run(mnemosyne_instance)
#28	"""
#29
#30	provider_name = "supermemory"
#31
#32	def __init__(self, api_key: str = None, container_tag: str = None,
#33	**kwargs):
#34	super().__init__(**kwargs)
#35	self.api_key = api_key
#36	self.container_tag = container_tag
#37
#38	def extract(self) -> List[Dict]:
#39	"""Extract memories from SuperMemory."""
#40	try:
#41	return self._extract_via_sdk()
#42	except ImportError:
#43	pass
#44	try:
#45	return self._extract_via_rest()
#46	except Exception:
#47	pass
#48	raise RuntimeError(
#49	"Could not extract from SuperMemory. Install: pip install supermemory"
#50	)
#51
#52	def _extract_via_sdk(self) -> List[Dict]:
#53	"""Extract using SuperMemory Python SDK."""
#54	from supermemory import SuperMemory
#55
#56	client = SuperMemory(api_key=self.api_key)
#57
#58	all_items = []
#59
#60	# Get documents
#61	try:
#62	docs = client.documents.list()
#63	if isinstance(docs, list):
#64	for doc in docs:
#65	content = doc.get("content", doc.get("text", ""))
#66	if content:
#67	all_items.append({
#68	"content": content,
#69	"source": "supermemory_document",
#70	"container_tag": doc.get("containerTag", ""),
#71	"is_static": doc.get("isStatic", False),
#72	"timestamp": doc.get("createdAt"),
#73	"metadata": doc.get("metadata", {}),
#74	})
#75	except Exception:
#76	pass
#77
#78	# Search for memories by container
#79	if self.container_tag:
#80	try:
#81	resp = client.search.execute(
#82	q="*",
#83	containerTags=[self.container_tag],
#84	)
#85	results = resp.get("results", resp.get("memories", []))
#86	for mem in results:
#87	content = mem.get("content", mem.get("memory", ""))
#88	if content:
#89	all_items.append({
#90	"content": content,
#91	"source": "supermemory_memory",
#92	"container_tag": mem.get("containerTag", self.container_tag),
#93	"is_static": mem.get("isStatic", False),
#94	"timestamp": mem.get("createdAt"),
#95	"metadata": mem.get("metadata", {}),
#96	})
#97	except Exception:
#98	pass
#99
#100	return all_items
#101
#102	def _extract_via_rest(self) -> List[Dict]:
#103	"""Extract using SuperMemory REST API."""
#104	import urllib.request
#105
#106	base = "https://api.supermemory.ai"
#107	headers = {
#108	"Authorization": f"Bearer {self.api_key}",
#109	"Content-Type": "application/json",
#110	}
#111
#112	all_items = []
#113
#114	# List documents
#115	try:
#116	req = urllib.request.Request(f"{base}/v4/documents", headers=headers)
#117	with urllib.request.urlopen(req, timeout=30) as resp:
#118	docs = json.loads(resp.read().decode())
#119	if isinstance(docs, list):
#120	for doc in docs:
#121	content = doc.get("content", "")
#122	if content:
#123	all_items.append({
#124	"content": content,
#125	"source": "supermemory_document",
#126	"container_tag": doc.get("containerTag", ""),
#127	"timestamp": doc.get("createdAt"),
#128	"metadata": doc.get("metadata", {}),
#129	})
#130	except Exception:
#131	pass
#132
#133	# Add memories directly
#134	try:
#135	payload = json.dumps({"q": "*"}).encode()
#136	if self.container_tag:
#137	payload = json.dumps({
#138	"q": "*",
#139	"containerTags": [self.container_tag],
#140	}).encode()
#141	req = urllib.request.Request(
#142	f"{base}/v4/search",
#143	data=payload,
#144	headers=headers,
#145	method="POST",
#146	)
#147	with urllib.request.urlopen(req, timeout=30) as resp:
#148	data = json.loads(resp.read().decode())
#149	results = data.get("results", data.get("memories", []))
#150	for mem in results:
#151	content = mem.get("content", mem.get("memory", ""))
#152	if content:
#153	all_items.append({
#154	"content": content,
#155	"source": "supermemory_memory",
#156	"container_tag": mem.get("containerTag", ""),
#157	"is_static": mem.get("isStatic", False),
#158	"timestamp": mem.get("createdAt"),
#159	"metadata": mem.get("metadata", {}),
#160	})
#161	except Exception:
#162	pass
#163
#164	return all_items
#165
#166	def transform(self, raw_data: List[Dict]) -> List[Dict]:
#167	"""Transform SuperMemory data to Mnemosyne format."""
#168	memories = []
#169	for item in raw_data:
#170	content = item.get("content", "")
#171	if not content:
#172	continue
#173
#174	container = item.get("container_tag", "")
#175	is_static = item.get("is_static", False)
#176
#177	# Static memories are more important (identity traits)
#178	importance = 0.9 if is_static else 0.5
#179
#180	meta = item.get("metadata", {}) or {}
#181	meta["_supermemory_container"] = container
#182	meta["_supermemory_static"] = is_static
#183
#184	ts = item.get("timestamp")
#185	if ts:
#186	meta["_timestamp"] = ts
#187
#188	memories.append({
#189	"content": content,
#190	"source": item.get("source", "supermemory_import"),
#191	"importance": importance,
#192	"metadata": meta,
#193	"valid_until": None,
#194	"scope": "session",
#195	"_author_id": "supermemory_import",
#196	"_author_type": "system",
#197	"_channel_id": container or self.container_tag,
#198	"_timestamp": ts,
#199	})
#200
#201	return memories
#202
#203	def run(self, mnemosyne, dry_run=False, session_id=None, channel_id=None):
#204	"""Override run to handle identity-aware import."""
#205	result = ImporterResult(provider=self.provider_name,
#206	started_at=datetime.now().isoformat())
#207	try:
#208	raw_data = self.extract()
#209	result.total = len(raw_data)
#210	if result.total == 0:
#211	result.errors.append("No memories found in SuperMemory")
#212	return result
#213	if not self.validate(raw_data):
#214	result.errors.append("Validation failed")
#215	return result
#216
#217	memories = self.transform(raw_data)
#218	if dry_run:
#219	result.imported = len(memories)
#220	return result
#221
#222	for mem_dict in memories:
#223	try:
#224	author_id = mem_dict.pop("_author_id", None)
#225	author_type = mem_dict.pop("_author_type", None)
#226	chan = mem_dict.pop("_channel_id", None) or channel_id
#227	ts = mem_dict.pop("_timestamp", None)
#228	meta = mem_dict.get("metadata", {})
#229	if ts:
#230	meta["imported_at_original"] = ts
#231
#232	mid = mnemosyne.remember(
#233	content=mem_dict["content"],
#234	source=mem_dict.get("source", self.provider_name),
#235	importance=mem_dict.get("importance", 0.5),
#236	metadata=meta,
#237	valid_until=mem_dict.get("valid_until"),
#238	scope=mem_dict.get("scope", "session"),
#239	)
#240	if author_id or author_type or chan:
#241	try:
#242	mnemosyne.beam.conn.execute("""
#243	UPDATE working_memory
#244	SET author_id = COALESCE(author_id, ?),
#245	author_type = COALESCE(author_type, ?),
#246	channel_id = COALESCE(channel_id, ?)
#247	WHERE id = ?
#248	""", (author_id, author_type, chan, mid))
#249	mnemosyne.beam.conn.commit()
#250	except Exception:
#251	pass
#252	result.memory_ids.append(mid)
#253	result.imported += 1
#254	except Exception as e:
#255	result.failed += 1
#256	result.errors.append(f"Failed: {str(e)[:100]}")
#257	except Exception as e:
#258	result.errors.append(f"SuperMemory import failed: {e}")
#259	result.finished_at = datetime.now().isoformat()
#260	return result
#261

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public