# data_manager.py import os import json import datetime from typing import Dict, Any, Optional, List class DataManager: def __init__(self, base_path: str = '/data'): """Initialize the data manager""" self.base_path = base_path self._ensure_directories() def _ensure_directories(self): """Create necessary directory structure""" directories = [ self.base_path, os.path.join(self.base_path, 'blueprints'), os.path.join(self.base_path, 'book_content'), os.path.join(self.base_path, 'exports'), os.path.join(self.base_path, 'chapters'), os.path.join(self.base_path, 'manual_content') ] for dir_path in directories: os.makedirs(dir_path, exist_ok=True) def save_blueprint(self, blueprint: str) -> str: """Save blueprint to file""" timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"blueprint_{timestamp}.txt" filepath = os.path.join(self.base_path, 'blueprints', filename) try: with open(filepath, 'w') as f: f.write(blueprint) return filename except Exception as e: raise Exception(f"Error saving blueprint: {e}") def save_manual_content(self, part_idx: int, chapter_idx: int, content: str) -> str: """Save manual pre-written content""" filename = f"manual_content_part_{part_idx}_chapter_{chapter_idx}.txt" filepath = os.path.join(self.base_path, 'manual_content', filename) try: with open(filepath, 'w') as f: f.write(content) return filename except Exception as e: raise Exception(f"Error saving manual content: {e}") def load_manual_content(self, part_idx: int, chapter_idx: int) -> Optional[str]: """Load manual pre-written content if it exists""" filename = f"manual_content_part_{part_idx}_chapter_{chapter_idx}.txt" filepath = os.path.join(self.base_path, 'manual_content', filename) try: if os.path.exists(filepath): with open(filepath, 'r') as f: return f.read() return None except Exception as e: raise Exception(f"Error loading manual content: {e}") def save_book_content(self, content: Dict[str, Any]) -> str: """Save complete book content""" timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"book_content_{timestamp}.json" filepath = os.path.join(self.base_path, 'book_content', filename) try: with open(filepath, 'w') as f: json.dump(content, f, indent=4) return filename except Exception as e: raise Exception(f"Error saving book content: {e}") def save_chapter(self, part_idx: int, chapter_idx: int, content: Dict[str, Any]): """Save individual chapter content""" filename = f"part_{part_idx}_chapter_{chapter_idx}.json" filepath = os.path.join(self.base_path, 'chapters', filename) try: with open(filepath, 'w') as f: json.dump(content, f, indent=4) except Exception as e: raise Exception(f"Error saving chapter: {e}") def load_latest_blueprint(self) -> Optional[str]: """Load most recent blueprint""" blueprint_dir = os.path.join(self.base_path, 'blueprints') try: files = os.listdir(blueprint_dir) if not files: return None latest_file = max(files, key=lambda f: os.path.getctime( os.path.join(blueprint_dir, f) )) with open(os.path.join(blueprint_dir, latest_file), 'r') as f: return f.read() except Exception as e: raise Exception(f"Error loading blueprint: {e}") def load_latest_book_content(self) -> Optional[Dict[str, Any]]: """Load most recent book content""" content_dir = os.path.join(self.base_path, 'book_content') try: files = os.listdir(content_dir) if not files: return None latest_file = max(files, key=lambda f: os.path.getctime( os.path.join(content_dir, f) )) with open(os.path.join(content_dir, latest_file), 'r') as f: return json.load(f) except Exception as e: raise Exception(f"Error loading book content: {e}") def export_markdown(self, content: Dict[str, Any]) -> str: """Export book content as markdown""" timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"self_api_book_{timestamp}.md" filepath = os.path.join(self.base_path, 'exports', filename) try: markdown_content = ( f"# {content.get('book_info', {}).get('title', 'Book Title')}\n\n" "## Introduction\n\n" ) markdown_content += content.get('introduction', '') + "\n\n" for part_idx, part in enumerate(content.get('parts', [])): markdown_content += f"# Part {part_idx + 1}: {part['title']}\n\n" for chapter in part.get('chapters', []): markdown_content += f"## {chapter['title']}\n\n" markdown_content += chapter.get('content', '') + "\n\n" with open(filepath, 'w') as f: f.write(markdown_content) return filepath except Exception as e: raise Exception(f"Error exporting markdown: {e}")