7000pctAUTO
0edaf795db
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
113 lines
3.5 KiB
Python
113 lines
3.5 KiB
Python
"""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) |