"""Asset storage with file management. Metadata is stored in SQLite via BaseSqliteStore; actual files live in the assets directory on disk. """ import mimetypes import shutil import uuid from datetime import datetime, timezone from pathlib import Path from typing import List, Optional from wled_controller.storage.asset import Asset, asset_type_from_mime from wled_controller.storage.base_sqlite_store import BaseSqliteStore from wled_controller.storage.database import Database from wled_controller.utils import get_logger logger = get_logger(__name__) class AssetStore(BaseSqliteStore[Asset]): """Persistent storage for uploaded file assets.""" _table_name = "assets" _entity_name = "Asset" def __init__(self, db: Database, assets_dir: str | Path): self._assets_dir = Path(assets_dir) self._assets_dir.mkdir(parents=True, exist_ok=True) super().__init__(db, Asset.from_dict) # -- Aliases for consistency with other stores ---------------------------- get_all_assets = BaseSqliteStore.get_all get_asset = BaseSqliteStore.get def get_visible_assets(self) -> List[Asset]: """Return all assets that are not soft-deleted.""" return [a for a in self._items.values() if not a.deleted] def get_assets_by_type(self, asset_type: str) -> List[Asset]: """Return visible assets filtered by type.""" return [a for a in self._items.values() if not a.deleted and a.asset_type == asset_type] def get_file_path(self, asset_id: str) -> Optional[Path]: """Resolve the on-disk path for an asset's file. Returns None if missing.""" asset = self._items.get(asset_id) if asset is None or asset.deleted: return None path = self._assets_dir / asset.stored_filename return path if path.exists() else None def create_asset( self, name: str, filename: str, file_data: bytes, mime_type: Optional[str] = None, description: Optional[str] = None, tags: Optional[List[str]] = None, prebuilt: bool = False, ) -> Asset: """Create a new asset from uploaded file data. Args: name: Display name for the asset. filename: Original upload filename. file_data: Raw file bytes. mime_type: MIME type (auto-detected from filename if not provided). description: Optional description. tags: Optional tags. prebuilt: Whether this is a shipped prebuilt asset. Returns: The created Asset. """ self._check_name_unique(name) if not mime_type: mime_type = mimetypes.guess_type(filename)[0] or "application/octet-stream" asset_type = asset_type_from_mime(mime_type) # Generate unique stored filename ext = Path(filename).suffix stored_filename = f"{uuid.uuid4().hex}{ext}" # Write file to disk dest = self._assets_dir / stored_filename dest.write_bytes(file_data) asset_id = f"asset_{uuid.uuid4().hex[:8]}" now = datetime.now(timezone.utc) asset = Asset( id=asset_id, name=name, filename=filename, stored_filename=stored_filename, mime_type=mime_type, asset_type=asset_type, size_bytes=len(file_data), created_at=now, updated_at=now, description=description, tags=tags or [], prebuilt=prebuilt, ) self._items[asset_id] = asset self._save_item(asset_id, asset) logger.info(f"Created asset: {name} ({asset_id}, type={asset_type}, {len(file_data)} bytes)") return asset def update_asset( self, asset_id: str, name: Optional[str] = None, description: Optional[str] = None, tags: Optional[List[str]] = None, ) -> Asset: """Update asset metadata (not the file itself).""" asset = self.get(asset_id) if name is not None: self._check_name_unique(name, exclude_id=asset_id) asset.name = name if description is not None: asset.description = description if tags is not None: asset.tags = tags asset.updated_at = datetime.now(timezone.utc) self._save_item(asset_id, asset) logger.info(f"Updated asset: {asset_id}") return asset def delete_asset(self, asset_id: str) -> None: """Delete an asset. Prebuilt assets are soft-deleted; others are fully removed.""" asset = self.get(asset_id) if asset.prebuilt: # Soft-delete: mark as deleted, remove file but keep metadata asset.deleted = True asset.updated_at = datetime.now(timezone.utc) self._save_item(asset_id, asset) file_path = self._assets_dir / asset.stored_filename if file_path.exists(): file_path.unlink() logger.info(f"Soft-deleted prebuilt asset: {asset_id}") else: # Hard delete: remove file and metadata file_path = self._assets_dir / asset.stored_filename if file_path.exists(): file_path.unlink() del self._items[asset_id] self._delete_item(asset_id) logger.info(f"Deleted asset: {asset_id}") def restore_prebuilt(self, prebuilt_dir: Path) -> List[Asset]: """Re-import any soft-deleted prebuilt assets from the prebuilt directory. Returns list of restored assets. """ restored = [] for asset in list(self._items.values()): if asset.prebuilt and asset.deleted: # Find original file in prebuilt dir src = prebuilt_dir / asset.filename if src.exists(): dest = self._assets_dir / asset.stored_filename shutil.copy2(src, dest) asset.deleted = False asset.size_bytes = dest.stat().st_size asset.updated_at = datetime.now(timezone.utc) self._save_item(asset.id, asset) restored.append(asset) logger.info(f"Restored prebuilt asset: {asset.name} ({asset.id})") return restored def import_prebuilt_sounds(self, prebuilt_dir: Path) -> List[Asset]: """Import prebuilt sound files that don't already exist as assets. Called on startup. Skips files that are already imported (by original filename match with prebuilt=True), including soft-deleted ones. Returns list of newly imported assets. """ if not prebuilt_dir.exists(): return [] # Build set of known prebuilt filenames (including deleted ones) known_filenames = { a.filename for a in self._items.values() if a.prebuilt } imported = [] for src in sorted(prebuilt_dir.iterdir()): if not src.is_file(): continue if src.name in known_filenames: continue file_data = src.read_bytes() # Derive a friendly name from filename: "chime.wav" -> "Chime" friendly_name = src.stem.replace("_", " ").replace("-", " ").title() asset = self.create_asset( name=friendly_name, filename=src.name, file_data=file_data, prebuilt=True, ) imported.append(asset) logger.info(f"Imported prebuilt sound: {src.name} -> {asset.id}") return imported