Compare commits

...

3 Commits

Author SHA1 Message Date
semantic-release 493e0902b6 chore(release): 0.4.0 [skip ci] 2025-02-16 19:49:15 +00:00
Paul Hernandez 56f47d6812 feat: import chatgpt conversation data (#9)
Co-authored-by: phernandez <phernandez@basicmachines.co>
2025-02-16 13:47:16 -06:00
Paul Hernandez a15c346d5e feat: Import claude.ai data (#8)
Import Claude.ai conversation and project data to basic-memory Markdown
format.

---------

Co-authored-by: phernandez <phernandez@basicmachines.co>
2025-02-16 12:47:41 -06:00
15 changed files with 1423 additions and 17 deletions
+19
View File
@@ -1,6 +1,25 @@
# CHANGELOG
## v0.4.0 (2025-02-16)
### Features
- Import chatgpt conversation data ([#9](https://github.com/basicmachines-co/basic-memory/pull/9),
[`56f47d6`](https://github.com/basicmachines-co/basic-memory/commit/56f47d6812982437f207629e6ac9a82e0e56514e))
Co-authored-by: phernandez <phernandez@basicmachines.co>
- Import claude.ai data ([#8](https://github.com/basicmachines-co/basic-memory/pull/8),
[`a15c346`](https://github.com/basicmachines-co/basic-memory/commit/a15c346d5ebd44344b76bad877bb4d1073fcbc3b))
Import Claude.ai conversation and project data to basic-memory Markdown format.
---------
Co-authored-by: phernandez <phernandez@basicmachines.co>
## v0.3.0 (2025-02-15)
### Bug Fixes
+86
View File
@@ -253,6 +253,92 @@ Basic Memory is built on some key ideas:
- Simple text patterns can capture rich meaning
- Local-first doesn't mean feature-poor
## Importing data
Basic memory has cli commands to import data from several formats into Markdown files
### Claude.ai
First, request an export of your data from your Claude account. The data will be emailed to you in several files,
including
`conversations.json` and `projects.json`.
Import Claude.ai conversation data
```bash
basic-memory import claude conversations conversations.json
```
The conversations will be turned into Markdown files and placed in the "conversations" folder by default (this can be
changed with the --folder arg).
Example:
```bash
Importing chats from conversations.json...writing to .../basic-memory
Reading chat data... ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100%
╭────────────────────────────╮
│ Import complete! │
│ │
│ Imported 307 conversations │
│ Containing 7769 messages │
╰────────────────────────────╯
```
Next, you can run the `sync` command to import the data into basic-memory
```bash
basic-memory sync
```
You can also import project data from Claude.ai
```bash
➜ basic-memory import claude projects
Importing projects from projects.json...writing to .../basic-memory/projects
Reading project data... ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100%
╭────────────────────────────────╮
│ Import complete! │
│ │
│ Imported 101 project documents │
│ Imported 32 prompt templates │
╰────────────────────────────────╯
Run 'basic-memory sync' to index the new files.
```
### Chat Gpt
```bash
➜ basic-memory import chatgpt
Importing chats from conversations.json...writing to .../basic-memory/conversations
Reading chat data... ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100%
╭────────────────────────────╮
│ Import complete! │
│ │
│ Imported 198 conversations │
│ Containing 11777 messages │
╰────────────────────────────╯
```
### Memory json
```bash
➜ basic-memory import memory-json
Importing from memory.json...writing to .../basic-memory
Reading memory.json... ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100%
Creating entities... ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100%
╭──────────────────────╮
│ Import complete! │
│ │
│ Created 126 entities │
│ Added 252 relations │
╰──────────────────────╯
```
## License
AGPL-3.0
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "basic-memory"
version = "0.3.0"
version = "0.4.0"
description = "Local-first knowledge management combining Zettelkasten with knowledge graphs"
readme = "README.md"
requires-python = ">=3.12.1"
+1 -1
View File
@@ -1,3 +1,3 @@
"""basic-memory - Local-first knowledge management combining Zettelkasten with knowledge graphs"""
__version__ = "0.3.0"
__version__ = "0.4.0"
+8 -1
View File
@@ -10,4 +10,11 @@ setup_logging(log_file=".basic-memory/basic-memory-cli.log") # pragma: no cover
asyncio.run(db.run_migrations(config))
app = typer.Typer()
app = typer.Typer(name="basic-memory")
import_app = typer.Typer()
app.add_typer(import_app, name="import")
claude_app = typer.Typer()
import_app.add_typer(claude_app, name="claude")
@@ -0,0 +1,255 @@
"""Import command for ChatGPT conversations."""
import asyncio
import json
from datetime import datetime
from pathlib import Path
from typing import Dict, Any, List, Annotated, Set, Optional
import typer
from loguru import logger
from rich.console import Console
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn
from basic_memory.cli.app import import_app
from basic_memory.config import config
from basic_memory.markdown import EntityParser, MarkdownProcessor
from basic_memory.markdown.schemas import EntityMarkdown, EntityFrontmatter
console = Console()
def clean_filename(text: str) -> str:
"""Convert text to safe filename."""
clean = "".join(c if c.isalnum() else "-" for c in text.lower()).strip("-")
return clean
def format_timestamp(ts: float) -> str:
"""Format Unix timestamp for display."""
dt = datetime.fromtimestamp(ts)
return dt.strftime("%Y-%m-%d %H:%M:%S")
def get_message_content(message: Dict[str, Any]) -> str:
"""Extract clean message content."""
if not message or "content" not in message:
return "" # pragma: no cover
content = message["content"]
if content.get("content_type") == "text":
return "\n".join(content.get("parts", []))
elif content.get("content_type") == "code":
return f"```{content.get('language', '')}\n{content.get('text', '')}\n```"
return "" # pragma: no cover
def traverse_messages(
mapping: Dict[str, Any], root_id: Optional[str], seen: Set[str]
) -> List[Dict[str, Any]]:
"""Traverse message tree and return messages in order."""
messages = []
node = mapping.get(root_id) if root_id else None
while node:
if node["id"] not in seen and node.get("message"):
seen.add(node["id"])
messages.append(node["message"])
# Follow children
children = node.get("children", [])
for child_id in children:
child_msgs = traverse_messages(mapping, child_id, seen)
messages.extend(child_msgs)
break # Don't follow siblings
return messages
def format_chat_markdown(
title: str, mapping: Dict[str, Any], root_id: Optional[str], created_at: float, modified_at: float
) -> str:
"""Format chat as clean markdown."""
# Start with title
lines = [f"# {title}\n"]
# Traverse message tree
seen_msgs = set()
messages = traverse_messages(mapping, root_id, seen_msgs)
# Format each message
for msg in messages:
# Skip hidden messages
if msg.get("metadata", {}).get("is_visually_hidden_from_conversation"):
continue
# Get author and timestamp
author = msg["author"]["role"].title()
ts = format_timestamp(msg["create_time"]) if msg.get("create_time") else ""
# Add message header
lines.append(f"### {author} ({ts})")
# Add message content
content = get_message_content(msg)
if content:
lines.append(content)
# Add spacing
lines.append("")
return "\n".join(lines)
def format_chat_content(folder: str, conversation: Dict[str, Any]) -> EntityMarkdown:
"""Convert chat conversation to Basic Memory entity."""
# Extract timestamps
created_at = conversation["create_time"]
modified_at = conversation["update_time"]
root_id = None
# Find root message
for node_id, node in conversation["mapping"].items():
if node.get("parent") is None:
root_id = node_id
break
# Generate permalink
date_prefix = datetime.fromtimestamp(created_at).strftime("%Y%m%d")
clean_title = clean_filename(conversation["title"])
# Format content
content = format_chat_markdown(
title=conversation["title"],
mapping=conversation["mapping"],
root_id=root_id,
created_at=created_at,
modified_at=modified_at,
)
# Create entity
entity = EntityMarkdown(
frontmatter=EntityFrontmatter(
metadata={
"type": "conversation",
"title": conversation["title"],
"created": format_timestamp(created_at),
"modified": format_timestamp(modified_at),
"permalink": f"{folder}/{date_prefix}-{clean_title}",
}
),
content=content,
)
return entity
async def process_chatgpt_json(
json_path: Path, folder: str, markdown_processor: MarkdownProcessor
) -> Dict[str, int]:
"""Import conversations from ChatGPT JSON format."""
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
BarColumn(),
TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
console=console,
) as progress:
read_task = progress.add_task("Reading chat data...", total=None)
# Read conversations
conversations = json.loads(json_path.read_text())
progress.update(read_task, total=len(conversations))
# Process each conversation
messages_imported = 0
chats_imported = 0
for chat in conversations:
# Convert to entity
entity = format_chat_content(folder, chat)
# Write file
file_path = config.home / f"{entity.frontmatter.metadata['permalink']}.md"
# logger.info(f"Writing file: {file_path.absolute()}")
await markdown_processor.write_file(file_path, entity)
# Count messages
msg_count = sum(
1
for node in chat["mapping"].values()
if node.get("message")
and not node.get("message", {})
.get("metadata", {})
.get("is_visually_hidden_from_conversation")
)
chats_imported += 1
messages_imported += msg_count
progress.update(read_task, advance=1)
return {"conversations": chats_imported, "messages": messages_imported}
async def get_markdown_processor() -> MarkdownProcessor:
"""Get MarkdownProcessor instance."""
entity_parser = EntityParser(config.home)
return MarkdownProcessor(entity_parser)
@import_app.command(name="chatgpt", help="Import conversations from ChatGPT JSON export.")
def import_chatgpt(
conversations_json: Annotated[
Path, typer.Option(..., help="Path to ChatGPT conversations.json file")
] = Path("conversations.json"),
folder: Annotated[
str, typer.Option(help="The folder to place the files in.")
] = "conversations",
):
"""Import chat conversations from ChatGPT JSON format.
This command will:
1. Read the complex tree structure of messages
2. Convert them to linear markdown conversations
3. Save as clean, readable markdown files
After importing, run 'basic-memory sync' to index the new files.
"""
try:
if conversations_json:
if not conversations_json.exists():
typer.echo(f"Error: File not found: {conversations_json}", err=True)
raise typer.Exit(1)
# Get markdown processor
markdown_processor = asyncio.run(get_markdown_processor())
# Process the file
base_path = config.home / folder
console.print(f"\nImporting chats from {conversations_json}...writing to {base_path}")
results = asyncio.run(
process_chatgpt_json(conversations_json, folder, markdown_processor)
)
# Show results
console.print(
Panel(
f"[green]Import complete![/green]\n\n"
f"Imported {results['conversations']} conversations\n"
f"Containing {results['messages']} messages",
expand=False,
)
)
console.print("\nRun 'basic-memory sync' to index the new files.")
except Exception as e:
logger.error("Import failed")
typer.echo(f"Error during import: {e}", err=True)
raise typer.Exit(1)
@@ -0,0 +1,211 @@
"""Import command for basic-memory CLI to import chat data from conversations2.json format."""
import asyncio
import json
from datetime import datetime
from pathlib import Path
from typing import Dict, Any, List, Annotated
import typer
from loguru import logger
from rich.console import Console
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn
from basic_memory.cli.app import claude_app
from basic_memory.config import config
from basic_memory.markdown import EntityParser, MarkdownProcessor
from basic_memory.markdown.schemas import EntityMarkdown, EntityFrontmatter
console = Console()
def clean_filename(text: str) -> str:
"""Convert text to safe filename."""
# Remove invalid characters and convert spaces
clean = "".join(c if c.isalnum() else "-" for c in text.lower()).strip("-")
return clean
def format_timestamp(ts: str) -> str:
"""Format ISO timestamp for display."""
dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
return dt.strftime("%Y-%m-%d %H:%M:%S")
def format_chat_markdown(
name: str, messages: List[Dict[str, Any]], created_at: str, modified_at: str, permalink: str
) -> str:
"""Format chat as clean markdown."""
# Start with frontmatter and title
lines = [
f"# {name}\n",
]
# Add messages
for msg in messages:
# Format timestamp
ts = format_timestamp(msg["created_at"])
# Add message header
lines.append(f"### {msg['sender'].title()} ({ts})")
# Handle message content
content = msg.get("text", "")
if msg.get("content"):
content = " ".join(c.get("text", "") for c in msg["content"])
lines.append(content)
# Handle attachments
attachments = msg.get("attachments", [])
for attachment in attachments:
if "file_name" in attachment:
lines.append(f"\n**Attachment: {attachment['file_name']}**")
if "extracted_content" in attachment:
lines.append("```")
lines.append(attachment["extracted_content"])
lines.append("```")
# Add spacing between messages
lines.append("")
return "\n".join(lines)
def format_chat_content(
base_path: Path, name: str, messages: List[Dict[str, Any]], created_at: str, modified_at: str
) -> EntityMarkdown:
"""Convert chat messages to Basic Memory entity format."""
# Generate permalink
date_prefix = datetime.fromisoformat(created_at.replace("Z", "+00:00")).strftime("%Y%m%d")
clean_title = clean_filename(name)
permalink = f"{base_path}/{date_prefix}-{clean_title}"
# Format content
content = format_chat_markdown(
name=name,
messages=messages,
created_at=created_at,
modified_at=modified_at,
permalink=permalink,
)
# Create entity
entity = EntityMarkdown(
frontmatter=EntityFrontmatter(
metadata={
"type": "conversation",
"title": name,
"created": created_at,
"modified": modified_at,
"permalink": permalink,
}
),
content=content,
)
return entity
async def process_conversations_json(
json_path: Path, base_path: Path, markdown_processor: MarkdownProcessor
) -> Dict[str, int]:
"""Import chat data from conversations2.json format."""
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
BarColumn(),
TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
console=console,
) as progress:
read_task = progress.add_task("Reading chat data...", total=None)
# Read chat data - handle array of arrays format
data = json.loads(json_path.read_text())
conversations = [chat for chat in data]
progress.update(read_task, total=len(conversations))
# Process each conversation
messages_imported = 0
chats_imported = 0
for chat in conversations:
# Convert to entity
entity = format_chat_content(
base_path=base_path,
name=chat["name"],
messages=chat["chat_messages"],
created_at=chat["created_at"],
modified_at=chat["updated_at"],
)
# Write file
file_path = Path(f"{entity.frontmatter.metadata['permalink']}.md")
await markdown_processor.write_file(file_path, entity)
chats_imported += 1
messages_imported += len(chat["chat_messages"])
progress.update(read_task, advance=1)
return {"conversations": chats_imported, "messages": messages_imported}
async def get_markdown_processor() -> MarkdownProcessor:
"""Get MarkdownProcessor instance."""
entity_parser = EntityParser(config.home)
return MarkdownProcessor(entity_parser)
@claude_app.command(name="conversations", help="Import chat conversations from Claude.ai.")
def import_claude(
conversations_json: Annotated[
Path, typer.Argument(..., help="Path to conversations.json file")
] = Path("conversations.json"),
folder: Annotated[
str, typer.Option(help="The folder to place the files in.")
] = "conversations",
):
"""Import chat conversations from conversations2.json format.
This command will:
1. Read chat data and nested messages
2. Create markdown files for each conversation
3. Format content in clean, readable markdown
After importing, run 'basic-memory sync' to index the new files.
"""
try:
if not conversations_json.exists():
typer.echo(f"Error: File not found: {conversations_json}", err=True)
raise typer.Exit(1)
# Get markdown processor
markdown_processor = asyncio.run(get_markdown_processor())
# Process the file
base_path = config.home / folder
console.print(f"\nImporting chats from {conversations_json}...writing to {base_path}")
results = asyncio.run(
process_conversations_json(conversations_json, base_path, markdown_processor)
)
# Show results
console.print(
Panel(
f"[green]Import complete![/green]\n\n"
f"Imported {results['conversations']} conversations\n"
f"Containing {results['messages']} messages",
expand=False,
)
)
console.print("\nRun 'basic-memory sync' to index the new files.")
except Exception as e:
logger.error("Import failed")
typer.echo(f"Error during import: {e}", err=True)
raise typer.Exit(1)
@@ -0,0 +1,195 @@
"""Import command for basic-memory CLI to import project data from Claude.ai."""
import asyncio
import json
from pathlib import Path
from typing import Dict, Any, Annotated, Optional
import typer
from loguru import logger
from rich.console import Console
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn
from basic_memory.cli.app import claude_app
from basic_memory.config import config
from basic_memory.markdown import EntityParser, MarkdownProcessor
from basic_memory.markdown.schemas import EntityMarkdown, EntityFrontmatter
console = Console()
def clean_filename(text: str) -> str:
"""Convert text to safe filename."""
clean = "".join(c if c.isalnum() else "-" for c in text.lower()).strip("-")
return clean
def format_project_markdown(project: Dict[str, Any], doc: Dict[str, Any]) -> EntityMarkdown:
"""Format a project document as a Basic Memory entity."""
# Extract timestamps
created_at = doc.get("created_at") or project["created_at"]
modified_at = project["updated_at"]
# Generate clean names for organization
project_dir = clean_filename(project["name"])
doc_file = clean_filename(doc["filename"])
# Create entity
entity = EntityMarkdown(
frontmatter=EntityFrontmatter(
metadata={
"type": "project_doc",
"title": doc["filename"],
"created": created_at,
"modified": modified_at,
"permalink": f"{project_dir}/docs/{doc_file}",
"project_name": project["name"],
"project_uuid": project["uuid"],
"doc_uuid": doc["uuid"],
}
),
content=doc["content"],
)
return entity
def format_prompt_markdown(project: Dict[str, Any]) -> Optional[EntityMarkdown]:
"""Format project prompt template as a Basic Memory entity."""
if not project.get("prompt_template"):
return None
# Extract timestamps
created_at = project["created_at"]
modified_at = project["updated_at"]
# Generate clean project directory name
project_dir = clean_filename(project["name"])
# Create entity
entity = EntityMarkdown(
frontmatter=EntityFrontmatter(
metadata={
"type": "prompt_template",
"title": f"Prompt Template: {project['name']}",
"created": created_at,
"modified": modified_at,
"permalink": f"{project_dir}/prompt-template",
"project_name": project["name"],
"project_uuid": project["uuid"],
}
),
content=f"# Prompt Template: {project['name']}\n\n{project['prompt_template']}",
)
return entity
async def process_projects_json(
json_path: Path, base_path: Path, markdown_processor: MarkdownProcessor
) -> Dict[str, int]:
"""Import project data from Claude.ai projects.json format."""
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
BarColumn(),
TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
console=console,
) as progress:
read_task = progress.add_task("Reading project data...", total=None)
# Read project data
data = json.loads(json_path.read_text())
progress.update(read_task, total=len(data))
# Track import counts
docs_imported = 0
prompts_imported = 0
# Process each project
for project in data:
project_dir = clean_filename(project["name"])
# Create project directories
docs_dir = base_path / project_dir / "docs"
docs_dir.mkdir(parents=True, exist_ok=True)
# Import prompt template if it exists
if prompt_entity := format_prompt_markdown(project):
file_path = base_path / f"{prompt_entity.frontmatter.metadata['permalink']}.md"
await markdown_processor.write_file(file_path, prompt_entity)
prompts_imported += 1
# Import project documents
for doc in project.get("docs", []):
entity = format_project_markdown(project, doc)
file_path = base_path / f"{entity.frontmatter.metadata['permalink']}.md"
await markdown_processor.write_file(file_path, entity)
docs_imported += 1
progress.update(read_task, advance=1)
return {"documents": docs_imported, "prompts": prompts_imported}
async def get_markdown_processor() -> MarkdownProcessor:
"""Get MarkdownProcessor instance."""
entity_parser = EntityParser(config.home)
return MarkdownProcessor(entity_parser)
@claude_app.command(name="projects", help="Import projects from Claude.ai.")
def import_projects(
projects_json: Annotated[Path, typer.Argument(..., help="Path to projects.json file")] = Path(
"projects.json"
),
base_folder: Annotated[
str, typer.Option(help="The base folder to place project files in.")
] = "projects",
):
"""Import project data from Claude.ai.
This command will:
1. Create a directory for each project
2. Store docs in a docs/ subdirectory
3. Place prompt template in project root
After importing, run 'basic-memory sync' to index the new files.
"""
try:
if projects_json:
if not projects_json.exists():
typer.echo(f"Error: File not found: {projects_json}", err=True)
raise typer.Exit(1)
# Get markdown processor
markdown_processor = asyncio.run(get_markdown_processor())
# Process the file
base_path = config.home / base_folder if base_folder else config.home
console.print(f"\nImporting projects from {projects_json}...writing to {base_path}")
results = asyncio.run(
process_projects_json(projects_json, base_path, markdown_processor)
)
# Show results
console.print(
Panel(
f"[green]Import complete![/green]\n\n"
f"Imported {results['documents']} project documents\n"
f"Imported {results['prompts']} prompt templates",
expand=False,
)
)
console.print("\nRun 'basic-memory sync' to index the new files.")
except Exception as e:
logger.error("Import failed")
typer.echo(f"Error during import: {e}", err=True)
raise typer.Exit(1)
@@ -3,7 +3,7 @@
import asyncio
import json
from pathlib import Path
from typing import Dict, Any, List
from typing import Dict, Any, List, Annotated
import typer
from loguru import logger
@@ -11,12 +11,11 @@ from rich.console import Console
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn
from basic_memory.cli.app import app
from basic_memory.cli.app import import_app
from basic_memory.config import config
from basic_memory.markdown import EntityParser, MarkdownProcessor
from basic_memory.markdown.schemas import EntityMarkdown, EntityFrontmatter, Observation, Relation
console = Console()
@@ -98,9 +97,11 @@ async def get_markdown_processor() -> MarkdownProcessor:
return MarkdownProcessor(entity_parser)
@app.command()
def import_json(
json_path: Path = typer.Argument(..., help="Path to memory.json file to import"),
@import_app.command()
def memory_json(
json_path: Annotated[Path, typer.Argument(..., help="Path to memory.json file")] = Path(
"memory.json"
),
):
"""Import entities and relations from a memory.json file.
@@ -138,6 +139,6 @@ def import_json(
console.print("\nRun 'basic-memory sync' to index the new files.")
except Exception as e:
logger.exception("Import failed")
logger.error("Import failed")
typer.echo(f"Error during import: {e}", err=True)
raise typer.Exit(1)
+10 -1
View File
@@ -4,7 +4,16 @@ from basic_memory.cli.app import app # pragma: no cover
from basic_memory.utils import setup_logging # pragma: no cover
# Register commands
from basic_memory.cli.commands import status, sync, db, import_memory_json, mcp # noqa: F401 # pragma: no cover
from basic_memory.cli.commands import ( # noqa: F401 # pragma: no cover
status,
sync,
db,
import_memory_json,
mcp,
import_claude_conversations,
import_claude_projects,
import_chatgpt,
)
# Set up logging when module is imported
+276
View File
@@ -0,0 +1,276 @@
"""Tests for import_chatgpt command."""
import json
import pytest
from typer.testing import CliRunner
from basic_memory.cli.app import import_app, app
from basic_memory.cli.commands import import_chatgpt
from basic_memory.config import config
from basic_memory.markdown import EntityParser, MarkdownProcessor
# Set up CLI runner
runner = CliRunner()
@pytest.fixture
def sample_conversation():
"""Sample ChatGPT conversation data for testing."""
return {
"title": "Test Conversation",
"create_time": 1736616594.24054, # Example timestamp
"update_time": 1736616603.164995,
"mapping": {
"root": {"id": "root", "message": None, "parent": None, "children": ["msg1"]},
"msg1": {
"id": "msg1",
"message": {
"id": "msg1",
"author": {"role": "user", "name": None, "metadata": {}},
"create_time": 1736616594.24054,
"content": {"content_type": "text", "parts": ["Hello, this is a test message"]},
"status": "finished_successfully",
"metadata": {},
},
"parent": "root",
"children": ["msg2"],
},
"msg2": {
"id": "msg2",
"message": {
"id": "msg2",
"author": {"role": "assistant", "name": None, "metadata": {}},
"create_time": 1736616603.164995,
"content": {"content_type": "text", "parts": ["This is a test response"]},
"status": "finished_successfully",
"metadata": {},
},
"parent": "msg1",
"children": [],
},
},
}
@pytest.fixture
def sample_conversation_with_code():
"""Sample conversation with code block."""
conversation = {
"title": "Code Test",
"create_time": 1736616594.24054,
"update_time": 1736616603.164995,
"mapping": {
"root": {"id": "root", "message": None, "parent": None, "children": ["msg1"]},
"msg1": {
"id": "msg1",
"message": {
"id": "msg1",
"author": {"role": "assistant", "name": None, "metadata": {}},
"create_time": 1736616594.24054,
"content": {
"content_type": "code",
"language": "python",
"text": "def hello():\n print('Hello world!')",
},
"status": "finished_successfully",
"metadata": {},
},
"parent": "root",
"children": [],
},
"msg2": {
"id": "msg2",
"message": {
"id": "msg2",
"author": {"role": "assistant", "name": None, "metadata": {}},
"create_time": 1736616594.24054,
"status": "finished_successfully",
"metadata": {},
},
"parent": "root",
"children": [],
},
},
}
return conversation
@pytest.fixture
def sample_conversation_with_hidden():
"""Sample conversation with hidden messages."""
conversation = {
"title": "Hidden Test",
"create_time": 1736616594.24054,
"update_time": 1736616603.164995,
"mapping": {
"root": {
"id": "root",
"message": None,
"parent": None,
"children": ["visible", "hidden"],
},
"visible": {
"id": "visible",
"message": {
"id": "visible",
"author": {"role": "user", "name": None, "metadata": {}},
"create_time": 1736616594.24054,
"content": {"content_type": "text", "parts": ["Visible message"]},
"status": "finished_successfully",
"metadata": {},
},
"parent": "root",
"children": [],
},
"hidden": {
"id": "hidden",
"message": {
"id": "hidden",
"author": {"role": "system", "name": None, "metadata": {}},
"create_time": 1736616594.24054,
"content": {"content_type": "text", "parts": ["Hidden message"]},
"status": "finished_successfully",
"metadata": {"is_visually_hidden_from_conversation": True},
},
"parent": "root",
"children": [],
},
},
}
return conversation
@pytest.fixture
def sample_chatgpt_json(tmp_path, sample_conversation):
"""Create a sample ChatGPT JSON file."""
json_file = tmp_path / "conversations.json"
with open(json_file, "w") as f:
json.dump([sample_conversation], f)
return json_file
@pytest.mark.asyncio
async def test_process_chatgpt_json(tmp_path, sample_chatgpt_json):
"""Test importing conversations from JSON."""
entity_parser = EntityParser(tmp_path)
processor = MarkdownProcessor(entity_parser)
results = await import_chatgpt.process_chatgpt_json(sample_chatgpt_json, tmp_path, processor)
assert results["conversations"] == 1
assert results["messages"] == 2
# Check conversation file exists
conv_path = tmp_path / "20250111-test-conversation.md"
assert conv_path.exists()
# Check content formatting
content = conv_path.read_text()
assert "# Test Conversation" in content
assert "### User" in content
assert "Hello, this is a test message" in content
assert "### Assistant" in content
assert "This is a test response" in content
@pytest.mark.asyncio
async def test_process_code_blocks(tmp_path, sample_conversation_with_code):
"""Test handling of code blocks."""
entity_parser = EntityParser(tmp_path)
processor = MarkdownProcessor(entity_parser)
# Create test file
json_file = tmp_path / "code_test.json"
with open(json_file, "w") as f:
json.dump([sample_conversation_with_code], f)
await import_chatgpt.process_chatgpt_json(json_file, tmp_path, processor)
# Check content
conv_path = tmp_path / "20250111-code-test.md"
content = conv_path.read_text()
assert "```python" in content
assert "def hello():" in content
assert "```" in content
@pytest.mark.asyncio
async def test_hidden_messages(tmp_path, sample_conversation_with_hidden):
"""Test handling of hidden messages."""
entity_parser = EntityParser(tmp_path)
processor = MarkdownProcessor(entity_parser)
# Create test file
json_file = tmp_path / "hidden_test.json"
with open(json_file, "w") as f:
json.dump([sample_conversation_with_hidden], f)
results = await import_chatgpt.process_chatgpt_json(json_file, tmp_path, processor)
# Should only count visible messages
assert results["messages"] == 1
# Check content
conv_path = tmp_path / "20250111-hidden-test.md"
content = conv_path.read_text()
assert "Visible message" in content
assert "Hidden message" not in content
def test_import_chatgpt_command_file_not_found(tmp_path):
"""Test error handling for nonexistent file."""
nonexistent = tmp_path / "nonexistent.json"
result = runner.invoke(app, ["import", "chatgpt", "--conversations-json", str(nonexistent)])
assert result.exit_code == 1
assert "File not found" in result.output
def test_import_chatgpt_command_success(tmp_path, sample_chatgpt_json, monkeypatch):
"""Test successful conversation import via command."""
# Set up test environment
monkeypatch.setenv("HOME", str(tmp_path))
# Run import
result = runner.invoke(
import_app, ["chatgpt", "--conversations-json", str(sample_chatgpt_json)]
)
assert result.exit_code == 0
assert "Import complete" in result.output
assert "Imported 1 conversations" in result.output
assert "Containing 2 messages" in result.output
def test_import_chatgpt_command_invalid_json(tmp_path):
"""Test error handling for invalid JSON."""
# Create invalid JSON file
invalid_file = tmp_path / "invalid.json"
invalid_file.write_text("not json")
result = runner.invoke(import_app, ["chatgpt", "--conversations-json", str(invalid_file)])
assert result.exit_code == 1
assert "Error during import" in result.output
def test_import_chatgpt_with_custom_folder(tmp_path, sample_chatgpt_json, monkeypatch):
"""Test import with custom conversations folder."""
# Set up test environment
config.home = tmp_path
conversations_folder = "chats"
# Run import
result = runner.invoke(
app,
[
"import",
"chatgpt",
"--conversations-json",
str(sample_chatgpt_json),
"--folder",
conversations_folder,
],
)
assert result.exit_code == 0
# Check files in custom folder
conv_path = tmp_path / conversations_folder / "20250111-test-conversation.md"
assert conv_path.exists()
@@ -0,0 +1,174 @@
"""Tests for import_claude command (chat conversations)."""
import json
import pytest
from typer.testing import CliRunner
from basic_memory.cli.app import app
from basic_memory.cli.commands import import_claude_conversations as import_claude
from basic_memory.config import config
from basic_memory.markdown import EntityParser, MarkdownProcessor
# Set up CLI runner
runner = CliRunner()
@pytest.fixture
def sample_conversation():
"""Sample conversation data for testing."""
return {
"uuid": "test-uuid",
"name": "Test Conversation",
"created_at": "2025-01-05T20:55:32.499880+00:00",
"updated_at": "2025-01-05T20:56:39.477600+00:00",
"chat_messages": [
{
"uuid": "msg-1",
"text": "Hello, this is a test",
"sender": "human",
"created_at": "2025-01-05T20:55:32.499880+00:00",
"content": [{"type": "text", "text": "Hello, this is a test"}],
},
{
"uuid": "msg-2",
"text": "Response to test",
"sender": "assistant",
"created_at": "2025-01-05T20:55:40.123456+00:00",
"content": [{"type": "text", "text": "Response to test"}],
},
],
}
@pytest.fixture
def sample_conversations_json(tmp_path, sample_conversation):
"""Create a sample conversations.json file."""
json_file = tmp_path / "conversations.json"
with open(json_file, "w") as f:
json.dump([sample_conversation], f)
return json_file
@pytest.mark.asyncio
async def test_process_chat_json(tmp_path, sample_conversations_json):
"""Test importing conversations from JSON."""
entity_parser = EntityParser(tmp_path)
processor = MarkdownProcessor(entity_parser)
results = await import_claude.process_conversations_json(
sample_conversations_json, tmp_path, processor
)
assert results["conversations"] == 1
assert results["messages"] == 2
# Check conversation file
conv_path = tmp_path / "20250105-test-conversation.md"
assert conv_path.exists()
content = conv_path.read_text()
# Check content formatting
assert "### Human" in content
assert "Hello, this is a test" in content
assert "### Assistant" in content
assert "Response to test" in content
def test_import_conversations_command_file_not_found(tmp_path):
"""Test error handling for nonexistent file."""
nonexistent = tmp_path / "nonexistent.json"
result = runner.invoke(app, ["import", "claude", "conversations", str(nonexistent)])
assert result.exit_code == 1
assert "File not found" in result.output
def test_import_conversations_command_success(tmp_path, sample_conversations_json, monkeypatch):
"""Test successful conversation import via command."""
# Set up test environment
monkeypatch.setenv("HOME", str(tmp_path))
# Run import
result = runner.invoke(
app, ["import", "claude", "conversations", str(sample_conversations_json)]
)
assert result.exit_code == 0
assert "Import complete" in result.output
assert "Imported 1 conversations" in result.output
assert "Containing 2 messages" in result.output
def test_import_conversations_command_invalid_json(tmp_path):
"""Test error handling for invalid JSON."""
# Create invalid JSON file
invalid_file = tmp_path / "invalid.json"
invalid_file.write_text("not json")
result = runner.invoke(app, ["import", "claude", "conversations", str(invalid_file)])
assert result.exit_code == 1
assert "Error during import" in result.output
def test_import_conversations_with_custom_folder(tmp_path, sample_conversations_json, monkeypatch):
"""Test import with custom conversations folder."""
# Set up test environment
config.home = tmp_path
conversations_folder = "chats"
# Run import
result = runner.invoke(
app,
[
"import",
"claude",
"conversations",
str(sample_conversations_json),
"--folder",
conversations_folder,
],
)
assert result.exit_code == 0
# Check files in custom folder
conv_path = tmp_path / conversations_folder / "20250105-test-conversation.md"
assert conv_path.exists()
def test_import_conversation_with_attachments(tmp_path):
"""Test importing conversation with attachments."""
# Create conversation with attachments
conversation = {
"uuid": "test-uuid",
"name": "Test With Attachments",
"created_at": "2025-01-05T20:55:32.499880+00:00",
"updated_at": "2025-01-05T20:56:39.477600+00:00",
"chat_messages": [
{
"uuid": "msg-1",
"text": "Here's a file",
"sender": "human",
"created_at": "2025-01-05T20:55:32.499880+00:00",
"content": [{"type": "text", "text": "Here's a file"}],
"attachments": [
{"file_name": "test.txt", "extracted_content": "Test file content"}
],
}
],
}
json_file = tmp_path / "with_attachments.json"
with open(json_file, "w") as f:
json.dump([conversation], f)
# Set up environment
config.home = tmp_path
# Run import
result = runner.invoke(app, ["import", "claude", "conversations", str(json_file)])
assert result.exit_code == 0
# Check attachment formatting
conv_path = tmp_path / "conversations/20250105-test-with-attachments.md"
content = conv_path.read_text()
assert "**Attachment: test.txt**" in content
assert "```" in content
assert "Test file content" in content
+173
View File
@@ -0,0 +1,173 @@
"""Tests for import_claude_projects command."""
import json
import pytest
from typer.testing import CliRunner
from basic_memory.cli.app import app
from basic_memory.cli.commands import import_claude_projects
from basic_memory.config import config
from basic_memory.markdown import EntityParser, MarkdownProcessor
# Set up CLI runner
runner = CliRunner()
@pytest.fixture
def sample_project():
"""Sample project data for testing."""
return {
"uuid": "test-uuid",
"name": "Test Project",
"created_at": "2025-01-05T20:55:32.499880+00:00",
"updated_at": "2025-01-05T20:56:39.477600+00:00",
"prompt_template": "# Test Prompt\n\nThis is a test prompt.",
"docs": [
{
"uuid": "doc-uuid-1",
"filename": "Test Document",
"content": "# Test Document\n\nThis is test content.",
"created_at": "2025-01-05T20:56:39.477600+00:00",
},
{
"uuid": "doc-uuid-2",
"filename": "Another Document",
"content": "# Another Document\n\nMore test content.",
"created_at": "2025-01-05T20:56:39.477600+00:00",
},
],
}
@pytest.fixture
def sample_projects_json(tmp_path, sample_project):
"""Create a sample projects.json file."""
json_file = tmp_path / "projects.json"
with open(json_file, "w") as f:
json.dump([sample_project], f)
return json_file
@pytest.mark.asyncio
async def test_process_projects_json(tmp_path, sample_projects_json):
"""Test importing projects from JSON."""
entity_parser = EntityParser(tmp_path)
processor = MarkdownProcessor(entity_parser)
results = await import_claude_projects.process_projects_json(
sample_projects_json, tmp_path, processor
)
assert results["documents"] == 2
assert results["prompts"] == 1
# Check project directory structure
project_dir = tmp_path / "test-project"
assert project_dir.exists()
assert (project_dir / "docs").exists()
assert (project_dir / "prompt-template.md").exists()
# Check document files
doc1 = project_dir / "docs/test-document.md"
assert doc1.exists()
content1 = doc1.read_text()
assert "# Test Document" in content1
assert "This is test content" in content1
# Check prompt template
prompt = project_dir / "prompt-template.md"
assert prompt.exists()
prompt_content = prompt.read_text()
assert "# Test Prompt" in prompt_content
assert "This is a test prompt" in prompt_content
def test_import_projects_command_file_not_found(tmp_path):
"""Test error handling for nonexistent file."""
nonexistent = tmp_path / "nonexistent.json"
result = runner.invoke(app, ["import", "claude", "projects", str(nonexistent)])
assert result.exit_code == 1
assert "File not found" in result.output
def test_import_projects_command_success(tmp_path, sample_projects_json, monkeypatch):
"""Test successful project import via command."""
# Set up test environment
config.home = tmp_path
# Run import
result = runner.invoke(app, ["import", "claude", "projects", str(sample_projects_json)])
assert result.exit_code == 0
assert "Import complete" in result.output
assert "Imported 2 project documents" in result.output
assert "Imported 1 prompt templates" in result.output
def test_import_projects_command_invalid_json(tmp_path):
"""Test error handling for invalid JSON."""
# Create invalid JSON file
invalid_file = tmp_path / "invalid.json"
invalid_file.write_text("not json")
result = runner.invoke(app, ["import", "claude", "projects", str(invalid_file)])
assert result.exit_code == 1
assert "Error during import" in result.output
def test_import_projects_with_base_folder(tmp_path, sample_projects_json, monkeypatch):
"""Test import with custom base folder."""
# Set up test environment
config.home = tmp_path
base_folder = "claude-exports"
# Run import
result = runner.invoke(
app,
[
"import",
"claude",
"projects",
str(sample_projects_json),
"--base-folder",
base_folder,
],
)
assert result.exit_code == 0
# Check files in base folder
project_dir = tmp_path / base_folder / "test-project"
assert project_dir.exists()
assert (project_dir / "docs").exists()
assert (project_dir / "prompt-template.md").exists()
def test_import_project_without_prompt(tmp_path):
"""Test importing project without prompt template."""
# Create project without prompt
project = {
"uuid": "test-uuid",
"name": "No Prompt Project",
"created_at": "2025-01-05T20:55:32.499880+00:00",
"updated_at": "2025-01-05T20:56:39.477600+00:00",
"docs": [
{
"uuid": "doc-uuid-1",
"filename": "Test Document",
"content": "# Test Document\n\nContent.",
"created_at": "2025-01-05T20:56:39.477600+00:00",
}
],
}
json_file = tmp_path / "no_prompt.json"
with open(json_file, "w") as f:
json.dump([project], f)
# Set up environment
config.home = tmp_path
# Run import
result = runner.invoke(app, ["import", "claude", "projects", str(json_file)])
assert result.exit_code == 0
assert "Imported 1 project documents" in result.output
assert "Imported 0 prompt templates" in result.output
+5 -5
View File
@@ -4,7 +4,7 @@ import json
import pytest
from typer.testing import CliRunner
from basic_memory.cli.app import app
from basic_memory.cli.app import import_app
from basic_memory.cli.commands import import_memory_json
from basic_memory.markdown import EntityParser, MarkdownProcessor
@@ -72,7 +72,7 @@ async def test_get_markdown_processor(tmp_path, monkeypatch):
def test_import_json_command_file_not_found(tmp_path):
"""Test error handling for nonexistent file."""
nonexistent = tmp_path / "nonexistent.json"
result = runner.invoke(app, ["import-json", str(nonexistent)])
result = runner.invoke(import_app, ["memory-json", str(nonexistent)])
assert result.exit_code == 1
assert "File not found" in result.output
@@ -83,7 +83,7 @@ def test_import_json_command_success(tmp_path, sample_json_file, monkeypatch):
monkeypatch.setenv("HOME", str(tmp_path))
# Run import
result = runner.invoke(app, ["import-json", str(sample_json_file)])
result = runner.invoke(import_app, ["memory-json", str(sample_json_file)])
assert result.exit_code == 0
assert "Import complete" in result.output
assert "Created 1 entities" in result.output
@@ -96,7 +96,7 @@ def test_import_json_command_invalid_json(tmp_path):
invalid_file = tmp_path / "invalid.json"
invalid_file.write_text("not json")
result = runner.invoke(app, ["import-json", str(invalid_file)])
result = runner.invoke(import_app, ["memory-json", str(invalid_file)])
assert result.exit_code == 1
assert "Error during import" in result.output
@@ -129,6 +129,6 @@ def test_import_json_command_handle_old_format(tmp_path):
monkeypatch.setenv("HOME", str(tmp_path))
# Run import
result = runner.invoke(app, ["import-json", str(json_file)])
result = runner.invoke(import_app, ["memory-json", str(json_file)])
assert result.exit_code == 0
assert "Import complete" in result.output
Generated
+1 -1
View File
@@ -70,7 +70,7 @@ wheels = [
[[package]]
name = "basic-memory"
version = "0.2.21"
version = "0.3.0"
source = { editable = "." }
dependencies = [
{ name = "aiosqlite" },