mirror of
https://github.com/zenlm/claude-collector.git
synced 2026-07-26 22:30:45 +00:00
feat: full agentic data extraction
- Include ALL tool calls and metadata - Preserve agent invocations - Keep full context for agentic training - Minimal sanitization (only critical PII) - Extract 4B+ tokens with all structure Perfect for training coding agents that use tools.
This commit is contained in:
Hanzo Dev
Binary file not shown.
Binary file not shown.
+121
-199
@@ -1,7 +1,7 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Claude Collector CLI
|
||||
Extract and sanitize Claude Code conversations
|
||||
Claude Collector - Full Agentic Data Extraction
|
||||
Preserves ALL tool calls, metadata, and context for agent training
|
||||
"""
|
||||
|
||||
import json
|
||||
@@ -12,246 +12,168 @@ import click
|
||||
from rich.console import Console
|
||||
from rich.progress import track
|
||||
from rich.table import Table
|
||||
from rich import print as rprint
|
||||
|
||||
console = Console()
|
||||
|
||||
def sanitize_text(text):
|
||||
"""Remove PII from text"""
|
||||
def sanitize(text):
|
||||
"""Minimal sanitization - only critical PII"""
|
||||
if not isinstance(text, str):
|
||||
return str(text)
|
||||
|
||||
replacements = [
|
||||
(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL]'),
|
||||
(r'sk-[a-zA-Z0-9]{48}', '[API_KEY]'),
|
||||
(r'xox[baprs]-[a-zA-Z0-9-]+', '[SLACK_TOKEN]'),
|
||||
(r'/Users/[^/\s]+', '/Users/[USER]'),
|
||||
(r'/home/[^/\s]+', '/home/[USER]'),
|
||||
(r'C:\\Users\\[^\\]+', 'C:\\Users\\[USER]'),
|
||||
(r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', '[IP]'),
|
||||
(r'oauth[_-]?token[_-]?secret?["\']?\s*[:=]\s*["\']?[a-f0-9:]+', 'oauth_token=[REDACTED]'),
|
||||
(r'(password|passwd|pwd)["\']?\s*[:=]\s*["\'][^"\']+["\']', r'\1=[REDACTED]'),
|
||||
]
|
||||
|
||||
for pattern, replacement in replacements:
|
||||
text = re.sub(pattern, replacement, text, flags=re.IGNORECASE)
|
||||
|
||||
# Only sanitize truly sensitive data
|
||||
text = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL]', text)
|
||||
text = re.sub(r'sk-[a-zA-Z0-9]{48}', '[API_KEY]', text)
|
||||
text = re.sub(r'/Users/([^/\s]+)', r'/Users/[USER]', text) # Keep structure
|
||||
return text
|
||||
|
||||
def extract_text_content(content):
|
||||
"""Extract text from content (string or array)"""
|
||||
if isinstance(content, str):
|
||||
return content
|
||||
if isinstance(content, list):
|
||||
texts = []
|
||||
for item in content:
|
||||
if isinstance(item, dict):
|
||||
if item.get('type') == 'text':
|
||||
texts.append(item.get('text', ''))
|
||||
elif 'text' in item:
|
||||
texts.append(item['text'])
|
||||
elif isinstance(item, str):
|
||||
texts.append(item)
|
||||
return '\n'.join(filter(None, texts))
|
||||
return str(content)
|
||||
|
||||
def process_jsonl_file(file_path):
|
||||
"""Extract conversations from a JSONL file"""
|
||||
messages = []
|
||||
total_tokens = 0
|
||||
def process_file_full(path):
|
||||
"""Extract EVERYTHING - full agentic data"""
|
||||
entries, tokens = [], 0
|
||||
|
||||
try:
|
||||
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
|
||||
for line in f:
|
||||
if not line.strip():
|
||||
for line in open(path, 'r', errors='ignore'):
|
||||
if not line.strip():
|
||||
continue
|
||||
try:
|
||||
data = json.loads(line)
|
||||
|
||||
# Keep EVERYTHING except pure metadata
|
||||
entry_type = data.get('type', '')
|
||||
|
||||
# Skip only pure metadata/snapshots
|
||||
if entry_type in ['summary', 'file-history-snapshot', 'queue-operation']:
|
||||
continue
|
||||
|
||||
try:
|
||||
data = json.loads(line)
|
||||
|
||||
# Only process actual user/assistant messages
|
||||
if data.get('type') not in ['user', 'assistant']:
|
||||
continue
|
||||
|
||||
msg = data.get('message', {})
|
||||
if not msg:
|
||||
continue
|
||||
|
||||
# Extract content
|
||||
content = extract_text_content(msg.get('content', ''))
|
||||
if not content or len(content) < 10:
|
||||
continue
|
||||
|
||||
# Get usage data (in message.usage or top-level)
|
||||
usage = msg.get('usage') or data.get('usage') or {}
|
||||
|
||||
token_count = (
|
||||
usage.get('input_tokens', 0) +
|
||||
usage.get('output_tokens', 0) +
|
||||
usage.get('cache_creation_input_tokens', 0) +
|
||||
usage.get('cache_read_input_tokens', 0)
|
||||
)
|
||||
|
||||
total_tokens += token_count
|
||||
|
||||
messages.append({
|
||||
'role': data['type'],
|
||||
'content': sanitize_text(content),
|
||||
'timestamp': data.get('timestamp'),
|
||||
'tokens': token_count,
|
||||
'usage': usage
|
||||
})
|
||||
|
||||
except (json.JSONDecodeError, KeyError, TypeError):
|
||||
continue
|
||||
|
||||
except Exception:
|
||||
# Sanitize content if present
|
||||
if 'message' in data and isinstance(data['message'], dict):
|
||||
if 'content' in data['message']:
|
||||
content = data['message']['content']
|
||||
if isinstance(content, str):
|
||||
data['message']['content'] = sanitize(content)
|
||||
elif isinstance(content, list):
|
||||
for item in content:
|
||||
if isinstance(item, dict) and 'text' in item:
|
||||
item['text'] = sanitize(item['text'])
|
||||
|
||||
# Count tokens
|
||||
usage = data.get('message', {}).get('usage', {}) or data.get('usage', {})
|
||||
if usage:
|
||||
tokens += (usage.get('input_tokens', 0) +
|
||||
usage.get('output_tokens', 0) +
|
||||
usage.get('cache_creation_input_tokens', 0) +
|
||||
usage.get('cache_read_input_tokens', 0))
|
||||
|
||||
# Keep the FULL entry with all metadata
|
||||
entries.append(data)
|
||||
|
||||
except:
|
||||
pass
|
||||
except:
|
||||
pass
|
||||
|
||||
return messages, total_tokens
|
||||
|
||||
def find_claude_directories():
|
||||
"""Find all Claude Code data directories"""
|
||||
locations = []
|
||||
home = Path.home()
|
||||
|
||||
# Check common locations
|
||||
candidates = [
|
||||
home / '.claude' / 'projects',
|
||||
home / '.config' / 'claude' / 'projects',
|
||||
]
|
||||
|
||||
for loc in candidates:
|
||||
if loc.exists() and loc.is_dir():
|
||||
locations.append(loc)
|
||||
|
||||
return locations
|
||||
return entries, tokens
|
||||
|
||||
@click.command()
|
||||
@click.option('--input', '-i', 'input_dir', type=click.Path(exists=True),
|
||||
help='Input directory (auto-detects ~/.claude/projects if not specified)')
|
||||
@click.option('--output', '-o', 'output_file', type=click.Path(),
|
||||
help='Output file (default: claude_dataset_YYYYMMDD.jsonl)')
|
||||
@click.option('--input', '-i', type=click.Path(exists=True),
|
||||
help='Input directory (default: ~/.claude/projects)')
|
||||
@click.option('--output', '-o', type=click.Path(),
|
||||
help='Output file (default: claude_full_YYYYMMDD.jsonl)')
|
||||
@click.option('--dry-run', is_flag=True, help='Show stats without saving')
|
||||
@click.option('--no-sanitize', is_flag=True, help='Skip PII sanitization (NOT RECOMMENDED)')
|
||||
@click.option('--min-tokens', type=int, default=0, help='Minimum tokens per conversation')
|
||||
def main(input_dir, output_file, dry_run, no_sanitize, min_tokens):
|
||||
@click.option('--full', is_flag=True, default=True,
|
||||
help='Include ALL data (tool calls, metadata) for agentic training')
|
||||
def main(input, output, dry_run, full):
|
||||
"""
|
||||
🤖 Claude Collector - Extract Claude Code conversations
|
||||
🤖 Claude Collector - Extract Claude Code for Agent Training
|
||||
|
||||
Extracts and sanitizes your Claude Code conversation history
|
||||
into a training dataset.
|
||||
Extracts FULL agentic data including:
|
||||
- User/assistant conversations
|
||||
- Tool calls and responses
|
||||
- Agent invocations
|
||||
- All metadata and context
|
||||
|
||||
Perfect for training coding agents!
|
||||
|
||||
\b
|
||||
Quick start:
|
||||
uvx claude-collector
|
||||
|
||||
\b
|
||||
Custom output:
|
||||
uvx claude-collector --output my-dataset.jsonl
|
||||
|
||||
\b
|
||||
Check what you have:
|
||||
uvx claude-collector --dry-run
|
||||
Usage:
|
||||
uvx claude-collector # Extract everything
|
||||
uvx claude-collector --dry-run # Just show stats
|
||||
uvx claude-collector -o my-data.jsonl # Custom output
|
||||
"""
|
||||
|
||||
console.print("\n[bold cyan]🤖 Claude Collector v0.1.0[/bold cyan]")
|
||||
console.print("[dim]Extract & sanitize Claude Code conversations[/dim]\n")
|
||||
console.print("[dim]Full Agentic Data Extraction[/dim]\n")
|
||||
|
||||
# Find Claude directories
|
||||
if input_dir:
|
||||
dirs = [Path(input_dir)]
|
||||
# Find data
|
||||
if input:
|
||||
dir = Path(input)
|
||||
else:
|
||||
dirs = find_claude_directories()
|
||||
dir = Path.home() / '.claude' / 'projects'
|
||||
if not dir.exists():
|
||||
dir = Path.home() / '.config' / 'claude' / 'projects'
|
||||
|
||||
if not dirs:
|
||||
console.print("[red]❌ No Claude Code data found![/red]")
|
||||
console.print("\nChecked:")
|
||||
console.print(" • ~/.claude/projects")
|
||||
console.print(" • ~/.config/claude/projects")
|
||||
console.print("\nSpecify manually: --input /path/to/projects")
|
||||
if not dir.exists():
|
||||
console.print(f"[red]❌ No Claude data found at {dir}[/red]")
|
||||
console.print("\nTry: --input /path/to/claude/projects")
|
||||
return
|
||||
|
||||
console.print(f"[green]✓[/green] Found Claude data: {dirs[0]}\n")
|
||||
console.print(f"[green]✓[/green] Found: {dir}")
|
||||
console.print(f"[yellow]⚙️[/yellow] Mode: FULL (includes tool calls & metadata)\n")
|
||||
|
||||
# Find all JSONL files
|
||||
all_files = []
|
||||
for directory in dirs:
|
||||
all_files.extend(directory.rglob('*.jsonl'))
|
||||
files = list(dir.rglob('*.jsonl'))
|
||||
console.print(f"[cyan]📂 Processing {len(files)} files...[/cyan]\n")
|
||||
|
||||
console.print(f"[cyan]📂 Processing {len(all_files)} files...[/cyan]\n")
|
||||
all_entries, total_tokens, file_count = [], 0, 0
|
||||
entry_types = {}
|
||||
|
||||
# Process files
|
||||
all_messages = []
|
||||
grand_total_tokens = 0
|
||||
files_with_data = 0
|
||||
|
||||
for file_path in track(all_files, description="Extracting"):
|
||||
messages, tokens = process_jsonl_file(file_path)
|
||||
if messages:
|
||||
all_messages.extend(messages)
|
||||
grand_total_tokens += tokens
|
||||
files_with_data += 1
|
||||
|
||||
# Create training examples (user + assistant pairs)
|
||||
examples = []
|
||||
i = 0
|
||||
while i < len(all_messages) - 1:
|
||||
if all_messages[i]['role'] == 'user' and all_messages[i+1]['role'] == 'assistant':
|
||||
example_tokens = all_messages[i+1]['tokens']
|
||||
for f in track(files, description="Extracting"):
|
||||
entries, toks = process_file_full(f)
|
||||
if entries:
|
||||
all_entries.extend(entries)
|
||||
total_tokens += toks
|
||||
file_count += 1
|
||||
|
||||
if example_tokens >= min_tokens:
|
||||
examples.append({
|
||||
'messages': [
|
||||
{'role': 'user', 'content': all_messages[i]['content']},
|
||||
{'role': 'assistant', 'content': all_messages[i+1]['content']}
|
||||
],
|
||||
'metadata': {
|
||||
'timestamp': all_messages[i]['timestamp'],
|
||||
'tokens': all_messages[i+1]['usage']
|
||||
}
|
||||
})
|
||||
i += 2
|
||||
else:
|
||||
i += 1
|
||||
# Track entry types
|
||||
for e in entries:
|
||||
t = e.get('type', 'unknown')
|
||||
entry_types[t] = entry_types.get(t, 0) + 1
|
||||
|
||||
# Display results
|
||||
table = Table(title="Extraction Results")
|
||||
table.add_column("Metric", style="cyan")
|
||||
table.add_column("Value", style="green")
|
||||
|
||||
table.add_row("Files scanned", f"{len(all_files):,}")
|
||||
table.add_row("Files with data", f"{files_with_data:,}")
|
||||
table.add_row("Total messages", f"{len(all_messages):,}")
|
||||
table.add_row("Training examples", f"{len(examples):,}")
|
||||
table.add_row("Total tokens", f"{grand_total_tokens:,} ({grand_total_tokens/1e9:.2f}B)")
|
||||
# Results table
|
||||
t = Table(title="Extraction Results")
|
||||
t.add_column("Metric", style="cyan")
|
||||
t.add_column("Value", style="green")
|
||||
t.add_row("Files scanned", f"{len(files):,}")
|
||||
t.add_row("Files with data", f"{file_count:,}")
|
||||
t.add_row("Total entries", f"{len(all_entries):,}")
|
||||
t.add_row("Total tokens", f"{total_tokens:,} ({total_tokens/1e9:.2f}B)")
|
||||
|
||||
console.print()
|
||||
console.print(table)
|
||||
console.print(t)
|
||||
|
||||
# Show entry types
|
||||
console.print("\n[cyan]Entry Types:[/cyan]")
|
||||
for etype, count in sorted(entry_types.items(), key=lambda x: x[1], reverse=True)[:10]:
|
||||
console.print(f" • {etype}: {count:,}")
|
||||
console.print()
|
||||
|
||||
if dry_run:
|
||||
console.print("[yellow]🔍 DRY RUN - No files saved[/yellow]")
|
||||
console.print("[yellow]🔍 DRY RUN - No files saved[/yellow]\n")
|
||||
return
|
||||
|
||||
# Save dataset
|
||||
if not output_file:
|
||||
output_file = f"claude_dataset_{datetime.now().strftime('%Y%m%d_%H%M')}.jsonl"
|
||||
# Save FULL dataset
|
||||
out = Path(output) if output else Path(f"claude_full_{datetime.now().strftime('%Y%m%d_%H%M')}.jsonl")
|
||||
out.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
output_path = Path(output_file)
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(out, 'w') as f:
|
||||
for entry in all_entries:
|
||||
f.write(json.dumps(entry) + '\n')
|
||||
|
||||
with open(output_path, 'w') as f:
|
||||
for example in examples:
|
||||
f.write(json.dumps(example) + '\n')
|
||||
size = out.stat().st_size / 1024 / 1024
|
||||
|
||||
size_mb = output_path.stat().st_size / 1024 / 1024
|
||||
|
||||
console.print(f"[green]✅ Dataset saved![/green]")
|
||||
console.print(f" File: [cyan]{output_path.name}[/cyan]")
|
||||
console.print(f" Size: [cyan]{size_mb:.2f} MB[/cyan]")
|
||||
console.print(f" Examples: [cyan]{len(examples):,}[/cyan]")
|
||||
console.print(f"\n[bold green]🎉 Ready for training![/bold green]\n")
|
||||
console.print(f"[green]✅ Full dataset saved![/green]")
|
||||
console.print(f" File: [cyan]{out.name}[/cyan]")
|
||||
console.print(f" Size: [cyan]{size:.2f} MB[/cyan]")
|
||||
console.print(f" Entries: [cyan]{len(all_entries):,}[/cyan]")
|
||||
console.print(f" Tokens: [cyan]{total_tokens:,} ({total_tokens/1e9:.2f}B)[/cyan]")
|
||||
console.print(f"\n[bold green]🎉 Ready for agentic training![/bold green]\n")
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
||||
Reference in New Issue
Block a user