95459fb4c8
- Fixed undefined 'tool' variable in display_history function - Changed '[tool]' markup tag usage to proper Rich syntax - All tests now pass (38/38 unit tests) - Type checking passes with mypy --strict
137 lines
3.9 KiB
Python
137 lines
3.9 KiB
Python
"""History management module."""
|
|
|
|
import json
|
|
from datetime import datetime
|
|
|
|
from shell_speak.config import ensure_data_dir, get_history_file
|
|
from shell_speak.models import HistoryEntry
|
|
|
|
|
|
class HistoryManager:
|
|
"""Manages command history storage and retrieval."""
|
|
|
|
def __init__(self) -> None:
|
|
self._entries: list[HistoryEntry] = []
|
|
self._loaded = False
|
|
|
|
def load(self) -> None:
|
|
"""Load history from file."""
|
|
history_file = get_history_file()
|
|
if not history_file.exists():
|
|
self._entries = []
|
|
self._loaded = True
|
|
return
|
|
|
|
try:
|
|
with open(history_file) as f:
|
|
data = json.load(f)
|
|
self._entries = []
|
|
for item in data.get("entries", []):
|
|
entry = HistoryEntry(
|
|
query=item.get("query", ""),
|
|
command=item.get("command", ""),
|
|
tool=item.get("tool", ""),
|
|
timestamp=datetime.fromisoformat(item.get("timestamp", datetime.now().isoformat())),
|
|
explanation=item.get("explanation", ""),
|
|
)
|
|
self._entries.append(entry)
|
|
except Exception:
|
|
self._entries = []
|
|
|
|
self._loaded = True
|
|
|
|
def save(self) -> None:
|
|
"""Save history to file."""
|
|
ensure_data_dir()
|
|
history_file = get_history_file()
|
|
|
|
data = {
|
|
"version": "1.0",
|
|
"entries": [
|
|
{
|
|
"query": entry.query,
|
|
"command": entry.command,
|
|
"tool": entry.tool,
|
|
"timestamp": entry.timestamp.isoformat(),
|
|
"explanation": entry.explanation,
|
|
}
|
|
for entry in self._entries
|
|
],
|
|
}
|
|
|
|
with open(history_file, 'w') as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
def add(self, query: str, command: str, tool: str, explanation: str = "") -> None:
|
|
"""Add a new entry to history."""
|
|
if not self._loaded:
|
|
self.load()
|
|
|
|
entry = HistoryEntry(
|
|
query=query,
|
|
command=command,
|
|
tool=tool,
|
|
timestamp=datetime.now(),
|
|
explanation=explanation,
|
|
)
|
|
self._entries.append(entry)
|
|
|
|
if len(self._entries) > 1000:
|
|
self._entries = self._entries[-1000:]
|
|
|
|
self.save()
|
|
|
|
def get_all(self) -> list[HistoryEntry]:
|
|
"""Get all history entries."""
|
|
if not self._loaded:
|
|
self.load()
|
|
return self._entries.copy()
|
|
|
|
def get_recent(self, limit: int = 20) -> list[HistoryEntry]:
|
|
"""Get recent history entries."""
|
|
if not self._loaded:
|
|
self.load()
|
|
return self._entries[-limit:]
|
|
|
|
def search(self, query: str, tool: str | None = None) -> list[HistoryEntry]:
|
|
"""Search history entries."""
|
|
if not self._loaded:
|
|
self.load()
|
|
|
|
results = []
|
|
query_lower = query.lower()
|
|
|
|
for entry in self._entries:
|
|
if query_lower in entry.query.lower() or query_lower in entry.command.lower():
|
|
if tool is None or entry.tool == tool:
|
|
results.append(entry)
|
|
|
|
return results
|
|
|
|
def get_last_command(self, tool: str | None = None) -> HistoryEntry | None:
|
|
"""Get the last command from history."""
|
|
if not self._loaded:
|
|
self.load()
|
|
|
|
for entry in reversed(self._entries):
|
|
if tool is None or entry.tool == tool:
|
|
return entry
|
|
|
|
return None
|
|
|
|
def clear(self) -> None:
|
|
"""Clear all history."""
|
|
self._entries = []
|
|
self.save()
|
|
|
|
|
|
_history_manager: HistoryManager | None = None
|
|
|
|
|
|
def get_history_manager() -> HistoryManager:
|
|
"""Get the global history manager."""
|
|
global _history_manager
|
|
if _history_manager is None:
|
|
_history_manager = HistoryManager()
|
|
return _history_manager
|