xiaoxia-saas/packages/adapters/sqlalchemy_impl/voice_library_repository.py

"""SQLAlchemy implementation of VoiceLibraryRepository."""

from __future__ import annotations

from typing import Optional

from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.models import VoiceLibraryModel
from packages.domain.voice_library import VoiceLibraryItem


class SQLAlchemyVoiceLibraryRepository:
    """SQLAlchemy 配音库仓储"""

    def __init__(self, session: Session) -> None:
        self.session = session

    def list_by_user(
        self,
        user_id: str,
        *,
        status: Optional[str] = None,
        skip: int = 0,
        limit: int = 50,
    ) -> list[VoiceLibraryItem]:
        query = self.session.query(VoiceLibraryModel).filter(
            VoiceLibraryModel.user_id == user_id,
        )
        if status:
            query = query.filter(VoiceLibraryModel.status == status)
        query = query.order_by(VoiceLibraryModel.created_at.desc())
        models = query.offset(skip).limit(limit).all()
        return [self._model_to_entity(m) for m in models]

    def get(self, voice_id: str, user_id: str) -> Optional[VoiceLibraryItem]:
        model = (
            self.session.query(VoiceLibraryModel)
            .filter(
                VoiceLibraryModel.id == voice_id,
                VoiceLibraryModel.user_id == user_id,
            )
            .first()
        )
        if model is None:
            return None
        return self._model_to_entity(model)

    def create(self, item: VoiceLibraryItem) -> VoiceLibraryItem:
        model = VoiceLibraryModel(
            id=item.id,
            user_id=item.user_id,
            project_id=item.project_id or "",
            name=item.name,
            text=item.text,
            voice_provider=item.voice_provider,
            voice_id=item.voice_id,
            voice_name=item.voice_name,
            audio_url=item.audio_url,
            duration=item.duration,
            file_size=item.file_size,
            status=item.status,
            tags=item.tags,
            extra_meta=item.metadata_,
        )
        self.session.add(model)
        self.session.commit()
        self.session.refresh(model)
        return self._model_to_entity(model)

    def update(self, item: VoiceLibraryItem) -> VoiceLibraryItem:
        model = (
            self.session.query(VoiceLibraryModel)
            .filter(
                VoiceLibraryModel.id == item.id,
                VoiceLibraryModel.user_id == item.user_id,
            )
            .first()
        )
        if model is None:
            raise ValueError(f"VoiceLibraryItem {item.id} not found")
        model.name = item.name
        model.text = item.text
        model.voice_provider = item.voice_provider
        model.voice_id = item.voice_id
        model.voice_name = item.voice_name
        model.audio_url = item.audio_url
        model.duration = item.duration
        model.file_size = item.file_size
        model.status = item.status
        model.tags = item.tags
        model.extra_meta = item.metadata_
        self.session.commit()
        self.session.refresh(model)
        return self._model_to_entity(model)

    def delete(self, voice_id: str, user_id: str) -> bool:
        model = (
            self.session.query(VoiceLibraryModel)
            .filter(
                VoiceLibraryModel.id == voice_id,
                VoiceLibraryModel.user_id == user_id,
            )
            .first()
        )
        if model is None:
            return False
        # Soft delete by setting status to deleted
        model.status = "deleted"
        self.session.commit()
        return True

    def count_by_user(self, user_id: str, *, status: Optional[str] = None) -> int:
        query = self.session.query(VoiceLibraryModel).filter(
            VoiceLibraryModel.user_id == user_id,
            VoiceLibraryModel.status != "deleted",
        )
        if status:
            query = query.filter(VoiceLibraryModel.status == status)
        return query.count()

    @staticmethod
    def _model_to_entity(model: VoiceLibraryModel) -> VoiceLibraryItem:
        return VoiceLibraryItem(
            id=model.id,
            user_id=model.user_id,
            name=model.name,
            text=model.text,
            voice_provider=model.voice_provider,
            voice_id=model.voice_id,
            voice_name=model.voice_name,
            audio_url=model.audio_url,
            duration=model.duration or 0,
            file_size=model.file_size or 0,
            status=model.status,
            project_id=model.project_id if model.project_id else None,
            tags=model.tags or [],
            metadata_=model.extra_meta or {},
            created_at=model.created_at,
            updated_at=model.updated_at,
        )