119 lines
4.1 KiB
Python
119 lines
4.1 KiB
Python
import subprocess
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import aiofiles
|
|
from fastapi import UploadFile
|
|
from PIL import Image
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
class LocalStorageService:
|
|
def __init__(self, root: str):
|
|
self.root = Path(root)
|
|
self.root.mkdir(parents=True, exist_ok=True)
|
|
|
|
async def save_upload(self, upload: UploadFile, subfolder: str) -> tuple[str, int]:
|
|
dest_dir = self.root / subfolder
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
ext = Path(upload.filename or "file").suffix
|
|
filename = f"{uuid.uuid4()}{ext}"
|
|
dest_path = dest_dir / filename
|
|
content = await upload.read()
|
|
async with aiofiles.open(dest_path, "wb") as handle:
|
|
await handle.write(content)
|
|
return str(dest_path.relative_to(self.root)).replace("\\", "/"), len(content)
|
|
|
|
def save_bytes(self, data: bytes, subfolder: str, filename: str) -> str:
|
|
dest_dir = self.root / subfolder
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
dest_path = dest_dir / filename
|
|
with open(dest_path, "wb") as handle:
|
|
handle.write(data)
|
|
return str(dest_path.relative_to(self.root)).replace("\\", "/")
|
|
|
|
def absolute_path(self, relative_path: str) -> Path:
|
|
return self.root / relative_path
|
|
|
|
def delete_relative_path(self, relative_path: Optional[str]) -> None:
|
|
if not relative_path:
|
|
return
|
|
abs_path = self.absolute_path(relative_path)
|
|
try:
|
|
if abs_path.exists():
|
|
abs_path.unlink()
|
|
except Exception:
|
|
pass
|
|
|
|
def generate_thumbnail(self, image_path: str, thumb_subfolder: str) -> Optional[str]:
|
|
try:
|
|
abs_path = self.absolute_path(image_path)
|
|
with Image.open(abs_path) as img:
|
|
img.thumbnail((400, 400))
|
|
thumb_dir = self.root / thumb_subfolder
|
|
thumb_dir.mkdir(parents=True, exist_ok=True)
|
|
thumb_name = f"thumb_{Path(image_path).stem}.jpg"
|
|
thumb_path = thumb_dir / thumb_name
|
|
img.convert("RGB").save(thumb_path, "JPEG", quality=80)
|
|
return str(thumb_path.relative_to(self.root)).replace("\\", "/")
|
|
except Exception:
|
|
return None
|
|
|
|
def generate_video_thumbnail(self, video_path: str, thumb_subfolder: str) -> Optional[str]:
|
|
abs_path = self.absolute_path(video_path)
|
|
thumb_dir = self.root / thumb_subfolder
|
|
thumb_dir.mkdir(parents=True, exist_ok=True)
|
|
thumb_name = f"thumb_{Path(video_path).stem}.jpg"
|
|
thumb_path = thumb_dir / thumb_name
|
|
try:
|
|
subprocess.run(
|
|
[
|
|
"ffmpeg",
|
|
"-y",
|
|
"-i",
|
|
str(abs_path),
|
|
"-ss",
|
|
"00:00:00.500",
|
|
"-vframes",
|
|
"1",
|
|
str(thumb_path),
|
|
],
|
|
capture_output=True,
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
if thumb_path.exists():
|
|
return str(thumb_path.relative_to(self.root)).replace("\\", "/")
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
def detect_duration_seconds(self, relative_path: str) -> Optional[float]:
|
|
abs_path = self.absolute_path(relative_path)
|
|
try:
|
|
result = subprocess.run(
|
|
[
|
|
"ffprobe",
|
|
"-v",
|
|
"error",
|
|
"-show_entries",
|
|
"format=duration",
|
|
"-of",
|
|
"default=noprint_wrappers=1:nokey=1",
|
|
str(abs_path),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=20,
|
|
check=True,
|
|
)
|
|
return round(float(result.stdout.strip()), 3)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
asset_storage = LocalStorageService(settings.ASSET_STORAGE_ROOT)
|
|
output_storage = LocalStorageService(settings.OUTPUT_STORAGE_ROOT)
|