77 lines
2.0 KiB
Python
77 lines
2.0 KiB
Python
import os
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import ffmpeg
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import settings
|
|
from app.models.recording import Recording, RecordingSource, RecordingStatus
|
|
|
|
|
|
def get_upload_path(filename: str) -> str:
|
|
ext = Path(filename).suffix
|
|
unique_name = f"{uuid.uuid4().hex}{ext}"
|
|
return os.path.join(settings.upload_dir, unique_name)
|
|
|
|
|
|
def get_media_duration(file_path: str) -> float | None:
|
|
try:
|
|
probe = ffmpeg.probe(file_path)
|
|
return float(probe["format"]["duration"])
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def get_format_from_filename(filename: str) -> str:
|
|
return Path(filename).suffix.lstrip(".").lower()
|
|
|
|
|
|
async def create_recording(
|
|
db: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
title: str,
|
|
file_path: str,
|
|
file_size: int,
|
|
format: str,
|
|
source: RecordingSource,
|
|
duration: float | None = None,
|
|
) -> Recording:
|
|
recording = Recording(
|
|
user_id=user_id,
|
|
title=title,
|
|
file_path=file_path,
|
|
file_size=file_size,
|
|
format=format,
|
|
source=source,
|
|
duration=duration,
|
|
)
|
|
db.add(recording)
|
|
await db.commit()
|
|
await db.refresh(recording)
|
|
return recording
|
|
|
|
|
|
async def get_recordings(
|
|
db: AsyncSession,
|
|
user_id: uuid.UUID | None = None,
|
|
skip: int = 0,
|
|
limit: int = 50,
|
|
) -> tuple[list[Recording], int]:
|
|
query = select(Recording).order_by(Recording.created_at.desc())
|
|
count_query = select(func.count(Recording.id))
|
|
|
|
if user_id:
|
|
query = query.where(Recording.user_id == user_id)
|
|
count_query = count_query.where(Recording.user_id == user_id)
|
|
|
|
total = (await db.execute(count_query)).scalar()
|
|
result = await db.execute(query.offset(skip).limit(limit))
|
|
return list(result.scalars().all()), total
|
|
|
|
|
|
async def get_recording_by_id(db: AsyncSession, recording_id: uuid.UUID) -> Recording | None:
|
|
result = await db.execute(select(Recording).where(Recording.id == recording_id))
|
|
return result.scalar_one_or_none()
|