Initial commit: Add shell-memory-cli project
A CLI tool that learns from terminal command patterns to automate repetitive workflows. Features: - Command recording with tags and descriptions - Pattern detection for command sequences - Session recording and replay - Natural language script generation
This commit is contained in:
113
shell_memory/sessions.py
Normal file
113
shell_memory/sessions.py
Normal file
@@ -0,0 +1,113 @@
|
||||
"""Session recording and replay functionality."""
|
||||
|
||||
import time
|
||||
import subprocess
|
||||
from typing import List, Optional, Dict, Any
|
||||
from datetime import datetime
|
||||
|
||||
from .models import Session
|
||||
from .database import Database
|
||||
|
||||
|
||||
class SessionRecorder:
|
||||
"""Records and replays terminal sessions."""
|
||||
|
||||
def __init__(self, db: Database):
|
||||
self.db = db
|
||||
self.current_session = None
|
||||
self.recorded_commands = []
|
||||
|
||||
def start_session(self, name: Optional[str] = None) -> Session:
|
||||
session_name = name or f"Session {datetime.now().strftime('%Y%m%d_%H%M%S')}"
|
||||
self.current_session = Session(
|
||||
name=session_name,
|
||||
commands=[],
|
||||
start_time=datetime.now(),
|
||||
)
|
||||
self.recorded_commands = []
|
||||
return self.current_session
|
||||
|
||||
def record_command(self, command: str, output: str = "", success: bool = True):
|
||||
if self.current_session is None:
|
||||
return
|
||||
|
||||
cmd_entry = {
|
||||
"command": command,
|
||||
"output": output,
|
||||
"success": success,
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
}
|
||||
self.recorded_commands.append(cmd_entry)
|
||||
self.current_session.commands = self.recorded_commands
|
||||
|
||||
def stop_session(self) -> Optional[Session]:
|
||||
if self.current_session is None:
|
||||
return None
|
||||
|
||||
self.current_session.end_time = datetime.now()
|
||||
session_id = self.db.add_session(self.current_session)
|
||||
self.current_session.id = session_id
|
||||
|
||||
session = self.current_session
|
||||
self.current_session = None
|
||||
self.recorded_commands = []
|
||||
|
||||
return session
|
||||
|
||||
def get_sessions(self) -> List[Session]:
|
||||
return self.db.get_sessions()
|
||||
|
||||
def get_session(self, session_id: int) -> Optional[Session]:
|
||||
return self.db.get_session(session_id)
|
||||
|
||||
def replay_session(self, session_id: int, dry_run: bool = False, delay: float = 0.5) -> List[Dict[str, Any]]:
|
||||
session = self.db.get_session(session_id)
|
||||
if session is None:
|
||||
return []
|
||||
|
||||
results = []
|
||||
for cmd_entry in session.commands:
|
||||
result = {
|
||||
"command": cmd_entry["command"],
|
||||
"executed": False,
|
||||
"output": "",
|
||||
"success": False,
|
||||
}
|
||||
|
||||
if not dry_run:
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
cmd_entry["command"],
|
||||
shell=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
result["output"] = proc.stdout + proc.stderr
|
||||
result["success"] = proc.returncode == 0
|
||||
result["executed"] = True
|
||||
except subprocess.TimeoutExpired:
|
||||
result["output"] = "Command timed out"
|
||||
except Exception as e:
|
||||
result["output"] = str(e)
|
||||
|
||||
results.append(result)
|
||||
|
||||
if delay > 0 and not dry_run:
|
||||
time.sleep(delay)
|
||||
|
||||
return results
|
||||
|
||||
def export_session(self, session_id: int) -> Optional[str]:
|
||||
session = self.db.get_session(session_id)
|
||||
if session is None:
|
||||
return None
|
||||
|
||||
lines = ["#!/bin/bash", "", f"# Session: {session.name}", f"# Recorded: {session.start_time.isoformat()}", ""]
|
||||
|
||||
for cmd_entry in session.commands:
|
||||
lines.append(f"# {cmd_entry['timestamp']}")
|
||||
lines.append(cmd_entry["command"])
|
||||
lines.append("")
|
||||
|
||||
return "\n".join(lines)
|
||||
Reference in New Issue
Block a user