import json import os from typing import List, Dict, Any, Optional from datetime import datetime import logging logger = logging.getLogger(__name__) class Database: """ Eine einfache JSON-basierte Datenbank für die Datenspeicherung. In einer Produktionsumgebung würde hier eine richtige Datenbank verwendet werden. """ def __init__(self, data_dir="./data"): self.data_dir = data_dir os.makedirs(data_dir, exist_ok=True) # Standardpfade für die Datendateien self.workspaces_file = os.path.join(data_dir, "workspaces.json") self.agents_file = os.path.join(data_dir, "agents.json") self.files_file = os.path.join(data_dir, "files.json") self.prompts_file = os.path.join(data_dir, "prompts.json") # Initiale Daten erstellen, falls die Dateien nicht existieren self._initialize_data() def _initialize_data(self): """Initialisiert die Datendateien, falls sie nicht existieren""" # Workspaces if not os.path.exists(self.workspaces_file): self._save_data(self.workspaces_file, [ { "id": "workspace_001", "name": "Datenanalyse-Projekt", "created_at": datetime.now().isoformat(), "prompts": [], "agents": [], "dataObjectReferences": [] }, { "id": "workspace_002", "name": "Marktforschung", "created_at": datetime.now().isoformat(), "prompts": [], "agents": [], "dataObjectReferences": [] } ]) # Agenten if not os.path.exists(self.agents_file): self._save_data(self.agents_file, [ { "id": "agent_001", "name": "Datenanalyse-Agent", "type": "analyzer", "workspace_id": "workspace_001", "capabilities": ["Datenanalyse", "Statistik", "Trendanalyse"], "description": "Spezialisiert auf die Analyse von strukturierten Daten" }, { "id": "agent_002", "name": "Visualisierungs-Agent", "type": "visualizer", "workspace_id": "workspace_001", "capabilities": ["Datenvisualisierung", "Diagrammerstellung"], "description": "Erstellt visuelle Darstellungen aus Daten" }, { "id": "agent_003", "name": "Text-Generator", "type": "writer", "workspace_id": "workspace_001", "capabilities": ["Texterstellung", "Zusammenfassung"], "description": "Verfasst verständliche Berichte und Zusammenfassungen" }, { "id": "agent_004", "name": "Web-Scraper", "type": "scraper", "workspace_id": "workspace_002", "capabilities": ["Datensammlung", "Webanalyse"], "description": "Sammelt Daten aus verschiedenen Webquellen" }, { "id": "agent_005", "name": "Marktanalyse-Agent", "type": "analyzer", "workspace_id": "workspace_002", "capabilities": ["Wettbewerbsanalyse", "Markttrends"], "description": "Spezialisiert auf Wettbewerbsanalyse und Markttrends" } ]) # Dateien if not os.path.exists(self.files_file): self._save_data(self.files_file, [ { "id": "file_001", "name": "Quartalsbericht Q1 2025.pdf", "type": "document", "content_type": "application/pdf", "size": 2500000, # ca. 2.4 MB "upload_date": datetime.now().isoformat(), "path": "uploads/dummy_file1.pdf" }, { "id": "file_002", "name": "Marktanalyse-Diagramm.png", "type": "image", "content_type": "image/png", "size": 1150000, # ca. 1.1 MB "upload_date": datetime.now().isoformat(), "path": "uploads/dummy_file2.png" }, { "id": "file_003", "name": "Finanzdaten_2024-2025.xlsx", "type": "document", "content_type": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "size": 3880000, # ca. 3.7 MB "upload_date": datetime.now().isoformat(), "path": "uploads/dummy_file3.xlsx" } ]) # Prompts if not os.path.exists(self.prompts_file): self._save_data(self.prompts_file, [ { "id": "prompt_001", "content": "Analysiere die Quartalsdaten und erstelle eine Zusammenfassung mit wichtigsten Trends", "workspace_id": "workspace_001", "created_at": datetime.now().isoformat() }, { "id": "prompt_002", "content": "Erstelle eine Prognose für das nächste Quartal basierend auf historischen Daten", "workspace_id": "workspace_001", "created_at": datetime.now().isoformat() } ]) def _load_data(self, file_path): """Lädt Daten aus einer JSON-Datei""" try: if os.path.exists(file_path): with open(file_path, 'r', encoding='utf-8') as f: return json.load(f) return [] except Exception as e: logger.error(f"Fehler beim Laden der Daten aus {file_path}: {e}") return [] def _save_data(self, file_path, data): """Speichert Daten in einer JSON-Datei""" try: with open(file_path, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) except Exception as e: logger.error(f"Fehler beim Speichern der Daten in {file_path}: {e}") # Workspace-Methoden def get_all_workspaces(self) -> List[Dict[str, Any]]: """Gibt alle Workspaces zurück""" workspaces = self._load_data(self.workspaces_file) # Füge Agenten und Prompts hinzu for workspace in workspaces: workspace_id = workspace["id"] workspace["agents"] = self.get_agents_by_workspace(workspace_id) workspace["prompts"] = self.get_prompts_by_workspace(workspace_id) return workspaces def get_workspace(self, workspace_id: str) -> Optional[Dict[str, Any]]: """Gibt einen Workspace anhand seiner ID zurück""" workspaces = self._load_data(self.workspaces_file) for workspace in workspaces: if workspace["id"] == workspace_id: # Füge Agenten und Prompts hinzu workspace["agents"] = self.get_agents_by_workspace(workspace_id) workspace["prompts"] = self.get_prompts_by_workspace(workspace_id) return workspace return None def create_workspace(self, workspace_data: Dict[str, Any]) -> Dict[str, Any]: """Erstellt einen neuen Workspace""" workspaces = self._load_data(self.workspaces_file) workspaces.append(workspace_data) self._save_data(self.workspaces_file, workspaces) return workspace_data # Agent-Methoden def get_all_agents(self) -> List[Dict[str, Any]]: """Gibt alle Agenten zurück""" return self._load_data(self.agents_file) def get_agents_by_workspace(self, workspace_id: str) -> List[Dict[str, Any]]: """Gibt alle Agenten eines Workspaces zurück""" agents = self._load_data(self.agents_file) return [agent for agent in agents if agent.get("workspace_id") == workspace_id] def get_agent(self, agent_id: str) -> Optional[Dict[str, Any]]: """Gibt einen Agenten anhand seiner ID zurück""" agents = self._load_data(self.agents_file) for agent in agents: if agent["id"] == agent_id: return agent return None def create_agent(self, agent_data: Dict[str, Any], workspace_id: str) -> Dict[str, Any]: """Erstellt einen neuen Agenten""" agents = self._load_data(self.agents_file) agent_data["workspace_id"] = workspace_id agents.append(agent_data) self._save_data(self.agents_file, agents) # Aktualisiere die Workspace-Referenz workspaces = self._load_data(self.workspaces_file) for workspace in workspaces: if workspace["id"] == workspace_id: if "agents" not in workspace: workspace["agents"] = [] workspace["agents"].append(agent_data["id"]) break self._save_data(self.workspaces_file, workspaces) return agent_data # Datei-Methoden def get_all_files(self) -> List[Dict[str, Any]]: """Gibt alle Dateien zurück""" files = self._load_data(self.files_file) return files def get_file(self, file_id: str) -> Optional[Dict[str, Any]]: """Gibt eine Datei anhand ihrer ID zurück""" files = self._load_data(self.files_file) for file in files: if file["id"] == file_id: return file return None def create_file(self, file_data: Dict[str, Any]) -> Dict[str, Any]: """Erstellt einen neuen Dateieintrag""" files = self._load_data(self.files_file) files.append(file_data) self._save_data(self.files_file, files) return file_data # Prompt-Methoden def get_all_prompts(self) -> List[Dict[str, Any]]: """Gibt alle Prompts zurück""" return self._load_data(self.prompts_file) def get_prompts_by_workspace(self, workspace_id: str) -> List[Dict[str, Any]]: """Gibt alle Prompts eines Workspaces zurück""" prompts = self._load_data(self.prompts_file) return [prompt for prompt in prompts if prompt.get("workspace_id") == workspace_id] def get_prompt(self, prompt_id: str) -> Optional[Dict[str, Any]]: """Gibt einen Prompt anhand seiner ID zurück""" prompts = self._load_data(self.prompts_file) for prompt in prompts: if prompt["id"] == prompt_id: return prompt return None def create_prompt(self, prompt_data: Dict[str, Any], workspace_id: str) -> Dict[str, Any]: """Erstellt einen neuen Prompt""" prompts = self._load_data(self.prompts_file) prompt_data["workspace_id"] = workspace_id prompts.append(prompt_data) self._save_data(self.prompts_file, prompts) # Aktualisiere die Workspace-Referenz workspaces = self._load_data(self.workspaces_file) for workspace in workspaces: if workspace["id"] == workspace_id: if "prompts" not in workspace: workspace["prompts"] = [] workspace["prompts"].append(prompt_data["id"]) break self._save_data(self.workspaces_file, workspaces) return prompt_data # Singleton-Instanz der Datenbank _db_instance = None def get_db(): """Gibt eine Singleton-Instanz der Datenbank zurück""" global _db_instance if _db_instance is None: _db_instance = Database() return _db_instance