414 lines
15 KiB
Python
414 lines
15 KiB
Python
"""
|
|
Mercedes-Benz MYP - Datei-Management-System
|
|
Organisierte Speicherung von hochgeladenen Dateien mit Verzeichniskonventionen
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
from datetime import datetime
|
|
from werkzeug.utils import secure_filename
|
|
from typing import Optional, Tuple, Dict, List
|
|
from config.settings import UPLOAD_FOLDER, ALLOWED_EXTENSIONS
|
|
|
|
class FileManager:
|
|
"""
|
|
Zentrales Datei-Management-System für die MYP-Platform
|
|
Organisiert Uploads in strukturierte Unterverzeichnisse
|
|
"""
|
|
|
|
# Verzeichniskonventionen
|
|
DIRECTORIES = {
|
|
'jobs': 'jobs', # Druckjob-Dateien
|
|
'guests': 'guests', # Gastauftrags-Dateien
|
|
'avatars': 'avatars', # Benutzer-Avatare
|
|
'temp': 'temp', # Temporäre Dateien
|
|
'backups': 'backups', # Backup-Dateien
|
|
'logs': 'logs', # Exportierte Logs
|
|
'assets': 'assets' # Statische Assets
|
|
}
|
|
|
|
def __init__(self, base_upload_folder: str = UPLOAD_FOLDER):
|
|
"""
|
|
Initialisiert den FileManager
|
|
|
|
Args:
|
|
base_upload_folder: Basis-Upload-Verzeichnis
|
|
"""
|
|
self.base_folder = base_upload_folder
|
|
self.ensure_directories()
|
|
|
|
def ensure_directories(self) -> None:
|
|
"""Erstellt alle erforderlichen Verzeichnisse"""
|
|
try:
|
|
# Basis-Upload-Ordner erstellen
|
|
os.makedirs(self.base_folder, exist_ok=True)
|
|
|
|
# Alle Unterverzeichnisse erstellen
|
|
for category, subdir in self.DIRECTORIES.items():
|
|
dir_path = os.path.join(self.base_folder, subdir)
|
|
os.makedirs(dir_path, exist_ok=True)
|
|
|
|
# Jahres-/Monatsverzeichnisse für organisierte Speicherung
|
|
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:
|
|
print(f"Fehler beim Erstellen der Verzeichnisse: {e}")
|
|
|
|
def allowed_file(self, filename: str) -> bool:
|
|
"""
|
|
Prüft, ob eine Datei erlaubt ist
|
|
|
|
Args:
|
|
filename: Name der Datei
|
|
|
|
Returns:
|
|
bool: True wenn erlaubt
|
|
"""
|
|
if '.' not in filename:
|
|
return False
|
|
|
|
extension = filename.rsplit('.', 1)[1].lower()
|
|
return extension in ALLOWED_EXTENSIONS
|
|
|
|
def generate_unique_filename(self, original_filename: str, prefix: str = "") -> str:
|
|
"""
|
|
Generiert einen eindeutigen Dateinamen
|
|
|
|
Args:
|
|
original_filename: Ursprünglicher Dateiname
|
|
prefix: Optionaler Präfix
|
|
|
|
Returns:
|
|
str: Eindeutiger Dateiname
|
|
"""
|
|
# Dateiname sicher machen
|
|
secure_name = secure_filename(original_filename)
|
|
|
|
# Timestamp hinzufügen für Eindeutigkeit
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
|
# Dateiname und Erweiterung trennen
|
|
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
|
|
|
|
Args:
|
|
file: Werkzeug FileStorage Objekt
|
|
category: Kategorie (jobs, guests, avatars, etc.)
|
|
user_id: Benutzer-ID für Pfad-Organisation
|
|
prefix: Dateiname-Präfix
|
|
metadata: Zusätzliche Metadaten
|
|
|
|
Returns:
|
|
Tuple[str, str, Dict]: (relativer_pfad, absoluter_pfad, metadaten) oder None bei Fehler
|
|
"""
|
|
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}")
|
|
|
|
# Verzeichnisstruktur aufbauen
|
|
current_date = datetime.now()
|
|
category_dir = self.DIRECTORIES[category]
|
|
year_dir = str(current_date.year)
|
|
month_dir = f"{current_date.month:02d}"
|
|
|
|
# Benutzer-spezifischen Unterordner hinzufügen wenn user_id vorhanden
|
|
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)
|
|
|
|
# Vollständigen Pfad erstellen
|
|
full_dir = os.path.join(self.base_folder, relative_dir)
|
|
os.makedirs(full_dir, exist_ok=True)
|
|
|
|
# Eindeutigen Dateinamen generieren
|
|
unique_filename = self.generate_unique_filename(file.filename, prefix)
|
|
|
|
# Pfade definieren
|
|
relative_path = os.path.join(relative_dir, unique_filename).replace('\\', '/')
|
|
absolute_path = os.path.join(full_dir, unique_filename)
|
|
|
|
# Datei speichern
|
|
file.save(absolute_path)
|
|
|
|
# Metadaten sammeln
|
|
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'
|
|
}
|
|
|
|
# Zusätzliche Metadaten hinzufügen
|
|
if metadata:
|
|
file_metadata.update(metadata)
|
|
|
|
return relative_path, absolute_path, file_metadata
|
|
|
|
except Exception as e:
|
|
print(f"Fehler beim Speichern der Datei: {e}")
|
|
return None
|
|
|
|
def delete_file(self, relative_path: str) -> bool:
|
|
"""
|
|
Löscht eine Datei
|
|
|
|
Args:
|
|
relative_path: Relativer Pfad zur Datei
|
|
|
|
Returns:
|
|
bool: True wenn erfolgreich gelöscht
|
|
"""
|
|
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:
|
|
print(f"Fehler beim Löschen der Datei {relative_path}: {e}")
|
|
return False
|
|
|
|
def move_file(self, old_relative_path: str, new_category: str,
|
|
new_prefix: str = "") -> Optional[str]:
|
|
"""
|
|
Verschiebt eine Datei in eine andere Kategorie
|
|
|
|
Args:
|
|
old_relative_path: Alter relativer Pfad
|
|
new_category: Neue Kategorie
|
|
new_prefix: Neuer Präfix
|
|
|
|
Returns:
|
|
str: Neuer relativer Pfad oder None bei Fehler
|
|
"""
|
|
try:
|
|
old_absolute_path = os.path.join(self.base_folder, old_relative_path)
|
|
|
|
if not os.path.exists(old_absolute_path):
|
|
return None
|
|
|
|
# Dateiname extrahieren
|
|
filename = os.path.basename(old_absolute_path)
|
|
|
|
# Neuen Pfad generieren
|
|
current_date = datetime.now()
|
|
new_category_dir = self.DIRECTORIES.get(new_category)
|
|
if not new_category_dir:
|
|
return None
|
|
|
|
year_dir = str(current_date.year)
|
|
month_dir = f"{current_date.month:02d}"
|
|
new_relative_dir = os.path.join(new_category_dir, year_dir, month_dir)
|
|
new_full_dir = os.path.join(self.base_folder, new_relative_dir)
|
|
|
|
os.makedirs(new_full_dir, exist_ok=True)
|
|
|
|
# Neuen Dateinamen generieren falls Präfix angegeben
|
|
if new_prefix:
|
|
new_filename = self.generate_unique_filename(filename, new_prefix)
|
|
else:
|
|
new_filename = filename
|
|
|
|
new_relative_path = os.path.join(new_relative_dir, new_filename).replace('\\', '/')
|
|
new_absolute_path = os.path.join(new_full_dir, new_filename)
|
|
|
|
# Datei verschieben
|
|
shutil.move(old_absolute_path, new_absolute_path)
|
|
|
|
return new_relative_path
|
|
|
|
except Exception as e:
|
|
print(f"Fehler beim Verschieben der Datei: {e}")
|
|
return None
|
|
|
|
def get_file_info(self, relative_path: str) -> Optional[Dict]:
|
|
"""
|
|
Gibt Informationen über eine Datei zurück
|
|
|
|
Args:
|
|
relative_path: Relativer Pfad zur Datei
|
|
|
|
Returns:
|
|
Dict: Datei-Informationen oder None
|
|
"""
|
|
try:
|
|
if not relative_path:
|
|
return None
|
|
|
|
absolute_path = os.path.join(self.base_folder, relative_path)
|
|
|
|
if not os.path.exists(absolute_path):
|
|
return None
|
|
|
|
stat = os.stat(absolute_path)
|
|
|
|
return {
|
|
'filename': os.path.basename(absolute_path),
|
|
'relative_path': relative_path,
|
|
'absolute_path': absolute_path,
|
|
'size': stat.st_size,
|
|
'created': datetime.fromtimestamp(stat.st_ctime).isoformat(),
|
|
'modified': datetime.fromtimestamp(stat.st_mtime).isoformat(),
|
|
'exists': True
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"Fehler beim Abrufen der Datei-Informationen: {e}")
|
|
return None
|
|
|
|
def cleanup_temp_files(self, max_age_hours: int = 24) -> int:
|
|
"""
|
|
Räumt temporäre Dateien auf
|
|
|
|
Args:
|
|
max_age_hours: Maximales Alter in Stunden
|
|
|
|
Returns:
|
|
int: Anzahl gelöschte Dateien
|
|
"""
|
|
try:
|
|
temp_dir = os.path.join(self.base_folder, self.DIRECTORIES['temp'])
|
|
if not os.path.exists(temp_dir):
|
|
return 0
|
|
|
|
deleted_count = 0
|
|
max_age_seconds = max_age_hours * 3600
|
|
current_time = datetime.now().timestamp()
|
|
|
|
for root, dirs, files in os.walk(temp_dir):
|
|
for file in files:
|
|
file_path = os.path.join(root, file)
|
|
try:
|
|
file_age = current_time - os.path.getmtime(file_path)
|
|
if file_age > max_age_seconds:
|
|
os.remove(file_path)
|
|
deleted_count += 1
|
|
except Exception:
|
|
continue
|
|
|
|
return deleted_count
|
|
|
|
except Exception as e:
|
|
print(f"Fehler beim Aufräumen temporärer Dateien: {e}")
|
|
return 0
|
|
|
|
def get_category_stats(self) -> Dict[str, Dict]:
|
|
"""
|
|
Gibt Statistiken für alle Kategorien zurück
|
|
|
|
Returns:
|
|
Dict: Statistiken pro Kategorie
|
|
"""
|
|
stats = {}
|
|
|
|
try:
|
|
for category, subdir in self.DIRECTORIES.items():
|
|
category_path = os.path.join(self.base_folder, subdir)
|
|
|
|
if not os.path.exists(category_path):
|
|
stats[category] = {'file_count': 0, 'total_size': 0}
|
|
continue
|
|
|
|
file_count = 0
|
|
total_size = 0
|
|
|
|
for root, dirs, files in os.walk(category_path):
|
|
for file in files:
|
|
file_path = os.path.join(root, file)
|
|
try:
|
|
total_size += os.path.getsize(file_path)
|
|
file_count += 1
|
|
except Exception:
|
|
continue
|
|
|
|
stats[category] = {
|
|
'file_count': file_count,
|
|
'total_size': total_size,
|
|
'total_size_mb': round(total_size / (1024 * 1024), 2)
|
|
}
|
|
|
|
return stats
|
|
|
|
except Exception as e:
|
|
print(f"Fehler beim Abrufen der Kategorie-Statistiken: {e}")
|
|
return {}
|
|
|
|
|
|
# Globale FileManager-Instanz
|
|
file_manager = FileManager()
|
|
|
|
# Convenience-Funktionen
|
|
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_asset_file(file, user_id: int, metadata: Dict = None) -> Optional[Tuple[str, str, Dict]]:
|
|
"""Speichert eine Asset-Datei"""
|
|
return file_manager.save_file(file, 'assets', user_id, 'asset', metadata)
|
|
|
|
def save_log_file(file, user_id: int, metadata: Dict = None) -> Optional[Tuple[str, str, Dict]]:
|
|
"""Speichert eine Log-Datei"""
|
|
return file_manager.save_file(file, 'logs', user_id, 'log', metadata)
|
|
|
|
def save_backup_file(file, user_id: int, metadata: Dict = None) -> Optional[Tuple[str, str, Dict]]:
|
|
"""Speichert eine Backup-Datei"""
|
|
return file_manager.save_file(file, 'backups', user_id, 'backup', metadata)
|
|
|
|
def save_temp_file(file, user_id: int, 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 delete_file(relative_path: str) -> bool:
|
|
"""Löscht eine Datei"""
|
|
return file_manager.delete_file(relative_path)
|
|
|
|
def get_file_info(relative_path: str) -> Optional[Dict]:
|
|
"""Gibt Datei-Informationen zurück"""
|
|
return file_manager.get_file_info(relative_path) |