369 lines
14 KiB
Python
369 lines
14 KiB
Python
#!/usr/bin/env python3.11
|
|
"""
|
|
Data Management - Konsolidierte Datenverwaltung
|
|
=============================================
|
|
|
|
Migration Information:
|
|
- Ursprünglich: file_manager.py, file_utils.py, backup_manager.py
|
|
- Konsolidiert am: 2025-06-09
|
|
- Funktionalitäten: File Management, File Utils, Backup Management,
|
|
Safe Deletion, Data Organization
|
|
- Breaking Changes: Keine - Alle Original-APIs bleiben verfügbar
|
|
|
|
MASSIVE KONSOLIDIERUNG für Projektarbeit MYP
|
|
Author: MYP Team - Till Tomczak
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
import platform
|
|
import subprocess
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from werkzeug.utils import secure_filename
|
|
from typing import Union, List, Optional, Tuple, Dict
|
|
|
|
from utils.logging_config import get_logger
|
|
|
|
# Logger
|
|
data_logger = get_logger("data_management")
|
|
|
|
# ===== FILE MANAGER =====
|
|
|
|
class FileManager:
|
|
"""Zentrales Datei-Management-System für die MYP-Platform"""
|
|
|
|
DIRECTORIES = {
|
|
'jobs': 'jobs',
|
|
'guests': 'guests',
|
|
'avatars': 'avatars',
|
|
'temp': 'temp',
|
|
'backups': 'backups',
|
|
'logs': 'logs',
|
|
'assets': 'assets'
|
|
}
|
|
|
|
def __init__(self, base_upload_folder: str = None):
|
|
try:
|
|
from utils.utilities_collection import UPLOAD_FOLDER, ALLOWED_EXTENSIONS
|
|
self.base_folder = base_upload_folder or UPLOAD_FOLDER
|
|
self.allowed_extensions = ALLOWED_EXTENSIONS
|
|
except ImportError:
|
|
self.base_folder = "uploads"
|
|
self.allowed_extensions = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif', 'gcode'}
|
|
|
|
self.ensure_directories()
|
|
|
|
def ensure_directories(self) -> None:
|
|
"""Erstellt alle erforderlichen Verzeichnisse"""
|
|
try:
|
|
os.makedirs(self.base_folder, exist_ok=True)
|
|
|
|
for category, subdir in self.DIRECTORIES.items():
|
|
dir_path = os.path.join(self.base_folder, subdir)
|
|
os.makedirs(dir_path, exist_ok=True)
|
|
|
|
current_date = datetime.now()
|
|
year_dir = os.path.join(dir_path, str(current_date.year))
|
|
month_dir = os.path.join(year_dir, f"{current_date.month:02d}")
|
|
|
|
os.makedirs(year_dir, exist_ok=True)
|
|
os.makedirs(month_dir, exist_ok=True)
|
|
|
|
except Exception as e:
|
|
data_logger.error(f"Fehler beim Erstellen der Verzeichnisse: {e}")
|
|
|
|
def allowed_file(self, filename: str) -> bool:
|
|
"""Prüft, ob eine Datei erlaubt ist"""
|
|
if '.' not in filename:
|
|
return False
|
|
|
|
extension = filename.rsplit('.', 1)[1].lower()
|
|
return extension in self.allowed_extensions
|
|
|
|
def generate_unique_filename(self, original_filename: str, prefix: str = "") -> str:
|
|
"""Generiert einen eindeutigen Dateinamen"""
|
|
secure_name = secure_filename(original_filename)
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
|
if '.' in secure_name:
|
|
name, ext = secure_name.rsplit('.', 1)
|
|
if prefix:
|
|
unique_name = f"{prefix}_{name}_{timestamp}.{ext}"
|
|
else:
|
|
unique_name = f"{name}_{timestamp}.{ext}"
|
|
else:
|
|
if prefix:
|
|
unique_name = f"{prefix}_{secure_name}_{timestamp}"
|
|
else:
|
|
unique_name = f"{secure_name}_{timestamp}"
|
|
|
|
return unique_name
|
|
|
|
def save_file(self, file, category: str, user_id: int = None,
|
|
prefix: str = "", metadata: Dict = None) -> Optional[Tuple[str, str, Dict]]:
|
|
"""Speichert eine Datei in der organisierten Struktur"""
|
|
try:
|
|
if not file or not file.filename:
|
|
return None
|
|
|
|
if not self.allowed_file(file.filename):
|
|
raise ValueError(f"Dateityp nicht erlaubt: {file.filename}")
|
|
|
|
if category not in self.DIRECTORIES:
|
|
raise ValueError(f"Unbekannte Kategorie: {category}")
|
|
|
|
current_date = datetime.now()
|
|
category_dir = self.DIRECTORIES[category]
|
|
year_dir = str(current_date.year)
|
|
month_dir = f"{current_date.month:02d}"
|
|
|
|
if user_id:
|
|
relative_dir = os.path.join(category_dir, year_dir, month_dir, f"user_{user_id}")
|
|
else:
|
|
relative_dir = os.path.join(category_dir, year_dir, month_dir)
|
|
|
|
full_dir = os.path.join(self.base_folder, relative_dir)
|
|
os.makedirs(full_dir, exist_ok=True)
|
|
|
|
unique_filename = self.generate_unique_filename(file.filename, prefix)
|
|
|
|
relative_path = os.path.join(relative_dir, unique_filename).replace('\\', '/')
|
|
absolute_path = os.path.join(full_dir, unique_filename)
|
|
|
|
file.save(absolute_path)
|
|
|
|
file_metadata = {
|
|
'original_filename': file.filename,
|
|
'unique_filename': unique_filename,
|
|
'relative_path': relative_path,
|
|
'absolute_path': absolute_path,
|
|
'category': category,
|
|
'user_id': user_id,
|
|
'file_size': os.path.getsize(absolute_path),
|
|
'upload_timestamp': current_date.isoformat(),
|
|
'mime_type': file.content_type or 'application/octet-stream'
|
|
}
|
|
|
|
if metadata:
|
|
file_metadata.update(metadata)
|
|
|
|
return relative_path, absolute_path, file_metadata
|
|
|
|
except Exception as e:
|
|
data_logger.error(f"Fehler beim Speichern der Datei: {e}")
|
|
return None
|
|
|
|
def delete_file(self, relative_path: str) -> bool:
|
|
"""Löscht eine Datei"""
|
|
try:
|
|
if not relative_path:
|
|
return False
|
|
|
|
absolute_path = os.path.join(self.base_folder, relative_path)
|
|
|
|
if os.path.exists(absolute_path) and os.path.isfile(absolute_path):
|
|
os.remove(absolute_path)
|
|
return True
|
|
|
|
return False
|
|
|
|
except Exception as e:
|
|
data_logger.error(f"Fehler beim Löschen der Datei {relative_path}: {e}")
|
|
return False
|
|
|
|
# ===== SAFE FILE HANDLER =====
|
|
|
|
class SafeFileHandler:
|
|
"""Sichere Datei-Operationen ohne externe Abhängigkeiten"""
|
|
|
|
def __init__(self):
|
|
self.platform = platform.system().lower()
|
|
self.is_windows = self.platform == 'windows'
|
|
|
|
def move_to_trash(self, file_path: Union[str, Path]) -> bool:
|
|
"""Verschiebt eine Datei in den Papierkorb"""
|
|
file_path = Path(file_path)
|
|
|
|
if not file_path.exists():
|
|
data_logger.warning(f"Datei nicht gefunden: {file_path}")
|
|
return False
|
|
|
|
try:
|
|
if self.is_windows:
|
|
return self._move_to_trash_windows(file_path)
|
|
else:
|
|
return self._move_to_trash_unix(file_path)
|
|
|
|
except Exception as e:
|
|
data_logger.error(f"Fehler beim Verschieben in Papierkorb: {e}")
|
|
return False
|
|
|
|
def _move_to_trash_windows(self, file_path: Path) -> bool:
|
|
"""Windows-spezifische Papierkorb-Implementation"""
|
|
try:
|
|
from utils.core_system import safe_subprocess_run
|
|
cmd = [
|
|
'powershell', '-Command',
|
|
f'Add-Type -AssemblyName Microsoft.VisualBasic; '
|
|
f'[Microsoft.VisualBasic.FileIO.FileSystem]::DeleteFile("{file_path}", '
|
|
f'"OnlyErrorDialogs", "SendToRecycleBin")'
|
|
]
|
|
|
|
result = safe_subprocess_run(cmd, capture_output=True, timeout=30)
|
|
|
|
if result and result.returncode == 0:
|
|
data_logger.info(f"Datei erfolgreich in Papierkorb verschoben: {file_path}")
|
|
return True
|
|
else:
|
|
error_msg = result.stderr if result else "Unbekannter Fehler"
|
|
data_logger.error(f"PowerShell-Fehler: {error_msg}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
data_logger.error(f"Windows Papierkorb-Fehler: {e}")
|
|
return False
|
|
|
|
def _move_to_trash_unix(self, file_path: Path) -> bool:
|
|
"""Unix-spezifische Papierkorb-Implementation"""
|
|
try:
|
|
from utils.core_system import safe_subprocess_run
|
|
tools = ['gio', 'gvfs-trash', 'trash-put']
|
|
|
|
for tool in tools:
|
|
if shutil.which(tool):
|
|
if tool == 'gio':
|
|
cmd = ['gio', 'trash', str(file_path)]
|
|
elif tool == 'gvfs-trash':
|
|
cmd = ['gvfs-trash', str(file_path)]
|
|
elif tool == 'trash-put':
|
|
cmd = ['trash-put', str(file_path)]
|
|
|
|
result = safe_subprocess_run(cmd, capture_output=True, timeout=10)
|
|
|
|
if result and result.returncode == 0:
|
|
data_logger.info(f"Datei erfolgreich in Papierkorb verschoben ({tool}): {file_path}")
|
|
return True
|
|
|
|
# Fallback: Direkte Löschung
|
|
return self._delete_permanently(file_path)
|
|
|
|
except Exception as e:
|
|
data_logger.error(f"Unix Papierkorb-Fehler: {e}")
|
|
return False
|
|
|
|
def _delete_permanently(self, file_path: Path) -> bool:
|
|
"""Permanente Löschung als Fallback"""
|
|
try:
|
|
if file_path.is_file():
|
|
file_path.unlink()
|
|
elif file_path.is_dir():
|
|
shutil.rmtree(file_path)
|
|
|
|
data_logger.info(f"Datei permanent gelöscht: {file_path}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
data_logger.error(f"Permanente Löschung fehlgeschlagen: {e}")
|
|
return False
|
|
|
|
def safe_delete(self, file_path: Union[str, Path], use_trash: bool = True) -> bool:
|
|
"""Sichere Datei-Löschung"""
|
|
file_path = Path(file_path)
|
|
|
|
if not file_path.exists():
|
|
return True
|
|
|
|
if use_trash:
|
|
if self.move_to_trash(file_path):
|
|
return True
|
|
else:
|
|
return self._delete_permanently(file_path)
|
|
else:
|
|
return self._delete_permanently(file_path)
|
|
|
|
# ===== BACKUP MANAGER =====
|
|
|
|
class BackupManager:
|
|
"""Backup-Management-System"""
|
|
|
|
def __init__(self):
|
|
self.enabled = True
|
|
|
|
def create_backup(self, backup_type="manual"):
|
|
"""Erstellt ein Backup"""
|
|
try:
|
|
from utils.database_utils import DatabaseBackupManager
|
|
backup_manager = DatabaseBackupManager()
|
|
backup_path = backup_manager.create_backup(compress=True)
|
|
|
|
return {
|
|
"success": True,
|
|
"message": f"Backup erfolgreich erstellt: {backup_type}",
|
|
"backup_path": backup_path,
|
|
"backup_type": backup_type
|
|
}
|
|
|
|
except Exception as e:
|
|
data_logger.error(f"Backup-Fehler: {e}")
|
|
return {
|
|
"success": False,
|
|
"message": f"Backup fehlgeschlagen: {e}",
|
|
"error": str(e)
|
|
}
|
|
|
|
# ===== GLOBALE INSTANZEN =====
|
|
|
|
file_manager = FileManager()
|
|
file_handler = SafeFileHandler()
|
|
backup_manager = BackupManager()
|
|
|
|
# ===== CONVENIENCE FUNCTIONS =====
|
|
|
|
def save_job_file(file, user_id: int, metadata: Dict = None) -> Optional[Tuple[str, str, Dict]]:
|
|
"""Speichert eine Druckjob-Datei"""
|
|
return file_manager.save_file(file, 'jobs', user_id, 'job', metadata)
|
|
|
|
def save_guest_file(file, metadata: Dict = None) -> Optional[Tuple[str, str, Dict]]:
|
|
"""Speichert eine Gastauftrags-Datei"""
|
|
return file_manager.save_file(file, 'guests', None, 'guest', metadata)
|
|
|
|
def save_avatar_file(file, user_id: int) -> Optional[Tuple[str, str, Dict]]:
|
|
"""Speichert eine Avatar-Datei"""
|
|
return file_manager.save_file(file, 'avatars', user_id, 'avatar')
|
|
|
|
def save_temp_file(file, user_id: int = None, metadata: Dict = None) -> Optional[Tuple[str, str, Dict]]:
|
|
"""Speichert eine temporäre Datei"""
|
|
return file_manager.save_file(file, 'temp', user_id, 'temp', metadata)
|
|
|
|
def save_asset_file(file, metadata: Dict = None) -> Optional[Tuple[str, str, Dict]]:
|
|
"""Speichert eine Asset-Datei"""
|
|
return file_manager.save_file(file, 'assets', None, 'asset', metadata)
|
|
|
|
def save_log_file(file, metadata: Dict = None) -> Optional[Tuple[str, str, Dict]]:
|
|
"""Speichert eine Log-Datei"""
|
|
return file_manager.save_file(file, 'logs', None, 'log', metadata)
|
|
|
|
def save_backup_file(file, metadata: Dict = None) -> Optional[Tuple[str, str, Dict]]:
|
|
"""Speichert eine Backup-Datei"""
|
|
return file_manager.save_file(file, 'backups', None, 'backup', metadata)
|
|
|
|
def delete_file(relative_path: str) -> bool:
|
|
"""Löscht eine Datei"""
|
|
return file_manager.delete_file(relative_path)
|
|
|
|
def move_to_trash(file_path: Union[str, Path]) -> bool:
|
|
"""Verschiebt Datei in Papierkorb"""
|
|
return file_handler.move_to_trash(file_path)
|
|
|
|
def safe_delete(file_path: Union[str, Path], use_trash: bool = True) -> bool:
|
|
"""Sichere Datei-Löschung"""
|
|
return file_handler.safe_delete(file_path, use_trash)
|
|
|
|
def send2trash(path: Union[str, Path]) -> None:
|
|
"""Kompatibilitätsfunktion für Send2Trash"""
|
|
if not move_to_trash(path):
|
|
raise OSError(f"Konnte Datei nicht in Papierkorb verschieben: {path}")
|
|
|
|
data_logger.info("✅ Data Management Module initialisiert")
|
|
data_logger.info("📊 Massive Konsolidierung: 3 Dateien → 1 Datei (67% Reduktion)") |