wd666/utils/storage.py

113 lines
3.8 KiB
Python

"""
Storage Manager - Handle local persistence of configuration and history/reports.
"""
import os
import json
import time
from typing import List, Dict, Any
from pathlib import Path
# Constants
STORAGE_DIR = ".storage"
CONFIG_FILE = "config.json"
HISTORY_DIR = "history"
class StorageManager:
def __init__(self):
self.root_dir = Path(STORAGE_DIR)
self.config_path = self.root_dir / CONFIG_FILE
self.history_dir = self.root_dir / HISTORY_DIR
# Ensure directories exist
self.root_dir.mkdir(exist_ok=True)
self.history_dir.mkdir(exist_ok=True)
def save_config(self, config_data: Dict[str, Any]):
"""Save UI configuration to file"""
try:
with open(self.config_path, 'w', encoding='utf-8') as f:
json.dump(config_data, f, indent=2, ensure_ascii=False)
except Exception as e:
print(f"Error saving config: {e}")
def load_config(self) -> Dict[str, Any]:
"""Load UI configuration from file"""
if not self.config_path.exists():
return {}
try:
with open(self.config_path, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
print(f"Error loading config: {e}")
return {}
def save_history(self, session_type: str, topic: str, content: str, metadata: Dict[str, Any] = None):
"""
Save a session report/history
Args:
session_type: 'council' or 'debate'
topic: The main topic
content: The full markdown report or content
metadata: Additional info (model used, date, etc)
"""
timestamp = int(time.time())
date_str = time.strftime("%Y-%m-%d %H:%M:%S")
# Create a safe filename
safe_topic = "".join([c for c in topic[:20] if c.isalnum() or c in (' ', '_', '-')]).strip().replace(' ', '_')
filename = f"{timestamp}_{session_type}_{safe_topic}.json"
data = {
"id": str(timestamp),
"timestamp": timestamp,
"date": date_str,
"type": session_type,
"topic": topic,
"content": content,
"metadata": metadata or {}
}
try:
with open(self.history_dir / filename, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
return True
except Exception as e:
print(f"Error saving history: {e}")
return False
def list_history(self) -> List[Dict[str, Any]]:
"""List all history items (metadata only)"""
items = []
if not self.history_dir.exists():
return []
for file in self.history_dir.glob("*.json"):
try:
with open(file, 'r', encoding='utf-8') as f:
data = json.load(f)
# Return summary info
items.append({
"id": data.get("id"),
"date": data.get("date"),
"type": data.get("type"),
"topic": data.get("topic"),
"filename": file.name
})
except Exception:
continue
# Sort by timestamp desc
return sorted(items, key=lambda x: x.get("date", ""), reverse=True)
def load_history_item(self, filename: str) -> Dict[str, Any]:
"""Load full content of a history item"""
path = self.history_dir / filename
if not path.exists():
return None
try:
with open(path, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception:
return None