Initial commit - Video Service for Coolify

app/services/storage.py

@@ -0,0 +1,168 @@
+"""
+Storage Service
+Handles file storage for uploaded and converted videos
+"""
+
+import json
+import shutil
+import uuid
+from datetime import UTC, datetime
+from pathlib import Path
+
+from app.config import Settings, get_settings
+from app.models.schemas import VideoMetadata, VideoStatus
+
+
+class StorageService:
+    """Manage video file storage."""
+
+    def __init__(self, settings: Settings | None = None):
+        self.settings = settings or get_settings()
+        self.base_path = Path(self.settings.storage_path)
+        self.uploads_path = self.base_path / "uploads"
+        self.converted_path = self.base_path / "converted"
+
+        # Ensure directories exist
+        self.uploads_path.mkdir(parents=True, exist_ok=True)
+        self.converted_path.mkdir(parents=True, exist_ok=True)
+
+    def generate_video_id(self) -> str:
+        """Generate unique video ID."""
+        return str(uuid.uuid4())[:12]
+
+    def get_upload_path(self, video_id: str, filename: str) -> Path:
+        """Get path for uploaded video file."""
+        video_dir = self.uploads_path / video_id
+        video_dir.mkdir(parents=True, exist_ok=True)
+        return video_dir / filename
+
+    def get_converted_path(self, video_id: str) -> Path:
+        """Get path for converted video directory."""
+        video_dir = self.converted_path / video_id
+        video_dir.mkdir(parents=True, exist_ok=True)
+        return video_dir
+
+    def get_hls_path(self, video_id: str) -> Path:
+        """Get path for HLS output."""
+        return self.get_converted_path(video_id) / "hls"
+
+    def get_thumbnail_path(self, video_id: str) -> Path:
+        """Get path for thumbnail."""
+        return self.get_converted_path(video_id) / "thumbnail.jpg"
+
+    def get_metadata_path(self, video_id: str) -> Path:
+        """Get path for metadata JSON."""
+        return self.get_converted_path(video_id) / "metadata.json"
+
+    def save_metadata(
+        self,
+        video_id: str,
+        kurs_id: int,
+        title: str,
+        status: VideoStatus,
+        metadata: VideoMetadata | None = None,
+        error_message: str | None = None,
+    ) -> dict:
+        """Save video metadata to JSON file."""
+        meta_path = self.get_metadata_path(video_id)
+
+        data = {
+            "video_id": video_id,
+            "kurs_id": kurs_id,
+            "title": title,
+            "status": status.value,
+            "created_at": datetime.now(UTC).isoformat(),
+            "updated_at": datetime.now(UTC).isoformat(),
+            "error_message": error_message,
+        }
+
+        if metadata:
+            data["metadata"] = metadata.model_dump()
+
+        # Load existing data to preserve created_at
+        if meta_path.exists():
+            with meta_path.open() as f:
+                existing = json.load(f)
+                data["created_at"] = existing.get("created_at", data["created_at"])
+
+        with meta_path.open("w") as f:
+            json.dump(data, f, indent=2, default=str)
+
+        return data
+
+    def load_metadata(self, video_id: str) -> dict | None:
+        """Load video metadata from JSON file."""
+        meta_path = self.get_metadata_path(video_id)
+        if not meta_path.exists():
+            return None
+
+        with meta_path.open() as f:
+            return json.load(f)
+
+    def update_status(
+        self,
+        video_id: str,
+        status: VideoStatus,
+        progress: int = 0,
+        error_message: str | None = None,
+    ) -> dict | None:
+        """Update video status in metadata."""
+        meta = self.load_metadata(video_id)
+        if not meta:
+            return None
+
+        meta["status"] = status.value
+        meta["progress"] = progress
+        meta["updated_at"] = datetime.now(UTC).isoformat()
+        if error_message:
+            meta["error_message"] = error_message
+
+        meta_path = self.get_metadata_path(video_id)
+        with meta_path.open("w") as f:
+            json.dump(meta, f, indent=2, default=str)
+
+        return meta
+
+    def delete_video(self, video_id: str) -> bool:
+        """Delete all files for a video."""
+        upload_dir = self.uploads_path / video_id
+        converted_dir = self.converted_path / video_id
+
+        deleted = False
+        if upload_dir.exists():
+            shutil.rmtree(upload_dir)
+            deleted = True
+        if converted_dir.exists():
+            shutil.rmtree(converted_dir)
+            deleted = True
+
+        return deleted
+
+    def get_video_ids(self) -> list[str]:
+        """Get all video IDs."""
+        ids = set()
+        if self.uploads_path.exists():
+            ids.update(d.name for d in self.uploads_path.iterdir() if d.is_dir())
+        if self.converted_path.exists():
+            ids.update(d.name for d in self.converted_path.iterdir() if d.is_dir())
+        return sorted(ids)
+
+    def is_writable(self) -> bool:
+        """Check if storage is writable."""
+        test_file = self.base_path / ".write_test"
+        try:
+            test_file.write_text("test")
+            test_file.unlink()
+            return True
+        except (OSError, PermissionError):
+            return False
+
+    def get_disk_usage(self) -> dict:
+        """Get disk usage statistics."""
+        total, used, free = shutil.disk_usage(self.base_path)
+        return {
+            "total_gb": round(total / (1024**3), 2),
+            "used_gb": round(used / (1024**3), 2),
+            "free_gb": round(free / (1024**3), 2),
+            "percent_used": round((used / total) * 100, 1),
+        }