299 lines
No EOL
12 KiB
Python
299 lines
No EOL
12 KiB
Python
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 |