""" 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)